Moves the creation of OpenSearch clients back into the initalize() method. (#6855)

Data Prepper doesn't have strong lifecycles beyond initialize() and shutdown(). So creating multiple pipelines or attempting to initialize over time resulted in running out of files. This reverts the change to use the constructor until a better way to close plugins is available.

Signed-off-by: David Venable <dlv@amazon.com>

Author: dlvenable

data-prepper-plugins/opensearch/src/main/java/org/opensearch/dataprepper/plugins/sink/opensearch/OpenSearchSink.java

@@ -65,12 +65,14 @@ public class OpenSearchSink extends AbstractSink<Record<Event>> {
   private final IndexManagerFactory indexManagerFactory;
   private final IndexType indexType;
   private final PluginConfigObservable pluginConfigObservable;
-  private final Ingester ingester;
-
-  private final RestHighLevelClient restHighLevelClient;
-  private final OpenSearchClient openSearchClient;
-  private final OpenSearchClientRefresher openSearchClientRefresher;
-  private final ConnectionConfiguration connectionConfiguration;
+  private final ExpressionEvaluator expressionEvaluator;
+  private final SinkContext sinkContext;
+  private final String pipeline;
+  private Ingester ingester;
+
+  private RestHighLevelClient restHighLevelClient;
+  private OpenSearchClient openSearchClient;
+  private OpenSearchClientRefresher openSearchClientRefresher;
   private IndexManager indexManager;
   private volatile boolean initialized;
 
@@ -85,40 +87,16 @@ public OpenSearchSink(final PluginSetting pluginSetting,
     super(pluginSetting, Integer.MAX_VALUE, INITIALIZE_RETRY_WAIT_TIME_MS);
     this.awsCredentialsSupplier = awsCredentialsSupplier;
     this.pluginConfigObservable = pluginConfigObservable;
+    this.expressionEvaluator = expressionEvaluator;
 
-    final SinkContext resolvedSinkContext = sinkContext != null ? sinkContext :
+    this.sinkContext = sinkContext != null ? sinkContext :
             new SinkContext(null, Collections.emptyList(), Collections.emptyList(), Collections.emptyList());
 
     this.openSearchSinkConfig = OpenSearchSinkConfiguration.readOSConfig(openSearchSinkConfiguration, expressionEvaluator);
     this.indexType = openSearchSinkConfig.getIndexConfiguration().getIndexType();
     this.indexManagerFactory = new IndexManagerFactory(new ClusterSettingsParser());
+    this.pipeline = pipelineDescription.getPipelineName();
     this.initialized = false;
-
-    connectionConfiguration = openSearchSinkConfig.getConnectionConfiguration();
-    restHighLevelClient = connectionConfiguration.createClient(awsCredentialsSupplier);
-    openSearchClient = connectionConfiguration.createOpenSearchClient(restHighLevelClient, awsCredentialsSupplier);
-    final Function<ConnectionConfiguration, OpenSearchClient> clientFunction =
-            (connConfig) -> {
-      final RestHighLevelClient client = connConfig.createClient(awsCredentialsSupplier);
-      return connConfig.createOpenSearchClient(client, awsCredentialsSupplier).withTransportOptions(
-              TransportOptions.builder()
-                      .setParameter("filter_path", "errors,took,items.*.error,items.*.status,items.*._index,items.*._id")
-                      .build());
-    };
-    openSearchClientRefresher = new OpenSearchClientRefresher(
-            pluginMetrics, connectionConfiguration, clientFunction);
-
-    final String pipeline = pipelineDescription.getPipelineName();
-    final EventActionResolver eventActionResolver = new EventActionResolver(
-            openSearchSinkConfig.getIndexConfiguration().getAction(),
-            openSearchSinkConfig.getIndexConfiguration().getActions(),
-            expressionEvaluator);
-
-    this.ingester = new BulkIngester(openSearchSinkConfig, expressionEvaluator, resolvedSinkContext,
-            pluginMetrics, pipeline, eventActionResolver,
-            openSearchClient, () -> openSearchClientRefresher.get(),
-            this::getIndexManager, this::getFailurePipeline,
-            new CustomDocumentBuilderFactory().create(this.indexType));
   }
 
   @Override
@@ -145,6 +123,20 @@ public void doInitialize() {
   private void doInitializeInternal() throws IOException {
     LOG.info("Initializing OpenSearch sink");
 
+    final ConnectionConfiguration connectionConfiguration = openSearchSinkConfig.getConnectionConfiguration();
+    restHighLevelClient = connectionConfiguration.createClient(awsCredentialsSupplier);
+    openSearchClient = connectionConfiguration.createOpenSearchClient(restHighLevelClient, awsCredentialsSupplier);
+    final Function<ConnectionConfiguration, OpenSearchClient> clientFunction =
+            (connConfig) -> {
+      final RestHighLevelClient client = connConfig.createClient(awsCredentialsSupplier);
+      return connConfig.createOpenSearchClient(client, awsCredentialsSupplier).withTransportOptions(
+              TransportOptions.builder()
+                      .setParameter("filter_path", "errors,took,items.*.error,items.*.status,items.*._index,items.*._id")
+                      .build());
+    };
+    openSearchClientRefresher = new OpenSearchClientRefresher(
+            pluginMetrics, connectionConfiguration, clientFunction);
+
     pluginConfigObservable.addPluginConfigObserver(
             newOpenSearchSinkConfig -> openSearchClientRefresher.update((OpenSearchSinkConfig) newOpenSearchSinkConfig));
 
@@ -167,6 +159,17 @@ private void doInitializeInternal() throws IOException {
 
     indexManager.setupIndex();
 
+    final EventActionResolver eventActionResolver = new EventActionResolver(
+            openSearchSinkConfig.getIndexConfiguration().getAction(),
+            openSearchSinkConfig.getIndexConfiguration().getActions(),
+            expressionEvaluator);
+
+    ingester = new BulkIngester(openSearchSinkConfig, expressionEvaluator, sinkContext,
+            pluginMetrics, pipeline, eventActionResolver,
+            openSearchClient, () -> openSearchClientRefresher.get(),
+            this::getIndexManager, this::getFailurePipeline,
+            new CustomDocumentBuilderFactory().create(this.indexType));
+
     ingester.initialize();
 
     this.initialized = true;
@@ -190,7 +193,9 @@ public void doOutput(final Collection<Record<Event>> records) {
   @Override
   public void shutdown() {
     super.shutdown();
-    ingester.shutdown();
+    if (ingester != null) {
+      ingester.shutdown();
+    }
     if (restHighLevelClient != null) {
       try {
         restHighLevelClient.close();

data-prepper-plugins/opensearch/src/test/java/org/opensearch/dataprepper/plugins/sink/opensearch/OpenSearchSinkTest.java

@@ -69,6 +69,7 @@
 import static org.opensearch.dataprepper.plugins.sink.opensearch.OpenSearchSink.INVALID_VERSION_EXPRESSION_DROPPED_EVENTS;
 import static org.opensearch.dataprepper.plugins.sink.opensearch.configuration.OpenSearchSinkConfig.DEFAULT_BULK_SIZE;
 import static org.opensearch.dataprepper.plugins.sink.opensearch.configuration.OpenSearchSinkConfig.DEFAULT_FLUSH_TIMEOUT;
+import static org.opensearch.dataprepper.test.helper.ReflectivelySetField.setField;
 
 @ExtendWith(MockitoExtension.class)
 public class OpenSearchSinkTest {
@@ -165,8 +166,7 @@ private OpenSearchSink createObjectUnderTest() throws IOException {
              final MockedStatic<PluginMetrics> pluginMetricsMockedStatic = mockStatic(PluginMetrics.class);
              final MockedConstruction<IndexManagerFactory> indexManagerFactoryMockedConstruction = mockConstruction(IndexManagerFactory.class, (mock, context) -> {
                  indexManagerFactory = mock;
-             });
-             final MockedConstruction<BulkIngester> bulkIngesterMockedConstruction = mockConstruction(BulkIngester.class)) {
+             })) {
             pluginMetricsMockedStatic.when(() -> PluginMetrics.fromPluginSetting(pluginSetting)).thenReturn(pluginMetrics);
             openSearchSinkConfigurationMockedStatic.when(() -> OpenSearchSinkConfiguration.readOSConfig(openSearchSinkConfig, expressionEvaluator))
                     .thenReturn(openSearchSinkConfiguration);
@@ -181,7 +181,10 @@ void test_initialization() throws IOException {
         when(indexManagerFactory.getIndexManager(any(IndexType.class), eq(openSearchClient), any(RestHighLevelClient.class), eq(openSearchSinkConfiguration), any(TemplateStrategy.class), any()))
                 .thenReturn(indexManager);
         doNothing().when(indexManager).setupIndex();
-        objectUnderTest.initialize();
+
+        try (final MockedConstruction<BulkIngester> ignored = mockConstruction(BulkIngester.class)) {
+            objectUnderTest.initialize();
+        }
 
         verify(pluginConfigObservable).addPluginConfigObserver(any());
         assertThat(objectUnderTest.isReady(), equalTo(true));
@@ -193,16 +196,20 @@ void test_initialization_with_failure_and_retry() throws IOException {
         when(indexManagerFactory.getIndexManager(any(IndexType.class), eq(openSearchClient), any(RestHighLevelClient.class), eq(openSearchSinkConfiguration), any(TemplateStrategy.class), any()))
                 .thenThrow(RuntimeException.class).thenReturn(indexManager);
         doNothing().when(indexManager).setupIndex();
-        objectUnderTest.initialize();
-        objectUnderTest.initialize();
+
+        try (final MockedConstruction<BulkIngester> ignored = mockConstruction(BulkIngester.class)) {
+            objectUnderTest.initialize();
+            objectUnderTest.initialize();
+        }
         verify(pluginConfigObservable, times(2)).addPluginConfigObserver(any());
     }
 
     @Test
     void doOutput_delegates_to_ingester() throws Exception {
         final OpenSearchSink objectUnderTest = createObjectUnderTest();
 
-        final Ingester ingester = getField(objectUnderTest, "ingester");
+        final Ingester ingester = mock(Ingester.class);
+        setField(OpenSearchSink.class, objectUnderTest, "ingester", ingester);
         final List<Record<Event>> records = Collections.emptyList();
         objectUnderTest.doOutput(records);
 
@@ -213,16 +220,17 @@ void doOutput_delegates_to_ingester() throws Exception {
     void shutdown_delegates_to_ingester() throws Exception {
         final OpenSearchSink objectUnderTest = createObjectUnderTest();
 
-        final Ingester ingester = getField(objectUnderTest, "ingester");
+        final Ingester ingester = mock(Ingester.class);
+        setField(OpenSearchSink.class, objectUnderTest, "ingester", ingester);
         objectUnderTest.shutdown();
 
         verify(ingester).shutdown();
     }
 
-    @SuppressWarnings("unchecked")
-    private static <T> T getField(final Object target, final String fieldName) throws Exception {
-        java.lang.reflect.Field field = target.getClass().getDeclaredField(fieldName);
-        field.setAccessible(true);
-        return (T) field.get(target);
+    @Test
+    void shutdown_without_initialization_does_not_throw() throws Exception {
+        final OpenSearchSink objectUnderTest = createObjectUnderTest();
+        objectUnderTest.shutdown();
     }
+
 }