add dispatch-timeout-checker switch

Author: 苏义超

dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/config/MasterConfig.java

@@ -108,11 +108,11 @@ public void validate(Object target, Errors errors) {
         // Validate dispatch timeout checker config
         MasterDispatchTimeoutCheckerConfig timeoutChecker = masterConfig.getDispatchTimeoutChecker();
         if (timeoutChecker != null && timeoutChecker.isEnabled()) {
-            if (timeoutChecker.getTimeoutDuration() == null) {
-                errors.rejectValue("master-dispatch-timeout-checker.timeout-duration", null,
+            if (timeoutChecker.getMaxTaskDispatchDuration() == null) {
+                errors.rejectValue("dispatch-timeout-checker.max-task-dispatch-duration", null,
                         "must be specified when dispatch timeout checker is enabled");
-            } else if (timeoutChecker.getTimeoutDuration().toMillis() <= 0) {
-                errors.rejectValue("master-dispatch-timeout-checker.timeout-duration", null,
+            } else if (timeoutChecker.getMaxTaskDispatchDuration().toMillis() <= 0) {
+                errors.rejectValue("dispatch-timeout-checker.max-task-dispatch-duration", null,
                         "must be a positive duration (e.g., '2m', '5m', '30m')");
             }
         }
@@ -142,6 +142,7 @@ private void printConfig() {
                         "\n  command-fetch-strategy: " + commandFetchStrategy +
                         "\n  worker-load-balancer-configuration-properties: "
                         + workerLoadBalancerConfigurationProperties +
+                        "\n dispatchTimeoutChecker: " + dispatchTimeoutChecker +
                         "\n****************************Master Configuration**************************************";
         log.info(config);
     }

dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/config/MasterDispatchTimeoutCheckerConfig.java

@@ -23,7 +23,7 @@
 
 /**
  * Configuration for the master's task dispatch timeout checker.
- * If enabled, tasks that remain in the dispatch queue longer than {@link #timeoutDuration} will be marked as failed to prevent indefinite queuing.
+ * If enabled, tasks that remain in the dispatch queue longer than {@link #maxTaskDispatchDuration} will be marked as failed to prevent indefinite queuing.
  */
 @Data
 public class MasterDispatchTimeoutCheckerConfig {
@@ -38,5 +38,5 @@ public class MasterDispatchTimeoutCheckerConfig {
      * Tasks exceeding this duration in the dispatch queue will be failed.
      * Examples: "2m", "5m", "30m". Defaults to 5 minutes.
      */
-    private Duration timeoutDuration = Duration.ofMinutes(5);
+    private Duration maxTaskDispatchDuration = Duration.ofMinutes(5);
 }

dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/task/client/PhysicalTaskExecutorClientDelegator.java

@@ -28,9 +28,9 @@
 import org.apache.dolphinscheduler.server.master.cluster.loadbalancer.IWorkerLoadBalancer;
 import org.apache.dolphinscheduler.server.master.config.MasterConfig;
 import org.apache.dolphinscheduler.server.master.engine.task.runnable.ITaskExecutionRunnable;
+import org.apache.dolphinscheduler.server.master.exception.dispatch.NoAvailableWorkerException;
 import org.apache.dolphinscheduler.server.master.exception.dispatch.TaskDispatchException;
 import org.apache.dolphinscheduler.server.master.exception.dispatch.WorkerGroupNotFoundException;
-import org.apache.dolphinscheduler.server.master.exception.dispatch.WorkerNotFoundException;
 import org.apache.dolphinscheduler.task.executor.eventbus.ITaskExecutorLifecycleEventReporter;
 import org.apache.dolphinscheduler.task.executor.operations.TaskExecutorDispatchRequest;
 import org.apache.dolphinscheduler.task.executor.operations.TaskExecutorDispatchResponse;
@@ -67,20 +67,13 @@ public void dispatch(final ITaskExecutionRunnable taskExecutionRunnable) throws
         final String taskName = taskExecutionContext.getTaskName();
         final String workerGroup = taskExecutionContext.getWorkerGroup();
         if (!clusterManager.getWorkerClusters().containsWorkerGroup(workerGroup)) {
-            throw new WorkerGroupNotFoundException(
-                    String.format("Cannot find worker group to dispatch Task[id=%s, name=%s, workerGroup=%s]",
-                            taskExecutionContext.getTaskInstanceId(), taskName,
-                            workerGroup));
+            throw new WorkerGroupNotFoundException(workerGroup);
         }
         final String physicalTaskExecutorAddress = workerLoadBalancer
                 .select(workerGroup)
                 .map(Host::of)
                 .map(Host::getAddress)
-                .orElseThrow(() -> new WorkerNotFoundException(
-                        String.format(
-                                "Cannot find available worker host to dispatch Task[id=%s, name=%s, workerGroup=%s]",
-                                taskExecutionContext.getTaskInstanceId(), taskName,
-                                workerGroup)));
+                .orElseThrow(() -> new NoAvailableWorkerException(workerGroup));
 
         taskExecutionContext.setHost(physicalTaskExecutorAddress);
         taskExecutionRunnable.getTaskInstance().setHost(physicalTaskExecutorAddress);

dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/task/dispatcher/WorkerGroupDispatcher.java

@@ -95,21 +95,21 @@ public void run() {
     }
 
     private void doDispatchTask(ITaskExecutionRunnable taskExecutionRunnable) {
-        final int taskId = taskExecutionRunnable.getId();
+        final int taskInstanceId = taskExecutionRunnable.getId();
         final TaskExecutionContext taskExecutionContext = taskExecutionRunnable.getTaskExecutionContext();
         try {
-            if (!waitingDispatchTaskIds.remove(taskId)) {
+            if (!waitingDispatchTaskIds.remove(taskInstanceId)) {
                 log.info(
                         "The task: {} doesn't exist in waitingDispatchTaskIds(it might be paused or killed), will skip dispatch",
-                        taskId);
+                        taskInstanceId);
                 return;
             }
             taskExecutorClient.dispatch(taskExecutionRunnable);
         } catch (TaskDispatchException ex) {
             if (dispatchTimeoutChecker.isEnabled()) {
                 // Checks whether the given task has exceeded its allowed dispatch timeout.
-                long timeoutMs = this.dispatchTimeoutChecker.getTimeoutDuration().toMillis();
-                long elapsed = System.currentTimeMillis() - taskExecutionContext.getFirstDispatchEnqueueTimeMs();
+                long timeoutMs = this.dispatchTimeoutChecker.getMaxTaskDispatchDuration().toMillis();
+                long elapsed = System.currentTimeMillis() - taskExecutionContext.getFirstDispatchTime();
                 if (elapsed > timeoutMs) {
                     handleDispatchFailure(taskExecutionRunnable, ex, elapsed, timeoutMs);
                     return;
@@ -122,7 +122,7 @@ private void doDispatchTask(ITaskExecutionRunnable taskExecutionRunnable) {
             long waitingTimeMillis = Math.min(
                     taskExecutionRunnable.getTaskExecutionContext().increaseDispatchFailTimes() * 1_000L, 60_000L);
             dispatchTask(taskExecutionRunnable, waitingTimeMillis);
-            log.warn("Dispatch Task: {} failed will retry after: {}/ms", taskId,
+            log.warn("Dispatch Task: {} failed will retry after: {}/ms", taskInstanceId,
                     waitingTimeMillis, ex);
         }
     }
@@ -138,31 +138,26 @@ private void doDispatchTask(ITaskExecutionRunnable taskExecutionRunnable) {
      */
     private void handleDispatchFailure(ITaskExecutionRunnable taskExecutionRunnable, TaskDispatchException exception,
                                        long elapsed, long timeoutMs) {
-        int taskId = taskExecutionRunnable.getId();
-        int workflowId = taskExecutionRunnable.getWorkflowInstance().getId();
+        final String taskName = taskExecutionRunnable.getName();
 
-        log.warn("[DISPATCH_FAILED] taskId: {}, workflowId: {}, timed out after {} ms (limit: {} ms))", taskId,
-                workflowId, elapsed, timeoutMs);
+        log.warn("[DISPATCH_FAILED] taskName: {}, timed out after {} ms (limit: {} ms))", taskName, elapsed, timeoutMs);
 
         if (ExceptionUtils.isWorkerGroupNotFoundException(exception)) {
-            log.error("[DISPATCH_FAILED] Worker group not found. taskId: {}, workflowId: {}", taskId, workflowId,
-                    exception);
+            log.error("[DISPATCH_FAILED] taskName: {}, Worker group not found.", taskName, exception);
             final TaskFatalLifecycleEvent taskFatalEvent = TaskFatalLifecycleEvent.builder()
                     .taskExecutionRunnable(taskExecutionRunnable)
                     .endTime(new Date())
                     .build();
             taskExecutionRunnable.getWorkflowEventBus().publish(taskFatalEvent);
         } else if (ExceptionUtils.isWorkerNotFoundException(exception)) {
-            log.error("[DISPATCH_FAILED] No available workers. taskId: {}, workflowId: {}", taskId, workflowId,
-                    exception);
+            log.error("[DISPATCH_FAILED] taskName: {}, No available worker.", taskName, exception);
             final TaskFailedLifecycleEvent taskFailedEvent = TaskFailedLifecycleEvent.builder()
                     .taskExecutionRunnable(taskExecutionRunnable)
                     .endTime(new Date())
                     .build();
             taskExecutionRunnable.getWorkflowEventBus().publish(taskFailedEvent);
         } else {
-            log.error("[DISPATCH_FAILED] Unexpected dispatch error. taskId: {}, workflowId: {}", taskId, workflowId,
-                    exception);
+            log.error("[DISPATCH_FAILED] taskName: {}, Unexpected dispatch error.", taskName, exception);
             final TaskFailedLifecycleEvent taskFailedEvent = TaskFailedLifecycleEvent.builder()
                     .taskExecutionRunnable(taskExecutionRunnable)
                     .endTime(new Date())

…on/dispatch/WorkerNotFoundException.java …dispatch/NoAvailableWorkerException.javadolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/exception/dispatch/WorkerNotFoundException.java renamed to dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/exception/dispatch/NoAvailableWorkerException.java dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/exception/dispatch/WorkerNotFoundException.java renamed to dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/exception/dispatch/NoAvailableWorkerException.java

@@ -17,9 +17,9 @@
 
 package org.apache.dolphinscheduler.server.master.exception.dispatch;
 
-public class WorkerNotFoundException extends TaskDispatchException {
+public class NoAvailableWorkerException extends TaskDispatchException {
 
-    public WorkerNotFoundException(String message) {
-        super(message);
+    public NoAvailableWorkerException(String workerGroup) {
+        super("Cannot find available worker under worker group: " + workerGroup);
     }
 }

dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/exception/dispatch/WorkerGroupNotFoundException.java

@@ -19,7 +19,7 @@
 
 public class WorkerGroupNotFoundException extends TaskDispatchException {
 
-    public WorkerGroupNotFoundException(String message) {
-        super(message);
+    public WorkerGroupNotFoundException(String workerGroup) {
+        super("Cannot find worker group: " + workerGroup);
     }
 }

dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/utils/ExceptionUtils.java

@@ -18,7 +18,6 @@
 package org.apache.dolphinscheduler.server.master.utils;
 
 import org.apache.dolphinscheduler.server.master.exception.dispatch.WorkerGroupNotFoundException;
-import org.apache.dolphinscheduler.server.master.exception.dispatch.WorkerNotFoundException;
 
 import org.springframework.dao.DataAccessResourceFailureException;
 

dolphinscheduler-master/src/main/resources/application.yaml

@@ -115,7 +115,7 @@ master:
   # When enabled, tasks not dispatched within this duration are marked as failed.
   dispatch-timeout-checker:
     enabled: false
-    timeout-duration: 5m
+    max-task-dispatch-duration: 5m
   command-fetch-strategy:
     type: ID_SLOT_BASED
     config:

dolphinscheduler-master/src/test/resources/application.yaml

@@ -73,6 +73,11 @@ master:
       cpu-usage-weight: 30
       task-thread-pool-usage-weight: 30
   worker-group-refresh-interval: 5m
+  # Task dispatch timeout check (currently disabled).
+  # When enabled, tasks not dispatched within this duration are marked as failed.
+  dispatch-timeout-checker:
+    enabled: false
+    max-task-dispatch-duration: 5m
   command-fetch-strategy:
     type: ID_SLOT_BASED
     config:

dolphinscheduler-task-plugin/dolphinscheduler-task-api/src/main/java/org/apache/dolphinscheduler/plugin/task/api/TaskExecutionContext.java

@@ -127,9 +127,9 @@ public class TaskExecutionContext implements Serializable {
     private boolean failover;
 
     /**
-     * Timestamp (ms) when the task was first enqueued for dispatch.
+     * Timestamp (ms) when the task was first dispatched.
      */
-    private final long firstDispatchEnqueueTimeMs = System.currentTimeMillis();
+    private final long firstDispatchTime = System.currentTimeMillis();
 
     public int increaseDispatchFailTimes() {
         return ++dispatchFailTimes;