[Fix-17834] [Master] Fix workflow failure strategy cannot work (#17851)

ruanwenjun · web-flow · commit 6b038383265b · 2026-01-10T11:49:04.000+08:00
diff --git a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/task/runnable/ITaskExecutionRunnable.java b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/task/runnable/ITaskExecutionRunnable.java
@@ -67,6 +67,8 @@ default String getName() {
      */
     boolean isTaskInstanceCanRetry();
 
+    boolean isFailure();
+
     /**
      * Retry the TaskExecutionRunnable.
      * <p> Will create retry task instance and start it.
diff --git a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/task/runnable/TaskExecutionRunnable.java b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/task/runnable/TaskExecutionRunnable.java
@@ -96,6 +96,11 @@ public boolean isTaskInstanceCanRetry() {
         return taskInstance.getRetryTimes() < taskInstance.getMaxRetryTimes();
     }
 
+    @Override
+    public boolean isFailure() {
+        return isTaskInstanceInitialized() && !isTaskInstanceCanRetry() && taskInstance.getState().isFailure();
+    }
+
     @Override
     public void retry() {
         checkState(isTaskInstanceInitialized(), "The task instance is not initialized, can't initialize retry task.");
diff --git a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/task/statemachine/AbstractTaskStateAction.java b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/task/statemachine/AbstractTaskStateAction.java
@@ -136,7 +136,7 @@ public void onPausedEvent(final IWorkflowExecutionRunnable workflowExecutionRunn
         releaseTaskInstanceResourcesIfNeeded(taskExecutionRunnable);
         persistentTaskInstancePausedEventToDB(taskExecutionRunnable, taskPausedEvent);
         taskExecutionRunnable.getWorkflowExecutionGraph().markTaskExecutionRunnableChainPause(taskExecutionRunnable);
-        publishWorkflowInstanceTopologyLogicalTransitionEvent(taskExecutionRunnable);
+        publishWorkflowInstanceTopologyLogicalTransitionEvent(workflowExecutionRunnable, taskExecutionRunnable);
     }
 
     private void persistentTaskInstancePausedEventToDB(final ITaskExecutionRunnable taskExecutionRunnable,
@@ -153,7 +153,7 @@ public void onKilledEvent(final IWorkflowExecutionRunnable workflowExecutionRunn
         releaseTaskInstanceResourcesIfNeeded(taskExecutionRunnable);
         persistentTaskInstanceKilledEventToDB(taskExecutionRunnable, taskInstanceKillEvent);
         taskExecutionRunnable.getWorkflowExecutionGraph().markTaskExecutionRunnableChainKill(taskExecutionRunnable);
-        publishWorkflowInstanceTopologyLogicalTransitionEvent(taskExecutionRunnable);
+        publishWorkflowInstanceTopologyLogicalTransitionEvent(workflowExecutionRunnable, taskExecutionRunnable);
     }
 
     private void persistentTaskInstanceKilledEventToDB(final ITaskExecutionRunnable taskExecutionRunnable,
@@ -181,11 +181,12 @@ public void onFailedEvent(final IWorkflowExecutionRunnable workflowExecutionRunn
         final IWorkflowExecutionGraph workflowExecutionGraph = taskExecutionRunnable.getWorkflowExecutionGraph();
         if (workflowExecutionGraph.isAllSuccessorsAreConditionTask(taskExecutionRunnable)) {
             mergeTaskVarPoolToWorkflow(workflowExecutionRunnable, taskExecutionRunnable);
-            publishWorkflowInstanceTopologyLogicalTransitionEvent(taskExecutionRunnable);
+            publishWorkflowInstanceTopologyLogicalTransitionEvent(workflowExecutionRunnable, taskExecutionRunnable);
             return;
         }
+        // todo: Use Plugin to extend the act strategy on xxEvent.
         taskExecutionRunnable.getWorkflowExecutionGraph().markTaskExecutionRunnableChainFailure(taskExecutionRunnable);
-        publishWorkflowInstanceTopologyLogicalTransitionEvent(taskExecutionRunnable);
+        publishWorkflowInstanceTopologyLogicalTransitionEvent(workflowExecutionRunnable, taskExecutionRunnable);
     }
 
     private void persistentTaskInstanceFailedEventToDB(final ITaskExecutionRunnable taskExecutionRunnable,
@@ -203,7 +204,7 @@ public void onSucceedEvent(final IWorkflowExecutionRunnable workflowExecutionRun
         releaseTaskInstanceResourcesIfNeeded(taskExecutionRunnable);
         persistentTaskInstanceSuccessEventToDB(taskExecutionRunnable, taskSuccessEvent);
         mergeTaskVarPoolToWorkflow(workflowExecutionRunnable, taskExecutionRunnable);
-        publishWorkflowInstanceTopologyLogicalTransitionEvent(taskExecutionRunnable);
+        publishWorkflowInstanceTopologyLogicalTransitionEvent(workflowExecutionRunnable, taskExecutionRunnable);
     }
 
     protected void mergeTaskVarPoolToWorkflow(final IWorkflowExecutionRunnable workflowExecutionRunnable,
@@ -244,9 +245,9 @@ protected void tryToDispatchTask(final ITaskExecutionRunnable taskExecutionRunna
         taskExecutionRunnable.getWorkflowEventBus().publish(TaskDispatchLifecycleEvent.of(taskExecutionRunnable));
     }
 
-    protected void publishWorkflowInstanceTopologyLogicalTransitionEvent(final ITaskExecutionRunnable taskExecutionRunnable) {
-        final Integer workflowInstanceId = taskExecutionRunnable.getWorkflowInstance().getId();
-        final IWorkflowExecutionRunnable workflowExecutionRunnable = workflowRepository.get(workflowInstanceId);
+    protected void publishWorkflowInstanceTopologyLogicalTransitionEvent(
+                                                                         final IWorkflowExecutionRunnable workflowExecutionRunnable,
+                                                                         final ITaskExecutionRunnable taskExecutionRunnable) {
         taskExecutionRunnable
                 .getWorkflowEventBus()
                 .publish(
diff --git a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/task/statemachine/TaskKillStateAction.java b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/task/statemachine/TaskKillStateAction.java
@@ -47,7 +47,7 @@ public void onStartEvent(final IWorkflowExecutionRunnable workflowExecutionRunna
                              final TaskStartLifecycleEvent taskStartEvent) {
         throwExceptionIfStateIsNotMatch(taskExecutionRunnable);
         taskExecutionRunnable.getWorkflowExecutionGraph().markTaskExecutionRunnableChainKill(taskExecutionRunnable);
-        publishWorkflowInstanceTopologyLogicalTransitionEvent(taskExecutionRunnable);
+        publishWorkflowInstanceTopologyLogicalTransitionEvent(workflowExecutionRunnable, taskExecutionRunnable);
     }
 
     @Override
diff --git a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/task/statemachine/TaskPauseStateAction.java b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/task/statemachine/TaskPauseStateAction.java
@@ -47,7 +47,7 @@ public void onStartEvent(final IWorkflowExecutionRunnable workflowExecutionRunna
                              final TaskStartLifecycleEvent taskStartEvent) {
         throwExceptionIfStateIsNotMatch(taskExecutionRunnable);
         taskExecutionRunnable.getWorkflowExecutionGraph().markTaskExecutionRunnableChainPause(taskExecutionRunnable);
-        publishWorkflowInstanceTopologyLogicalTransitionEvent(taskExecutionRunnable);
+        publishWorkflowInstanceTopologyLogicalTransitionEvent(workflowExecutionRunnable, taskExecutionRunnable);
     }
 
     @Override
diff --git a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/workflow/policy/ContinueWorkflowFailureStrategy.java b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/workflow/policy/ContinueWorkflowFailureStrategy.java
@@ -0,0 +1,41 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.dolphinscheduler.server.master.engine.workflow.policy;
+
+import org.apache.dolphinscheduler.server.master.engine.task.runnable.ITaskExecutionRunnable;
+import org.apache.dolphinscheduler.server.master.engine.workflow.runnable.IWorkflowExecutionRunnable;
+
+/**
+ * The strategy used to deal with {@link org.apache.dolphinscheduler.common.enums.FailureStrategy#CONTINUE} when task failure occurs.
+ * <p> Will wait the active the tasks finished.
+ */
+public class ContinueWorkflowFailureStrategy implements IWorkflowFailureStrategy {
+
+    @Override
+    public void onTaskFailure(IWorkflowExecutionRunnable workflowExecutionRunnable,
+                              ITaskExecutionRunnable taskExecutionRunnable) {
+        // do nothing, just continue workflow execution
+    }
+
+    @Override
+    public boolean canTriggerSuccessor(IWorkflowExecutionRunnable workflowExecutionRunnable,
+                                       ITaskExecutionRunnable taskExecutionRunnable) {
+        return true;
+    }
+
+}
diff --git a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/workflow/policy/EndWorkflowFailureStrategy.java b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/workflow/policy/EndWorkflowFailureStrategy.java
@@ -0,0 +1,46 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.dolphinscheduler.server.master.engine.workflow.policy;
+
+import org.apache.dolphinscheduler.server.master.engine.task.runnable.ITaskExecutionRunnable;
+import org.apache.dolphinscheduler.server.master.engine.workflow.runnable.IWorkflowExecutionRunnable;
+
+import lombok.extern.slf4j.Slf4j;
+
+/**
+ * The strategy used to deal with {@link org.apache.dolphinscheduler.common.enums.FailureStrategy#END} when task failure occurs.
+ * <p> Will kill the active tasks and end the workflow execution.
+ */
+@Slf4j
+public class EndWorkflowFailureStrategy implements IWorkflowFailureStrategy {
+
+    @Override
+    public void onTaskFailure(IWorkflowExecutionRunnable workflowExecutionRunnable,
+                              ITaskExecutionRunnable taskExecutionRunnable) {
+        log.info("The workflow instance: [{}] using END failure strategy, will kill the active tasks.",
+                workflowExecutionRunnable.getName());
+        workflowExecutionRunnable.killActiveTasks();
+    }
+
+    @Override
+    public boolean canTriggerSuccessor(IWorkflowExecutionRunnable workflowExecutionRunnable,
+                                       ITaskExecutionRunnable taskExecutionRunnable) {
+        return !workflowExecutionRunnable.getWorkflowExecutionGraph().isExistFailureTaskExecutionRunnableChain();
+    }
+
+}
diff --git a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/workflow/policy/IWorkflowFailureStrategy.java b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/workflow/policy/IWorkflowFailureStrategy.java
@@ -0,0 +1,34 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.dolphinscheduler.server.master.engine.workflow.policy;
+
+import org.apache.dolphinscheduler.server.master.engine.task.runnable.ITaskExecutionRunnable;
+import org.apache.dolphinscheduler.server.master.engine.workflow.runnable.IWorkflowExecutionRunnable;
+
+/**
+ * Used to deal with {@link org.apache.dolphinscheduler.common.enums.FailureStrategy} when task failure occurs
+ */
+public interface IWorkflowFailureStrategy {
+
+    void onTaskFailure(IWorkflowExecutionRunnable workflowExecutionRunnable,
+                       ITaskExecutionRunnable taskExecutionRunnable);
+
+    boolean canTriggerSuccessor(IWorkflowExecutionRunnable workflowExecutionRunnable,
+                                ITaskExecutionRunnable taskExecutionRunnable);
+
+}
diff --git a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/workflow/policy/WorkflowFailureStrategyFactory.java b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/workflow/policy/WorkflowFailureStrategyFactory.java
@@ -0,0 +1,31 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.dolphinscheduler.server.master.engine.workflow.policy;
+
+import org.apache.dolphinscheduler.common.enums.FailureStrategy;
+
+public class WorkflowFailureStrategyFactory {
+
+    public static IWorkflowFailureStrategy getStrategy(final FailureStrategy failureStrategy) {
+        if (failureStrategy == FailureStrategy.END) {
+            return new EndWorkflowFailureStrategy();
+        }
+        return new ContinueWorkflowFailureStrategy();
+    }
+
+}
diff --git a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/workflow/runnable/IWorkflowExecutionRunnable.java b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/workflow/runnable/IWorkflowExecutionRunnable.java
@@ -21,7 +21,9 @@
 import org.apache.dolphinscheduler.dao.entity.WorkflowInstance;
 import org.apache.dolphinscheduler.server.master.engine.WorkflowEventBus;
 import org.apache.dolphinscheduler.server.master.engine.graph.IWorkflowExecutionGraph;
+import org.apache.dolphinscheduler.server.master.engine.task.runnable.ITaskExecutionRunnable;
 import org.apache.dolphinscheduler.server.master.engine.workflow.listener.IWorkflowLifecycleListener;
+import org.apache.dolphinscheduler.server.master.engine.workflow.policy.IWorkflowFailureStrategy;
 import org.apache.dolphinscheduler.server.master.runner.IWorkflowExecuteContext;
 
 import java.util.List;
@@ -68,6 +70,13 @@ default boolean isWorkflowReadyStop() {
         return workflowExecutionStatus == WorkflowExecutionStatus.READY_STOP;
     }
 
+    /**
+     * Kill the active tasks of the WorkflowExecutionRunnable.
+     */
+    default void killActiveTasks() {
+        getWorkflowExecutionGraph().getActiveTaskExecutionRunnable().forEach(ITaskExecutionRunnable::kill);
+    }
+
     /**
      * Get the WorkflowExecuteContext belongs to the WorkflowExecutionRunnable.
      */
@@ -111,4 +120,5 @@ default IWorkflowExecutionGraph getWorkflowExecutionGraph() {
      */
     void registerWorkflowInstanceLifecycleListener(IWorkflowLifecycleListener listener);
 
+    IWorkflowFailureStrategy getWorkflowFailureStrategy();
 }
diff --git a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/workflow/runnable/WorkflowExecutionRunnable.java b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/workflow/runnable/WorkflowExecutionRunnable.java
@@ -23,8 +23,11 @@
 import org.apache.dolphinscheduler.server.master.engine.workflow.lifecycle.event.WorkflowPauseLifecycleEvent;
 import org.apache.dolphinscheduler.server.master.engine.workflow.lifecycle.event.WorkflowStopLifecycleEvent;
 import org.apache.dolphinscheduler.server.master.engine.workflow.listener.IWorkflowLifecycleListener;
+import org.apache.dolphinscheduler.server.master.engine.workflow.policy.IWorkflowFailureStrategy;
+import org.apache.dolphinscheduler.server.master.engine.workflow.policy.WorkflowFailureStrategyFactory;
 import org.apache.dolphinscheduler.server.master.runner.IWorkflowExecuteContext;
 
+import java.util.ArrayList;
 import java.util.List;
 
 import lombok.Getter;
@@ -39,9 +42,15 @@ public class WorkflowExecutionRunnable implements IWorkflowExecutionRunnable {
     @Getter
     private final List<IWorkflowLifecycleListener> workflowInstanceLifecycleListeners;
 
+    @Getter
+    private final IWorkflowFailureStrategy workflowFailureStrategy;
+
     public WorkflowExecutionRunnable(WorkflowExecutionRunnableBuilder workflowExecutionRunnableBuilder) {
         this.workflowExecuteContext = workflowExecutionRunnableBuilder.getWorkflowExecuteContextBuilder().build();
-        this.workflowInstanceLifecycleListeners = workflowExecuteContext.getWorkflowInstanceLifecycleListeners();
+        this.workflowInstanceLifecycleListeners =
+                new ArrayList<>(workflowExecuteContext.getWorkflowInstanceLifecycleListeners());
+        this.workflowFailureStrategy = WorkflowFailureStrategyFactory
+                .getStrategy(workflowExecuteContext.getWorkflowInstance().getFailureStrategy());
     }
 
     @Override
diff --git a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/workflow/statemachine/AbstractWorkflowStateAction.java b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/workflow/statemachine/AbstractWorkflowStateAction.java
@@ -34,6 +34,7 @@
 import org.apache.dolphinscheduler.server.master.engine.task.runnable.ITaskExecutionRunnable;
 import org.apache.dolphinscheduler.server.master.engine.workflow.lifecycle.event.WorkflowFinalizeLifecycleEvent;
 import org.apache.dolphinscheduler.server.master.engine.workflow.lifecycle.event.WorkflowTopologyLogicalTransitionWithTaskFinishLifecycleEvent;
+import org.apache.dolphinscheduler.server.master.engine.workflow.policy.IWorkflowFailureStrategy;
 import org.apache.dolphinscheduler.server.master.engine.workflow.runnable.IWorkflowExecutionRunnable;
 import org.apache.dolphinscheduler.server.master.utils.WorkflowInstanceUtils;
 import org.apache.dolphinscheduler.service.alert.WorkflowAlertManager;
@@ -103,18 +104,6 @@ protected void triggerTasks(final IWorkflowExecutionRunnable workflowExecutionRu
         }
     }
 
-    protected void killActiveTask(final IWorkflowExecutionRunnable workflowExecutionRunnable) {
-        try {
-            LogUtils.setWorkflowInstanceIdMDC(workflowExecutionRunnable.getId());
-            workflowExecutionRunnable
-                    .getWorkflowExecutionGraph()
-                    .getActiveTaskExecutionRunnable()
-                    .forEach(ITaskExecutionRunnable::kill);
-        } finally {
-            LogUtils.removeWorkflowInstanceIdMDC();
-        }
-    }
-
     protected void pauseActiveTask(final IWorkflowExecutionRunnable workflowExecutionRunnable) {
         try {
             LogUtils.setWorkflowInstanceIdMDC(workflowExecutionRunnable.getId());
@@ -129,21 +118,25 @@ protected void pauseActiveTask(final IWorkflowExecutionRunnable workflowExecutio
 
     protected void tryToTriggerSuccessorsAfterTaskFinish(final IWorkflowExecutionRunnable workflowExecutionRunnable,
                                                          final ITaskExecutionRunnable taskExecutionRunnable) {
+        successorFlowAdjuster.adjustSuccessorFlow(taskExecutionRunnable);
+
+        final IWorkflowFailureStrategy workflowFailureStrategy = workflowExecutionRunnable.getWorkflowFailureStrategy();
+        if (taskExecutionRunnable.isFailure()) {
+            workflowFailureStrategy.onTaskFailure(workflowExecutionRunnable, taskExecutionRunnable);
+        }
+
         final IWorkflowExecutionGraph workflowExecutionGraph = workflowExecutionRunnable.getWorkflowExecutionGraph();
         if (workflowExecutionGraph.isEndOfTaskChain(taskExecutionRunnable)) {
             emitWorkflowFinishedEventIfApplicable(workflowExecutionRunnable);
             return;
         }
 
-        successorFlowAdjuster.adjustSuccessorFlow(taskExecutionRunnable);
-        final List<ITaskExecutionRunnable> successors = workflowExecutionGraph.getSuccessors(taskExecutionRunnable);
-        if (successors.isEmpty()) {
-            log.debug("The task: {} has no successor, try to emit workflow finished event",
-                    taskExecutionRunnable.getName());
+        if (!workflowFailureStrategy.canTriggerSuccessor(workflowExecutionRunnable, taskExecutionRunnable)) {
             emitWorkflowFinishedEventIfApplicable(workflowExecutionRunnable);
             return;
         }
-        triggerTasks(workflowExecutionRunnable, successors);
+
+        triggerTasks(workflowExecutionRunnable, workflowExecutionGraph.getSuccessors(taskExecutionRunnable));
     }
 
     protected void workflowFinish(final IWorkflowExecutionRunnable workflowExecutionRunnable,
diff --git a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/workflow/statemachine/WorkflowReadyStopStateAction.java b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/workflow/statemachine/WorkflowReadyStopStateAction.java
@@ -85,7 +85,7 @@ public void onPausedEvent(final IWorkflowExecutionRunnable workflowExecutionRunn
     public void onStopEvent(final IWorkflowExecutionRunnable workflowExecutionRunnable,
                             final WorkflowStopLifecycleEvent workflowStopEvent) {
         throwExceptionIfStateIsNotMatch(workflowExecutionRunnable);
-        super.killActiveTask(workflowExecutionRunnable);
+        workflowExecutionRunnable.killActiveTasks();
     }
 
     @Override
diff --git a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/workflow/statemachine/WorkflowRunningStateAction.java b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/workflow/statemachine/WorkflowRunningStateAction.java
diff --git a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/runner/WorkflowExecuteContext.java b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/runner/WorkflowExecuteContext.java
diff --git a/dolphinscheduler-master/src/test/java/org/apache/dolphinscheduler/server/master/integration/WorkflowOperator.java b/dolphinscheduler-master/src/test/java/org/apache/dolphinscheduler/server/master/integration/WorkflowOperator.java
diff --git a/dolphinscheduler-master/src/test/java/org/apache/dolphinscheduler/server/master/integration/cases/WorkflowStartTestCase.java b/dolphinscheduler-master/src/test/java/org/apache/dolphinscheduler/server/master/integration/cases/WorkflowStartTestCase.java
diff --git a/dolphinscheduler-master/src/test/resources/it/start/workflow_with_three_parallel_fake_task_using_failure_strategy.yaml b/dolphinscheduler-master/src/test/resources/it/start/workflow_with_three_parallel_fake_task_using_failure_strategy.yaml
diff --git a/dolphinscheduler-task-plugin/dolphinscheduler-task-api/src/main/java/org/apache/dolphinscheduler/plugin/task/api/enums/TaskExecutionStatus.java b/dolphinscheduler-task-plugin/dolphinscheduler-task-api/src/main/java/org/apache/dolphinscheduler/plugin/task/api/enums/TaskExecutionStatus.java

Original file line number	Diff line number	Diff line change
`@@ -47,7 +47,7 @@ public void onStartEvent(final IWorkflowExecutionRunnable workflowExecutionRunna`
`47`	`47`	`final TaskStartLifecycleEvent taskStartEvent) {`
`48`	`48`	`throwExceptionIfStateIsNotMatch(taskExecutionRunnable);`
`49`	`49`	`taskExecutionRunnable.getWorkflowExecutionGraph().markTaskExecutionRunnableChainKill(taskExecutionRunnable);`
`50`		`- publishWorkflowInstanceTopologyLogicalTransitionEvent(taskExecutionRunnable);`
	`50`	`+ publishWorkflowInstanceTopologyLogicalTransitionEvent(workflowExecutionRunnable, taskExecutionRunnable);`
`51`	`51`	`}`
`52`	`52`
`53`	`53`	`@Override`