[Improvement-17795][Master] Add dispatch timeout checking logic to handle cases where the worker group does not exist or no workers are available

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

@@ -74,6 +74,12 @@ public class MasterConfig implements Validator {
      */
     private String masterRegistryPath;
 
+    /**
+     * Maximum time allowed for a task to be successfully dispatched.
+     * Default: 5 minutes.
+     */
+    private Duration dispatchTimeout = Duration.ofMinutes(5);
+
     @Override
     public boolean supports(Class<?> clazz) {
         return MasterConfig.class.isAssignableFrom(clazz);

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

@@ -24,10 +24,13 @@
 import org.apache.dolphinscheduler.extract.base.utils.Host;
 import org.apache.dolphinscheduler.extract.worker.IPhysicalTaskExecutorOperator;
 import org.apache.dolphinscheduler.plugin.task.api.TaskExecutionContext;
+import org.apache.dolphinscheduler.server.master.cluster.ClusterManager;
 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.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;
@@ -55,18 +58,29 @@ public class PhysicalTaskExecutorClientDelegator implements ITaskExecutorClientD
     @Autowired
     private IWorkerLoadBalancer workerLoadBalancer;
 
+    @Autowired
+    private ClusterManager clusterManager;
+
     @Override
     public void dispatch(final ITaskExecutionRunnable taskExecutionRunnable) throws TaskDispatchException {
         final TaskExecutionContext taskExecutionContext = taskExecutionRunnable.getTaskExecutionContext();
         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));
+        }
         final String physicalTaskExecutorAddress = workerLoadBalancer
-                .select(taskExecutionContext.getWorkerGroup())
+                .select(workerGroup)
                 .map(Host::of)
                 .map(Host::getAddress)
-                .orElseThrow(() -> new TaskDispatchException(
-                        String.format("Cannot find the host to dispatch Task[id=%s, name=%s, workerGroup=%s]",
+                .orElseThrow(() -> new WorkerNotFoundException(
+                        String.format(
+                                "Cannot find available worker host to dispatch Task[id=%s, name=%s, workerGroup=%s]",
                                 taskExecutionContext.getTaskInstanceId(), taskName,
-                                taskExecutionContext.getWorkerGroup())));
+                                workerGroup)));
 
         taskExecutionContext.setHost(physicalTaskExecutorAddress);
         taskExecutionRunnable.getTaskInstance().setHost(physicalTaskExecutorAddress);

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

@@ -18,12 +18,19 @@
 package org.apache.dolphinscheduler.server.master.engine.task.dispatcher;
 
 import org.apache.dolphinscheduler.common.thread.BaseDaemonThread;
+import org.apache.dolphinscheduler.plugin.task.api.TaskExecutionContext;
 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.dispatcher.event.TaskDispatchableEvent;
+import org.apache.dolphinscheduler.server.master.engine.task.lifecycle.event.TaskFailedLifecycleEvent;
+import org.apache.dolphinscheduler.server.master.engine.task.lifecycle.event.TaskFatalLifecycleEvent;
 import org.apache.dolphinscheduler.server.master.engine.task.runnable.ITaskExecutionRunnable;
+import org.apache.dolphinscheduler.server.master.exception.dispatch.TaskDispatchException;
+import org.apache.dolphinscheduler.server.master.utils.ExceptionUtils;
 import org.apache.dolphinscheduler.task.executor.log.TaskExecutorMDCUtils;
 
+import java.time.Duration;
+import java.util.Date;
 import java.util.Set;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.atomic.AtomicBoolean;
@@ -48,11 +55,15 @@ public class WorkerGroupDispatcher extends BaseDaemonThread {
 
     private final AtomicBoolean runningFlag = new AtomicBoolean(false);
 
-    public WorkerGroupDispatcher(String workerGroupName, ITaskExecutorClient taskExecutorClient) {
+    private final Duration dispatchTimeout;
+
+    public WorkerGroupDispatcher(String workerGroupName, ITaskExecutorClient taskExecutorClient,
+                                 Duration dispatchTimeout) {
         super("WorkerGroupTaskDispatcher-" + workerGroupName);
         this.taskExecutorClient = taskExecutorClient;
         this.workerGroupEventBus = new TaskDispatchableEventBus<>();
         this.waitingDispatchTaskIds = ConcurrentHashMap.newKeySet();
+        this.dispatchTimeout = dispatchTimeout;
         log.info("Initialize WorkerGroupDispatcher: {}", this.getName());
     }
 
@@ -84,23 +95,77 @@ public void run() {
     }
 
     private void doDispatchTask(ITaskExecutionRunnable taskExecutionRunnable) {
+        final int taskId = taskExecutionRunnable.getId();
+        final TaskExecutionContext taskExecutionContext = taskExecutionRunnable.getTaskExecutionContext();
+        final long timeoutMs = this.dispatchTimeout.toMillis();
         try {
-            if (!waitingDispatchTaskIds.remove(taskExecutionRunnable.getId())) {
+            if (!waitingDispatchTaskIds.remove(taskId)) {
                 log.info(
                         "The task: {} doesn't exist in waitingDispatchTaskIds(it might be paused or killed), will skip dispatch",
-                        taskExecutionRunnable.getId());
+                        taskId);
                 return;
             }
             taskExecutorClient.dispatch(taskExecutionRunnable);
-        } catch (Exception e) {
+        } catch (TaskDispatchException ex) {
+            // Checks whether the given task has exceeded its allowed dispatch timeout.
+            long elapsed = System.currentTimeMillis() - taskExecutionContext.getFirstDispatchEnqueueTimeMs();
+            if (elapsed > timeoutMs) {
+                handleDispatchFailure(taskExecutionRunnable, ex, elapsed, timeoutMs);
+                return;
+            }
+
             // If dispatch failed, will put the task back to the queue
             // The task will be dispatched after waiting time.
             // the waiting time will increase multiple of times, but will not exceed 60 seconds
-            long waitingTimeMills = Math.min(
+            long waitingTimeMillis = Math.min(
                     taskExecutionRunnable.getTaskExecutionContext().increaseDispatchFailTimes() * 1_000L, 60_000L);
-            dispatchTask(taskExecutionRunnable, waitingTimeMills);
-            log.error("Dispatch Task: {} failed will retry after: {}/ms", taskExecutionRunnable.getId(),
-                    waitingTimeMills, e);
+            dispatchTask(taskExecutionRunnable, waitingTimeMillis);
+            log.warn("Dispatch Task: {} failed will retry after: {}/ms", taskId,
+                    waitingTimeMillis, ex);
+        }
+    }
+
+    /**
+     * Marks the specified task as fatally failed due to an unrecoverable dispatch error,such as timeout or persistent client failure.
+     * Once this method is called, the task is considered permanently failed and will not be retried.
+     *
+     * @param taskExecutionRunnable the task to mark as fatally failed; must not be null
+     * @param exception             the dispatch exception that triggered this failure handling; must not be null
+     * @param elapsed               the time (in milliseconds) already spent attempting to dispatch the task
+     * @param timeoutMs             the configured dispatch timeout threshold (in milliseconds)
+     */
+    private void handleDispatchFailure(ITaskExecutionRunnable taskExecutionRunnable, TaskDispatchException exception,
+                                       long elapsed, long timeoutMs) {
+        int taskId = taskExecutionRunnable.getId();
+        int workflowId = taskExecutionRunnable.getWorkflowInstance().getId();
+
+        log.warn("[DISPATCH_FAILED] taskId: {}, workflowId: {}, timed out after {} ms (limit: {} ms))", taskId,
+                workflowId, elapsed, timeoutMs);
+
+        if (ExceptionUtils.isWorkerGroupNotFoundException(exception)) {
+            log.error("[DISPATCH_FAILED] Worker group not found. taskId: {}, workflowId: {}", taskId, workflowId,
+                    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);
+            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);
+            final TaskFailedLifecycleEvent taskFailedEvent = TaskFailedLifecycleEvent.builder()
+                    .taskExecutionRunnable(taskExecutionRunnable)
+                    .endTime(new Date())
+                    .build();
+            taskExecutionRunnable.getWorkflowEventBus().publish(taskFailedEvent);
         }
     }
 

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

@@ -17,6 +17,7 @@
 
 package org.apache.dolphinscheduler.server.master.engine.task.dispatcher;
 
+import org.apache.dolphinscheduler.server.master.config.MasterConfig;
 import org.apache.dolphinscheduler.server.master.engine.task.client.ITaskExecutorClient;
 import org.apache.dolphinscheduler.server.master.engine.task.runnable.ITaskExecutionRunnable;
 
@@ -41,8 +42,11 @@ public class WorkerGroupDispatcherCoordinator implements AutoCloseable {
 
     private final ConcurrentHashMap<String, WorkerGroupDispatcher> workerGroupDispatcherMap;
 
-    public WorkerGroupDispatcherCoordinator() {
+    private final MasterConfig masterConfig;
+
+    public WorkerGroupDispatcherCoordinator(final MasterConfig masterConfig) {
         workerGroupDispatcherMap = new ConcurrentHashMap<>();
+        this.masterConfig = masterConfig;
     }
 
     public void start() {
@@ -99,7 +103,8 @@ public void close() throws Exception {
 
     private WorkerGroupDispatcher getOrCreateWorkerGroupDispatcher(String workerGroup) {
         return workerGroupDispatcherMap.computeIfAbsent(workerGroup, wg -> {
-            WorkerGroupDispatcher workerGroupDispatcher = new WorkerGroupDispatcher(wg, taskExecutorClient);
+            WorkerGroupDispatcher workerGroupDispatcher =
+                    new WorkerGroupDispatcher(wg, taskExecutorClient, masterConfig.getDispatchTimeout());
             workerGroupDispatcher.start();
             return workerGroupDispatcher;
         });

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 workerGroup) {
-        super("Cannot find worker group: " + workerGroup);
+    public WorkerGroupNotFoundException(String message) {
+        super(message);
     }
 }

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

@@ -0,0 +1,25 @@
+/*
+ * 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.exception.dispatch;
+
+public class WorkerNotFoundException extends TaskDispatchException {
+
+    public WorkerNotFoundException(String message) {
+        super(message);
+    }
+}

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

@@ -17,6 +17,9 @@
 
 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;
 
 public class ExceptionUtils {
@@ -25,4 +28,11 @@ public static boolean isDatabaseConnectedFailedException(Throwable e) {
         return e instanceof DataAccessResourceFailureException;
     }
 
+    public static boolean isWorkerGroupNotFoundException(Throwable e) {
+        return e instanceof WorkerGroupNotFoundException;
+    }
+
+    public static boolean isWorkerNotFoundException(Throwable e) {
+        return e instanceof WorkerNotFoundException;
+    }
 }

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

@@ -111,6 +111,9 @@ master:
     # Master max concurrent workflow instances, when the master's workflow instance count exceeds this value, master server will be marked as busy.
     max-concurrent-workflow-instances: 2147483647
   worker-group-refresh-interval: 5m
+  # Timeout for task dispatching. If a task cannot be successfully dispatched to any worker within this duration,
+  # it will be marked as failed. Prevents tasks from being stuck indefinitely in the dispatch queue.
+  dispatch-timeout: 5m
   command-fetch-strategy:
     type: ID_SLOT_BASED
     config:

dolphinscheduler-master/src/test/java/org/apache/dolphinscheduler/server/master/engine/task/dispatcher/WorkerGroupDispatcherTest.java

@@ -28,6 +28,7 @@
 
 import org.apache.dolphinscheduler.dao.entity.TaskInstance;
 import org.apache.dolphinscheduler.plugin.task.api.TaskExecutionContext;
+import org.apache.dolphinscheduler.server.master.config.MasterConfig;
 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.exception.dispatch.TaskDispatchException;
@@ -46,7 +47,8 @@ class WorkerGroupDispatcherTest {
     @BeforeEach
     void setUp() {
         taskExecutorClient = mock(ITaskExecutorClient.class);
-        dispatcher = new WorkerGroupDispatcher("TestGroup", taskExecutorClient);
+        final MasterConfig masterConfig = new MasterConfig();
+        dispatcher = new WorkerGroupDispatcher("TestGroup", taskExecutorClient, masterConfig.getDispatchTimeout());
     }
 
     @Test

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

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