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

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open

njnu-seafish wants to merge 48 commits into apache:dev from njnu-seafish:Improvement-17795

+728 −18

docs/docs/en/architecture/configuration.md

-Original file line number
+Diff line change
@@ Expand Up / @@ -291,6 +291,8 @@ Location: `master-server/conf/application.yaml` @@
     | master.command-fetch-strategy.type                                          | ID_SLOT_BASED                | The command fetch strategy, only support `ID_SLOT_BASED`                                                                                                 |
     | master.command-fetch-strategy.config.id-step                                | 1                            | The id auto incremental step of t_ds_command in db                                                                                                       |
     | master.command-fetch-strategy.config.fetch-size                             | 10                           | The number of commands fetched by master                                                                                                                 |
+    | master.task-dispatch-policy.dispatch-timeout-enabled                        | false                        | Indicates whether the dispatch timeout checking mechanism is enabled                                                                                     |
+    | master.task-dispatch-policy.max-task-dispatch-duration                      | 1h                           | The maximum allowed duration a task may wait in the dispatch queue before being assigned to a worker                                                     |
     ### Worker Server related configuration
@@ Expand Down @@

docs/docs/zh/architecture/configuration.md

-Original file line number
+Diff line change
@@ Expand Up @@
     | master.command-fetch-strategy.type                                          | ID_SLOT_BASED                | Command拉取策略, 目前仅支持 `ID_SLOT_BASED`                                                      |
     | master.command-fetch-strategy.config.id-step                                | 1                            | 数据库中t_ds_command的id自增步长                                                                 |
     | master.command-fetch-strategy.config.fetch-size                             | 10                           | master拉取command数量                                                                       |
+    | master.task-dispatch-policy.dispatch-timeout-enabled                        | false                        | 是否开启master分派超时检测功能                                                                      |
+    | master.task-dispatch-policy.max-task-dispatch-duration                      | 1h                           | master分派检测的超时时长，默认为一小时                                                                  |
     ## Worker Server相关配置
@@ Expand Down @@

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

-Original file line number
+Diff line change
@@ Expand Up / @@ -74,6 +74,8 @@ public class MasterConfig implements Validator { @@
          */
         private String masterRegistryPath;
+        private TaskDispatchPolicy taskDispatchPolicy = new TaskDispatchPolicy();
         @Override
         public boolean supports(Class<?> clazz) {
             return MasterConfig.class.isAssignableFrom(clazz);
@@ Expand All / @@ -97,6 +99,18 @@ public void validate(Object target, Errors errors) { @@
             if (masterConfig.getWorkerGroupRefreshInterval().getSeconds() < 10) {
                 errors.rejectValue("worker-group-refresh-interval", null, "should >= 10s");
             }
+            TaskDispatchPolicy dispatchPolicy = masterConfig.getTaskDispatchPolicy();
+            if (dispatchPolicy.isDispatchTimeoutEnabled()) {
+                if (dispatchPolicy.getMaxTaskDispatchDuration() == null) {
+                    errors.rejectValue("dispatch-timeout-checker.max-task-dispatch-duration", null,
+                            "must be specified when dispatch timeout checker is enabled");
+                } else if (dispatchPolicy.getMaxTaskDispatchDuration().toMillis() <= 0) {
+                    errors.rejectValue("dispatch-timeout-checker.max-task-dispatch-duration", null,
+                            "must be a positive duration (e.g., '10m', '30m', '1h')");
+                }
+            }
             if (StringUtils.isEmpty(masterConfig.getMasterAddress())) {
                 masterConfig.setMasterAddress(NetUtils.getAddr(masterConfig.getListenPort()));
             }
@@ Expand All / @@ -122,6 +136,7 @@ private void printConfig() { @@
                             "\n  command-fetch-strategy: " + commandFetchStrategy +
                             "\n  worker-load-balancer-configuration-properties: "
                             + workerLoadBalancerConfigurationProperties +
+                            "\n  taskDispatchPolicy: " + taskDispatchPolicy +
                             "\n****************************Master Configuration**************************************";
             log.info(config);
         }
@@ Expand Down @@

...er/src/main/java/org/apache/dolphinscheduler/server/master/config/TaskDispatchPolicy.java

-Original file line number
+Diff line change
@@ -0,0 +1,43 @@
+    /*
+     * 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.config;
+    import java.time.Duration;
+    import lombok.Data;
+    /**
+     * Configuration for the master's task dispatch policy.
+     * When enabled, tasks that remain in the dispatch queue longer than
+     * {@link #maxTaskDispatchDuration} will be marked as failed to prevent indefinite queuing.
+     */
+    @Data
+    public class TaskDispatchPolicy {
+        /**
+         * Indicates whether the dispatch timeout checking mechanism is enabled.
+         */
+        private boolean dispatchTimeoutEnabled = false;
+        /**
+         * The maximum allowed duration a task may wait in the dispatch queue before being assigned to a worker.
+         * Tasks that exceed this duration will be marked as failed.
+         * Examples: {@code "10m"}, {@code "30m"}, {@code "1h"}.
+         */
+        private Duration maxTaskDispatchDuration;
+    }

...olphinscheduler/server/master/engine/task/client/PhysicalTaskExecutorClientDelegator.java

            
                      Original file line number
                      Diff line number
                      Diff line change
                  
    @@ -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.NoAvailableWorkerException;

    import org.apache.dolphinscheduler.server.master.exception.dispatch.TaskDispatchException;

    import org.apache.dolphinscheduler.server.master.exception.dispatch.WorkerGroupNotFoundException;

    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,26 @@ 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();

            // workerGroup not exist

            if (!clusterManager.getWorkerClusters().containsWorkerGroup(workerGroup)) {

                throw new WorkerGroupNotFoundException(workerGroup);

            }

            // select an available worker from the worker group; throws NoAvailableWorkerException if none is available.

            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]",

                                    taskExecutionContext.getTaskInstanceId(), taskName,

                                    taskExecutionContext.getWorkerGroup())));

                    .orElseThrow(() -> new NoAvailableWorkerException(workerGroup));

            taskExecutionContext.setHost(physicalTaskExecutorAddress);

            taskExecutionRunnable.getTaskInstance().setHost(physicalTaskExecutorAddress);

...g/apache/dolphinscheduler/server/master/engine/task/dispatcher/WorkerGroupDispatcher.java

            
                      Original file line number
                      Diff line number
                      Diff line change
                  
    @@ -18,12 +18,16 @@
  
    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.config.TaskDispatchPolicy;

    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.runnable.ITaskExecutionRunnable;

    import org.apache.dolphinscheduler.task.executor.log.TaskExecutorMDCUtils;

    import java.util.Date;

    import java.util.Set;

    import java.util.concurrent.ConcurrentHashMap;

    import java.util.concurrent.atomic.AtomicBoolean;

    @@ -48,11 +52,22 @@ public class WorkerGroupDispatcher extends BaseDaemonThread {
  
        private final AtomicBoolean runningFlag = new AtomicBoolean(false);

        public WorkerGroupDispatcher(String workerGroupName, ITaskExecutorClient taskExecutorClient) {

        private final TaskDispatchPolicy taskDispatchPolicy;

        private final long maxTaskDispatchMillis;

        public WorkerGroupDispatcher(String workerGroupName, ITaskExecutorClient taskExecutorClient,

                                     TaskDispatchPolicy taskDispatchPolicy) {

            super("WorkerGroupTaskDispatcher-" + workerGroupName);

            this.taskExecutorClient = taskExecutorClient;

            this.workerGroupEventBus = new TaskDispatchableEventBus<>();

            this.waitingDispatchTaskIds = ConcurrentHashMap.newKeySet();

            this.taskDispatchPolicy = taskDispatchPolicy;

            if (taskDispatchPolicy.isDispatchTimeoutEnabled()) {

                this.maxTaskDispatchMillis = taskDispatchPolicy.getMaxTaskDispatchDuration().toMillis();

            } else {

                this.maxTaskDispatchMillis = 0L;

            }

            log.info("Initialize WorkerGroupDispatcher: {}", this.getName());

        }

    @@ -84,26 +99,54 @@ public void run() {
  
        }

        private void doDispatchTask(ITaskExecutionRunnable taskExecutionRunnable) {

            final int taskInstanceId = taskExecutionRunnable.getId();

            final TaskExecutionContext taskExecutionContext = taskExecutionRunnable.getTaskExecutionContext();

            try {

                if (!waitingDispatchTaskIds.remove(taskExecutionRunnable.getId())) {

                if (!waitingDispatchTaskIds.remove(taskInstanceId)) {

                    log.info(

                            "The task: {} doesn't exist in waitingDispatchTaskIds(it might be paused or killed), will skip dispatch",

                            taskExecutionRunnable.getId());

                            taskInstanceId);

                    return;

                }

                taskExecutorClient.dispatch(taskExecutionRunnable);

            } catch (Exception e) {

            } catch (Exception ex) {

                if (taskDispatchPolicy.isDispatchTimeoutEnabled()) {

                    // If a dispatch timeout occurs, the task will not be put back into the queue.

                    long elapsed = System.currentTimeMillis() - taskExecutionContext.getFirstDispatchTime();

                    if (elapsed > maxTaskDispatchMillis) {

                        onDispatchTimeout(taskExecutionRunnable, ex, elapsed, maxTaskDispatchMillis);

                        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", taskInstanceId,

                        waitingTimeMillis, ex);

            }

        }

        /**

         * Marks a task as permanently failed due to dispatch timeout.

         * Once called, the task is considered permanently failed and will not be retried.

         */

        private void onDispatchTimeout(ITaskExecutionRunnable taskExecutionRunnable, Exception ex,

                                       long elapsed, long timeout) {

            String taskName = taskExecutionRunnable.getName();

            log.error("Task: {} dispatch timeout after {}ms (limit: {}ms)",

                    taskName, elapsed, timeout, ex);

            final TaskFailedLifecycleEvent taskFailedEvent = TaskFailedLifecycleEvent.builder()

                    .taskExecutionRunnable(taskExecutionRunnable)

                    .endTime(new Date())

                    .build();

            taskExecutionRunnable.getWorkflowEventBus().publish(taskFailedEvent);

        }

        /**

         * Adds a task to the worker group queue.

         * This method wraps the given task execution object into a priority and delay-based task entry and adds it to the worker group queue.

...lphinscheduler/server/master/engine/task/dispatcher/WorkerGroupDispatcherCoordinator.java

-Original file line number
+Diff line change
@@ Expand Up / @@ -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;
@@ Expand All @@
         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() {
@@ Expand Down Expand Up / @@ -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.getTaskDispatchPolicy());
                 workerGroupDispatcher.start();
                 return workerGroupDispatcher;
             });
@@ Expand Down @@

.../apache/dolphinscheduler/server/master/exception/dispatch/NoAvailableWorkerException.java

-Original file line number
+Diff line change
@@ -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 NoAvailableWorkerException extends TaskDispatchException {
+        public NoAvailableWorkerException(String workerGroup) {
+            super("Cannot find available worker under worker group: " + workerGroup);
+        }
+    }

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

-Original file line number
+Diff line change
@@ Expand Up / @@ -18,6 +18,8 @@ @@
     package org.apache.dolphinscheduler.server.master.utils;
     import org.apache.dolphinscheduler.server.master.exception.TaskExecutionContextCreateException;
+    import org.apache.dolphinscheduler.server.master.exception.dispatch.NoAvailableWorkerException;
+    import org.apache.dolphinscheduler.server.master.exception.dispatch.WorkerGroupNotFoundException;
     import org.springframework.dao.DataAccessResourceFailureException;
@@ Expand All @@
         public static boolean isTaskExecutionContextCreateException(Throwable e) {
             return e instanceof TaskExecutionContextCreateException;
         }
+        public static boolean isWorkerGroupNotFoundException(Throwable e) {
+            return e instanceof WorkerGroupNotFoundException;
+        }
+        public static boolean isNoAvailableWorkerException(Throwable e) {
+            return e instanceof NoAvailableWorkerException;
+        }
     }

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

-Original file line number
+Diff line change
@@ Expand Up / @@ -111,6 +111,11 @@ 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
+      # Task dispatch timeout check (currently disabled).
+      # When enabled, tasks not dispatched within this duration are marked as failed.
+      task-dispatch-policy:
+        dispatch-timeout-enabled: false
+        max-task-dispatch-duration: 1h
       command-fetch-strategy:
         type: ID_SLOT_BASED
         config:
@@ Expand Down @@

...ster/src/test/java/org/apache/dolphinscheduler/server/master/config/MasterConfigTest.java

-Original file line number
+Diff line change
@@ Expand Up / @@ -24,6 +24,8 @@ @@
     import org.apache.dolphinscheduler.server.master.cluster.loadbalancer.WorkerLoadBalancerConfigurationProperties;
     import org.apache.dolphinscheduler.server.master.cluster.loadbalancer.WorkerLoadBalancerType;
+    import java.time.Duration;
     import org.junit.jupiter.api.Test;
     import org.springframework.beans.factory.annotation.Autowired;
     import org.springframework.boot.test.autoconfigure.web.servlet.AutoConfigureMockMvc;
@@ Expand Down Expand Up / @@ -72,4 +74,13 @@ public void getWorkerLoadBalancerConfigurationProperties() { @@
             assertThat(dynamicWeightConfigProperties.getCpuUsageWeight()).isEqualTo(30);
             assertThat(dynamicWeightConfigProperties.getTaskThreadPoolUsageWeight()).isEqualTo(30);
         }
+        @Test
+        public void getTaskDispatchPolicy() {
+            TaskDispatchPolicy policy = masterConfig.getTaskDispatchPolicy();
+            assertThat(policy).isNotNull();
+            assertThat(policy.isDispatchTimeoutEnabled()).isFalse();
+            assertThat(policy.getMaxTaskDispatchDuration()).isEqualTo(Duration.ofHours(1));
+        }
     }

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

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

Diff view

Diff view

There are no files selected for viewing

ruanwenjun Jan 29, 2026

Uh oh!

njnu-seafish Jan 29, 2026

Uh oh!

Uh oh!

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

Are you sure you want to change the base?

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

Uh oh!

Uh oh!

Diff view

Diff view

There are no files selected for viewing

ruanwenjun Jan 29, 2026

Choose a reason for hiding this comment

Uh oh!

njnu-seafish Jan 29, 2026

Choose a reason for hiding this comment

Uh oh!

Uh oh!