You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@dolphinscheduler.apache.org by "csurong (via GitHub)" <gi...@apache.org> on 2026/10/23 09:16:23 UTC
[PR] [Fix-18735][Master] Recover terminal ACKs after workflow finalization [dolphinscheduler]
csurong opened a new pull request, #18736:
URL: https://github.com/apache/dolphinscheduler/pull/18736
## Was this PR generated or assisted by AI?
YES. AI assisted with the investigation, implementation, tests, review, and this description.
## Purpose of the pull request
Fixes #18735.
If a terminal ACK is lost, the workflow can finish and leave the Master's cache before the executor retries its report. The retry then fails because the listener can no longer find the workflow. The result is already in the database, but the executor keeps waiting for an ACK and cannot finish cleaning up.
This change lets the Master confirm that persisted result and resend the ACK after cache eviction, so the executor can follow its normal finalization path.
## Brief change log
- On a terminal report with no cached workflow, check the persisted task, workflow, executor host, and terminal state before sending the ACK. This path does not update task state or replay workflow events.
- Accept the state changes that can happen after the original report: a failed task waiting for retry can be paused or killed, and a failed or killed task can be forced to success. Unrelated or unfinished states are still rejected.
- Allow the executor client to send an ACK from a persisted TaskInstance. Keep success alerts before the ACK, as in the existing handler.
- Add listener/client tests and three Spring integration cases covering a lost SUCCESS ACK and a lost FAILED ACK followed by pause or stop. Each integration case waits for cache eviction before restoring ACK delivery, then checks that the executor resources are released.
## Verify this pull request
The Maven reactor build passed with 68 selected test cases, including the three Spring/H2 integration cases:
```sh
mvn -B -ntp -pl dolphinscheduler-master -am \
-Dtest=TaskExecutorEventListenerImplTest,TaskExecutorClientTest,TaskSuccessPersistenceRecoveryTest,TaskSuccessLifecycleEventHandlerTest,TaskTerminalAckRecoveryTestCase \
-Dsurefire.failIfNoSpecifiedTests=false \
-Djacoco.skip=true -Dspotless.skip=true \
-Dmaster.server-load-protection.enabled=false test
```
The first run hit the local machine's load-protection threshold, which prevented workflow commands from being consumed. The command above disables that check for the test process; no repository configuration was changed. These are targeted tests, not the full test suite. The integration cases use logic executors and do not exercise physical Worker directory deletion.
`./mvnw -pl dolphinscheduler-master spotless:apply`, the standard `spotless:check`, and `git diff --check` also passed.
## Pull Request Notice
[Pull Request Notice](https://github.com/apache/dolphinscheduler/blob/dev/docs/docs/en/contribute/join/pull-request.md)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@dolphinscheduler.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Fix-18735][TaskExecutor] Release completed executors without waiting for ACKs [dolphinscheduler]
Posted by "csurong (via GitHub) via commits" <co...@dolphinscheduler.apache.org>.
csurong commented on PR #18736:
URL: https://github.com/apache/dolphinscheduler/pull/18736#issuecomment-6556259024
Thanks @ruanwenjun, updated in 8c8a2e2. I've removed `finishedTime` and now use the terminal event's creation time, with a 6h default on both Master and Worker.
I kept the preceding events until that deadline because they can be prerequisites: if an undelivered DISPATCHED expires first, the Master can remain in SUBMITTED_SUCCESS and ignore a later SUCCESS while still ACKing it. So the reporter keeps retrying in order, then discards the whole pending channel when the terminal event expires. I've added an integration test for this sequence.
The ordering change makes sure execution tracking stops before the terminal-event listener can finalize the executor and release its slot. The targeted tests pass locally.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@dolphinscheduler.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Fix-18735][Master] Recover terminal ACKs after workflow finalization [dolphinscheduler]
Posted by "csurong (via GitHub)" <gi...@apache.org>.
csurong commented on PR #18736:
URL: https://github.com/apache/dolphinscheduler/pull/18736#issuecomment-6432921598
Thanks @ruanwenjun, I've moved the recovery logic into `TaskExecutorTerminalAckRecovery` to make its purpose clearer, and restored the existing `ITaskExecution` client interfaces.
The recovery path checks the task state in the DB before sending the ACK. For a FAILED report, it also allows PAUSE/KILL/FORCED_SUCCESS, since those can follow the original failure. I've added integration coverage for ACK loss and retries after cache eviction; the targeted tests pass locally. Could you take another look when you have a chance?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@dolphinscheduler.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Fix-18735][TaskExecutor] Release completed executors without waiting for ACKs [dolphinscheduler]
Posted by "ruanwenjun (via GitHub)" <gi...@apache.org>.
ruanwenjun commented on code in PR #18736:
URL: https://github.com/apache/dolphinscheduler/pull/18736#discussion_r4422229880
##########
dolphinscheduler-task-executor/src/main/java/org/apache/dolphinscheduler/task/executor/eventbus/TaskExecutorLifecycleEventRemoteReporter.java:
##########
@@ -179,13 +191,22 @@ public Map<Integer, ReportableTaskExecutorLifecycleEventChannel> getEventChannel
return eventChannels;
}
- private void finalizeTaskExecutor(final Integer taskExecutorId) {
- final Optional<ITaskExecutor> taskExecutorOptional = taskExecutorRepository.get(taskExecutorId);
- if (taskExecutorOptional.isPresent()) {
- taskExecutorOptional.get().getTaskExecutorEventBus()
- .publish(TaskExecutorFinalizeLifecycleEvent.of(taskExecutorOptional.get()));
- } else {
- log.warn("TaskExecutor is not exists: {}", taskExecutorId);
+ private boolean expireFinishedTaskEventChannel(final ReportableTaskExecutorLifecycleEventChannel eventChannel) {
+ eventChannelsLock.lock();
+ try {
+ if (eventChannel.finishedTime == null
+ || System.currentTimeMillis() - eventChannel.finishedTime < taskEventReportMaxDurationMillis) {
+ return false;
+ }
+ if (eventChannels.remove(eventChannel.getTaskExecutorId(), eventChannel)) {
+ log.error("Task executor {} result was not acknowledged within {} ms after completion. "
+ + "Discarding pending events for master {}",
+ eventChannel.getTaskExecutorId(), taskEventReportMaxDurationMillis,
+ eventChannel.workflowInstanceHost);
+ }
+ return true;
+ } finally {
+ eventChannelsLock.unlock();
}
}
Review Comment:
```suggestion
private boolean removeExpiredEvent(final ReportableTaskExecutorLifecycleEventChannel eventChannel) {
...
}
```
We can loop the event from the channel, if the current time - event's create time > event max lifecycle, then means the event is expired, we can remove the event.
##########
dolphinscheduler-task-executor/src/main/java/org/apache/dolphinscheduler/task/executor/worker/AbstractTaskExecutorWorker.java:
##########
@@ -60,26 +60,30 @@ protected void trackTaskExecutorState(final ITaskExecutor taskExecutor) {
protected void onTaskExecutorSuccess(final ITaskExecutor taskExecutor) {
taskExecutor.getTaskExecutionContext().setEndTime(System.currentTimeMillis());
- taskExecutor.getTaskExecutorEventBus().publish(TaskExecutorSuccessLifecycleEvent.of(taskExecutor));
+ final TaskExecutorSuccessLifecycleEvent event = TaskExecutorSuccessLifecycleEvent.of(taskExecutor);
onTaskExecutorFinished(taskExecutor);
+ taskExecutor.getTaskExecutorEventBus().publish(event);
}
Review Comment:
Why change the order?
##########
dolphinscheduler-task-executor/src/main/java/org/apache/dolphinscheduler/task/executor/eventbus/TaskExecutorLifecycleEventRemoteReporter.java:
##########
@@ -291,14 +317,21 @@ public static class ReportableTaskExecutorLifecycleEventChannel {
private final LinkedBlockingQueue<IReportableTaskExecutorLifecycleEvent> taskExecutionEventsQueue;
- // todo: remove the master address from the channel, we need to get the master address from the TaskExecutor
- public ReportableTaskExecutorLifecycleEventChannel(int taskExecutorId) {
+ private volatile String workflowInstanceHost;
+
+ private volatile Long finishedTime;
Review Comment:
We don't need to add this filed, looks strange.
##########
dolphinscheduler-master/src/main/resources/application.yaml:
##########
@@ -99,6 +99,9 @@ master:
# The number of threads used to execute logic task.
# task-executor-thread-count: 10
max-heartbeat-interval: 10s
+ # Stop retrying and discard unacknowledged events after this duration from task completion.
+ # Expiration is logged as an error; it does not mean the result was persisted by the master.
+ task-event-report-max-duration: 1h
Review Comment:
```suggestion
task-event-report-max-duration: 6h
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@dolphinscheduler.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Fix-18735][TaskExecutor] Release completed executors without waiting for ACKs [dolphinscheduler]
Posted by "csurong (via GitHub) via commits" <co...@dolphinscheduler.apache.org>.
csurong commented on PR #18736:
URL: https://github.com/apache/dolphinscheduler/pull/18736#issuecomment-6576211591
Thanks @SbloodyS @ruanwenjun. I traced this again, and the coordinator retry was not well justified. The test mocked a temporary enqueue rejection, but I hadn't established a corresponding failure path in the real reporter.
The normal sequence is:
1. The worker stops tracking execution, but the executor remains in the repository.
2. The coordinator passes the terminal event to the reporter's in-memory queue.
3. Once accepted, the listener finalizes the executor. Remote delivery and ACK retries run independently.
The explicit missing-executor exception is a lifecycle check; retrying the event would not restore the missing executor. Other local exceptions are still possible, so I shouldn't have presented this as general recovery.
I've removed the coordinator retry in 6174097, which also removes the unbounded 50ms retry path. The tests now use the real worker/coordinator/reporter chain, with failure injected at remote delivery, and verify resource release and slot reuse for all four terminal states. I also added coverage for recreating a channel after its previous events are acknowledged. The 34 targeted tests pass locally.
Does keeping this PR focused on resource release after remote delivery or ACK failure make sense to you?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@dolphinscheduler.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Fix-18735][TaskExecutor] Release completed executors without waiting for ACKs [dolphinscheduler]
Posted by "ruanwenjun (via GitHub)" <gi...@apache.org>.
ruanwenjun commented on code in PR #18736:
URL: https://github.com/apache/dolphinscheduler/pull/18736#discussion_r4422194050
##########
dolphinscheduler-task-executor/src/main/java/org/apache/dolphinscheduler/task/executor/eventbus/TaskExecutorLifecycleEventRemoteReporter.java:
##########
@@ -291,14 +317,21 @@ public static class ReportableTaskExecutorLifecycleEventChannel {
private final LinkedBlockingQueue<IReportableTaskExecutorLifecycleEvent> taskExecutionEventsQueue;
- // todo: remove the master address from the channel, we need to get the master address from the TaskExecutor
- public ReportableTaskExecutorLifecycleEventChannel(int taskExecutorId) {
+ private volatile String workflowInstanceHost;
+
+ private volatile Long finishedTime;
Review Comment:
We don't need to add `finishedTime` here, looks strange.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@dolphinscheduler.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Fix-18735][Master] Recover terminal ACKs after workflow finalization [dolphinscheduler]
Posted by "ruanwenjun (via GitHub)" <gi...@apache.org>.
ruanwenjun commented on code in PR #18736:
URL: https://github.com/apache/dolphinscheduler/pull/18736#discussion_r4375815381
##########
dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/rpc/TaskExecutorEventListenerImpl.java:
##########
@@ -161,23 +207,90 @@ public void onTaskExecutorKilled(final TaskExecutorKilledLifecycleEvent taskExec
public void onTaskExecutorPaused(final TaskExecutorPausedLifecycleEvent taskExecutorPausedLifecycleEvent) {
LogUtils.setWorkflowInstanceIdMDC(taskExecutorPausedLifecycleEvent.getWorkflowInstanceId());
try {
+ final IWorkflowExecution workflowExecution =
+ workflowRepository.get(taskExecutorPausedLifecycleEvent.getWorkflowInstanceId());
+ if (workflowExecution == null) {
+ ackTaskExecutorTerminalEvent(taskExecutorPausedLifecycleEvent, TaskExecutionStatus.PAUSE,
+ taskExecutorPausedLifecycleEvent.getTaskInstanceHost());
+ return;
+ }
final ITaskExecution taskExecution =
- getTaskExecution(taskExecutorPausedLifecycleEvent);
+ getTaskExecution(taskExecutorPausedLifecycleEvent, workflowExecution);
final TaskPausedLifecycleEvent taskPausedEvent = TaskPausedLifecycleEvent.of(taskExecution);
taskExecution.getWorkflowEventBus().publish(taskPausedEvent);
} finally {
LogUtils.removeWorkflowInstanceIdMDC();
}
}
+ private void ackTaskExecutorTerminalEvent(final IReportableTaskExecutorLifecycleEvent event,
Review Comment:
Is there a better way to implement this? Right now, it's hard to tell that this logic is intended for scenarios like failover.
##########
dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/task/client/ITaskExecutorClient.java:
##########
@@ -72,4 +73,12 @@ void ackTaskExecutorLifecycleEvent(
final ITaskExecution taskExecution,
final ITaskExecutorLifecycleEventReporter.TaskExecutorLifecycleEventAck taskExecutorLifecycleEventAck);
+ /**
+ * Send an acknowledgement using a persisted task instance after its workflow has been released.
+ * <p> Like the execution-based overload, a failed send is retried by the executor engine.
+ */
+ void ackTaskExecutorLifecycleEvent(
+ final TaskInstance taskInstance,
Review Comment:
```suggestion
final ITaskExecution taskExecution,
```
Donot change the interface style here.
##########
dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/rpc/TaskExecutorEventListenerImpl.java:
##########
@@ -129,8 +161,15 @@ public void onTaskExecutorSuccess(final TaskExecutorSuccessLifecycleEvent taskEx
public void onTaskExecutorFailed(final TaskExecutorFailedLifecycleEvent taskExecutorFailedLifecycleEvent) {
LogUtils.setWorkflowInstanceIdMDC(taskExecutorFailedLifecycleEvent.getWorkflowInstanceId());
try {
+ final IWorkflowExecution workflowExecution =
+ workflowRepository.get(taskExecutorFailedLifecycleEvent.getWorkflowInstanceId());
+ if (workflowExecution == null) {
+ ackTaskExecutorTerminalEvent(taskExecutorFailedLifecycleEvent, TaskExecutionStatus.FAILURE,
+ taskExecutorFailedLifecycleEvent.getTaskInstanceHost());
+ return;
+ }
Review Comment:
Need to check if the task instance status is failed in DB.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@dolphinscheduler.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Fix-18735][TaskExecutor] Release completed executors without waiting for ACKs [dolphinscheduler]
Posted by "csurong (via GitHub) via commits" <co...@dolphinscheduler.apache.org>.
csurong commented on PR #18736:
URL: https://github.com/apache/dolphinscheduler/pull/18736#issuecomment-6570109748
Thanks @SbloodyS, fixed in bdafbe2. The coordinator now requeues the original terminal event if handling fails while the executor is still in the repository. It scans the repository, so it can retry even after the worker stops tracking the task.
I kept the retry there so we can reuse the completed result without generating new terminal events or resetting their timestamps. Once the reporter accepts it, the executor finalizes as before. I've added worker-to-coordinator tests for all four terminal states, including recovery after enqueue failure and reuse of the released slot. The targeted tests pass locally.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@dolphinscheduler.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Fix-18735][TaskExecutor] Release completed executors without waiting for ACKs [dolphinscheduler]
Posted by "ruanwenjun (via GitHub) via commits" <co...@dolphinscheduler.apache.org>.
ruanwenjun commented on PR #18736:
URL: https://github.com/apache/dolphinscheduler/pull/18736#issuecomment-6730478571
@csurong Please look my previous review https://github.com/apache/dolphinscheduler/pull/18736/changes#r4422229880, we need to make the code clean.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@dolphinscheduler.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Fix-18735][Master] Recover terminal ACKs after workflow finalization [dolphinscheduler]
Posted by "ruanwenjun (via GitHub)" <gi...@apache.org>.
ruanwenjun commented on PR #18736:
URL: https://github.com/apache/dolphinscheduler/pull/18736#issuecomment-6436743233
@csurong
Another approach to the fix this is for the worker to trigger `finalize` immediately after the task completes, freeing up resources such as threads, and then send the task result event. If no ACK is received, it resends the event until the maximum retry time—for example, 1 hour (configurable)—is exceeded.
In this case, if the master successfully processes the result event but the ACK is lost, it will not cause a serious problem.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@dolphinscheduler.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Fix-18735][Master] Recover terminal ACKs after workflow finalization [dolphinscheduler]
Posted by "ruanwenjun (via GitHub)" <gi...@apache.org>.
ruanwenjun commented on PR #18736:
URL: https://github.com/apache/dolphinscheduler/pull/18736#issuecomment-6436487115
@csurong There is another scenario: when the task instance status has already been persisted to the database but the ACK has not yet been sent to the worker, and the master node crashes, triggering fault tolerance, if the master host changes, the ACK will be permanently lost.
In this scenario, when the master goes offline, it may be necessary to send the result event to the coordinator, or to randomly select a master to send the result event.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@dolphinscheduler.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Fix-18735][TaskExecutor] Release completed executors without waiting for ACKs [dolphinscheduler]
Posted by "csurong (via GitHub)" <gi...@apache.org>.
csurong commented on PR #18736:
URL: https://github.com/apache/dolphinscheduler/pull/18736#issuecomment-6519218654
Thanks @ruanwenjun, I've updated the PR following your suggestion. Completed tasks now finalize as soon as their result is queued, without waiting for an ACK. The reporter retries independently for up to 1 hour (configurable), then logs an error and drops any unacknowledged events. I've also added tests for resource release, retries after cleanup, and expiration; the targeted tests pass locally. Could you take another look when you have a chance?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@dolphinscheduler.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Fix-18735][TaskExecutor] Release completed executors without waiting for ACKs [dolphinscheduler]
Posted by "ruanwenjun (via GitHub) via commits" <co...@dolphinscheduler.apache.org>.
ruanwenjun commented on code in PR #18736:
URL: https://github.com/apache/dolphinscheduler/pull/18736#discussion_r4439389958
##########
dolphinscheduler-task-executor/src/main/java/org/apache/dolphinscheduler/task/executor/eventbus/TaskExecutorEventBusCoordinator.java:
##########
@@ -193,6 +193,11 @@ private void doFireTaskExecutorEventBus(final ITaskExecutor taskExecutor) {
JSONUtils.toJsonString(taskExecutorLifecycleEvent));
} catch (Exception e) {
log.error("Fire TaskExecutorLifecycleEvent: {} error", taskExecutorLifecycleEvent, e);
+ if (taskExecutorLifecycleEvent.getType().isFinished()
+ && taskExecutorRepository.contains(taskExecutor.getId())) {
+ // The worker has stopped tracking; retain the original result until the listener can finalize it.
+ taskExecutorEventBus.publish(headEventOptional.get());
+ }
Review Comment:
The logic here seems strange—the event shouldn’t be executed repeatedly, as this can easily lead to memory leaks. If you’re concerned about the event being rejected, you should remove the rejection logic from `reportTaskExecutorLifecycleEvent`. I’m actually not sure why `reportTaskExecutorLifecycleEvent` needs to throw an exception, or why a retry is necessary here. Could you explain the issue here using a sequence diagram or some other method?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@dolphinscheduler.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org