removed additional listener in provision action

Signed-off-by: Amit Galitzky <[email protected]>

src/main/java/org/opensearch/flowframework/transport/ProvisionWorkflowTransportAction.java

@@ -134,7 +134,7 @@ protected void doExecute(Task task, WorkflowRequest request, ActionListener<Work
 
                 // Respond to rest action then execute provisioning workflow async
                 listener.onResponse(new WorkflowResponse(workflowId));
-                executeWorkflowAsync(workflowId, provisionProcessSequence);
+                executeWorkflowAsync(workflowId, provisionProcessSequence, listener);
 
             }, exception -> {
                 if (exception instanceof FlowFrameworkException) {
@@ -155,59 +155,22 @@ protected void doExecute(Task task, WorkflowRequest request, ActionListener<Work
      * Retrieves a thread from the provision thread pool to execute a workflow
      * @param workflowId The id of the workflow
      * @param workflowSequence The sorted workflow to execute
+     * @param listener ActionListener for any failures that don't get caught earlier in below step
      */
-    private void executeWorkflowAsync(String workflowId, List<ProcessNode> workflowSequence) {
-        // TODO : Update Action listener type to State index Request
-        ActionListener<String> provisionWorkflowListener = ActionListener.wrap(response -> {
-            logger.info("Provisioning completed successfully for workflow {}", workflowId);
-            flowFrameworkIndicesHandler.updateFlowFrameworkSystemIndexDoc(
-                workflowId,
-                ImmutableMap.of(
-                    STATE_FIELD,
-                    State.COMPLETED,
-                    PROVISIONING_PROGRESS_FIELD,
-                    ProvisioningProgress.DONE,
-                    PROVISION_END_TIME_FIELD,
-                    Instant.now().toEpochMilli()
-                ),
-                ActionListener.wrap(updateResponse -> {
-                    logger.info("updated workflow {} state to {}", workflowId, State.COMPLETED);
-                }, exception -> { logger.error("Failed to update workflow state : {}", exception.getMessage()); })
-            );
-
-        }, exception -> {
-            logger.error("Provisioning failed for workflow {} : {}", workflowId, exception);
-
-            flowFrameworkIndicesHandler.updateFlowFrameworkSystemIndexDoc(
-                workflowId,
-                ImmutableMap.of(
-                    STATE_FIELD,
-                    State.FAILED,
-                    ERROR_FIELD,
-                    exception.getMessage(), // TODO: potentially improve the error message here
-                    PROVISIONING_PROGRESS_FIELD,
-                    ProvisioningProgress.FAILED,
-                    PROVISION_END_TIME_FIELD,
-                    Instant.now().toEpochMilli()
-                ),
-                ActionListener.wrap(updateResponse -> {
-                    logger.info("updated workflow {} state to {}", workflowId, State.COMPLETED);
-                }, exceptionState -> { logger.error("Failed to update workflow state : {}", exceptionState.getMessage()); })
-            );
-        });
+    private void executeWorkflowAsync(String workflowId, List<ProcessNode> workflowSequence, ActionListener<WorkflowResponse> listener) {
         try {
-            threadPool.executor(PROVISION_THREAD_POOL).execute(() -> { executeWorkflow(workflowSequence, provisionWorkflowListener); });
+            threadPool.executor(PROVISION_THREAD_POOL).execute(() -> { executeWorkflow(workflowSequence, workflowId); });
         } catch (Exception exception) {
-            provisionWorkflowListener.onFailure(new FlowFrameworkException(exception.getMessage(), ExceptionsHelper.status(exception)));
+            listener.onFailure(new FlowFrameworkException(exception.getMessage(), ExceptionsHelper.status(exception)));
         }
     }
 
     /**
      * Executes the given workflow sequence
      * @param workflowSequence The topologically sorted workflow to execute
-     * @param workflowListener The listener that updates the status of a workflow execution
+     * @param workflowId The workflowId associated with the workflow that is executing
      */
-    private void executeWorkflow(List<ProcessNode> workflowSequence, ActionListener<String> workflowListener) {
+    private void executeWorkflow(List<ProcessNode> workflowSequence, String workflowId) {
         try {
 
             List<CompletableFuture<?>> workflowFutureList = new ArrayList<>();
@@ -232,12 +195,39 @@ private void executeWorkflow(List<ProcessNode> workflowSequence, ActionListener<
             // Attempt to join each workflow step future, may throw a CompletionException if any step completes exceptionally
             workflowFutureList.forEach(CompletableFuture::join);
 
-            workflowListener.onResponse("READY");
-
-        } catch (IllegalArgumentException e) {
-            workflowListener.onFailure(new FlowFrameworkException(e.getMessage(), RestStatus.BAD_REQUEST));
+            logger.info("Provisioning completed successfully for workflow {}", workflowId);
+            flowFrameworkIndicesHandler.updateFlowFrameworkSystemIndexDoc(
+                workflowId,
+                ImmutableMap.of(
+                    STATE_FIELD,
+                    State.COMPLETED,
+                    PROVISIONING_PROGRESS_FIELD,
+                    ProvisioningProgress.DONE,
+                    PROVISION_END_TIME_FIELD,
+                    Instant.now().toEpochMilli()
+                ),
+                ActionListener.wrap(updateResponse -> {
+                    logger.info("updated workflow {} state to {}", workflowId, State.COMPLETED);
+                }, exception -> { logger.error("Failed to update workflow state : {}", exception.getMessage()); })
+            );
         } catch (Exception ex) {
-            workflowListener.onFailure(new FlowFrameworkException(ex.getMessage(), ExceptionsHelper.status(ex)));
+            logger.error("Provisioning failed for workflow {} : {}", workflowId, ex);
+            flowFrameworkIndicesHandler.updateFlowFrameworkSystemIndexDoc(
+                workflowId,
+                ImmutableMap.of(
+                    STATE_FIELD,
+                    State.FAILED,
+                    ERROR_FIELD,
+                    ex.getMessage(), // TODO: potentially improve the error message here
+                    PROVISIONING_PROGRESS_FIELD,
+                    ProvisioningProgress.FAILED,
+                    PROVISION_END_TIME_FIELD,
+                    Instant.now().toEpochMilli()
+                ),
+                ActionListener.wrap(updateResponse -> {
+                    logger.info("updated workflow {} state to {}", workflowId, State.COMPLETED);
+                }, exceptionState -> { logger.error("Failed to update workflow state : {}", exceptionState.getMessage()); })
+            );
         }
     }