Vivek1106-04 commented on PR #12165:
URL: https://github.com/apache/seatunnel/pull/12165#issuecomment-5613175735
@DanielLeens Thanks for the very thorough review — tracing the
single-pending-checkpoint guard, the master-failover restore path and the MDC
propagation was exactly what I was hoping a reviewer would check on this one.
First, a CI update. You mentioned you'd look again once CI had run: it did,
and it caught a real problem in this PR that I've now fixed.
`CheckpointRestoreWithStopIT#testRestoreFromCheckpointAfterStop` and
`SavepointRestoreIT#testRestoreFromSavepoint` both failed with:
```
There are still threads running in the container:
"checkpoint-timer-1" ... ScheduledThreadPoolExecutor$DelayedWorkQueue.take
"checkpoint-timer-0" ... ScheduledThreadPoolExecutor$DelayedWorkQueue.take
==> expected: <true> but was: <false> within 2 minutes.
```
Root cause is a direct consequence of the change this PR makes.
`SeaTunnelContainer#doExecuteJob` asserts that no non-system thread outlives a
finished job. Before this PR the timer pool was per-coordinator
(`checkpoint-coordinator-<pipelineId>/<jobId>`) and was shut down when the job
ended, so it never tripped that check. Now the pools are owned by
`SeaTunnelEngineContext` and torn down only at member shutdown, so the first
job executed in a container creates `checkpoint-timer-%d` /
`checkpoint-dispatcher-%d`, they are absent from `beforeThreads`, and they get
reported as leaked. That also explains the spread of failures across unrelated
connector e2e jobs — it hit whichever test happened to run first in its
container, and jobs running second or later already had the threads in
`beforeThreads`.
The fix treats both pools as system threads in `isSystemThread`, alongside
the existing member-scoped `seatunnel-coordinator-service` entry. Verified with
`spotless:apply` and a clean `test-compile`. Pushed as `fd2841fd7`; CI is
re-running now.
On your three non-blocking notes — I agree with all three and plan to
address them in follow-up commits on this branch rather than leave them as
TODOs:
1. **`PipelineCheckpointTask` cancellation race.** Agreed the current
behaviour is correct and I'd rather not change the cancellation contract to
close a race that every caller already absorbs. I'll add the comment on
`run()`/`cancel()` documenting that the residual race is deliberately tolerated
because callers re-validate coordinator/checkpoint state before acting, so it
doesn't get "fixed" later in a way that alters the contract.
2. **Unbounded `checkpoint-dispatcher` pool.** Fair point that this is now a
member-wide shared resource rather than a per-pipeline isolated one, so the
blast radius of a future blocking body is different in kind, not just degree.
I'll bound `maximumPoolSize` and document the tradeoff (including why a
`SynchronousQueue`-style hand-off is still the right shape here), rather than
relying on the dispatched body staying non-blocking by convention.
3. **`Thread.sleep(800)` in `testCancelBeforeTimerFiresStopsTheBody`.**
Agreed, and it's the one inconsistency in an otherwise `Thread.sleep`-free
suite. I'll switch it to an Awaitility-based negative assertion to match the
rest.
@nzw921rx thanks for the savepoint question, and @DanielLeens thanks for the
detailed trace in reply — that matches my understanding: `startSavepoint()`
waits on `pendingCounter` under the per-pipeline `lock` and then calls
`startTriggerPendingCheckpoint()` directly, so it never enters the periodic
re-arm path, and the only thing it asks the shared scheduler for is that
checkpoint's timeout watchdog. This PR doesn't change when or whether a
checkpoint of any type is triggered, only where the timer pool lives.
--
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]
