[Improvement-17795][Master] Add dispatch timeout checking logic to handle cases where the worker group does not exist or no workers are available by njnu-seafish · Pull Request #17796 · apache/dolphinscheduler

njnu-seafish · 2025-12-15T12:01:50Z

Purpose of the pull request

close #17795

Brief change log

Add dispatch timeout checking logic to handle cases where the worker group does not exist or no workers are available

Verify this pull request

Successfully tested and verified in a real-world environment.

Pull Request Notice

If your pull request contains incompatible change, you should also add it to docs/docs/en/guide/upgrade/incompatible.md

SbloodyS · 2025-12-16T01:37:01Z

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

+     * Maximum time allowed for a task to be successfully dispatched.
+     * Default: 5 minutes.
+     */
+    private Duration dispatchTimeout = Duration.ofMinutes(5);


We should make it configurable by users in the configuration file and it can be turned off.

We should make it configurable by users in the configuration file and it can be turned off.

good idea!

ruanwenjun · 2025-12-16T12:07:08Z

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

+     * Maximum time allowed for a task to be successfully dispatched.
+     * Default: 5 minutes.
+     */
+    private Duration dispatchTimeout = Duration.ofMinutes(5);


Suggested change

private Duration dispatchTimeout = Duration.ofMinutes(5);

private Duration maxTaskDispatchDuration;

dispatchTimeout might confuse with single RPC timeout when dispatch task, and the default value should be null or a large duration, should compatibility with the previous behavior.

And you need to add this in document.

maxTaskDispatchDuration

ok

ruanwenjun · 2025-12-16T12:14:07Z

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

    private void doDispatchTask(ITaskExecutionRunnable taskExecutionRunnable) {
+        final int taskId = taskExecutionRunnable.getId();
+        final TaskExecutionContext taskExecutionContext = taskExecutionRunnable.getTaskExecutionContext();
+        final long timeoutMs = this.dispatchTimeout.toMillis();


It's better to store the mills rather than store duration, then you can avoid execute dispatchTimeout.toMillis() here.

It's better to store the mills rather than store duration, then you can avoid execute dispatchTimeout.toMillis() here.

Using a duration format like '5m' is more concise and readable. Most other configurations I’ve seen also avoid millisecond-level granularity.

You can translate the duration into milliseconds in the constructor.

ruanwenjun · 2025-12-16T12:14:25Z

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

    }

    private void doDispatchTask(ITaskExecutionRunnable taskExecutionRunnable) {
+        final int taskId = taskExecutionRunnable.getId();


Suggested change

final int taskId = taskExecutionRunnable.getId();

final int taskExecutionRunnableId = taskExecutionRunnable.getId();

Avoid confuse with task definition id.

taskExecutionRunnableId

final int taskInstanceId = taskExecutionRunnable.getId(); is it better?

/**
* Get the task instance id.
*
Need to know the id might change since the task instance might be regenerated.
*/
default int getId() {
return getTaskInstance().getId();
}

It's better to use taskExecutionRunnableId , since the outer layer is unaware that this is the taskInstanceId.

ruanwenjun · 2025-12-16T12:20:24Z

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

+     * @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,


Please don't print the taskId and workflowId here, all ids should already be added by MDC. We should only need to print the exception here, the exception already contains failure message.

Please don't print the taskId and workflowId here, all ids should already be added by MDC. We should only need to print the exception here, the exception already contains failure message.

ok

ruanwenjun · 2025-12-16T12:22:54Z

...g/apache/dolphinscheduler/server/master/exception/dispatch/WorkerGroupNotFoundException.java


-    public WorkerGroupNotFoundException(String workerGroup) {
-        super("Cannot find worker group: " + workerGroup);
+    public WorkerGroupNotFoundException(String message) {


Why do this change?

Why do this change?

restore, remove the task info

ruanwenjun · 2025-12-16T12:24:13Z

...va/org/apache/dolphinscheduler/server/master/exception/dispatch/WorkerNotFoundException.java

+
+public class WorkerNotFoundException extends TaskDispatchException {
+
+    public WorkerNotFoundException(String message) {


Suggested change

public WorkerNotFoundException(String message) {

public NoAvailableWorkerException(String workerGroup) {

super("Cannot find available worker under worker group: " + workerGroup);

}

ruanwenjun · 2025-12-16T12:25:29Z

.../apache/dolphinscheduler/server/master/engine/task/dispatcher/WorkerGroupDispatcherTest.java

+        final MasterConfig masterConfig = new MasterConfig();
+        dispatcher = new WorkerGroupDispatcher("TestGroup", taskExecutorClient, masterConfig.getDispatchTimeout());


You should add a validated test case.

You should add a validated test case.

ok, add test for dispatch timeout checker

You should add a validated test case.

I've already validated and tested it in a real environment.
The LogicFakeTask type runs on the Master and does not involve WorkerGroup dispatching.
I track the fix in a separate issue: #17872

ruanwenjun · 2025-12-16T12:28:06Z

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

+    /**
+     * Timestamp (ms) when the task was first enqueued for dispatch.
+     */
+    private final long firstDispatchEnqueueTimeMs = System.currentTimeMillis();


Suggested change

/**

* Timestamp (ms) when the task was first enqueued for dispatch.

*/

private final long firstDispatchEnqueueTimeMs = System.currentTimeMillis();

private final long firstDispatchTime = System.currentTimeMillis();

This is more like the creation time of the taskExecutionContext.

This is more like the creation time of the taskExecutionContext.

ok

njnu-seafish · 2026-01-15T09:37:22Z

@ruanwenjun @SbloodyS Whenever you have time, I’d be grateful for your review. Thanks so much!

ruanwenjun · 2026-01-22T07:04:01Z

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

+        TaskDispatchPolicy dispatchPolicy = masterConfig.getTaskDispatchPolicy();
+        if (dispatchPolicy != null && dispatchPolicy.isDispatchTimeoutFailedEnabled()) {
+            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., '2m', '5m', '30m')");
+            }
+        }


Can dispatchPolicy be null?

Can dispatchPolicy be null?
It indeed cannot be null.
private TaskDispatchPolicy taskDispatchPolicy = new TaskDispatchPolicy();

ruanwenjun · 2026-01-22T07:06:21Z

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

+        } catch (Exception ex) {
+            if (taskDispatchPolicy.isDispatchTimeoutFailedEnabled()) {
+                // If a dispatch timeout occurs, the task will not be put back into the queue.
+                long timeoutMs = this.taskDispatchPolicy.getMaxTaskDispatchDuration().toMillis();


#17796 (comment)

#17796 (comment)

ok

ruanwenjun · 2026-01-22T07:08:23Z

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

+     * Marks the specified task as fatally failed due to an unrecoverable dispatch error,such as timeout
+     * Once this method is called, the task is considered permanently failed and will not be retried.
+     */
+    private void handleDispatchFailure(ITaskExecutionRunnable taskExecutionRunnable, Exception ex,


Suggested change

private void handleDispatchFailure(ITaskExecutionRunnable taskExecutionRunnable, Exception ex,

private void onDispatchTimeout(ITaskExecutionRunnable taskExecutionRunnable, Exception ex,

onDispatchTimeout

better

ruanwenjun · 2026-01-22T07:09:39Z

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

+        if (ExceptionUtils.isWorkerGroupNotFoundException(ex)) {
+            log.error("Dispatch fail, taskName: {}, Worker group not found.", taskName, ex);
+            final TaskFailedLifecycleEvent taskFailedEvent = TaskFailedLifecycleEvent.builder()
+                    .taskExecutionRunnable(taskExecutionRunnable)
+                    .endTime(new Date())
+                    .build();
+            taskExecutionRunnable.getWorkflowEventBus().publish(taskFailedEvent);
+        } else if (ExceptionUtils.isNoAvailableWorkerException(ex)) {
+            log.error("Dispatch fail, taskName: {}, No available worker.", taskName, ex);
+            final TaskFailedLifecycleEvent taskFailedEvent = TaskFailedLifecycleEvent.builder()
+                    .taskExecutionRunnable(taskExecutionRunnable)
+                    .endTime(new Date())
+                    .build();
+            taskExecutionRunnable.getWorkflowEventBus().publish(taskFailedEvent);
+        } else {
+            log.error("Dispatch fail, taskName: {}, Unexpected dispatch error.", taskName, ex);
+            final TaskFailedLifecycleEvent taskFailedEvent = TaskFailedLifecycleEvent.builder()
+                    .taskExecutionRunnable(taskExecutionRunnable)
+                    .endTime(new Date())
+                    .build();
+            taskExecutionRunnable.getWorkflowEventBus().publish(taskFailedEvent);


Suggested change

if (ExceptionUtils.isWorkerGroupNotFoundException(ex)) {

log.error("Dispatch fail, taskName: {}, Worker group not found.", taskName, ex);

final TaskFailedLifecycleEvent taskFailedEvent = TaskFailedLifecycleEvent.builder()

.taskExecutionRunnable(taskExecutionRunnable)

.endTime(new Date())

.build();

taskExecutionRunnable.getWorkflowEventBus().publish(taskFailedEvent);

} else if (ExceptionUtils.isNoAvailableWorkerException(ex)) {

log.error("Dispatch fail, taskName: {}, No available worker.", taskName, ex);

final TaskFailedLifecycleEvent taskFailedEvent = TaskFailedLifecycleEvent.builder()

.taskExecutionRunnable(taskExecutionRunnable)

.endTime(new Date())

.build();

taskExecutionRunnable.getWorkflowEventBus().publish(taskFailedEvent);

} else {

log.error("Dispatch fail, taskName: {}, Unexpected dispatch error.", taskName, ex);

final TaskFailedLifecycleEvent taskFailedEvent = TaskFailedLifecycleEvent.builder()

.taskExecutionRunnable(taskExecutionRunnable)

.endTime(new Date())

.build();

taskExecutionRunnable.getWorkflowEventBus().publish(taskFailedEvent);

log.error("Task: {} dispatch timeout.", taskName, ex);

final TaskFailedLifecycleEvent taskFailedEvent = TaskFailedLifecycleEvent.builder()

.taskExecutionRunnable(taskExecutionRunnable)

.endTime(new Date())

.build();

taskExecutionRunnable.getWorkflowEventBus().publish(taskFailedEvent);

log.error("Task: {} dispatch timeout.", taskName, ex);
final TaskFailedLifecycleEvent taskFailedEvent = TaskFailedLifecycleEvent.builder()
.taskExecutionRunnable(taskExecutionRunnable)
.endTime(new Date())
.build();
taskExecutionRunnable.getWorkflowEventBus().publish(taskFailedEvent);

ok

ruanwenjun · 2026-01-22T07:10:39Z

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

+  # When enabled, tasks not dispatched within this duration are marked as failed.
+  task-dispatch-policy:
+    dispatch-timeout-failed-enabled: false
+    max-task-dispatch-duration: 5m


Suggested change

max-task-dispatch-duration: 5m

max-task-dispatch-duration: 1h

We can remove the max-task-dispatch-duration here, since these is not used when dispatch-timeout-failed-enabled is false, only need to add these on the doc.

ruanwenjun · 2026-01-22T07:11:48Z

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

+    /**
+     * Indicates whether the dispatch timeout checking mechanism is enabled.
+     */
+    private boolean dispatchTimeoutFailedEnabled = false;


Suggested change

private boolean dispatchTimeoutFailedEnabled = false;

private boolean dispatchTimeoutEnabled = false;

dispatchTimeoutEnabled

better

ruanwenjun · 2026-01-29T01:45:53Z

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

+  task-dispatch-policy:
+    dispatch-timeout-enabled: false
+    max-task-dispatch-duration: 1h


Please add to the doc.

Please add to the doc.

ok, add it.

ruanwenjun

LGTM.

njnu-seafish · 2026-01-30T06:56:41Z

@SbloodyS Whenever you have time, I’d be grateful for your review. Thanks so much!

ruanwenjun · 2026-02-01T15:04:48Z

#17932 could provide a more complete solution, effectively preventing abnormal tasks from consuming system resources.

njnu-seafish · 2026-02-02T01:28:59Z

#17932 could provide a more complete solution, effectively preventing abnormal tasks from consuming system resources.

I personally don't think there's a conflict here. The dispatch timeout shouldn't be very long—such exceptions should raise errors as early as possible (e.g., after 10 minutes). In contrast, your workflow or task execution timeout should be configured with a sufficiently long duration, as some weekly tasks may easily run for more than a day.

njnu-seafish · 2026-02-02T01:32:35Z

#17932 could provide a more complete solution, effectively preventing abnormal tasks from consuming system resources.

For the vast majority of tasks and scenarios, task dispatch should not take long, whereas task execution can legitimately run for a very long time. Therefore, the timeout configuration for task dispatch and task execution should not be handled with a single unified setting. It’s normal for a task to run for several hours or even days during execution, but if a task remains in the dispatch phase for hours or days, it most likely indicates an issue.

Dispatch timeout is intended to avoid wasting Master resources, while workflow or task timeout is primarily aimed at preventing Worker resource waste. Wouldn't it be better to have both?

ruanwenjun · 2026-02-02T07:43:10Z

We can keep the both config.

sonarqubecloud · 2026-02-02T08:29:48Z

Quality Gate passed

Issues
8 New issues
0 Accepted issues

Measures
0 Security Hotspots
82.3% Coverage on New Code
0.0% Duplication on New Code

See analysis details on SonarQube Cloud

SbloodyS

+1

Add dispatch timeout checking logic

00b8dc6

njnu-seafish requested review from Gallardot, SbloodyS, caishunfeng and ruanwenjun as code owners December 15, 2025 12:01

github-actions bot added backend test labels Dec 15, 2025

github-actions bot assigned njnu-seafish Dec 15, 2025

SbloodyS requested changes Dec 16, 2025

View reviewed changes

SbloodyS added the improvement make more easy to user or prompt friendly label Dec 16, 2025

SbloodyS added this to the 3.4.0 milestone Dec 16, 2025

ruanwenjun requested changes Dec 16, 2025

View reviewed changes

ruanwenjun added the feature new feature label Dec 16, 2025

苏义超 added 4 commits December 17, 2025 11:52

add dispatch-timeout-checker switch

bb089f3

add dispatch-timeout-checker switch

fbd7edf

update ExceptionUtils

ab84035

add test for dispatch timeout checker

f193af6

njnu-seafish requested review from SbloodyS and ruanwenjun December 17, 2025 11:56

njnu-seafish and others added 10 commits December 19, 2025 14:08

Merge branch 'apache:dev' into Improvement-17795

ab750bb

add TaskFatalLifecycleEvent and handler

584ad2d

add TaskFatalLifecycleEvent and handler

877db06

Add dispatch timeout checking logic

cb44db7

add dispatch-timeout-checker switch

1aa12df

add dispatch-timeout-checker switch

0c29a82

update ExceptionUtils

6ad77d1

add test for dispatch timeout checker

b760003

add TaskFatalLifecycleEvent and handler

39b40c1

add TaskFatalLifecycleEvent and handler

d7bf4fa

ruanwenjun force-pushed the Improvement-17795 branch from 877db06 to d7bf4fa Compare December 21, 2025 04:17

update test

d6f8ccd

njnu-seafish and others added 3 commits January 22, 2026 10:33

Merge branch 'apache:dev' into Improvement-17795

5a47e02

remove redundant tests and comments

743dcbc

remove redundant comments

62cfb30

ruanwenjun reviewed Jan 22, 2026

View reviewed changes

njnu-seafish and others added 2 commits January 23, 2026 11:25

Merge branch 'apache:dev' into Improvement-17795

edf818c

update comment and test

f772a55

njnu-seafish requested a review from ruanwenjun January 23, 2026 07:28

苏义超 and others added 2 commits January 26, 2026 10:03

update test

5f19c7b

Merge branch 'dev' into Improvement-17795

e12af5d

ruanwenjun reviewed Jan 29, 2026

View reviewed changes

add dispatch timeout config in doc

65e7359

njnu-seafish requested a review from EricGao888 as a code owner January 29, 2026 03:12

github-actions bot added the document label Jan 29, 2026

njnu-seafish requested a review from ruanwenjun January 29, 2026 03:13

ruanwenjun approved these changes Jan 29, 2026

View reviewed changes

Merge branch 'dev' into Improvement-17795

6f103b3

Merge branch 'dev' into Improvement-17795

099d3df

Merge branch 'dev' into Improvement-17795

0c295a7

Merge branch 'dev' into Improvement-17795

aed9460

SbloodyS approved these changes Feb 2, 2026

View reviewed changes

SbloodyS merged commit b12b682 into apache:dev Feb 2, 2026
75 checks passed

ruanwenjun mentioned this pull request Feb 18, 2026

[Chore] Bump testcontainer to 1.21.4 to fix could not find a valid Docker environment at CI #17978

Merged

	private Duration dispatchTimeout = Duration.ofMinutes(5);
	private Duration maxTaskDispatchDuration;

	final int taskId = taskExecutionRunnable.getId();
	final int taskExecutionRunnableId = taskExecutionRunnable.getId();


		public class WorkerNotFoundException extends TaskDispatchException {

		public WorkerNotFoundException(String message) {

-    public WorkerNotFoundException(String message) {
+    public NoAvailableWorkerException(String workerGroup) {
+        super("Cannot find available worker under worker group: " + workerGroup);
+    }

		final MasterConfig masterConfig = new MasterConfig();
		dispatcher = new WorkerGroupDispatcher("TestGroup", taskExecutorClient, masterConfig.getDispatchTimeout());

	private void handleDispatchFailure(ITaskExecutionRunnable taskExecutionRunnable, Exception ex,
	private void onDispatchTimeout(ITaskExecutionRunnable taskExecutionRunnable, Exception ex,

	max-task-dispatch-duration: 5m
	max-task-dispatch-duration: 1h

	private boolean dispatchTimeoutFailedEnabled = false;
	private boolean dispatchTimeoutEnabled = false;

Conversation

njnu-seafish commented Dec 15, 2025

Purpose of the pull request

Brief change log

Verify this pull request

Pull Request Notice

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

njnu-seafish Jan 13, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

njnu-seafish commented Jan 15, 2026

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

njnu-seafish Jan 13, 2026 •

edited

Loading

ruanwenjun commented Feb 1, 2026 •

edited

Loading

njnu-seafish commented Feb 2, 2026 •

edited

Loading