Refactor: Use factory pattern for WorkflowExecutionGraph creation

Address reviewer feedback: command handlers should not be concerned with
how instances are initialized. Changes:

- Add WorkflowExecutionGraphFactory to centralize all graph creation logic
- Remove IWorkflowExecutionGraphAssembler interface (no longer needed)
- Remove assembleWorkflowExecutionGraphAssembler() from AbstractCommandHandler
- Remove createWorkflowExecutionGraphAssembler() from all command handlers
- Update WorkflowRunningStateAction to use factory for graph initialization
- Update IWorkflowExecuteContext: add getProject(), change to setWorkflowExecutionGraph()
- Update WorkflowExecuteContext to remove assembler, add setter

The factory handles all command-type-specific graph creation strategies:
- START_PROCESS: Fresh start with configured start nodes
- REPEAT_RUNNING: Rerun entire workflow
- START_FAILURE_TASK_PROCESS: Recover failed/paused tasks
- RECOVER_TOLERANCE_FAULT_PROCESS: Failover with existing task instances
- RECOVER_SERIAL_WAIT: Serial wait recovery

Author: GaneshPatil7517

dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/command/handler/AbstractCommandHandler.java

@@ -32,7 +32,6 @@
 import org.apache.dolphinscheduler.extract.master.command.ICommandParam;
 import org.apache.dolphinscheduler.server.master.engine.WorkflowEventBus;
 import org.apache.dolphinscheduler.server.master.engine.command.ICommandHandler;
-import org.apache.dolphinscheduler.server.master.engine.graph.IWorkflowExecutionGraphAssembler;
 import org.apache.dolphinscheduler.server.master.engine.graph.IWorkflowGraph;
 import org.apache.dolphinscheduler.server.master.engine.graph.WorkflowGraphFactory;
 import org.apache.dolphinscheduler.server.master.engine.workflow.listener.IWorkflowLifecycleListener;
@@ -81,7 +80,6 @@ public WorkflowExecutionRunnable handleCommand(final Command command) {
         assembleWorkflowInstance(workflowExecuteContextBuilder);
         assembleWorkflowInstanceLifecycleListeners(workflowExecuteContextBuilder);
         assembleWorkflowEventBus(workflowExecuteContextBuilder);
-        assembleWorkflowExecutionGraphAssembler(workflowExecuteContextBuilder);
 
         final WorkflowExecutionRunnableBuilder workflowExecutionRunnableBuilder = WorkflowExecutionRunnableBuilder
                 .builder()
@@ -126,32 +124,6 @@ protected void assembleWorkflowGraph(
     protected abstract void assembleWorkflowInstance(
                                                      final WorkflowExecuteContextBuilder workflowExecuteContextBuilder);
 
-    /**
-     * Assemble the workflow execution graph assembler.
-     * <p>
-     * The assembler is used to defer the initialization of the WorkflowExecutionGraph
-     * until the WorkflowStartLifecycleEvent is fired. This reduces transaction time
-     * during command processing.
-     */
-    protected void assembleWorkflowExecutionGraphAssembler(
-                                                           final WorkflowExecuteContextBuilder workflowExecuteContextBuilder) {
-        final IWorkflowExecutionGraphAssembler assembler = createWorkflowExecutionGraphAssembler(
-                workflowExecuteContextBuilder);
-        workflowExecuteContextBuilder.setWorkflowExecutionGraphAssembler(assembler);
-    }
-
-    /**
-     * Create the workflow execution graph assembler.
-     * <p>
-     * Subclasses should implement this method to provide the logic for building
-     * the WorkflowExecutionGraph. The returned assembler will be invoked when
-     * the WorkflowStartLifecycleEvent is fired.
-     *
-     * @return the assembler for creating the WorkflowExecutionGraph, or null if no graph is needed
-     */
-    protected abstract IWorkflowExecutionGraphAssembler createWorkflowExecutionGraphAssembler(
-                                                                                              final WorkflowExecuteContextBuilder workflowExecuteContextBuilder);
-
     protected List<String> parseStartNodesFromWorkflowInstance(
                                                                final WorkflowExecuteContextBuilder workflowExecuteContextBuilder) {
         final WorkflowInstance workflowInstance = workflowExecuteContextBuilder.getWorkflowInstance();

dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/command/handler/ReRunWorkflowCommandHandler.java

@@ -20,16 +20,12 @@
 import org.apache.dolphinscheduler.common.enums.CommandType;
 import org.apache.dolphinscheduler.common.enums.WorkflowExecutionStatus;
 import org.apache.dolphinscheduler.dao.entity.Command;
-import org.apache.dolphinscheduler.dao.entity.TaskInstance;
 import org.apache.dolphinscheduler.dao.entity.WorkflowInstance;
-import org.apache.dolphinscheduler.dao.repository.TaskInstanceDao;
 import org.apache.dolphinscheduler.dao.repository.WorkflowInstanceDao;
 import org.apache.dolphinscheduler.server.master.config.MasterConfig;
-import org.apache.dolphinscheduler.server.master.engine.graph.IWorkflowExecutionGraphAssembler;
 import org.apache.dolphinscheduler.server.master.runner.WorkflowExecuteContext.WorkflowExecuteContextBuilder;
 
 import java.util.Date;
-import java.util.List;
 
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Component;
@@ -43,9 +39,6 @@ public class ReRunWorkflowCommandHandler extends RunWorkflowCommandHandler {
     @Autowired
     private WorkflowInstanceDao workflowInstanceDao;
 
-    @Autowired
-    private TaskInstanceDao taskInstanceDao;
-
     @Autowired
     private MasterConfig masterConfig;
 
@@ -80,22 +73,6 @@ protected void assembleWorkflowInstance(final WorkflowExecuteContextBuilder work
         workflowExecuteContextBuilder.setWorkflowInstance(workflowInstance);
     }
 
-    @Override
-    protected IWorkflowExecutionGraphAssembler createWorkflowExecutionGraphAssembler(
-                                                                                     final WorkflowExecuteContextBuilder workflowExecuteContextBuilder) {
-        // Capture parent assembler
-        final IWorkflowExecutionGraphAssembler parentAssembler =
-                super.createWorkflowExecutionGraphAssembler(workflowExecuteContextBuilder);
-        final WorkflowInstance workflowInstance = workflowExecuteContextBuilder.getWorkflowInstance();
-
-        return () -> {
-            // Mark all task instances as invalid before creating the new graph
-            final List<TaskInstance> taskInstances = getValidTaskInstance(workflowInstance);
-            taskInstanceDao.markTaskInstanceInvalid(taskInstances);
-            return parentAssembler.assemble();
-        };
-    }
-
     @Override
     public CommandType commandType() {
         return CommandType.REPEAT_RUNNING;

dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/command/handler/RecoverFailureTaskCommandHandler.java

@@ -20,36 +20,14 @@
 import org.apache.dolphinscheduler.common.enums.CommandType;
 import org.apache.dolphinscheduler.common.enums.WorkflowExecutionStatus;
 import org.apache.dolphinscheduler.dao.entity.Command;
-import org.apache.dolphinscheduler.dao.entity.TaskInstance;
 import org.apache.dolphinscheduler.dao.entity.WorkflowInstance;
-import org.apache.dolphinscheduler.dao.repository.TaskInstanceDao;
 import org.apache.dolphinscheduler.dao.repository.WorkflowInstanceDao;
-import org.apache.dolphinscheduler.plugin.task.api.enums.TaskExecutionStatus;
 import org.apache.dolphinscheduler.server.master.config.MasterConfig;
-import org.apache.dolphinscheduler.server.master.engine.graph.IWorkflowExecutionGraphAssembler;
-import org.apache.dolphinscheduler.server.master.engine.graph.IWorkflowGraph;
-import org.apache.dolphinscheduler.server.master.engine.graph.WorkflowExecutionGraph;
-import org.apache.dolphinscheduler.server.master.engine.graph.WorkflowGraphTopologyLogicalVisitor;
-import org.apache.dolphinscheduler.server.master.engine.task.runnable.TaskExecutionRunnable;
-import org.apache.dolphinscheduler.server.master.engine.task.runnable.TaskExecutionRunnableBuilder;
-import org.apache.dolphinscheduler.server.master.engine.task.runnable.TaskInstanceFactories;
 import org.apache.dolphinscheduler.server.master.runner.WorkflowExecuteContext.WorkflowExecuteContextBuilder;
 
-import java.util.ArrayList;
-import java.util.HashSet;
-import java.util.List;
-import java.util.Map;
-import java.util.Set;
-import java.util.function.BiConsumer;
-import java.util.function.Function;
-import java.util.stream.Collectors;
-
 import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.context.ApplicationContext;
 import org.springframework.stereotype.Component;
 
-import com.google.common.collect.Lists;
-
 /**
  * This handler used to handle {@link CommandType#START_FAILURE_TASK_PROCESS}.
  * <p> Will start the failure/pause/killed and other task instance which is behind success tasks instance but not been triggered.
@@ -60,15 +38,6 @@ public class RecoverFailureTaskCommandHandler extends AbstractCommandHandler {
     @Autowired
     private WorkflowInstanceDao workflowInstanceDao;
 
-    @Autowired
-    private TaskInstanceDao taskInstanceDao;
-
-    @Autowired
-    private ApplicationContext applicationContext;
-
-    @Autowired
-    private TaskInstanceFactories taskInstanceFactories;
-
     @Autowired
     private MasterConfig masterConfig;
 
@@ -101,146 +70,6 @@ protected void assembleWorkflowInstance(
         workflowExecuteContextBuilder.setWorkflowInstance(workflowInstance);
     }
 
-    @Override
-    protected IWorkflowExecutionGraphAssembler createWorkflowExecutionGraphAssembler(
-                                                                                     final WorkflowExecuteContextBuilder workflowExecuteContextBuilder) {
-        // Capture the context needed for deferred graph assembly
-        final IWorkflowGraph workflowGraph = workflowExecuteContextBuilder.getWorkflowGraph();
-        final WorkflowInstance workflowInstance = workflowExecuteContextBuilder.getWorkflowInstance();
-        final List<String> startNodes = parseStartNodesFromWorkflowInstance(workflowExecuteContextBuilder);
-
-        return () -> {
-            final Map<String, TaskInstance> taskInstanceMap = dealWithHistoryTaskInstances(
-                    workflowExecuteContextBuilder)
-                            .stream()
-                            .collect(Collectors.toMap(TaskInstance::getName, Function.identity()));
-
-            final WorkflowExecutionGraph workflowExecutionGraph = new WorkflowExecutionGraph();
-
-            final BiConsumer<String, Set<String>> taskExecutionRunnableCreator = (task, successors) -> {
-                final TaskExecutionRunnableBuilder taskExecutionRunnableBuilder =
-                        TaskExecutionRunnableBuilder
-                                .builder()
-                                .workflowExecutionGraph(workflowExecutionGraph)
-                                .workflowDefinition(workflowExecuteContextBuilder.getWorkflowDefinition())
-                                .project(workflowExecuteContextBuilder.getProject())
-                                .workflowInstance(workflowInstance)
-                                .taskDefinition(workflowGraph.getTaskNodeByName(task))
-                                .taskInstance(taskInstanceMap.get(task))
-                                .workflowEventBus(workflowExecuteContextBuilder.getWorkflowEventBus())
-                                .applicationContext(applicationContext)
-                                .build();
-                workflowExecutionGraph.addNode(new TaskExecutionRunnable(taskExecutionRunnableBuilder));
-                workflowExecutionGraph.addEdge(task, successors);
-            };
-
-            final WorkflowGraphTopologyLogicalVisitor workflowGraphTopologyLogicalVisitor =
-                    WorkflowGraphTopologyLogicalVisitor.builder()
-                            .taskDependType(workflowInstance.getTaskDependType())
-                            .onWorkflowGraph(workflowGraph)
-                            .fromTask(startNodes)
-                            .doVisitFunction(taskExecutionRunnableCreator)
-                            .build();
-            workflowGraphTopologyLogicalVisitor.visit();
-            workflowExecutionGraph.removeUnReachableEdge();
-
-            return workflowExecutionGraph;
-        };
-    }
-
-    /**
-     * Return the valid task instance which should not be recovered.
-     * <p> Will mark the failure/killed task instance as invalid.
-     */
-    private List<TaskInstance> dealWithHistoryTaskInstances(
-                                                            final WorkflowExecuteContextBuilder workflowExecuteContextBuilder) {
-        final WorkflowInstance workflowInstance = workflowExecuteContextBuilder.getWorkflowInstance();
-        final Map<String, TaskInstance> taskInstanceMap = super.getValidTaskInstance(workflowInstance)
-                .stream()
-                .collect(Collectors.toMap(TaskInstance::getName, Function.identity()));
-
-        final IWorkflowGraph workflowGraph = workflowExecuteContextBuilder.getWorkflowGraph();
-
-        final Set<String> needRecoverTasks = new HashSet<>();
-        final Set<String> markInvalidTasks = new HashSet<>();
-        final BiConsumer<String, Set<String>> historyTaskInstanceMarker = (task, successors) -> {
-            // If the parent is need recover
-            // Then the task should mark as invalid, and it's child should be mark as invalidated.
-            if (markInvalidTasks.contains(task)) {
-                if (taskInstanceMap.containsKey(task)) {
-                    taskInstanceDao.markTaskInstanceInvalid(Lists.newArrayList(taskInstanceMap.get(task)));
-                    taskInstanceMap.remove(task);
-                }
-                markInvalidTasks.addAll(successors);
-                return;
-            }
-
-            final TaskInstance taskInstance = taskInstanceMap.get(task);
-            if (taskInstance == null) {
-                return;
-            }
-
-            if (isTaskNeedRecreate(taskInstance) || isTaskCanRecover(taskInstance)) {
-                needRecoverTasks.add(task);
-                markInvalidTasks.addAll(successors);
-            }
-        };
-
-        final WorkflowGraphTopologyLogicalVisitor workflowGraphTopologyLogicalVisitor =
-                WorkflowGraphTopologyLogicalVisitor.builder()
-                        .onWorkflowGraph(workflowGraph)
-                        .taskDependType(workflowInstance.getTaskDependType())
-                        .fromTask(parseStartNodesFromWorkflowInstance(workflowExecuteContextBuilder))
-                        .doVisitFunction(historyTaskInstanceMarker)
-                        .build();
-        workflowGraphTopologyLogicalVisitor.visit();
-
-        for (String task : needRecoverTasks) {
-            final TaskInstance taskInstance = taskInstanceMap.get(task);
-            if (isTaskCanRecover(taskInstance)) {
-                taskInstanceMap.put(task, createRecoverTaskInstance(taskInstance));
-                continue;
-            }
-            if (isTaskNeedRecreate(taskInstance)) {
-                taskInstanceMap.put(task, createRecreatedTaskInstance(taskInstance));
-            }
-        }
-        return new ArrayList<>(taskInstanceMap.values());
-    }
-
-    /**
-     * Whether the task need to be recreated.
-     * <p> If the task state is FAILURE and KILL, then will mark the task invalid and recreate the task.
-     */
-    private boolean isTaskNeedRecreate(final TaskInstance taskInstance) {
-        if (taskInstance == null) {
-            return false;
-        }
-        return taskInstance.getState() == TaskExecutionStatus.FAILURE
-                || taskInstance.getState() == TaskExecutionStatus.KILL;
-    }
-
-    private TaskInstance createRecreatedTaskInstance(final TaskInstance taskInstance) {
-        return taskInstanceFactories.failedRecoverTaskInstanceFactory()
-                .builder()
-                .withTaskInstance(taskInstance)
-                .build();
-    }
-
-    private boolean isTaskCanRecover(final TaskInstance taskInstance) {
-        if (taskInstance == null) {
-            return false;
-        }
-        return taskInstance.getState() == TaskExecutionStatus.PAUSE;
-    }
-
-    private TaskInstance createRecoverTaskInstance(final TaskInstance taskInstance) {
-        return taskInstanceFactories.pauseRecoverTaskInstanceFactory()
-                .builder()
-                .withTaskInstance(taskInstance)
-                .build();
-    }
-
     @Override
     public CommandType commandType() {
         return CommandType.START_FAILURE_TASK_PROCESS;

dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/command/handler/RecoverSerialWaitCommandHandler.java

@@ -23,7 +23,6 @@
 import org.apache.dolphinscheduler.dao.entity.WorkflowInstance;
 import org.apache.dolphinscheduler.dao.repository.WorkflowInstanceDao;
 import org.apache.dolphinscheduler.server.master.config.MasterConfig;
-import org.apache.dolphinscheduler.server.master.engine.graph.IWorkflowExecutionGraphAssembler;
 import org.apache.dolphinscheduler.server.master.runner.WorkflowExecuteContext;
 
 import org.springframework.beans.factory.annotation.Autowired;
@@ -50,13 +49,6 @@ protected void assembleWorkflowInstance(WorkflowExecuteContext.WorkflowExecuteCo
         workflowExecuteContextBuilder.setWorkflowInstance(workflowInstance);
     }
 
-    @Override
-    protected IWorkflowExecutionGraphAssembler createWorkflowExecutionGraphAssembler(
-                                                                                     WorkflowExecuteContext.WorkflowExecuteContextBuilder workflowExecuteContextBuilder) {
-        // No execution graph needed for serial wait recovery
-        return null;
-    }
-
     @Override
     public CommandType commandType() {
         return CommandType.RECOVER_SERIAL_WAIT;