Move queryExecutor submission to the end of the initialization of opensearch sink

Signed-off-by: Taylor Gray <tylgry@amazon.com>

Author: graytaylor0

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

@@ -249,11 +249,6 @@ private void doInitializeInternal() throws IOException {
     openSearchClientRefresher = new OpenSearchClientRefresher(
             pluginMetrics, connectionConfiguration, clientFunction);
 
-    if (queryExecutorService != null) {
-      existingDocumentQueryManager = new ExistingDocumentQueryManager(openSearchSinkConfig.getIndexConfiguration(), pluginMetrics, openSearchClient);
-      queryExecutorService.submit(existingDocumentQueryManager);
-    }
-
     pluginConfigObservable.addPluginConfigObserver(
             newOpenSearchSinkConfig -> openSearchClientRefresher.update((OpenSearchSinkConfig) newOpenSearchSinkConfig));
     configuredIndexAlias = openSearchSinkConfig.getIndexConfiguration().getIndexAlias();
@@ -304,6 +299,11 @@ private void doInitializeInternal() throws IOException {
             PLUGIN_NAME,
             openSearchSinkConfig.getIndexConfiguration().getQueryOnBulkFailures() ? existingDocumentQueryManager : null);
 
+    if (queryExecutorService != null) {
+      existingDocumentQueryManager = new ExistingDocumentQueryManager(openSearchSinkConfig.getIndexConfiguration(), pluginMetrics, openSearchClient);
+      queryExecutorService.submit(existingDocumentQueryManager);
+    }
+
     this.initialized = true;
     LOG.info("Initialized OpenSearch sink");
   }