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

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-08T02:53:35Z

@SbloodyS When you have time, could you please help trigger the CI pipeline? I'm unable to run the integration tests locally. Thank you so much!

SbloodyS · 2026-01-12T09:32:40Z

@SbloodyS When you have time, could you please help trigger the CI pipeline? I'm unable to run the integration tests locally. Thank you so much!

Done.

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

sonarqubecloud · 2026-01-23T06:53:24Z

Quality Gate passed

Issues
8 New issues
0 Accepted issues

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

See analysis details on SonarQube Cloud

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

njnu-seafish and others added 2 commits January 8, 2026 09:50

Merge branch 'apache:dev' into Improvement-17795

42a5aa2

update test

879ba5e

njnu-seafish and others added 5 commits January 8, 2026 20:28

Merge branch 'dev' into Improvement-17795

77a5dfa

Merge branch 'dev' into Improvement-17795

9a6d664

Merge branch 'apache:dev' into Improvement-17795

cca935f

update test

e49549e

update test

b5a703b

苏义超 and others added 11 commits January 13, 2026 11:04

Remove this test case and track the fix in a separate issue

9f97478

add it test case by using SHELL task

ddb7eb7

update test comment

93e398c

Merge branch 'dev' into Improvement-17795

6864a2e

update task instance state

f7c7fe1

remove masterContainer.assertAllResourceReleased if workflow is running

7437249

update test

99d2640

Merge branch 'dev' into Improvement-17795

a6302d3

Resolve merge conflicts from dev branch

7a01fab

update test

8c4700a

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

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

	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;

[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

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 8, 2026

Uh oh!

SbloodyS commented Jan 12, 2026

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!

njnu-seafish Jan 13, 2026 •

edited

Loading