rebased to latest

Signed-off-by: Krishna Kondaka <[email protected]>
opensearch-project · Jul 23, 2024 · 3a70f10 · 3a70f10
1 parent 36a3e59
commit 3a70f10
Show file tree

Hide file tree

Showing 2 changed files with 9 additions and 5 deletions.
diff --git a/...in/java/org/opensearch/dataprepper/plugins/sink/opensearch/OpenSearchClientRefresher.java b/...in/java/org/opensearch/dataprepper/plugins/sink/opensearch/OpenSearchClientRefresher.java
@@ -30,7 +30,7 @@ public OpenSearchClientRefresher(final PluginMetrics pluginMetrics,
                                      final Function<ConnectionConfiguration, OpenSearchClient> clientFunction) {
         this.clientFunction = clientFunction;
         this.currentConfig = connectionConfiguration;
-        this.currentClient = clientFunction.apply(connectionConfiguration);
+        this.currentClient = null;
         credentialsChangeCounter = pluginMetrics.counter(CREDENTIALS_CHANGED);
         clientRefreshErrorsCounter = pluginMetrics.counter(CLIENT_REFRESH_ERRORS);
     }
@@ -44,6 +44,9 @@ public Class<OpenSearchClient> getComponentClass() {
     public OpenSearchClient get() {
         readWriteLock.readLock().lock();
         try {
+            if (currentClient == null) {
+                currentClient = clientFunction.apply(currentConfig);
+            }
             return currentClient;
         } finally {
             readWriteLock.readLock().unlock();
@@ -57,8 +60,8 @@ public void update(PluginSetting pluginSetting) {
             credentialsChangeCounter.increment();
             readWriteLock.writeLock().lock();
             try {
-                currentClient = clientFunction.apply(newConfig);
                 currentConfig = newConfig;
+                currentClient = clientFunction.apply(currentConfig);
             } catch (Exception e) {
                 clientRefreshErrorsCounter.increment();
                 LOG.error("Refreshing {} failed.", getComponentClass(), e);

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
@@ -186,6 +186,7 @@ public OpenSearchSink(final PluginSetting pluginSetting,
     this.bulkRequestMap = new ConcurrentHashMap<>();
     this.lastFlushTimeMap = new ConcurrentHashMap<>();
     this.pluginConfigObservable = pluginConfigObservable;
+    this.objectMapper = new ObjectMapper();
 
     final Optional<PluginModel> dlqConfig = openSearchSinkConfig.getRetryConfiguration().getDlq();
     if (dlqConfig.isPresent()) {
@@ -201,7 +202,7 @@ public void doInitialize() {
         doInitializeInternal();
     } catch (IOException e) {
         LOG.warn("Failed to initialize OpenSearch sink, retrying: {} ", e.getMessage());
-        closeFiles();
+        this.shutdown();
     } catch (InvalidPluginConfigurationException e) {
         LOG.error("Failed to initialize OpenSearch sink due to a configuration error.", e);
         this.shutdown();
@@ -212,7 +213,7 @@ public void doInitialize() {
         throw e;
     } catch (Exception e) {
         LOG.warn("Failed to initialize OpenSearch sink with a retryable exception. ", e);
-        closeFiles();
+        this.shutdown();
     }
   }
 
@@ -279,7 +280,6 @@ private void doInitializeInternal() throws IOException {
             bulkRequestSupplier,
             pluginSetting);
 
-    objectMapper = new ObjectMapper();
     this.initialized = true;
     LOG.info("Initialized OpenSearch sink");
   }
@@ -615,6 +615,7 @@ private void closeFiles() {
   public void shutdown() {
     super.shutdown();
     closeFiles();
+    openSearchClient.shutdown();
   }
 
   private void maybeUpdateServerlessNetworkPolicy() {