Retry TaskKillLifecycleEvent/TaskPauseLifecycleEvent when task instance is submitted but already dispatched (#17203)

ruanwenjun · web-flow · commit f9b0e044c32a · 2025-05-26T16:50:51.000+08:00
diff --git a/dolphinscheduler-eventbus/src/main/java/org/apache/dolphinscheduler/eventbus/AbstractDelayEvent.java b/dolphinscheduler-eventbus/src/main/java/org/apache/dolphinscheduler/eventbus/AbstractDelayEvent.java
@@ -34,6 +34,7 @@ public abstract class AbstractDelayEvent implements IEvent, Delayed {
 
     private static final long DEFAULT_DELAY_TIME = 0;
 
+    // In milliseconds
     protected long delayTime;
 
     @Builder.Default
diff --git a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/task/dispatcher/WorkerGroupDispatcher.java b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/task/dispatcher/WorkerGroupDispatcher.java
@@ -18,6 +18,7 @@
 package org.apache.dolphinscheduler.server.master.engine.task.dispatcher;
 
 import org.apache.dolphinscheduler.common.thread.BaseDaemonThread;
+import org.apache.dolphinscheduler.plugin.task.api.utils.LogUtils;
 import org.apache.dolphinscheduler.server.master.engine.task.client.ITaskExecutorClient;
 import org.apache.dolphinscheduler.server.master.engine.task.runnable.ITaskExecutionRunnable;
 import org.apache.dolphinscheduler.server.master.runner.queue.PriorityAndDelayBasedTaskEntry;
@@ -79,7 +80,10 @@ public void run() {
             try (
                     TaskExecutorMDCUtils.MDCAutoClosable ignore =
                             TaskExecutorMDCUtils.logWithMDC(taskExecutionRunnable.getId())) {
+                LogUtils.setWorkflowInstanceIdMDC(taskExecutionRunnable.getTaskInstance().getWorkflowInstanceId());
                 doDispatchTask(taskExecutionRunnable);
+            } finally {
+                LogUtils.removeWorkflowInstanceIdMDC();
             }
         }
     }
diff --git a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/task/lifecycle/event/TaskKillLifecycleEvent.java b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/task/lifecycle/event/TaskKillLifecycleEvent.java
@@ -22,17 +22,24 @@
 import org.apache.dolphinscheduler.server.master.engine.task.lifecycle.TaskLifecycleEventType;
 import org.apache.dolphinscheduler.server.master.engine.task.runnable.ITaskExecutionRunnable;
 
-import lombok.AllArgsConstructor;
 import lombok.Getter;
 
 @Getter
-@AllArgsConstructor
 public class TaskKillLifecycleEvent extends AbstractTaskLifecycleEvent {
 
     private final ITaskExecutionRunnable taskExecutionRunnable;
 
+    private TaskKillLifecycleEvent(ITaskExecutionRunnable taskExecutionRunnable, long delayTime) {
+        super(delayTime);
+        this.taskExecutionRunnable = taskExecutionRunnable;
+    }
+
     public static TaskKillLifecycleEvent of(final ITaskExecutionRunnable taskExecutionRunnable) {
-        return new TaskKillLifecycleEvent(taskExecutionRunnable);
+        return of(taskExecutionRunnable, 0);
+    }
+
+    public static TaskKillLifecycleEvent of(final ITaskExecutionRunnable taskExecutionRunnable, long delayTime) {
+        return new TaskKillLifecycleEvent(taskExecutionRunnable, delayTime);
     }
 
     @Override
diff --git a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/task/lifecycle/event/TaskPauseLifecycleEvent.java b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/task/lifecycle/event/TaskPauseLifecycleEvent.java
@@ -22,17 +22,24 @@
 import org.apache.dolphinscheduler.server.master.engine.task.lifecycle.TaskLifecycleEventType;
 import org.apache.dolphinscheduler.server.master.engine.task.runnable.ITaskExecutionRunnable;
 
-import lombok.AllArgsConstructor;
 import lombok.Getter;
 
 @Getter
-@AllArgsConstructor
 public class TaskPauseLifecycleEvent extends AbstractTaskLifecycleEvent {
 
     private final ITaskExecutionRunnable taskExecutionRunnable;
 
+    private TaskPauseLifecycleEvent(final ITaskExecutionRunnable taskExecutionRunnable, long delayTime) {
+        super(delayTime);
+        this.taskExecutionRunnable = taskExecutionRunnable;
+    }
+
     public static TaskPauseLifecycleEvent of(ITaskExecutionRunnable taskExecutionRunnable) {
-        return new TaskPauseLifecycleEvent(taskExecutionRunnable);
+        return of(taskExecutionRunnable, 0);
+    }
+
+    public static TaskPauseLifecycleEvent of(ITaskExecutionRunnable taskExecutionRunnable, long delayTime) {
+        return new TaskPauseLifecycleEvent(taskExecutionRunnable, delayTime);
     }
 
     @Override
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
@@ -229,7 +229,7 @@ protected void failoverTask(final ITaskExecutionRunnable taskExecutionRunnable)
     protected void tryToDispatchTask(final ITaskExecutionRunnable taskExecutionRunnable) {
         if (isTaskNeedAcquireTaskGroupSlot(taskExecutionRunnable)) {
             acquireTaskGroupSlot(taskExecutionRunnable);
-            log.info("Task{} using taskGroup, success acquire taskGroup slot", taskExecutionRunnable.getName());
+            log.info("Task[name={}] using taskGroup, success acquire taskGroup slot", taskExecutionRunnable.getName());
             return;
         }
         taskExecutionRunnable.getWorkflowEventBus().publish(TaskDispatchLifecycleEvent.of(taskExecutionRunnable));
@@ -262,7 +262,7 @@ protected void throwExceptionIfStateIsNotMatch(final ITaskExecutionRunnable task
     protected void logWarningIfCannotDoAction(final ITaskExecutionRunnable taskExecutionRunnable,
                                               final AbstractLifecycleEvent event) {
         final TaskInstance taskInstance = taskExecutionRunnable.getTaskInstance();
-        log.warn("Task {} state is {} cannot do action on event: {}",
+        log.warn("Task[name={}] state is {} cannot do action on event: {}",
                 taskInstance.getName(),
                 taskInstance.getState(),
                 event);
diff --git a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/task/statemachine/TaskDispatchStateAction.java b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/task/statemachine/TaskDispatchStateAction.java
@@ -17,7 +17,6 @@
 
 package org.apache.dolphinscheduler.server.master.engine.task.statemachine;
 
-import org.apache.dolphinscheduler.dao.repository.TaskInstanceDao;
 import org.apache.dolphinscheduler.plugin.task.api.enums.TaskExecutionStatus;
 import org.apache.dolphinscheduler.server.master.engine.task.client.TaskExecutorClient;
 import org.apache.dolphinscheduler.server.master.engine.task.lifecycle.event.TaskDispatchLifecycleEvent;
@@ -44,8 +43,6 @@
 @Component
 public class TaskDispatchStateAction extends AbstractTaskStateAction {
 
-    @Autowired
-    private TaskInstanceDao taskInstanceDao;
     @Autowired
     private TaskExecutorClient taskExecutorClient;
 
diff --git a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/task/statemachine/TaskSubmittedStateAction.java b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/task/statemachine/TaskSubmittedStateAction.java
@@ -37,6 +37,8 @@
 import org.apache.dolphinscheduler.server.master.engine.task.runnable.ITaskExecutionRunnable;
 import org.apache.dolphinscheduler.server.master.engine.workflow.runnable.IWorkflowExecutionRunnable;
 
+import java.util.concurrent.TimeUnit;
+
 import lombok.extern.slf4j.Slf4j;
 
 import org.springframework.beans.factory.annotation.Autowired;
@@ -128,7 +130,10 @@ public void pauseEventAction(final IWorkflowExecutionRunnable workflowExecutionR
             taskExecutionRunnable.getWorkflowEventBus().publish(TaskPausedLifecycleEvent.of(taskExecutionRunnable));
             return;
         }
-        logWarningIfCannotDoAction(taskExecutionRunnable, taskPauseEvent);
+        log.info("The task[id={}] is submitted and already dispatched, cannot pause, will try to pause it after 5s",
+                taskExecutionRunnable.getId());
+        taskExecutionRunnable.getWorkflowEventBus()
+                .publish(TaskPauseLifecycleEvent.of(taskExecutionRunnable, TimeUnit.SECONDS.toSeconds(5)));
     }
 
     @Override
@@ -145,11 +150,14 @@ public void killEventAction(final IWorkflowExecutionRunnable workflowExecutionRu
                                 final TaskKillLifecycleEvent taskKillEvent) {
         throwExceptionIfStateIsNotMatch(taskExecutionRunnable);
         if (workerGroupDispatcherCoordinator.removeTask(taskExecutionRunnable)) {
-            log.info("Success kill task: {} before dispatch", taskExecutionRunnable.getName());
+            log.info("Success kill task[id={}] before dispatch", taskExecutionRunnable.getId());
             taskExecutionRunnable.getWorkflowEventBus().publish(TaskKilledLifecycleEvent.of(taskExecutionRunnable));
             return;
         }
-        logWarningIfCannotDoAction(taskExecutionRunnable, taskKillEvent);
+        log.info("The task[id={}] is submitted and already dispatched, cannot kill, will kill it after 5s",
+                taskExecutionRunnable.getId());
+        taskExecutionRunnable.getWorkflowEventBus()
+                .publish(TaskKillLifecycleEvent.of(taskExecutionRunnable, TimeUnit.SECONDS.toSeconds(5)));
     }
 
     @Override
diff --git a/dolphinscheduler-task-executor/src/main/java/org/apache/dolphinscheduler/task/executor/log/TaskExecutorMDCUtils.java b/dolphinscheduler-task-executor/src/main/java/org/apache/dolphinscheduler/task/executor/log/TaskExecutorMDCUtils.java
@@ -17,7 +17,6 @@
 
 package org.apache.dolphinscheduler.task.executor.log;
 
-import org.apache.dolphinscheduler.common.constants.Constants;
 import org.apache.dolphinscheduler.task.executor.ITaskExecutor;
 
 import org.slf4j.MDC;
@@ -41,7 +40,7 @@ public static MDCAutoClosable logWithMDC(final int taskInstanceId, final String
             MDC.put(TASK_INSTANCE_LOG_FULL_PATH_MDC_KEY, logPath);
         }
 
-        MDC.put(Constants.TASK_INSTANCE_ID_MDC_KEY, String.valueOf(taskInstanceId));
+        MDC.put(TASK_INSTANCE_ID_MDC_KEY, String.valueOf(taskInstanceId));
 
         return () -> {
             MDC.remove(TASK_INSTANCE_LOG_FULL_PATH_MDC_KEY);

Original file line number	Diff line number	Diff line change
`@@ -18,6 +18,7 @@`
`18`	`18`	`package org.apache.dolphinscheduler.server.master.engine.task.dispatcher;`
`19`	`19`
`20`	`20`	`import org.apache.dolphinscheduler.common.thread.BaseDaemonThread;`
	`21`	`+import org.apache.dolphinscheduler.plugin.task.api.utils.LogUtils;`
`21`	`22`	`import org.apache.dolphinscheduler.server.master.engine.task.client.ITaskExecutorClient;`
`22`	`23`	`import org.apache.dolphinscheduler.server.master.engine.task.runnable.ITaskExecutionRunnable;`
`23`	`24`	`import org.apache.dolphinscheduler.server.master.runner.queue.PriorityAndDelayBasedTaskEntry;`
`@@ -79,7 +80,10 @@ public void run() {`
`79`	`80`	`try (`
`80`	`81`	`TaskExecutorMDCUtils.MDCAutoClosable ignore =`
`81`	`82`	`TaskExecutorMDCUtils.logWithMDC(taskExecutionRunnable.getId())) {`
	`83`	`+ LogUtils.setWorkflowInstanceIdMDC(taskExecutionRunnable.getTaskInstance().getWorkflowInstanceId());`
`82`	`84`	`doDispatchTask(taskExecutionRunnable);`
	`85`	`+ } finally {`
	`86`	`+ LogUtils.removeWorkflowInstanceIdMDC();`
`83`	`87`	`}`
`84`	`88`	`}`
`85`	`89`	`}`