Thanks for working on this. I reviewed the latest head 22262ee0ef59.

What this PR fixes

• User pain: A savepoint (or cancel) issued while a pipeline is reset for restore or not yet deployed leaves the job stuck in DOING_SAVEPOINT/CANCELING: PhysicalVertex.forceStop() moves the task to CANCELED but stateProcess() bails at if (!isRunning) (PhysicalVertex.java:594) before the CANCELED branch completes taskFuture, so SubPlan.addPhysicalVertexCallBack never increments finishedTaskNum and the pipeline never ends.
• Fix approach: forceStop() now does if (!isRunning) { startPhysicalVertex(); } (PhysicalVertex.java:558-560) before noticeTaskExecutionServiceCancel()/updateTaskState(CANCELED), mirroring the startPhysicalVertex(); cancel(); pairing in SubPlan's FAILING/CANCELING branch (SubPlan.java:727-734), so the CANCELED branch runs, calls stopPhysicalVertex() and completes the future; plus one vertex-level regression test in StateTransitionCleanupTest.
• One-line summary: The two-line guard is correct and minimal for the vertex-level future, but it does not address what happens around it -- the restore thread sleeping in prepareRestorePipeline() (SubPlan.java:490-504) holds the SubPlan monitor and will still release/re-apply slots and drive a pipeline whose tasks are all CANCELED through SCHEDULED/DEPLOYING/RUNNING before the callback can end it, isRunning is flipped to true and never reverted if updateTaskState(CANCELED) short-circuits (null/terminal IMap state at PhysicalVertex.java:354-379), the Javadoc promises more than the method guarantees, and the only test is a never-started vertex in CREATED rather than the reset-for-restore + forceStopPipeline() path from #12622.

Runtime chain I rechecked

PhysicalPlan.cancelJob/savepoint -> SubPlan.stopPipelineWithCheckpointFallback()   PhysicalPlan.java:260
  -> SubPlan.forceStopPipeline()   SubPlan.java:533-546 -> 413
    -> jobMaster.neverNeedRestore()   SubPlan.java:414
    -> PhysicalVertex.forceStop() for each coordinator/physical vertex   SubPlan.java:415-416
      -> getExecutionState() null/isEndState() -> return   PhysicalVertex.java:554-557
      -> [CHANGED] if (!isRunning) startPhysicalVertex()  (isRunning = true)   PhysicalVertex.java:558-560, 583-586
      -> noticeTaskExecutionServiceCancel()   PhysicalVertex.java:561 -> 415
         -> checkTaskGroupIsExecuting() false (no slot / not deployed) -> updateTaskState(CANCELED)   PhysicalVertex.java:418-420
      -> if (!taskFuture.isDone()) updateTaskState(CANCELED)   PhysicalVertex.java:562-564
        -> runningJobStateIMap.get(loc) null / equals target / isEndState -> return (isRunning stays true, future pending)   PhysicalVertex.java:353-379
        -> runningJobStateIMap.set(loc, CANCELED); stateProcess()   PhysicalVertex.java:383-400
          -> if (!isRunning) return   PhysicalVertex.java:594-596   <-- previously bailed here
          -> case CANCELED: stopPhysicalVertex(); taskFuture.complete(CANCELED)   PhysicalVertex.java:619-626
            -> SubPlan.addPhysicalVertexCallBack thenAcceptAsync: canceledTaskNum++, finishedTaskNum++ == total -> updatePipelineState(getPipelineEndState())   SubPlan.java:199-230
              (restore-wait case: the thread in SubPlan.stateProcess() FAILED/CANCELED -> prepareRestorePipeline() holds the SubPlan monitor and restoreLock through reset()/Thread.sleep(), then returns true -> releasePipelineResource / preApplyResources / restorePipeline() -> startSubPlanStateProcess() walks CREATED -> SCHEDULED -> DEPLOYING -> RUNNING with no CREATED task to deploy; only then does the blocked callback move the pipeline RUNNING -> CANCELED and end it)   SubPlan.java:489-515, 518-531, 648-652, 666-749

Comparison path this mirrors: SubPlan.stateProcess() FAILING/CANCELING -> task.startPhysicalVertex(); task.cancel()   SubPlan.java:723-735 -> PhysicalVertex.cancel() -> updateTaskState(CANCELING)   PhysicalVertex.java:409-413

Findings

Issue 1: Completing the task futures while SubPlan.prepareRestorePipeline() is sleeping does not stop the restore thread: it still releases/re-applies slots and restarts the state process for a pipeline whose tasks are all already CANCELED, because this base has no post-sleep isNeedRestore()/end-state re-check

• Location: seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/dag/physical/PhysicalVertex.java:558
• Why it matters: - Root cause: in the #12511 scenario this PR targets, forceStopPipeline() is invoked while SubPlan.prepareRestorePipeline() is inside Thread.sleep(pipelineRestoreIntervalSeconds * 1000) (SubPlan.java:504) with the tasks reset to CREATED and fresh futures already registered via addPipelineEndCallback. With startPhysicalVertex() now called from forceStop() (PhysicalVertex.java:558-559), every task future completes CANCELED and addPhysicalVertexCallBack drives finishedTaskNum to the task count and calls updatePipelineState(CANCELED).
• That call blocks: the restore thread entered prepareRestorePipeline() from the synchronized SubPlan.stateProcess() and holds the SubPlan monitor through the sleep, and updatePipelineState is synchronized too. When the sleep ends, prepareRestorePipeline() returns true unconditionally and stateProcess() continues with jobMaster.releasePipelineResource(this), jobMaster.preApplyResources(this) and restorePipeline() -> startSubPlanStateProcess() (SubPlan.java:739-743), which walks CREATED -> SCHEDULED -> DEPLOYING -> RUNNING: slots are applied in SCHEDULED, the DEPLOYING branch finds no CREATED task to deploy because they are all CANCELED, and the pipeline is reported RUNNING. Only after the monitor is released does the pending callback move it RUNNING -> CANCELED; checkNeedRestore is then false because neverNeedRestore() was already called, so the pipeline finally ends.
• Impact: on exactly the savepoint-failure path this PR is fixing, the Zeta master performs a full slot release/re-apply round-trip and a transient RUNNING transition for a pipeline that is already dead, delaying the job's end by the restore interval plus the slot allocation. This base does not contain the #12511 post-sleep isNeedRestore() re-check, so the PR description's claim that the two changes are independent holds only at the file level -- the clean behaviour needs both.
• Fix: either land this together with #12511's re-check, or make the restore path defensive on its own: after the sleep in prepareRestorePipeline() (and at the top of restorePipeline()), return early when !jobMaster.isNeedRestore() or every task is already in an end state, so the FAILED/CANCELED branch falls through to subPlanDone/pipelineFuture.complete instead of re-applying resources. Please also extend SavePointTest/SavePointBusySourceTest with the force-stop-during-restore-wait case so the end-to-end job status (CANCELED, not DOING_SAVEPOINT) is asserted.
• Evidence: seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/dag/physical/PhysicalVertex.java:559 startPhysicalVertex();; seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/dag/physical/PhysicalVertex.java:548 * <p>A task that was reset for a pipeline restore, or that was never deployed, has its state; seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/dag/physical/PhysicalVertex.java:549 * process stopped. Moving it to {@code CANCELED} would then leave its future pending, and its
• Suggested fix: This makes the task futures complete while SubPlan.prepareRestorePipeline() is still inside its Thread.sleep(...). The restore thread holds the SubPlan monitor through that sleep, so the callback's updatePipelineState(CANCELED) has to wait; meanwhile prepareRestorePipeline() returns true regardless and the thread goes on to releasePipelineResource / preApplyResources(this) / restorePipeline() -> startSubPlanStateProcess(), driving a pipeline whose tasks are all CANCELED through SCHEDULED (slot allocation) and DEPLOYING (nothing to deploy) to RUNNING before the callback can end it. The base here has no post-sleep isNeedRestore() re-check, so this change is not independent of #12511 in practice.

Could we either land both together or add a guard in SubPlan (after the sleep and at the top of restorePipeline(): bail out when !jobMaster.isNeedRestore() or all tasks are already terminal)? An assertion in SavePointTest that the job ends CANCELED after a force stop during the restore wait would pin the behavior.

• Severity: Medium

Issue 2: The new test covers a bare vertex in CREATED with no SubPlan or JobMaster; it does not exercise the #12622 scenario (pipeline in restore wait, forceStopPipeline(), job leaving DOING_SAVEPOINT), so the fix's end-to-end effect is unverified

• Location: seatunnel-engine/seatunnel-engine-server/src/test/java/org/apache/seatunnel/engine/server/dag/physical/StateTransitionCleanupTest.java:96
• Why it matters: - The test picks a vertex from a fresh PhysicalPlan, where isRunning is false simply because nothing ever started it. The Javadoc added to forceStop() names the reset-for-restore case first (PhysicalVertex.reset() after an end state), which is the path #12622 actually hits, and that path is not exercised: no test drives a vertex to a terminal state, calls reset(), then forceStop().
• Nothing asserts the pipeline-level outcome. The regression in #12622 is a pipeline that stays RUNNING and a job stuck in DOING_SAVEPOINT; SavePointTest/SavePointBusySourceTest already exist on dev where the bug is present, so listing them as green does not demonstrate the end-to-end fix. A test that reaches the FAILED/CANCELED branch of SubPlan.stateProcess() with a restore interval, then calls forceStopPipeline() during the wait and asserts the pipeline future completes, would pin the behaviour this PR is for.
• The test also does not assert that isRunning is back to false after completion (the CANCELED branch of stateProcess() calls stopPhysicalVertex()), which is the cheapest way to catch the isRunning leak described in the Javadoc finding.
• Fix: extend StateTransitionCleanupTest with a reset-then-forceStop case (move the vertex to FAILED via updateStateByExecutionService, call reset(), then forceStop()), add Assertions.assertFalse(physicalVertex.isRunning) after the future completes, and add a SubPlan-level case for the restore-wait stop.
• Evidence: seatunnel-engine/seatunnel-engine-server/src/test/java/org/apache/seatunnel/engine/server/dag/physical/StateTransitionCleanupTest.java:96 physicalVertex.forceStop();; seatunnel-engine/seatunnel-engine-server/src/test/java/org/apache/seatunnel/engine/server/dag/physical/StateTransitionCleanupTest.java:99 ExecutionState.CANCELED, taskFuture.get(10, TimeUnit.SECONDS).getExecutionState());; seatunnel-engine/seatunnel-engine-server/src/test/java/org/apache/seatunnel/engine/server/dag/physical/StateTransitionCleanupTest.java:96 physicalVertex.forceStop();
• Suggested fix: This verifies the vertex-level contract, which is good, but the bug in #12622 is observed at the pipeline/job level: SubPlan never counting the task as finished and the job staying in DOING_SAVEPOINT. Could you add a case that goes through SubPlan.forceStopPipeline() on a reset pipeline (tasks CREATED, state process stopped) and asserts the pipeline future from subPlan.initStateFuture() completes CANCELED? That is the path the production change is meant to unblock, and it is where the monitor ordering with prepareRestorePipeline() could regress unnoticed. An assertFalse(physicalVertex.isRunning) after the future completes here would also be cheap and useful.
• Severity: Medium

Issue 3: The new forceStop() Javadoc overstates the contract: the method can still block on the worker cancel round-trip and can still leave the future pending (with isRunning now flipped to true)

• Location: seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/dag/physical/PhysicalVertex.java:545
• Why it matters: - The Javadoc says the method "stops this task without waiting for it to report a terminal state, and completes its future with CANCELED". Neither clause is unconditional in the code below it.
• noticeTaskExecutionServiceCancel() (PhysicalVertex.java:426-456) issues a synchronous CancelTaskOperation via invoke().get() whenever the worker reports the task group as executing, and on failure retries with a 2 s sleep for as long as !taskFuture.isDone() and the member is still present. For a deployed task, forceStop() therefore blocks on the worker (potentially for several retry rounds) rather than returning immediately.
• If updateTaskState(CANCELED) returns early because runningJobStateIMap.get(taskGroupLocation) is null (state map already cleaned, PhysicalVertex.java:354-361) or already terminal (PhysicalVertex.java:376-379), the future is never completed; with this PR the vertex is additionally left with isRunning == true in a non-terminal state, which the Javadoc does not mention.
• Fix: reword the Javadoc to say the task is cancelled on the worker if it is executing (a blocking RPC with retries), and that the future is completed with CANCELED only when the state transition is still recorded in runningJobStateIMap; optionally guard the startPhysicalVertex() call so isRunning is restored when updateTaskState does not reach the CANCELED branch (e.g. check taskFuture.isDone() afterwards and call stopPhysicalVertex() if not).
• Evidence: seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/dag/physical/PhysicalVertex.java:546 * with {@code CANCELED}.; seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/dag/physical/PhysicalVertex.java:559 startPhysicalVertex();
• Suggested fix: Small accuracy point on the Javadoc: noticeTaskExecutionServiceCancel() below does a synchronous CancelTaskOperation invoke().get() when the worker says the task group is executing, and retries every 2 s on failure while !taskFuture.isDone(), so for a deployed task this method can block on the worker. And if updateTaskState(CANCELED) bails out on a null or terminal runningJobStateIMap entry, the future stays pending and, with the new startPhysicalVertex() call, isRunning is left true. Could you reword this to describe the actual contract (cancel on the worker if executing; complete with CANCELED when the state is still tracked), and consider resetting isRunning if the future is still not done after updateTaskState?
• Severity: Low

Issue 4: Javadoc cross-reference to SubPlan uses {@code} instead of {@link}; the referenced behaviour lives in the same package and should be navigable

• Location: seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/dag/physical/PhysicalVertex.java:550
• Why it matters: - The new Javadoc on forceStop() explains the fix by pointing at SubPlan's FAILING/CANCELING branch, which is exactly the context a future reader needs (SubPlan.java:725-735 calls task.startPhysicalVertex() before task.cancel()).
• SubPlan is in the same package as PhysicalVertex, so {@link SubPlan} resolves without an import and gives IDE/Javadoc navigation; {@code SubPlan} renders as plain monospace text. The taskFuture field Javadoc in PhysicalVertex already uses {@link SubPlan} for the same purpose (PhysicalVertex.java:104), so this would also keep the file consistent.
• Suggest ... as {@link SubPlan} does before cancelling its tasks.
• Suggested fix: Nit: SubPlan is in this package, so {@link SubPlan} would make the reference navigable and matches the {@link SubPlan} already used in the taskFuture field Javadoc above. The explanation itself is helpful, thanks for adding it.
• Severity: Low

Review conclusion

Conclusion: final approval remains pending on the previous comments

2. Remaining follow-ups

• Issue 1: Completing the task futures while SubPlan.prepareRestorePipeline() is sleeping does not stop the restore thread: it still releases/re-applies slots and restarts the state process for a pipeline whose tasks are all already CANCELED, because this base has no post-sleep isNeedRestore()/end-state re-check
• Issue 2: The new test covers a bare vertex in CREATED with no SubPlan or JobMaster; it does not exercise the #12622 scenario (pipeline in restore wait, forceStopPipeline(), job leaving DOING_SAVEPOINT), so the fix's end-to-end effect is unverified
• Issue 3: The new forceStop() Javadoc overstates the contract: the method can still block on the worker cancel round-trip and can still leave the future pending (with isRunning now flipped to true)
• Issue 4: Javadoc cross-reference to SubPlan uses {@code} instead of {@link}; the referenced behaviour lives in the same package and should be navigable

Thanks for the updates so far. Happy to discuss the remaining points.

Thanks for the follow-up. I checked the latest changes at f24ed531ed78 against my previous review comments.

The scope of this pass is limited to the findings from the previous review; it does not introduce a new set of review items.

Resolved previous findings

• PR12623-F3 The new forceStop() Javadoc overstates the contract: the method can still block on the worker cancel round-trip and can still leave the future pending (with isRunning now flipped to true)
• PR12623-F4 Javadoc cross-reference to SubPlan uses {@code} instead of {@link}; the referenced behaviour lives in the same package and should be navigable

Previous findings still requiring follow-up

Issue 1: PR12623-F1: Completing the task futures while SubPlan.prepareRestorePipeline() is sleeping does not stop the restore thread: it still releases/re-applies slots and restarts the state process for a pipeline whose tasks are all already CANCELED, because this base has no post-sleep isNeedRestore()/end-state re-check

• Location: seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/dag/physical/PhysicalVertex.java:558
• Why it matters: - Still present on this base; acknowledged in the reply and deferred to #12511.
• Mechanism: in the savepoint-failure scenario this PR targets, forceStopPipeline() runs while SubPlan.prepareRestorePipeline() is inside Thread.sleep(pipelineRestoreIntervalSeconds * 1000) (SubPlan.java:504), after reset() has put the tasks back in CREATED with their state process stopped and fresh futures registered via addPipelineEndCallback. forceStop() now starts the state process when isRunning is false before cancelling (PhysicalVertex.java:563-566), so every task future completes CANCELED and addPhysicalVertexCallBack drives finishedTaskNum to the task count and calls updatePipelineState(CANCELED).
• That call blocks: the restore thread entered prepareRestorePipeline() from the synchronized stateProcess() and holds the SubPlan monitor through the sleep, and updatePipelineState is synchronized too. When the sleep ends, prepareRestorePipeline() returns true unconditionally (SubPlan.java:505) and stateProcess() continues with jobMaster.releasePipelineResource(this), jobMaster.preApplyResources(this) and restorePipeline() -> startSubPlanStateProcess() (SubPlan.java:739-743). The pipeline walks CREATED -> SCHEDULED -> DEPLOYING -> RUNNING: slots are applied in SCHEDULED, DEPLOYING finds no CREATED task because they are all CANCELED, and the pipeline is reported RUNNING. Only after the monitor is released does the pending callback move it RUNNING -> CANCELED; checkNeedRestore is then false because neverNeedRestore() was already called, so the pipeline finally ends.
• Impact: on exactly the path this PR fixes, the master still performs a full slot release/re-apply round-trip and a transient RUNNING transition for a pipeline that is already dead, delaying the job's end by the restore interval plus slot allocation. The job does end rather than staying stuck, so the change is an improvement on its own, but the clean behaviour requires the post-sleep re-check you describe as living in #12511. This PR therefore depends on #12511 in practice, which the PR description does not yet state.
• Ask: record the merge-order dependency on #12511 explicitly in the PR description (replacing the "independent" wording, as you offered), or, if this PR may land first, add a minimal guard in SubPlan: after the sleep in prepareRestorePipeline() and at the top of restorePipeline(), return early when !jobMaster.isNeedRestore() or every task is already in an end state, so the FAILED/CANCELED branch falls through to subPlanDone/pipelineFuture.complete instead of re-applying resources. Either way, the end-to-end assertion (job ends CANCELED after a force stop during the restore wait) should be referenced from this PR so reviewers can find it.
• Evidence: seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/dag/physical/PhysicalVertex.java:559 startPhysicalVertex();; seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/dag/physical/PhysicalVertex.java:548 * <p>A task that was reset for a pipeline restore, or that was never deployed, has its state; seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/dag/physical/PhysicalVertex.java:549 * process stopped. Moving it to {@code CANCELED} would then leave its future pending, and its
• Suggested fix: Thanks for confirming. With this change the task futures complete while SubPlan.prepareRestorePipeline() is still inside its Thread.sleep(...). The restore thread holds the SubPlan monitor through that sleep, so the callback's updatePipelineState(CANCELED) has to wait; meanwhile prepareRestorePipeline() returns true regardless and the thread goes on to releasePipelineResource / preApplyResources(this) / restorePipeline() -> startSubPlanStateProcess(), driving a pipeline whose tasks are all CANCELED through SCHEDULED (slot allocation) and DEPLOYING (nothing to deploy) to RUNNING before the callback can end it.

Since the post-sleep re-check lives in #12511, this PR has a real merge-order dependency on it. Could you state that in the PR description (and drop the "independent" wording), or add a small guard here (after the sleep and at the top of restorePipeline(): bail out when !jobMaster.isNeedRestore() or all tasks are already terminal) so the PR is safe to land in either order?

• Severity: Medium

Issue 2: PR12623-F2: The new test covers a bare vertex in CREATED with no SubPlan or JobMaster; it does not exercise the #12622 scenario (pipeline in restore wait, forceStopPipeline(), job leaving DOING_SAVEPOINT), so the fix's end-to-end effect is unverified

• Location: seatunnel-engine/seatunnel-engine-server/src/test/java/org/apache/seatunnel/engine/server/dag/physical/StateTransitionCleanupTest.java:96
• Why it matters: - Partially addressed. testForceStopCompletesTheFutureOfATaskResetForRestore (StateTransitionCleanupTest.java:108-132) now drives the vertex to FAILED via updateStateByExecutionService, calls reset(), then forceStop(), which is the reset-for-restore path the Javadoc names first. Both force-stop tests now assert physicalVertex.isRunning is false after the future completes (lines 103 and 131), and testForceStopLeavesTheStateProcessStoppedWhenTheTransitionIsSkipped (lines 139-157) covers the skipped-transition branch.
• Still open: nothing in this PR asserts the pipeline-level outcome. The regression in #12622 is a pipeline that stays RUNNING and a job stuck in DOING_SAVEPOINT; the existing SavePointTest/SavePointBusySourceTest pass on dev where the bug is present, so they do not demonstrate the end-to-end fix. The reply defers that case to a test in #12511, which means this PR on its own has no assertion that SubPlan.forceStopPipeline() on a reset pipeline completes the pipeline future, and the monitor ordering with prepareRestorePipeline() (F1) could regress unnoticed if the PRs land in a different order.
• Ask: add a SubPlan-level case here that reaches the FAILED/CANCELED branch of stateProcess() with a restore interval, calls forceStopPipeline() during the wait, and asserts the future from subPlan.initStateFuture() completes CANCELED; or, if you keep it in #12511, link that test from the PR description and make the merge order explicit.
• Evidence: seatunnel-engine/seatunnel-engine-server/src/test/java/org/apache/seatunnel/engine/server/dag/physical/StateTransitionCleanupTest.java:96 physicalVertex.forceStop();; seatunnel-engine/seatunnel-engine-server/src/test/java/org/apache/seatunnel/engine/server/dag/physical/StateTransitionCleanupTest.java:99 ExecutionState.CANCELED, taskFuture.get(10, TimeUnit.SECONDS).getExecutionState());; seatunnel-engine/seatunnel-engine-server/src/test/java/org/apache/seatunnel/engine/server/dag/physical/StateTransitionCleanupTest.java:96 physicalVertex.forceStop();
• Suggested fix: Thanks for adding the reset-then-forceStop() case and the isRunning assertions; those cover the vertex-level contract well. The remaining gap is that nothing in this PR observes the pipeline/job level, which is where #12622 shows up (SubPlan never counting the task as finished, job staying in DOING_SAVEPOINT). If the pipeline-level case stays in #12511, could you link it from the PR description and make the merge order explicit? Otherwise a case that goes through SubPlan.forceStopPipeline() on a reset pipeline and asserts the future from subPlan.initStateFuture() completes CANCELED would pin the behaviour this change is meant to unblock.
• Severity: Medium

Review conclusion

Conclusion: final approval remains pending on the previous comments

2. Remaining follow-ups

• Issue 1: Completing the task futures while SubPlan.prepareRestorePipeline() is sleeping does not stop the restore thread: it still releases/re-applies slots and restarts the state process for a pipeline whose tasks are all already CANCELED, because this base has no post-sleep isNeedRestore()/end-state re-check
• Issue 2: The new test covers a bare vertex in CREATED with no SubPlan or JobMaster; it does not exercise the #12622 scenario (pipeline in restore wait, forceStopPipeline(), job leaving DOING_SAVEPOINT), so the fix's end-to-end effect is unverified

Thanks for the updates so far. Happy to discuss the remaining points.

Hi @Vivek1106-04, thank you for the detailed fix and for the careful write-up of how it relates to #12511. I reviewed head f24ed531e by tracing PhysicalVertex, SubPlan and PhysicalPlan on this branch and on dev, and I read the earlier review thread so I do not repeat it. This is a static review; I did not run the PR.

What problem does this PR solve

• Pain point: PhysicalVertex.forceStop() on a task whose state process is stopped (never deployed, or reset for a pipeline restore) moved the task to CANCELED but never completed taskFuture, because updateTaskState(CANCELED) -> stateProcess() returns at the if (!isRunning) check (PhysicalVertex.java:603) before the CANCELED branch that completes the future. SubPlan counts finished tasks off those futures, so the pipeline never ended and, in the savepoint scenario of #12511, the job stayed in DOING_SAVEPOINT.
• Approach: forceStop() now starts the state process first when it is not running (mirroring the startPhysicalVertex(); cancel(); pairing in SubPlan.stateProcess() for FAILING/CANCELING), and stops it again if the transition was skipped and the future is still pending.
• One sentence: a force-stopped task that is not running now completes its future with CANCELED so its pipeline can end.

1. Code change review

Core logic (before / after)

before: forceStop(): noticeTaskExecutionServiceCancel(); if (!future.isDone()) updateTaskState(CANCELED);
        -> stateProcess() returns early when isRunning == false -> future stays pending -> pipeline never ends
after : boolean started = !isRunning; if (started) startPhysicalVertex();
        noticeTaskExecutionServiceCancel(); if (!future.isDone()) updateTaskState(CANCELED);
        if (started && !future.isDone()) stopPhysicalVertex();

Runtime path (checked in source)

PhysicalPlan.cancelJob / savepoint failure -> SubPlan.forceStopPipeline()            SubPlan.java:413
  jobMaster.neverNeedRestore()
  -> PhysicalVertex.forceStop() for each coordinator / physical vertex               PhysicalVertex.java:558
       state null or end state -> return
       [NEW] if !isRunning -> startPhysicalVertex()
       noticeTaskExecutionServiceCancel(): not executing on a worker -> updateTaskState(CANCELED)
         updateTaskState: IMap set CANCELED -> stateProcess() -> case CANCELED: stopPhysicalVertex(); taskFuture.complete(CANCELED)
           -> SubPlan.addPhysicalVertexCallBack: canceledTaskNum++, finishedTaskNum == total -> updatePipelineState(CANCELED)

I traced the three reachable shapes of a non-running vertex and they all behave:

1. Never deployed (CREATED, state process stopped): the transition runs, the future completes CANCELED, and stateProcess() itself calls stopPhysicalVertex(), so isRunning ends false (asserted in test 1).
2. Reset for restore (reset() sets CREATED after an end state): same as above (test 2).
3. State entry already gone from the running-state IMap (job cleanup): updateTaskState returns early, the future is still pending, and the new trailing stopPhysicalVertex() puts isRunning back to what it was (test 3).

An already running task takes exactly the old path (started == false), so the common cancel path is unchanged.

Compatibility: fully compatible. No Hazelcast IMap value, serialization, config or protocol change; the only differences are an in-memory flag toggled under the vertex monitor and an earlier future completion for tasks that would otherwise hang.

Performance / side effects: none on the steady-state path. The side effect on the restore path is discussed in issue 1.

Error handling / logging: reuses the existing startPhysicalVertex() / stopPhysicalVertex() log lines. The Javadoc is accurate about the blocking cancel round trip and about when the future completes.

Issues found

1. Medium (sequencing, blocks merge order but not the code itself) - the fix is only clean once #12511 is in.
   Location: SubPlan.java:489-515 (prepareRestorePipeline, Thread.sleep(pipelineRestoreIntervalSeconds * 1000) while holding the SubPlan monitor) and SubPlan.java:739-743 (FAILED/CANCELED branch continues with releasePipelineResource, preApplyResources, restorePipeline).
   Problem: with this PR alone, a force stop during the restore wait completes the task futures, but updatePipelineState in the callback is synchronized and waits for the restore thread. When the sleep ends, the restore thread still releases and re-applies slots and walks the pipeline to a transient RUNNING, and only then does the pending callback move it to CANCELED. The same shape exists if the stop lands while the pipeline is in SCHEDULED (resource application) before the tasks are deployed.
   Risk: the job now ends instead of hanging, so this PR is strictly better than dev, but the end of the job is delayed by the restore interval plus a slot allocation round trip for a pipeline that is already dead. #12511 (abandonRestore re-check) removes the round trip for the restore-wait case. I agree with the author and the previous reviewer that duplicating that guard here would make whichever PR lands second remove it.
   Suggested fix: land #12511 first (it is currently open and BLOCKED), then rebase this PR and re-run SavePointTest, CheckpointErrorRestoreEndTest and StateTransitionCleanupTest together on the combined base. The PR description already states the dependency; please also keep the link to the two pipeline-level tests in #12511.

2. Low - the rollback in forceStop() relies on an unstated invariant.
   Location: PhysicalVertex.java:563-573.
   Problem: if the task were actually executing on a worker while isRunning is false, the trailing stopPhysicalVertex() would make stateProcess() drop the worker's later terminal report (stateProcess returns at the isRunning check) and the future could stay pending. By the deploy and restore code this combination does not occur: every deploy path calls startPhysicalVertex() first, restoreExecutionState() starts it, and a reset task has no deployment.
   Risk: none today; it would matter only if someone later introduces a deploy path that does not set the flag.
   Suggested fix: add a one-line code comment next to startedStateProcess stating "not running implies not deployed on a worker", so the rollback is understood as safe.

2. Code quality

• Standards: the Javadoc on forceStop() explains the contract, the blocking cancel round trip and the early return, and now uses {@link SubPlan}. The diff is small and focused (one production method, one test class).
• Tests (stability rating: Stable): three tests in StateTransitionCleanupTest use a real embedded Hazelcast instance via AbstractSeaTunnelServerTest, with taskFuture.get(10, TimeUnit.SECONDS) for the asynchronous part. No sleeps, no fixed ports of their own.
  • testForceStopCompletesTheFutureOfATaskWhoseStateProcessIsStopped and ...ResetForRestore: both fail on dev with a timeout (the future never completes), as the PR description shows; they would also fail if startPhysicalVertex() were removed from the new code. They also assert isRunning is false afterwards.
  • testForceStopLeavesTheStateProcessStoppedWhenTheTransitionIsSkipped: removes the IMap entry, so updateTaskState returns early; it passes only if the trailing stopPhysicalVertex() exists (the cached currExecutionState stays CREATED, so the method does enter the new path rather than returning early).
  • Environment note only: the base class bootstraps Hazelcast, which has a known node-start flake in some CI legs. That is infrastructure, not test logic.
  • There is no pipeline-level test in this PR by design. I agree with the reasoning: after #12511 a stop during the restore wait ends through its re-check regardless of this fix, so such a test could not isolate this change. The vertex-level contract tests are the right guard here.
• Docs: no user-visible change, so no docs/en / docs/zh update is required.

3. Architecture

• The fix is placed at the narrowest correct point: the one method whose contract ("move to CANCELED and complete the future") is violated for a stopped vertex. A change in stateProcess() itself would affect every state transition and is much riskier, so I agree with the placement.
• Maintainability: it mirrors the existing SubPlan pairing, and the rollback keeps forceStop() free of lasting flag changes.
• Historical compatibility: persistent state is untouched, so job recovery, savepoint and checkpoint formats are not affected.

4. Issue summary

5. Merge recommendation

Ready to merge after fixes: land #12511 first, rebase onto it, re-run the engine savepoint tests on the combined base, and optionally add the one-line invariant comment. The code change itself looks correct to me.

• Merge state is BLOCKED (review required); the branch is 36 commits behind dev and has no conflicts. #12511 touches SubPlan.java and checkpoint classes but not PhysicalVertex.java, so a textual conflict is not expected.
• CI diagnosis (fork run 37265238239, head f24ed531e): the Build workflow is green for this head (81 passed, 13 skipped), but it is run attempt 3, so earlier attempts failed. The author reports those were BackpressureSlowSinkIT, RocketMqIT, PostgresCDCIT and DatabendCDCSinkIT, none of which goes through forceStop(). CI is therefore not a blocker; the only open item is the merge ordering above.

Thanks again for the very thorough work on this one.

Thanks for the upstream sync and for making the #12511 merge order explicit. I re-read the full current diff and the task-to-pipeline completion chain. The vertex fix is sound, and the two previously discussed pipeline-level points must still be handled as a coordinated dependency rather than being hidden by green CI.

What Problem Does This PR Solve?

User pain point: A force-stopped task whose state processing is paused can enter CANCELED without completing its task future; the pipeline then never counts it as finished.

Fix approach: Start the vertex state process before cancellation when necessary, and restore the stopped flag when no terminal transition can be made.

One-sentence summary: A task reset for recovery, currently CREATED with isRunning=false, now completes its future as CANCELED when forceStop succeeds; a task removed from the state map remains stopped.

Review report time: 2026-10-29T11:26:50.631398-04:00. Current head caa5f098fc0a5c6dd887ba8632dc43d463eb4580; base dev at f7a575432fba5c35d36659c16e1590e83ad1f296. Checks: Signal pull request update=SUCCESS; Build=SUCCESS. Check references: Signal pull request update Build. Compare: diverged, ahead 3, behind 12; merge state blocked / BLOCKED.

1. Code Change Review

1.1 Core Logic Analysis

seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/dag/physical/PhysicalVertex.java:573-588 records whether it started processing, resumes before worker cancellation, requests CANCELED and stops again if the future is still unresolved. updateTaskState (:358-410) checks the distributed running-state map and invokes stateProcess; its terminal cases (:617-675) complete the future and stop processing. StateTransitionCleanupTest:170-247 covers never-started, reset and removed-state cases. All two changed files and the SubPlan/savepoint/restore callers were read.

Before:

noticeTaskExecutionServiceCancel();
if (!taskFuture.isDone()) { updateTaskState(ExecutionState.CANCELED); }

After:

boolean startedStateProcess = !isRunning;
if (startedStateProcess) { startPhysicalVertex(); }
noticeTaskExecutionServiceCancel();
if (!taskFuture.isDone()) { updateTaskState(ExecutionState.CANCELED); }
if (startedStateProcess && !taskFuture.isDone()) { stopPhysicalVertex(); }

Key Findings

• Normal already-running vertices keep their existing path; paused/reset/not-deployed vertices are the corrected boundary.
• The CANCELED state processor, not merely the state-map write, completes the future.
• The skipped-transition guard restores the stopped state, and the Javadoc now describes blocking worker RPC accurately.
• The current head still lacks the SubPlan post-sleep restore guard and #12511 remains open.

Complete runtime path

PhysicalPlan savepoint/cancel -> SubPlan stop fallback -> forceStopPipeline (:480-484)
 -> jobMaster.neverNeedRestore -> PhysicalVertex.forceStop (:573)
 -> start paused processor -> worker execution check / synchronous CancelTaskOperation
 -> updateTaskState(CANCELED) -> running-state map guard -> stateProcess (:617)
 -> stopPhysicalVertex -> taskFuture.complete(CANCELED)
 -> asynchronous SubPlan callback -> finished-task accounting -> pipeline/job future
Concurrent restore thread still needs #12511's post-wait guard before resource re-application.

The effects depend on the entry and preconditions above. Recovery, batching and connector lifecycle retain their existing contracts unless explicitly described here; source inspection is not runtime proof.

1.2 Compatibility Impact

Fully compatible. No API signature, configuration/default, enum ordinal, serialized task/checkpoint format or savepoint protocol changes. Correct terminal-future completion is a recovery/cancellation lifecycle correction. Existing live-worker cancellation remains synchronous and can retry; it is not converted into a new nonblocking RPC contract.

1.3 Performance / Side-Effect Analysis

Only the paused-processor path adds two boolean/lifecycle operations and existing INFO lifecycle messages. No executor, thread, timer, cache, queue or network stage is introduced. forceStop remains synchronized and retains the pre-existing worker RPC/retry under its monitor. The separate SubPlan restore resource round-trip remains a dependency gate.

1.4 Error Handling and Logging

Removed/untracked/terminal map transitions can still be skipped; the method now re-stops a processor it started when its future did not finish. Worker cancellation exceptions/retry ownership are unchanged. No new swallowed exception, credential logging or falsely completed absent state is introduced.

No additional numbered Daniel issue. Existing reviewer blockers remain: +1 to @SEZ9 PR12623-F1: #12511 must supply the restore-wait guard before this PR is merged; #12511 is still OPEN at a691fb1; +1 to @SEZ9 PR12623-F2: verify the named pipeline/job restore-wait tests in the prerequisite and retain a test for this PR's post-recheck paused-vertex window; bare vertex completion alone is not a whole-job proof

2. Code Quality Assessment

2.1 Coding Standards

The complete changed-file set, imports and removed lines were inspected. No new option, protocol field, public API or connector registration is added. The important behavior and its limits are explained above; test-boundary documentation should explain the regression it protects.

2.2 Test Coverage and Test Stability

Stability rating: Stable. StateTransitionCleanupTest.java:170-247 uses a real class-owned Hazelcast state-map fixture, isolated generated job IDs, bounded future.get waits, exact terminal/map values and isRunning assertions. No fixed sleep, order-dependent collection result or new port is added. The reset and removed-state cases improve vertex coverage. Existing whole-job tests passing on the base do not prove the affected restore-wait race. No new engine E2E is present; the named #12511 tests and merge ordering remain explicit gates.

Local verification results: exact-head git fetch and isolated review branch seatunnel-review-12623 created; full git diff f7a575432fba5c35d36659c16e1590e83ad1f296...caa5f098fc0a5c6dd887ba8632dc43d463eb4580 read (2 files); git diff --check passed; git show, rg and line-numbered source/caller/config/test reads used for the trace above. GitHub changed filenames are reconciled with this local diff before publication. No local compile, Maven verify, unit/IT/E2E, Docker, service or UI execution occurred, as required for this review task. No build/deployment artifact was reused; engine-ui was not built.

2.3 Documentation Updates

No new user configuration is introduced. The method Javadoc correctly qualifies worker blocking and tracked-state preconditions. The PR description now states merge after #12511 and names SavePointTest#testSavePointFailureDuringPipelineRestoreWaitEndsTheJob and CheckpointErrorRestoreEndTest#testCancelDuringPipelineRestoreWaitEndsTheJobCanceled.

3. Architectural Soundness

3.1 Elegance of the Solution

Precise fix.

3.2 Maintainability

Best change-location assessment: The task future must be completed through PhysicalVertex's state processor; directly completing it in SubPlan would duplicate state bookkeeping. The smallest safe fix is this processor guard plus the coordinated SubPlan prerequisite/test proof. The broader restore-thread cancellation policy belongs in #12511/SubPlan, not an unrelated worker redesign.

3.3 Extensibility

Keep existing connector identities and contracts. Reuse the current boundary and regression fixtures; no shared-framework redesign or unrelated dependency is needed for the behavior reviewed here.

3.4 Historical-Version Compatibility

No API signature, configuration/default, enum ordinal, serialized task/checkpoint format or savepoint protocol changes. Correct terminal-future completion is a recovery/cancellation lifecycle correction. Existing live-worker cancellation remains synchronous and can retry; it is not converted into a new nonblocking RPC contract. This review does not claim an executed upgrade/recovery validation.

4. Issue Summary

Completeness check: full diff and affected entry/call chain, failure/resource paths, compatibility, tests and stability, bilingual docs, configuration/registration and current CI/merge gates were inspected before enumerating these findings. No additional finding is reserved for a later round on this version.

Community review reconciliation: @SEZ9 reviews #5409636066 and #5423011679 already raised PR12623-F1/F2. I agree and do not duplicate them as new numbered Daniel issues. Their F3/F4 (overstated Javadoc/link and processor-state cleanup) are addressed. The author's description now records the dependency, but live #12511 is OPEN; current-head Build SUCCESS cannot substitute for that code/test prerequisite.

5. Merge Recommendation

Conclusion: Ready to merge after fixes

1. Blockers — must be fixed: Existing reviewer/merge-order blockers: +1 to @SEZ9 PR12623-F1: #12511 must supply the restore-wait guard before this PR is merged; #12511 is still OPEN at a691fb1; +1 to @SEZ9 PR12623-F2: verify the named pipeline/job restore-wait tests in the prerequisite and retain a test for this PR's post-recheck paused-vertex window; bare vertex completion alone is not a whole-job proof. Required current-head CI and effective maintainer review/merge gates must be satisfied.
2. Recommended fixes — non-blocking: No additional unrelated redesign is requested.

Overall assessment: The task future must be completed through PhysicalVertex's state processor; directly completing it in SubPlan would duplicate state bookkeeping. The smallest safe fix is this processor guard plus the coordinated SubPlan prerequisite/test proof. The broader restore-thread cancellation policy belongs in #12511/SubPlan, not an unrelated worker redesign. Current head caa5f098fc0a5c6dd887ba8632dc43d463eb4580; base dev at f7a575432fba5c35d36659c16e1590e83ad1f296. Checks: Signal pull request update=SUCCESS; Build=SUCCESS. Check references: Signal pull request update Build. Compare: diverged, ahead 3, behind 12; merge state blocked / BLOCKED. Behind/diverged status alone is synchronization metadata and does not justify a sync-and-rerun instruction. Daniel has READ permission: this COMMENT is a technical conclusion, not an effective APPROVED event or dismissal of historical requested changes. A write-capable maintainer owns final approval/dismissal where required. Current-head CI is green. The remaining dependency/test-proof gate is non-CI, so this is not approval-ready and must not be cleared merely because Build recovered. Thanks again; happy to help with the focused next revision.

Thanks @Vivek1106-04 for the sync, the clear merge-order notes, and for posting the fork Build result. I re-checked everything against the unchanged head caa5f098fc0a5c6dd887ba8632dc43d463eb4580, and I am happy to say my two earlier carried-over points are now resolved. I'm switching this to an approval.

CI. The fork Build run 40727927696 is success for exactly this SHA (the only non-success jobs are the intentionally skipped integration/website/UI jobs), and the apache-side Build and Signal pull request update checks pass. So CI is no longer a blocker.

Verification of the claims

• PhysicalVertex.forceStop() (PhysicalVertex.java around lines 573-588): records startedStateProcess = !isRunning, starts the state process before noticeTaskExecutionServiceCancel()/updateTaskState(CANCELED), and calls stopPhysicalVertex() again only if it started the process and the future is still pending. This matches the description and the Javadoc, and the already-running path is unchanged.
• Earlier F1 (restore thread after the sleep): confirmed in #12511 at a691fb195, SubPlan.stateProcess() FAILED/CANCELED branch -- after prepareRestorePipeline() returns, if (!jobMaster.isNeedRestore()) { abandonRestore(state, failure); return; } runs before releasePipelineResource/preApplyResources/restorePipeline. #12511 is OPEN, APPROVED, Build SUCCESS, just not yet merged.
• Earlier F2 (pipeline-level proof): SavePointTest#testSavePointFailureDuringPipelineRestoreWaitEndsTheJob and CheckpointErrorRestoreEndTest#testCancelDuringPipelineRestoreWaitEndsTheJobCanceled exist at a691fb195. Your reasoning for keeping this PR's tests at the vertex level is sound: after #12511, a pipeline-level test would pass with or without this change, whereas testForceStopCompletesTheFutureOfATaskWhoseStateProcessIsStopped, testForceStopCompletesTheFutureOfATaskResetForRestore and testForceStopLeavesTheStateProcessStoppedWhenTheTransitionIsSkipped pin exactly the post-recheck window and fail if the PhysicalVertex change is reverted. They use bounded waits and no Thread.sleep.

Compatibility / risk. Fully compatible: no API, config, serialization or savepoint-format change. On its own this PR is already a strict improvement (the job ends instead of hanging in DOING_SAVEPOINT/CANCELING); with #12511 the extra restore round-trip disappears as well.

Issue summary: no open source-level issues.

Conclusion: Ready to merge

Non-blocking request for the committer: please land #12511 first, as you and @SEZ9 agreed, so the two changes are exercised together. Write-capable maintainer approval is still required for the actual merge. Thanks again for the careful, well-argued work!