Fix threat intel multinode tests (#1274) (#1279)

* fix multinode tests * fix multinode --------- (cherry picked from commit f89c3ed) Signed-off-by: Joanne Wang <[email protected]> Signed-off-by: github-actions[bot] <github-actions[bot]@users.noreply.github.com> Co-authored-by: github-actions[bot] <github-actions[bot]@users.noreply.github.com>
opensearch-project · Sep 4, 2024 · ceb7107 · ceb7107
1 parent baa25a8
commit ceb7107
Showing 1 changed file with 15 additions and 5 deletions.
diff --git a/src/main/java/org/opensearch/securityanalytics/services/STIX2IOCFeedStore.java b/src/main/java/org/opensearch/securityanalytics/services/STIX2IOCFeedStore.java
@@ -10,6 +10,7 @@
 import org.apache.logging.log4j.Logger;
 import org.opensearch.OpenSearchException;
 import org.opensearch.action.DocWriteRequest;
+import org.opensearch.action.StepListener;
 import org.opensearch.action.admin.indices.create.CreateIndexRequest;
 import org.opensearch.action.admin.indices.create.CreateIndexResponse;
 import org.opensearch.action.bulk.BulkRequest;
@@ -80,7 +81,6 @@ public STIX2IOCFeedStore(
         this.baseListener = listener;
         batchSize = clusterService.getClusterSettings().get(SecurityAnalyticsSettings.BATCH_SIZE);
         newActiveIndex = getNewActiveIndex(saTifSourceConfig.getId());
-        initSourceConfigIndexes();
     }
 
     @Override
@@ -113,7 +113,15 @@ public void storeIOCs(Map<IOC, UpdateAction> actionToIOCs) {
     }
 
     public void indexIocs(List<STIX2IOC> iocs) throws IOException {
-        bulkIndexIocs(iocs, newActiveIndex);
+        StepListener<Void> initSourceConfigIndexesListener = new StepListener<>();
+        initSourceConfigIndexes(initSourceConfigIndexesListener);
+        initSourceConfigIndexesListener.whenComplete(r -> {
+            bulkIndexIocs(iocs, newActiveIndex);
+        }, e -> {
+            log.error("Failed to init source config indexes");
+            baseListener.onFailure(e);
+        });
+
     }
 
     private void bulkIndexIocs(List<STIX2IOC> iocs, String activeIndex) throws IOException {
@@ -197,7 +205,7 @@ public SATIFSourceConfig getSaTifSourceConfig() {
         return saTifSourceConfig;
     }
 
-    private void initSourceConfigIndexes() {
+    private void initSourceConfigIndexes(StepListener<Void> stepListener) {
         String iocIndexPattern = getAllIocIndexPatternById(saTifSourceConfig.getId());
         initFeedIndex(newActiveIndex, ActionListener.wrap(
                 r -> {
@@ -214,10 +222,10 @@ private void initSourceConfigIndexes() {
                             ((DefaultIocStoreConfig) saTifSourceConfig.getIocStoreConfig()).getIocToIndexDetails().add(iocToIndexDetails);
                         }
                     });
-
+                    stepListener.onResponse(null);
                 }, e-> {
                     log.error("Failed to initialize the IOC index and save the IOCs", e);
-                    baseListener.onFailure(e);
+                    stepListener.onFailure(e);
                 }
         ));
     }
@@ -237,6 +245,8 @@ private void initFeedIndex(String feedIndexName, ActionListener<CreateIndexRespo
                         listener.onFailure(e);
                     }
             ));
+        } else {
+            listener.onResponse(null);
         }
     }
 }