add more IT

Signed-off-by: Jackie Han <hnyng@amazon.com>

Author: jackiehanyang

src/main/java/org/opensearch/timeseries/indices/IndexManagement.java

@@ -1020,36 +1020,39 @@ public <T> void initCustomResultIndexAndExecute(String resultIndexOrAlias, Execu
      * creates flattened result index
      * @param flattenedResultIndexAlias the flattened result index alias
      * @param actionListener the action listener
-     * @throws IOException
      */
-    public void initFlattenedResultIndex(String flattenedResultIndexAlias, ActionListener<CreateIndexResponse> actionListener)
-        throws IOException {
-        String indexName = getCustomResultIndexPattern(flattenedResultIndexAlias);
-        logger.info("Initializing flattened result index: {}", indexName);
+    public void initFlattenedResultIndex(String flattenedResultIndexAlias, ActionListener<CreateIndexResponse> actionListener) {
+        try {
+            String indexName = getCustomResultIndexPattern(flattenedResultIndexAlias);
+            logger.info("Initializing flattened result index: {}", indexName);
 
-        CreateIndexRequest request = new CreateIndexRequest(indexName)
-            .mapping(getFlattenedResultMappings(), XContentType.JSON)
-            .settings(settings);
+            CreateIndexRequest request = new CreateIndexRequest(indexName)
+                .mapping(getFlattenedResultMappings(), XContentType.JSON)
+                .settings(settings);
 
-        if (flattenedResultIndexAlias != null) {
-            request.alias(new Alias(flattenedResultIndexAlias));
-        }
+            if (flattenedResultIndexAlias != null) {
+                request.alias(new Alias(flattenedResultIndexAlias));
+            }
 
-        choosePrimaryShards(request, false);
+            choosePrimaryShards(request, false);
 
-        adminClient.indices().create(request, ActionListener.wrap(response -> {
-            if (response.isAcknowledged()) {
-                logger.info("Successfully created flattened result index: {} with alias: {}", indexName, flattenedResultIndexAlias);
-                actionListener.onResponse(response);
-            } else {
-                String errorMsg = "Index creation not acknowledged for index: " + indexName;
-                logger.error(errorMsg);
-                actionListener.onFailure(new IllegalStateException(errorMsg));
-            }
-        }, exception -> {
-            logger.error("Failed to create flattened result index: {}", indexName, exception);
-            actionListener.onFailure(exception);
-        }));
+            adminClient.indices().create(request, ActionListener.wrap(response -> {
+                if (response.isAcknowledged()) {
+                    logger.info("Successfully created flattened result index: {} with alias: {}", indexName, flattenedResultIndexAlias);
+                    actionListener.onResponse(response);
+                } else {
+                    String errorMsg = "Index creation not acknowledged for index: " + indexName;
+                    logger.error(errorMsg);
+                    actionListener.onFailure(new IllegalStateException(errorMsg));
+                }
+            }, exception -> {
+                logger.error("Failed to create flattened result index: {}", indexName, exception);
+                actionListener.onFailure(exception);
+            }));
+        } catch (IOException e) {
+            logger.error("Error while building mappings for flattened result index: {}", flattenedResultIndexAlias, e);
+            actionListener.onFailure(e);
+        }
     }
 
     public String getFlattenedResultIndexAlias(String indexOrAliasName, String configId) {

src/main/java/org/opensearch/timeseries/rest/handler/AbstractTimeSeriesActionHandler.java

@@ -466,27 +466,31 @@ private void handlePostRequest(boolean indexingDryRun, ActionListener<T> listene
             if (shouldHandleFlattening(indexingDryRun, createConfigResponse)) {
                 IndexAnomalyDetectorResponse response = (IndexAnomalyDetectorResponse) createConfigResponse;
                 String configId = response.getId();
-                String flattenedResultIndexAlias = timeSeriesIndices
-                    .getFlattenedResultIndexAlias(config.getCustomResultIndexOrAlias(), configId);
-                String pipelineId = timeSeriesIndices.getFlattenResultIndexIngestPipelineId(configId);
-
-                timeSeriesIndices
-                    .initFlattenedResultIndex(
-                        flattenedResultIndexAlias,
-                        ActionListener.wrap(initResponse -> setupIngestPipeline(configId, ActionListener.wrap(pipelineResponse -> {
-                            updateResultIndexSetting(
-                                pipelineId,
-                                flattenedResultIndexAlias,
-                                ActionListener.wrap(updateResponse -> listener.onResponse(createConfigResponse), listener::onFailure)
-                            );
-                        }, listener::onFailure)), listener::onFailure)
-                    );
+                initFlattenedResultIndexAndPipeline(
+                    configId,
+                    ActionListener.wrap(initResponse -> listener.onResponse(createConfigResponse), listener::onFailure)
+                );
             } else {
                 listener.onResponse(createConfigResponse);
             }
         }, listener::onFailure));
     }
 
+    private void initFlattenedResultIndexAndPipeline(String configId, ActionListener<T> listener) {
+        String flattenedResultIndexAlias = timeSeriesIndices.getFlattenedResultIndexAlias(config.getCustomResultIndexOrAlias(), configId);
+        String pipelineId = timeSeriesIndices.getFlattenResultIndexIngestPipelineId(configId);
+
+        timeSeriesIndices.initFlattenedResultIndex(flattenedResultIndexAlias, ActionListener.wrap(initResponse -> {
+            setupIngestPipeline(configId, ActionListener.wrap(pipelineResponse -> {
+                updateResultIndexSetting(
+                    pipelineId,
+                    flattenedResultIndexAlias,
+                    ActionListener.wrap(updateResponse -> listener.onResponse(null), listener::onFailure)
+                );
+            }, listener::onFailure));
+        }, listener::onFailure));
+    }
+
     private boolean shouldHandleFlattening(boolean indexingDryRun, Object createConfigResponse) {
         Boolean flattenResultIndexMapping = config.getFlattenResultIndexMapping();
 
@@ -579,7 +583,7 @@ private void handleFlattenResultIndexMappingUpdate(ActionListener<T> listener) {
             return;
         }
         if (config.getFlattenResultIndexMapping() != null && config.getFlattenResultIndexMapping()) {
-            setupIngestPipeline(id, listener);
+            initFlattenedResultIndexAndPipeline(config.getId(), listener);
         } else {
             String pipelineId = timeSeriesIndices.getFlattenResultIndexIngestPipelineId(config.getId());
             client.admin().cluster().deletePipeline(new DeletePipelineRequest(pipelineId), new ActionListener<AcknowledgedResponse>() {