Vivek1106-04 commented on PR #12165: URL: https://github.com/apache/seatunnel/pull/12165#issuecomment-5691858060
@DanielLeens All three carried-over items are now in the diff, along with the blocker. Head is `e8f9f336b`. ### Blocker: CI and sync with `dev` Rebased onto current `dev` (`1325a44b2`). The branch was 106 commits behind, not 35, by the time I got to it. The rebase was conflict-free and I verified the patch content survived it unchanged: the added and removed lines are byte-identical to the pre-rebase branch, 703 of them. The only thing that moved was the E2E whitelist hunk, since `SeaTunnelContainer` shifted upstream and `isSystemThread` is now around line 677 rather than 420; I checked it landed in the right method. Locally on the new base: `spotless:check` clean, full `-am` build green, and the unit tests pass. CI is running. ### Issue 1: cancellation race now documented Javadoc on `PipelineCheckpointTask#run()` and `#cancel(boolean)`. It states that the single cancellation read before the body is deliberate, that cancellation is best-effort once a task has been dispatched and guaranteed only while it is still on the timer, and why a late body is harmless: the periodic trigger re-checks completed/shutdown state under the coordinator's lock, and the watchdog re-checks that its checkpoint is still pending and not fully acknowledged. I also added an explicit warning not to "fix" this by holding a lock across `body.run()`, since that is the tempting wrong move and it would serialise unrelated pipelines on the shared dispatch pool for the length of a checkpoint. ### Issue 2: dispatch pool bounded This one needed a decision rather than a one-liner, because the obvious bound is unsafe here. A bounded pool over a `SynchronousQueue` rejects when saturated, and a rejected task in this position is a dropped checkpoint trigger or a dropped timeout watchdog. That is the same objection the class Javadoc already makes against reusing the coordinator executor, so applying it to our own pool would have been inconsistent. What landed is a `ThreadPoolExecutor` with core equal to maximum, an unbounded queue, and `allowCoreThreadTimeOut(true)`. Core equals maximum because a pool backed by an unbounded queue never grows past core; allowing core threads to time out gives back the on-demand growth and 60s idle reaping the cached pool had. The bound is `max(8, availableProcessors * 2)`, so it tracks the host rather than the pipeline count, and no task can ever be rejected. The residual cost, stated plainly: if every dispatch thread is blocked simultaneously, further bodies queue instead of starting at once. That is a delay rather than a drop, it is only reachable when the member is already in trouble, and it is not expected in normal operation since the heavy barrier work already runs on the coordinator's executor via `thenApplyAsync`. ### Issue 3: `Thread.sleep(800)` replaced Now `Awaitility.await().during(800ms).atMost(10s).untilAsserted(...)`. Awaitility 4.2.0 was already a dependency of this module and is used across its tests, so this matches the existing idiom. It is also strictly stronger than the sleep: `during` re-checks throughout the window, so a body that runs at any point inside it fails immediately rather than only being caught if the flag is still set at the end. ### Two tests added beyond the three items - `testDispatchPoolRespectsItsBound` - submits twice the bound in blocking bodies and asserts the thread count stays within it. - `testBlockedPipelineDoesNotDelayOtherPipelines` - blocks all but one dispatch thread and asserts a different pipeline's trigger still runs. The second one covers the design's central claim directly, which nothing previously asserted. Both release their latch in a `finally` so a failure cannot hang the suite. `SharedCheckpointSchedulerTest` is now 10 tests, `CheckpointCoordinatorTest` still 16, all 26 passing locally. ### On sequencing against #12149 and #12096 Agreed this is a maintainer call rather than something I should resolve unilaterally. `gh pr view` still reports `MERGEABLE` after the rebase, so there is no textual conflict today, but whichever of the three lands last will need a careful pass over `cleanPendingCheckpoint()`. Happy to go last and rebase. ### STIP Per @nzw921rx's request this now has a proposal: [STIP-40](https://github.com/apache/seatunnel/issues/12339). Your Issue 2 is in it as a resolved design decision with the rejected alternatives written out, rather than as an open question, since the code now commits to an answer. -- 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]
