Merge pull request #118 from aodn/features/5277-avoid-flooding-system

Avoid flooding and cause geonetwork too busy error

indexer/src/main/java/au/org/aodn/esindexer/configuration/IndexerConfig.java

@@ -1,13 +1,19 @@
 package au.org.aodn.esindexer.configuration;
 
 import au.org.aodn.esindexer.utils.VocabsUtils;
+import org.springframework.beans.factory.annotation.Value;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 import org.springframework.retry.annotation.EnableRetry;
+import org.springframework.scheduling.annotation.EnableAsync;
+import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
+
+import java.util.concurrent.Executor;
 
 @Configuration
 @EnableRetry
+@EnableAsync
 public class IndexerConfig {
     /**
      * We need to create component here because we do not want to run test with real http connection
@@ -20,4 +26,22 @@ public class IndexerConfig {
     public VocabsUtils createVocabsUtils() {
         return new VocabsUtils();
     }
+    /**
+     * This executor is used to limit the number of concurrent call to index metadata so not to flood the
+     * geonetwork. This is useful because the geonetwork do not care about re-index call it invoke, hence
+     * the elastic of geonetwork may be flooded by its re-index call.
+     * @return - An async task executor
+     */
+    @Bean(name = "asyncIndexMetadata")
+    public Executor taskExecutor(
+            @Value("${app.indexing.pool.core:5}") Integer coreSize,
+            @Value("${app.indexing.pool.max:10}") Integer coreMax) {
+        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
+        executor.setCorePoolSize(coreSize);     // Number of concurrent threads
+        executor.setMaxPoolSize(coreMax);       // Max number of concurrent threads
+        executor.setQueueCapacity(5000);        // Size of the queue
+        executor.setThreadNamePrefix("Async-");
+        executor.initialize();
+        return executor;
+    }
 }

indexer/src/main/java/au/org/aodn/esindexer/controller/IndexerController.java

@@ -34,14 +34,6 @@ public class IndexerController {
     @Autowired
     GeoNetworkService geonetworkResourceService;
 
-//    @GetMapping(path="/gn_records/{uuid}", produces = "application/json")
-//    @Operation(description = "Get a document from GeoNetwork Elasticsearch by UUID")
-//    public ResponseEntity getMetadataRecordFromGeoNetworkElasticsearchByUUID(@PathVariable("uuid") String uuid) {
-//        logger.info("getting a document by UUID: " + uuid);
-//        JSONObject response =  geonetworkResourceService.searchMetadataBy(uuid);
-//        return ResponseEntity.status(HttpStatus.OK).body(response.toString());
-//    }
-
     @GetMapping(path="/records/{uuid}", produces = "application/json")
     @Operation(description = "Get a document from GeoNetwork by UUID directly - JSON format response")
     public ResponseEntity<String> getMetadataRecordFromGeoNetworkByUUID(@PathVariable("uuid") String uuid) {

indexer/src/main/java/au/org/aodn/esindexer/service/IndexerService.java

@@ -12,6 +12,7 @@
 import java.util.List;
 
 public interface IndexerService {
+    // Event call back to notify caller, this avoid gateway timeout as we have message back to browser
     interface Callback {
         void onProgress(Object update);
         void onComplete(Object result);
@@ -20,4 +21,6 @@ interface Callback {
     ResponseEntity<String> deleteDocumentByUUID(String uuid) throws IOException;
     List<BulkResponse> indexAllMetadataRecordsFromGeoNetwork(boolean confirm, Callback callback) throws IOException;
     Hit<ObjectNode> getDocumentByUUID(String uuid) throws IOException;
+    boolean isMetadataPublished(String uuid);
+    boolean isGeoNetworkInstanceReinstalled(long portalIndexDocumentsCount);
 }

indexer/src/main/java/au/org/aodn/esindexer/service/IndexerServiceImpl.java

@@ -3,7 +3,6 @@
 import au.org.aodn.esindexer.configuration.AppConstants;
 import au.org.aodn.esindexer.exception.*;
 import au.org.aodn.esindexer.utils.JaxbUtils;
-import au.org.aodn.esindexer.utils.StringUtil;
 import au.org.aodn.metadata.iso19115_3_2018.MDMetadataType;
 import au.org.aodn.stac.model.RecordSuggest;
 import au.org.aodn.stac.model.StacCollectionModel;
@@ -29,6 +28,7 @@
 import org.springframework.beans.factory.annotation.Value;
 import org.springframework.http.HttpStatus;
 import org.springframework.http.ResponseEntity;
+import org.springframework.scheduling.annotation.Async;
 import org.springframework.stereotype.Service;
 
 import java.io.ByteArrayInputStream;
@@ -101,15 +101,17 @@ public Hit<ObjectNode> getDocumentByUUID(String uuid) throws IOException {
         }
     }
 
-    protected boolean isGeoNetworkInstanceReinstalled(long portalIndexDocumentsCount) {
+    @Override
+    public boolean isGeoNetworkInstanceReinstalled(long portalIndexDocumentsCount) {
         /**
          * compare if GeoNetwork has 1 only metadata (the recently added one which triggered the indexer)
          * and the portal index has more than 0 documents (the most recent metadata yet indexed to portal index at this point)
          */
         return geoNetworkResourceService.isMetadataRecordsCountLessThan(2) && portalIndexDocumentsCount > 0;
     }
 
-    protected boolean isMetadataPublished(String uuid) {
+    @Override
+    public boolean isMetadataPublished(String uuid) {
         /* read for the published status from GN Elasticsearch index, the flag is not part of the XML body */
         try {
             geoNetworkResourceService.searchRecordBy(uuid);
@@ -170,7 +172,13 @@ protected StacCollectionModel getMappedMetadataValues(String metadataValues) thr
 
         return stacCollectionModel;
     }
-
+    /**
+     * Use to index a particular UUID, the async is used to limit the number of same function call to avoid flooding
+     * the system.
+     * @param metadataValues - The XML of the metadata
+     * @return - The STAC doc in string format.
+     */
+    @Async("asyncIndexMetadata")
     public ResponseEntity<String> indexMetadata(String metadataValues) {
         try {
             StacCollectionModel mappedMetadataValues = this.getMappedMetadataValues(metadataValues);

indexer/src/test/java/au/org/aodn/esindexer/service/IndexerServiceTests.java

@@ -27,7 +27,7 @@ public class IndexerServiceTests extends BaseTestClass {
     protected GeoNetworkServiceImpl geoNetworkService;
 
     @Autowired
-    protected IndexerServiceImpl indexerService;
+    protected IndexerService indexerService;
 
     @Autowired
     protected ObjectMapper indexerObjectMapper;

indexer/src/test/resources/application-test.yaml

@@ -1,5 +1,9 @@
 # Client calling the Indexer API must provide this token in the Authorization header
 app:
+  indexing:
+    pool:
+      core: 1
+      max: 1
   http:
     authToken: sample-auth-token