Repository navigation
Fix pipe task start failures: report, roll back and recover - #19119
Merged
jt2594838 merged 5 commits intoNov 4, 2026
Merged
Conversation
…ission PipeSubtaskExecutor.start() marked the subtask as "submitting self" before calling submitSelf(). When the submission was rejected, e.g. because the worker pool had been shut down, the flag stayed set, so every later start() took the "already running" branch, logged nothing and reported success while the subtask never ran until the node restarted. The flag is now rolled back on failure and the rejection is logged and rethrown, so the failure reaches the pipe task agent and the config node instead of being masked as success. PipeAbstractSinkSubtask also clears its "isSubmitted" sign when submitting is disallowed, so a submission that was never dispatched to the worker pool can be submitted again. The IoTConsensusV2 subtask executor is shared by all the consensus pipes of a node. Once it had been shut down, getConsensusExecutor() kept returning the dead pool, so every consensus pipe registered afterwards could never start its sink subtask. A shut-down executor is now recreated and installed back into IoTV2GlobalComponentContainer.
startPipeTask() let an exception of PipeTask.start() escape without rolling back the task it had just created. The task meta was already registered, so the following meta sync found an existing task meta with the same leader and never called start() again: the pipe stayed "RUNNING" in the metadata while running nothing until the node restarted. The failed task is now dropped together with its task meta, so that the next meta sync recreates and starts the task again, which gives the task another chance to start once the underlying problem is gone. The failure itself is logged at error level instead of being carried away by the meta sync loop.
…pping silently A sink subtask submits itself again after the end of each round. If the submission is rejected, e.g. because the worker pool is being shut down while the subtask is finishing a round, the rejection escaped from the callback thread where it was swallowed, and, for a delayed submission, the "submitted" sign had already been set before scheduling the delayed task, so it was never cleared. In both cases the subtask was never submitted again and stopped transferring events without leaving any trace in the log. The sign is now cleared and the rejection is reported whenever the submission or the delayed scheduling fails, so the failure is visible and a following start of the subtask can submit it again.
Caideyipi
reviewed
Nov 4, 2026
| // later allowSubmittingSelf() + submitSelf() can really submit the subtask again, even if the | ||
| // previous submission was never dispatched to the worker pool, e.g., because the submission was | ||
| // rejected or scheduled on an executor that has been shut down. | ||
| isSubmitted = false; |
Collaborator
There was a problem hiding this comment.
[P1] Preserve the submitted state while a sink is in flight. disallowSubmittingSelf() is also used by an ordinary stop, and PipeSubtaskExecutor.stop() only flips the stop flag without waiting for the worker. Clearing isSubmitted here lets a quick stop/start submit the same sink subtask while its previous worker is still running. A pending delayed lambda can also pass the shouldStopSubmittingSelf check after restart and submit in addition to the new run. That breaks the non-parallel invariant and can concurrently consume or send the same events. Please clear this flag only when the previous submission was definitely rejected, or otherwise prevent old work from resuming after restart.
PipeAbstractSinkSubtask#onDisallowSubmittingSelf cleared isSubmitted while a submitted call could still be running. A following start() then submitted the same subtask again, so two workers could execute one sink subtask concurrently, which PipeSubtask does not support. Keep isSubmitted as the single-flight flag: a call in flight is allowed to finish and re-submit itself once start() has cleared the stop flag. Instead, make PipeSubtaskExecutor#start fail fast when the executor has been shut down, before touching any subtask state. A shut-down executor can never run the subtask, so start() must not report a false success. This also covers a delayed submission that is still pending when the executor is shut down. Add regression tests for stopping and restarting a subtask while its first call is blocked, and for starting a subtask with a pending delayed submission after shutdown. Recreate the shared consensus executor after the manager tests so they do not leave it shut down for other tests.
Replace the verbose inline comments in the pipe task start fix with concise ones, and state the scenario and expected behavior in a short Javadoc above each new test.
luoluoyuyu
changed the title
jt2594838
approved these changes
Nov 4, 2026
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Motivation
A pipe sink subtask can fail to start and still be reported as started, and it may never recover by itself:
PipeSubtaskExecutor.start()callsallowSubmittingSelf()beforesubmitSelf(). WhensubmitSelf()throws - for example because the executor has been shut down - the "submitting self" flag has already been set. Every laterstart()then takes the "already running" branch, logs only at debug level and returns normally, soPipeSinkSubtaskLifeCycle.start()incrementsrunningTaskCountand the config node sees a successful start while the sink never transfers a single event.PipeAbstractSinkSubtask.submitSelf()setsisSubmittedbefore scheduling the delayed submission, and it does not handle a rejected submission at all. When the submission or the delayed scheduling is rejected, nothing is dispatched while the sign stays set, so every following submission is skipped; on the round-completion path the rejection is thrown from the callback thread, where it is swallowed. In both cases the subtask stops transferring events without leaving any trace in the log.IoTConsensusV2SubtaskExecutoris a node-wide singleton shared by all consensus pipes.PipeSubtaskExecutorManager.getConsensusExecutor()handed it out without checking whether it had been shut down, so after a shutdown every consensus pipe registered later was bound to a dead pool and could never start its sink.PipeTaskAgent.startPipeTask()let the exception ofPipeTask.start()escape without rolling back the task it had just created. The task meta was already registered, so the following meta sync found an existing task meta with an unchanged leader and never calledstart()again: the pipe stayedRUNNINGin the metadata while running nothing until the node restarted.isSubmittedmust not be cleared on an ordinary stop:stop()only flips the stop flag and does not wait for the worker, so a quickstop()/start()could submit the same sink subtask while its previous call is still running, and a pending delayed submission could pass the stop check after the restart and submit again. It may only be cleared when the previous call has completed or the submission was definitely rejected.These were observed together on a 2.0.11 cluster: the shared consensus executor was shut down at 20:14, 384 consensus pipes were re-created at 20:16, every sink submission was rejected, and the pipes stayed "running" without transferring any event until the node was restarted.
Modification
PipeSubtaskExecutor.start(): fail fast withRejectedExecutionExceptionwhen the executor has been shut down, before touching the subtask state, so a dead executor can never be reported as a successful start. WhensubmitSelf()still fails, roll back the "submitting self" state, log at error level and rethrow.PipeAbstractSinkSubtask: keepisSubmittedas the single-flight flag and do not clear it inonDisallowSubmittingSelf(). It is only cleared when the previous call has completed (onSuccess/onFailure) or when the submission was definitely rejected, i.e. no worker was queued. A rejected submission is reported as a warning instead of being swallowed.PipeSubtaskExecutorManager.getConsensusExecutor(): recreate the executor and install it back intoIoTV2GlobalComponentContainerwhen the current one has been shut down.PipeTaskAgent.startPipeTask(): drop the failed task together with its task meta, so that the next meta sync creates and starts it again, and log the failure instead of carrying it away.Behavior notes
start()on a shut-down executor now throws instead of logging a false success.PipeSinkSubtaskLifeCycle.start()only incrementsrunningTaskCountafter a successfulexecutor.start(), andPipeTaskAgent.startPipeTask()rolls the task back on failure.Tests
PipeSinkSubtaskExecutorTest#testStartWhileThePreviousCallIsStillRunning: stopping and starting a subtask while its first call is blocked does not start a second concurrent call.PipeSinkSubtaskExecutorTest#testStartAfterShutdownWithPendingDelayedSubmission: with a pending delayed submission,start()after shutdown still fails.PipeSinkSubtaskExecutorTest#testStartAfterShutdown: starting on a shut-down executor throws and does not leave the subtask in the submitting state.PipeSinkSubtaskExecutorTest#testSubmitSelfAfterShutdownand#testDelayedSubmitSelfAfterShutdown: a rejected submission is reported and clears the submitted flag, so a following submission is still attempted.PipeSubtaskExecutorTest#testStartWithRejectedSubmission: a rejected submission rolls back the submitting state and still fails.PipeSubtaskExecutorManagerTest: the consensus executor is a singleton while it is alive and is recreated after it has been shut down; the shared executor is recreated after each test so other tests are not affected.PipeDataNodeTaskAgentTest#testFailedTaskStartIsRolledBackForRetry: a failed task start drops the task and its task meta.Verification:
mvn spotless:check,mvn test-compile -DskipTestsandmvn test-compile -DskipTests -P with-zh-localepass, and the affected test classes (31 tests) pass. The new tests fail against the pre-fix code: the start/rejection tests fail against the pre-fixnode-commonsclasses, and the single-flight test fails when theonDisallowSubmittingSelf()override is restored. The datanode module could not be compiled locally against currentmasterbecause the Thrift compiler is not available in this environment, so its compilation is left to CI.