Vivek1106-04 opened a new issue, #12622: URL: https://github.com/apache/seatunnel/issues/12622
### Search before asking - [X] I had searched in the [issues](https://github.com/apache/seatunnel/issues?q=is%3Aissue+label%3A%22bug%22) and found no similar issues. ### What happened `PhysicalVertex.forceStop()` leaves the task future pending when the task's state process is stopped. The pipeline then never sees that task end, and the pipeline and its job can stay in their current state for good. `forceStop()` returns early only for an end state. For any other state it calls `noticeTaskExecutionServiceCancel()`. For a task that is not executing anywhere, that calls `updateTaskState(CANCELED)`. `updateTaskState` writes the state and then calls `stateProcess()`, but `stateProcess()` returns at once when `isRunning` is false (`state process is not start`). The `CANCELED` branch that completes `taskFuture` therefore never runs. The trailing `if (!taskFuture.isDone()) updateTaskState(CANCELED)` cannot help either, because the state already equals the target and the call is skipped. A task's state process is stopped whenever it is not deployed: - **Pipeline restore wait.** After a task ends, `stopPhysicalVertex()` clears `isRunning`. `SubPlan.prepareRestorePipeline()` then resets the tasks to `CREATED`, calls `initStateFuture()` (new task futures), and sleeps `pipelineRestoreIntervalSeconds`. - **Before deployment.** A fresh vertex has `isRunning = false` until `SubPlan` reaches `DEPLOYING`. `forceStop()` is reached through `SubPlan.forceStopPipeline()`. The callers are `PhysicalPlan.stopJob()` (non-savepoint path) and `SubPlan.stopPipelineWithCheckpointFallback()`, which `PhysicalPlan.savepointFailed()` uses. **Observed case.** A pipeline fails while its job is in `DOING_SAVEPOINT` and enters its restore wait. `JobMaster.savepointFailed()` then force-stops it. A log trace on `dev` (from the reproduction in #12511): ``` ... SplitEnumerator (1/1) turned from state CREATED to CANCELED. ... SplitEnumerator (1/1) state process is not start ... AggregatedCommitterTask (1/1) turned from state CREATED to CANCELED. ... AggregatedCommitterTask (1/1) state process is not start ... SourceTask (1/1) future complete with state CANCELED <- only this one reaches SubPlan (2 s later, the restore wait ends) ... Pipeline: [(1/2)] turned from state CREATED to SCHEDULED. ... Pipeline: [(1/2)] turned from state SCHEDULED to DEPLOYING. <- deploys nothing: no task is CREATED ... Pipeline: [(1/2)] turned from state DEPLOYING to RUNNING. ``` The pipeline counted 1 of 3 tasks ended. It restarts, deploys nothing because `DEPLOYING` starts only `CREATED` tasks, and stays `RUNNING`. The job never leaves `DOING_SAVEPOINT`. #12511 avoids this for a stop decided during the restore wait: `SubPlan` re-checks `jobMaster.isNeedRestore()` after the wait and ends the pipeline instead of restarting it. That check is point-in-time, though. A stop decided after it, while the restarted pipeline is still applying resources and before its tasks are deployed, hits the same `forceStop()` gap. Any other force stop of a task that is not deployed does too. **Expected.** A force-stopped task always completes its future with `CANCELED`, so its pipeline can end. `SubPlan`'s own `CANCELING`/`FAILING` branch already does this: it calls `task.startPhysicalVertex()` before `task.cancel()`. ### SeaTunnel Version dev (`af80a6704`) ### SeaTunnel Config ```conf Not config specific. Reproduced with stream_two_pipelines_savepoint_fails_during_restore.conf from #12511 (one sink fails the savepoint while the other is still taking it, so the failed pipeline is force-stopped during its restore wait). ``` ### Running Command ```shell Unit level: build a PhysicalPlan, take a PhysicalVertex that has not been deployed, call initStateFuture(), then forceStop(). The returned future never completes. ``` ### Error Exception ```log No exception. The task future stays pending, and the pipeline (and its job) never reaches a terminal state. ``` ### Zeta or Flink or Spark Version Zeta (dev) ### Java or Scala Version JDK 17 (also applies to JDK 8; the code path is version independent) ### Screenshots _No response_ ### Are you willing to submit PR? - [X] Yes I am willing to submit a PR! ### Code of Conduct - [X] I agree to follow this project's [Code of Conduct](https://www.apache.org/foundation/policies/conduct) -- 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: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
