OpenSearchSink should close open files before retrying initialization (…

…opensearch-project#2255) Signed-off-by: Krishna Kondaka <[email protected]> Co-authored-by: Krishna Kondaka <[email protected]>
asifsmohammed · Feb 9, 2023 · f782b66 · f782b66
1 parent e32e1c9
commit f782b66
Show file tree

Hide file tree

Showing 2 changed files with 11 additions and 4 deletions.
diff --git a/data-prepper-core/src/main/java/org/opensearch/dataprepper/DataPrepper.java b/data-prepper-core/src/main/java/org/opensearch/dataprepper/DataPrepper.java
@@ -94,7 +94,8 @@ public boolean execute() {
             }
         }
         if (waitingPipelineNames.size() > 0) {
-            LOG.info("One or more Pipelines are not ready even after {} retries.", numRetries);
+            LOG.info("One or more Pipelines are not ready even after {} retries. Shutting down pipelines", numRetries);
+            shutdown();
             throw new RuntimeException("Failed to start pipelines");
         }
         transformationPipelines.forEach((name, pipeline) -> {

diff --git a/...arch/src/main/java/org/opensearch/dataprepper/plugins/sink/opensearch/OpenSearchSink.java b/...arch/src/main/java/org/opensearch/dataprepper/plugins/sink/opensearch/OpenSearchSink.java
@@ -107,6 +107,7 @@ public void doInitialize() {
         doInitializeInternal();
     } catch (IOException e) {
         LOG.warn("Failed to initialize OpenSearch sink, retrying. Error - " + e.getCause());
+        closeFiles();
     } catch (InvalidPluginConfigurationException e) {
         LOG.error("Failed to initialize OpenSearch sink.");
         this.shutdown();
@@ -118,6 +119,7 @@ public void doInitialize() {
             throw e;
         }
         LOG.warn("Failed to initialize OpenSearch sink, retrying. Error - " + e.getCause());
+        closeFiles();
     }
   }
 
@@ -252,9 +254,7 @@ private void logFailure(final BulkOperation bulkOperation, final Throwable failu
     }
   }
 
-  @Override
-  public void shutdown() {
-    super.shutdown();
+  private void closeFiles() {
     // Close the client. This closes the low-level client which will close it for both high-level clients.
     if (restHighLevelClient != null) {
       try {
@@ -271,4 +271,10 @@ public void shutdown() {
       }
     }
   }
+
+  @Override
+  public void shutdown() {
+    super.shutdown();
+    closeFiles();
+  }
 }