[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

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

ruanwenjun · 2025-12-21T04:21:08Z

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

+    /**
+     * Configuration for the master's task dispatch timeout check mechanism.
+     * This controls whether the system enforces a time limit for dispatching tasks to workers,
+     * and if so, how long to wait before marking a task as failed due to dispatch timeout.
+     */
+    private MasterDispatchTimeoutCheckerConfig dispatchTimeoutChecker = new MasterDispatchTimeoutCheckerConfig();


Suggested change

/**

* Configuration for the master's task dispatch timeout check mechanism.

* This controls whether the system enforces a time limit for dispatching tasks to workers,

* and if so, how long to wait before marking a task as failed due to dispatch timeout.

*/

private MasterDispatchTimeoutCheckerConfig dispatchTimeoutChecker = new MasterDispatchTimeoutCheckerConfig();

private TaskDispatchPolicy taskDispatchPolicy = new TaskDispatchPolicy();

ruanwenjun · 2025-12-21T04:21:46Z

...ava/org/apache/dolphinscheduler/server/master/config/MasterDispatchTimeoutCheckerConfig.java

+ * 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 {


Suggested change

public class MasterDispatchTimeoutCheckerConfig {

public class TaskDispatchPolicy {

ruanwenjun · 2025-12-21T04:24:21Z

...ava/org/apache/dolphinscheduler/server/master/config/MasterDispatchTimeoutCheckerConfig.java

+    /**
+     * Whether to enable the dispatch timeout checking mechanism.
+     */
+    private boolean enabled = false;


Suggested change

private boolean enabled = false;

private boolean dispatchTimeoutFailedEnabled = false;

ruanwenjun · 2025-12-21T04:28:17Z

...ava/org/apache/dolphinscheduler/server/master/config/MasterDispatchTimeoutCheckerConfig.java

+     * Tasks exceeding this duration in the dispatch queue will be failed.
+     * Examples: "2m", "5m", "30m". Defaults to 5 minutes.
+     */
+    private Duration maxTaskDispatchDuration = Duration.ofMinutes(5);


Suggested change

private Duration maxTaskDispatchDuration = Duration.ofMinutes(5);

private Duration maxTaskDispatchDuration;

When we don't have a good suggested value, don't set a default value, as this can mislead many users.

When we don't have a good suggested value, don't set a default value, as this can mislead many users.

ok

ruanwenjun · 2025-12-21T04:30:04Z

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

            }
            taskExecutorClient.dispatch(taskExecutionRunnable);
-        } catch (Exception e) {
+        } catch (TaskDispatchException ex) {


Suggested change

} catch (TaskDispatchException ex) {

} catch (Exception ex) {

Currently, at least in this PR, you can't guarantee that other exceptions won't be thrown here, right?

ruanwenjun · 2025-12-21T04:31:40Z

...ache/dolphinscheduler/server/master/engine/task/lifecycle/event/TaskFatalLifecycleEvent.java

+@Data
+@Builder
+@AllArgsConstructor
+public class TaskFatalLifecycleEvent extends AbstractTaskLifecycleEvent {
+
+    private final ITaskExecutionRunnable taskExecutionRunnable;
+
+    private final Date endTime;
+
+    @Override
+    public ILifecycleEventType getEventType() {
+        return TaskLifecycleEventType.FATAL;
+    }
+
+    @Override
+    public String toString() {
+        return "TaskFatalLifecycleEvent{" +
+                "task=" + taskExecutionRunnable.getName() +
+                ", endTime=" + endTime +
+                '}';
+    }
+}


Please remove this.

Please remove this.

ok, will publish TaskFailedLifecycleEvent

ruanwenjun · 2025-12-21T04:35:21Z

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

                .untilAsserted(() -> verify(taskExecutorClient, times(2)).dispatch(taskExecutionRunnable));
    }

+    @Test


You should add UT or IT test case to verify the time out check is enabled. Your core logic has not been tested.

You should add UT or IT test case to verify the time out check is enabled. Your core logic has not been tested.

OK. I need to modify the master's configuration. I'll submit the changes and test them — I can't run the integration tests locally.

…heduler into Improvement-17795

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

ruanwenjun requested changes Dec 21, 2025

View reviewed changes

njnu-seafish and others added 3 commits December 22, 2025 17:02

Merge branch 'apache:dev' into Improvement-17795

0c8c93e

Merge branch 'Improvement-17795' of github.com:njnu-seafish/dolphinsc…

5e1dd52

…heduler into Improvement-17795

update TaskDispatchPolicy

5b2722d

SbloodyS modified the milestones: 3.4.0, 3.4.1 Dec 23, 2025

add it test case

33c16e8

njnu-seafish requested a review from ruanwenjun December 23, 2025 01:58

njnu-seafish and others added 3 commits December 23, 2025 09:59

Merge branch 'dev' into Improvement-17795

3c25528

update test

99a5d7c

Merge branch 'Improvement-17795' of github.com:njnu-seafish/dolphinsc…

9966047

…heduler into Improvement-17795

	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());

	public class MasterDispatchTimeoutCheckerConfig {
	public class TaskDispatchPolicy {

	private boolean enabled = false;
	private boolean dispatchTimeoutFailedEnabled = false;

	} catch (TaskDispatchException ex) {
	} catch (Exception ex) {

[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!

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!

Reviewers

Assignees

Labels

Projects