[Improve][Zeta] Share checkpoint trigger scheduling across active pipelines by Vivek1106-04 · Pull Request #12165 · apache/seatunnel

Vivek1106-04 · 2026-09-07T06:07:15Z

Purpose of this pull request

Every CheckpointCoordinator built its own two-thread scheduled pool (CheckpointCoordinator.java:227-238), and CheckpointManager builds one coordinator per entry of checkpointPlanMap (CheckpointManager.java:119-168). Timer threads therefore grew with the number of active checkpoint-enabled pipelines: a member running 500 of them held up to 1000 threads that spend nearly all their time asleep.

The pool was only ever used as a timer. There are two call sites — re-arming the periodic trigger in scheduleTriggerPendingCheckpoint() and arming the checkpoint-timeout watchdog stored via pendingCheckpoint.setCheckpointTimeOutFuture(...) — while the heavy barrier work already runs on the injected executorService through thenApplyAsync. That is what makes a shared timer safe here.

This PR introduces a member-wide scheduler and gives each coordinator a per-pipeline lease.

What changed

New checkpoint/scheduler/SharedCheckpointScheduler: one instance per member, owned by SeaTunnelEngineContext (built once at SeaTunnelServer.java:155) and closed with it. It is deliberately not a static singleton, because several members run in one JVM during unit and E2E tests.
Timing and execution are split. Two fixed timer threads (checkpoint-timer-%d, with setRemoveOnCancelPolicy(true)) do nothing but hand the task to a bounded dispatch pool (checkpoint-dispatcher-%d, at most max(8, 2 x cores) threads, created on demand and reaped after 60 s idle, over an unbounded queue so excess work is delayed rather than dropped), so a checkpoint body that blocks on an RPC delays neither its own pipeline's next timer nor any other pipeline's.
New PipelineCheckpointScheduler: the lease, identified by (jobId, pipelineId). It holds only its own outstanding tasks, so cancelAll() is scoped to one pipeline and a coordinator no longer shuts down a pool it does not own. The lease stays usable afterwards, so a coordinator restored after a master-failover reset schedules again without rebuilding any pool.
New PipelineCheckpointTask: the ScheduledFuture handed back to callers. Because the task crosses two pools, cancelling it has to work both before the timer fires and after the body has been handed to the dispatch pool; the body re-checks its own cancellation flag at run time rather than relying on the timer future, which is already complete by then.
CheckpointCoordinator now leases its timer from CheckpointManager instead of constructing a pool, and cleanPendingCheckpoint() cancels this pipeline's tasks instead of calling shutdownNow() and building a replacement.

Two deliberate choices worth reviewing

Checkpoint work is not dispatched onto the coordinator executor. createCoordinatorExecutor() (CoordinatorService.java:287-298) is a bounded ThreadPoolExecutor over a SynchronousQueue with a rejection handler, so a saturated coordinator pool could silently drop a checkpoint trigger or a timeout watchdog. Keeping a separate dispatch pool also leaves this change independent of the executor-bounding work in #12117 and #12121.

The timer pool size is a constant rather than a new engine config option. Adding an Option would create a user-facing config contract for something that should not need tuning, and would require matching docs/en and docs/zh updates. Happy to add one if reviewers prefer it.

Invariants preserved

The single-pending-checkpoint guard is untouched. The whole trigger path stays inside synchronized (lock), and the pendingCounter.get() > 0 guard still short-circuits a redundant trigger. Coordinators previously already had two threads that could enter that path concurrently, so sharing the timer adds no new race: extra dispatch threads queue on the same lock and bail through the same guard.

Note that the issue text cites this guard as :802, which matches the commit it was written against; on current dev line 802 is the re-schedule call and the guard sits at :791.

How it works

1. Before and after

The problem in one picture: one pool per pipeline, so timer threads grew with the
pipeline count even though they spend nearly all their time asleep.

Now the member owns the threads, and each coordinator gets a lease on them.

flowchart LR
    subgraph BEFORE["Before: one pool per pipeline"]
        direction TB
        C1["coordinator<br/>pipeline 1"] --> P1["own timer pool<br/>2 threads"]
        C2["coordinator<br/>pipeline 2"] --> P2["own timer pool<br/>2 threads"]
        CN["coordinator<br/>pipeline N"] --> PN["own timer pool<br/>2 threads"]
    end

    subgraph AFTER["After: pools shared by the member"]
        direction TB
        D1["coordinator<br/>pipeline 1"] --> L1["lease 1"]
        D2["coordinator<br/>pipeline 2"] --> L2["lease 2"]
        DN["coordinator<br/>pipeline N"] --> LN["lease N"]
        L1 --> SH["shared scheduler<br/>2 timer threads<br/>bounded dispatch pool"]
        L2 --> SH
        LN --> SH
    end

Two timer threads plus a bounded dispatch pool instead of 2N threads, whatever the pipeline count.

2. Who owns what

The idea that makes a shared pool safe: threads belong to the member, cancellation
belongs to the pipeline.

A lease owns no threads. It borrows them, and it keeps its own list of scheduled tasks.
That list is the thing a coordinator controls.

flowchart TD
    SRV["SeaTunnel member"] --> CTX["engine context<br/>lives as long as the member"]
    CTX --> SHARED["shared checkpoint scheduler"]
    SHARED --> TIMER["timer pool<br/>2 daemon threads<br/>only fires and hands off"]
    SHARED --> DISP["dispatch pool<br/>bounded, reaps idle threads<br/>runs the checkpoint body"]

    CTX --> MGR["checkpoint manager"]
    MGR -->|"one lease per pipeline"| LEASE1["lease: pipeline 1<br/>own task list"]
    MGR --> LEASE2["lease: pipeline 2<br/>own task list"]

    LEASE1 --> COORD1["coordinator 1"]
    LEASE2 --> COORD2["coordinator 2"]

    LEASE1 -.->|"borrows threads"| TIMER
    LEASE2 -.->|"borrows threads"| TIMER

What	Belongs to	Goes away when
timer and dispatch threads	the member	the member shuts down
a lease's task list	one pipeline	the pipeline finishes, or a failover reset
the coordinator's lock and counters	one coordinator	unchanged by this PR

Each lease keeps a private task set, so cancelling pipeline 1 can't touch pipeline 2's
timers.

3. The timer only fires; the dispatch pool does the work

This split is the reason a shared pool doesn't couple pipelines together. A timer thread
never runs checkpoint logic. It wakes up, hands the task to the dispatch pool, and goes
back to sleep.

So if a checkpoint blocks on a slow RPC, it holds a dispatch thread, not a timer thread.
No timer fires late because of it. It does hold a dispatch thread, and those are shared,
which is why what a dispatched body may block on matters: see
What this PR does not change.

sequenceDiagram
    autonumber
    participant L as lease
    participant T as timer thread
    participant D as dispatch thread

    L->>T: schedule the next trigger, after the checkpoint interval
    Note over T: waits out the delay
    T->>D: hand off the task
    Note over T: free again immediately
    D->>D: run the trigger, under this pipeline's lock
    D->>L: task done, drop it from the task list
    D->>T: schedule the next trigger

The lock in step 5 is per-coordinator, so a dispatch thread running pipeline A's
checkpoint only ever waits on other work for pipeline A. While it waits, though, it is a
shared thread that nobody else can use.

4. Savepoints don't queue behind the periodic timer

This came up in review: what happens when the savepoint API fires while the scheduler is
running the periodic trigger? A savepoint never enters the periodic loop.

The periodic path is the only one that asks the scheduler to re-arm itself. A savepoint
is triggered directly by the caller. It uses the scheduler for one thing only, the same
thing every checkpoint type uses it for: arming that checkpoint's timeout watchdog.

flowchart TD
    PER["periodic trigger<br/>tryTriggerPendingCheckpoint"] --> LOCK["take this pipeline's lock"]
    SAV["savepoint API<br/>startSavepoint"] --> LOCK

    LOCK --> BUSY{"a checkpoint already pending?"}

    BUSY -->|"periodic: yes"| REARM["give up, ask the timer<br/>to try again later"]
    BUSY -->|"savepoint: yes"| WAIT["wait here until it drains"]
    BUSY -->|"no"| FIRE["create and trigger the checkpoint"]

    WAIT --> FIRE
    FIRE --> WD["arm this checkpoint's<br/>timeout watchdog"]

    REARM -.->|"only the periodic path re-arms"| PER

So calling the savepoint API during a pending periodic checkpoint makes the savepoint
wait, then fire straight away. It never sits in the timer queue. And because a pipeline
allows only one pending checkpoint at a time — which this PR doesn't change — a
savepoint's watchdog and a periodic checkpoint's watchdog are never both live for the
same pipeline.

This PR changes where the timer threads live. It doesn't change when a checkpoint gets
triggered, or which kind wins.

One thing the diagram hides: "wait here until it drains" is a 500 ms sleep-poll inside
this pipeline's lock. That is existing dev behaviour, and this PR does not change it.
What does change is where a periodic trigger for that pipeline waits on it. See below.

5. Master failover reuses the lease

The old code threw its executor away on reset and built a replacement. The new code just
cancels that pipeline's scheduled tasks. The lease stays valid, so a restored
coordinator starts scheduling again with no pool to rebuild.

sequenceDiagram
    autonumber
    participant CO as coordinator
    participant LE as its lease
    participant SH as shared scheduler

    CO->>CO: restore after master failover
    CO->>LE: cancel everything this pipeline scheduled
    LE->>LE: clear the task list
    Note over SH: threads untouched, still 2 timer threads
    CO->>CO: clear pending checkpoints, mark itself running
    CO->>LE: schedule again on the same lease

Same coordinator, same lease, no thread churn.

6. Cancelling a task that has already been handed off

A task crosses two pools, so cancellation has to work in both halves. Cancelling the
timer entry is enough while the task is still waiting. Once the timer has fired, that
entry is already spent — so the task also checks a cancelled flag right before running
the body.

stateDiagram-v2
    [*] --> Waiting: scheduled
    Waiting --> Cancelled: cancelled before the timer fires
    Waiting --> HandedOff: timer fires

    HandedOff --> Skipped: sees the cancelled flag,<br/>body never runs
    HandedOff --> Running: not cancelled

    Running --> Finished: body completes
    Running --> Failed: body throws

    Finished --> [*]: dropped from the task list
    Failed --> [*]: dropped from the task list
    Skipped --> [*]: dropped from the task list
    Cancelled --> [*]: dropped from the task list

A body that throws completes the task's future exceptionally and is logged at ERROR with the
job and pipeline id. The dispatch thread survives and goes back to the pool.

There's a small window left: a task that already passed the flag check will finish its
body even if cancellation lands mid-run. That's harmless, because every body re-checks
the coordinator's state before doing anything. A periodic trigger bails out if the
coordinator is finished or shut down; a watchdog bails out if the checkpoint it was
watching is gone, and an expiry it already handed off is dropped if the coordinator was
restored in the meantime (see the watchdog section). Both checks share the same lock as the cancellation, so a late firing
does nothing.

Shutdown

The shared pools close once, with the engine context, when the member shuts down.
Anything scheduled in that window isn't lost quietly: the task is cancelled and the
rejection is logged at debug, since it is expected during shutdown rather than a fault.

Timeout watchdog: expiry runs off the dispatch thread

The watchdog used to run its expiry handling inline on the thread that fired it. That
handling updates the coordinator status in the IMap through a retry loop and takes the
coordinator lock through cleanPendingCheckpoint, which is too much to run on a shared
dispatch thread. Since da0cbbf52 the dispatch thread only checks whether the checkpoint
is still pending and hands the expiry to the coordinator executor. That executor aborts
when saturated, and a dropped expiry would leave the checkpoint pending with nothing left
to fail it, so a rejection falls back to running inline, with a WARN.

Because the expiry now runs later than the watchdog's check, restoreCoordinator() can
reuse the coordinator in between. Since 3cc4dea45 the expiry carries the expired
PendingCheckpoint and the coordinator future that was current when that checkpoint was
triggered (read in startTriggerPendingCheckpoint while the caller holds lock). It fails
the coordinator only if that future is still current and not done, and the checkpoint is
still the one tracked under its id; otherwise it is dropped with an INFO log. That check and
the FAILED transition hold a dedicated generationLock, which restoreCoordinator() also
holds while it swaps in the new future, so a stale expiry can never fail a restored
generation. It is not lock, because startSavepoint sleep-polls inside lock waiting for
the very checkpoint that may be expiring. handleCheckpointError is called after the lock is
released, since it takes the pipeline monitor that a restoring pipeline holds while it calls
restoreCoordinator().

Cost of the hand-off

The timer -> dispatcher hand-off is not free, and the scheduling benchmark from #12896 measures
it. Full analysis and raw data: #12165 (comment).

Steady cost: about 25 us per trigger. That is one extra cross-thread wake. A per-stage
trace shows the timer fires just as late on both sides (about 80 us at p50), and the whole
difference is the hop from timer thread to dispatcher thread.
Rare tail: 0.5-3 ms. If no CPU is idle when the dispatcher wakes, it waits for the running
thread's scheduler time slice to end. On the 4-vCPU CI runners the thread holding the CPU was
usually the benchmark's own busy-spinning JMH thread.
Proof that the hand-off is the cause. The same commit with only the hand-off removed
(timer.schedule(task, ...)) scores the same as dev on JDK 11 and 17, with no tail
(run 40071856278; builds alternated
on one runner, two rounds each).

JDK	pipelineNum	dev (us/op)	this PR	this PR without hand-off
11	10	96.0-97.0	124.7-138.7	95.8-96.5
11	100	92.6	131.6-137.6	93.8-94.1
17	10	87.5-89.5	126.0-138.6	85.8-89.7
17	100	83.6-83.9	118.4-122.1	84.0-84.4

The hand-off is kept on purpose. It keeps lock-taking checkpoint bodies off the two shared
timer threads, so one slow coordinator can't delay every pipeline's timers. The cost is
microseconds against checkpoint intervals of seconds, and it does not grow with the pipeline count.

What sharing buys, from the JDK 17 safepoint logs of run 40061505634: dev's 2 timer threads per
pipeline make safepoints slower as P grows. Time-to-safepoint p50 is 10 / 18 / 56 us at
P = 10 / 100 / 500, and the longest pause at P = 500 is 34-37 ms. This PR stays at about 8.5 us,
with a 6.6-6.9 ms longest pause, because its thread count does not depend on P.

What this PR does not change: savepoint lock holding (#12441)

startSavepoint holds the coordinator lock across its 500 ms sleep-poll until the
in-flight checkpoint drains, and tryTriggerPendingCheckpoint takes the same lock. A
periodic trigger that fires for that pipeline during a savepoint therefore blocks until
the savepoint gets going. CheckpointManager.triggerSavePoints does this for every
coordinator of a job at once.

On dev the blocked trigger holds one of that pipeline's own two threads. Here it holds
a shared dispatch thread. That is a new cross-pipeline coupling, and this PR does not
remove it.
It is bounded by how long a savepoint takes, not unbounded. To hold the whole dispatch
pool, enough pipelines' trigger and savepoint windows have to coincide, and I have not
been able to demonstrate that.
Moving the wait off the lock is [Bug][Zeta] startSavepoint holds the coordinator lock across its sleep-poll, blocking concurrent checkpoint triggers #12441, because it changes dev's savepoint path, not
this design. The regression lands with that fix, written red against the current
blocking lock first: fill the dispatch capacity with savepoint-contended coordinators,
then assert that an unrelated pipeline still gets its trigger and watchdog. This split
was agreed on [STIP-40][Improve][Zeta] Member-wide shared checkpoint trigger scheduling #12339.

The same caveat is in the SharedCheckpointScheduler class Javadoc and in the
Correction section of the STIP (#12339).

Does this PR introduce any user-facing change?

No.

No config option is added, removed, or renamed, and no default changes. The only externally observable difference is thread naming in stack dumps and profiles: the per-pipeline checkpoint-coordinator-<pipelineId>/<jobId> threads are replaced by shared checkpoint-timer-%d and checkpoint-dispatcher-%d threads. Job and pipeline context is still available on those threads, because each scheduled body is wrapped with MDCTracer.tracing(...) at schedule time, which is what the previous MDCTracer.tracing(scheduler) wrapper did. No documented behavior depends on the old thread names.

How was this patch tested?

New SharedCheckpointSchedulerTest covers the lease contract:

a scheduled body runs on the dispatch pool, not on a timer thread;
cancelAll() on one pipeline's lease leaves a sibling pipeline's timer running;
a task cancelled before its timer fires never reaches the dispatch pool;
a task cancelled after its timer fired still does not run its body, and cancel is idempotent;
a lease stays usable after cancelAll(), which is the master-failover reset path;
finished tasks remove themselves rather than accumulating on a long-lived pipeline;
scheduling after the shared scheduler is closed returns an already-cancelled future and runs nothing;
the dispatch pool never grows past its bound;
a pipeline whose checkpoint body is stuck does not delay another pipeline's trigger;
a body that throws completes the task's future exceptionally and does not kill its dispatch thread (verified red on 771fce221, where the rethrow terminated the worker). It waits for the task to settle (outstandingCount() == 0) and then for the worker to park in the dispatch queue's timed poll, rather than inferring settlement from the thread leaving RUNNABLE.

The acceptance criterion from the issue is asserted directly: 500 leases each scheduling a task keep the timer pool at 2 threads.

CheckpointCoordinatorTest needed one change. Nine tests build coordinators on Mockito.mock(CheckpointManager.class); a mocked manager returns null for the new leaseCheckpointScheduler(...), so those coordinators would NPE on cancel. They now go through a helper that hands out real leases from a test-scoped shared scheduler. CheckpointCoordinatorScheduledTriggerTest (from #12448 on dev) got the same treatment: its plain mock now hands out a lease from a test-scoped scheduler that is closed in finally. All of its assertions are unchanged, so it now also checks that #12448's wrapped trigger body still fails the coordinator when it runs on the shared dispatcher.

CheckpointCoordinatorTest#testCheckpointExpiryHandlingNeverRunsOnTheDispatchThread covers the watchdog change: an expired checkpoint's handling runs on the coordinator executor, not on the dispatch thread that fired the watchdog. It was verified red with da0cbbf52 reverted.

CheckpointCoordinatorTest#testExpiryHeldAcrossCoordinatorRestoreDoesNotFailTheRestoredGeneration holds a handed-off expiry behind a latch on a single-thread coordinator executor, restores the coordinator, puts the same PendingCheckpoint back so only the generation check can drop the expiry, then releases it and asserts the restored generation's future is not done and handleCheckpointError was never called. It was verified red twice: with the expiry reverted to the old unconditional handleCoordinatorError, and with only the owner-future check removed. testExpiryOfTheCurrentGenerationFailsTheCoordinator asserts that a genuine expiry still completes the current future FAILED and calls handleCheckpointError(1, false).

Local results at the current head a932e27a7 (after the dev merge feedd2506), JDK 17:

SharedCheckpointSchedulerTest: 11 tests, 0 failures
CheckpointCoordinatorTest: 20 tests, 0 failures
CheckpointCoordinatorScheduledTriggerTest: 1 test, 0 failures (it failed with an NPE on the null lease before the fix)
StateTransitionCleanupTest: 13 tests, 0 failures
all org.apache.seatunnel.engine.server.checkpoint.** tests plus SavePointTest and CheckpointErrorRestoreEndTest: 69 tests, 0 failures, 2 skipped (existing @Disabled)
the two new expiry tests and the changed failing-body test, 20 consecutive runs: all green

From an earlier head of this PR, on JDK 11, before the watchdog and trigger-body commits; not re-run since:

*Checkpoint*Test plus CoordinatorServiceTest: 83 tests, 0 failures, 3 skipped
full seatunnel-engine-server unit test suite: 513 tests, 0 failures, 0 errors, 5 skipped

./mvnw spotless:apply applied; the only formatting changes were whitespace in the files this PR already touches.

E2E thread-leak check

SeaTunnelContainer asserts that a finished job leaves no non-system thread behind. The
old per-coordinator pool shut down with the job, so it never tripped that assertion. The
shared pools outlive the job by design, so they do. The first job in a container
creates checkpoint-timer-%d and checkpoint-dispatcher-%d, those threads are absent
from the pre-job snapshot, and they are reported as leaked — which is why the failures
showed up in unrelated connector e2e jobs as well as in CheckpointRestoreWithStopIT
and SavepointRestoreIT. Fixed in fd2841fd7 by treating both prefixes as system
threads in isSystemThread(), alongside the existing member-scoped
seatunnel-coordinator-service entry.

Rebase note

#12149 is open on the same constructor and the same cleanPendingCheckpoint() lines, and #12096 also touches cleanPendingCheckpoint(). This change is complementary rather than a duplicate: #12149 fixes the lifecycle of the per-coordinator pool, while this removes the per-coordinator pool. I kept the diff narrow specifically so that rebasing is cheap, and I am happy to rebase onto either of those once they land, or to have this sequenced behind them — whichever reviewers prefer.

Check list

If any new Jar binary package adding in your PR, please add License Notice according
New License Guide — no new dependencies
If necessary, please update the documentation to describe the new feature. https://github.com/apache/seatunnel/tree/dev — no user-facing change, so no doc update
If necessary, please update incompatible-changes.md to describe the incompatibility caused by this PR — no incompatible change
If you are contributing the connector code, please check that the following files are updated — not a connector change

DanielLeens

What Problem Does This PR Solve?

User pain point: Every CheckpointCoordinator built its own dedicated two-thread ScheduledExecutorService (CheckpointCoordinator.java:227-238 on dev), and CheckpointManager builds one coordinator per pipeline (CheckpointManager.java:119-168). Timer threads therefore grew linearly with the number of active checkpoint-enabled pipelines on a member: a member running 500 pipelines held up to 1000 mostly-sleeping timer threads, purely to re-arm a periodic trigger and arm a timeout watchdog.

Fix approach: Introduce one SharedCheckpointScheduler per member (owned by SeaTunnelEngineContext, closed with it) backed by two pools: a tiny fixed checkpoint-timer pool that only fires and hands off, and an elastic checkpoint-dispatcher pool that actually runs the body, so a slow checkpoint body never delays another pipeline's timer. Each CheckpointCoordinator now leases a PipelineCheckpointScheduler from CheckpointManager instead of owning a pool; the lease tracks only its own outstanding tasks so cancelAll() stays scoped to one pipeline, and — critically — the lease is reusable after cancelAll(), so a coordinator restored on master failover (restoreCoordinator() -> cleanPendingCheckpoint(CHECKPOINT_COORDINATOR_RESET)) keeps working without rebuilding a pool.

One-sentence summary: This replaces N dedicated per-pipeline timer pools with one shared member-wide timer + elastic dispatch pool, cutting a 500-pipeline member's checkpoint-timer thread count from ~1000 to 2, with no behavioral change beyond thread names in stack dumps.

Example: Before vs After

A member running 500 checkpoint-enabled pipelines, each with checkpoint.interval set:

Before: 500 CheckpointCoordinator instances, each with its own Executors.newScheduledThreadPool(2, ...) named checkpoint-coordinator-<pipelineId>/<jobId> -> up to 1000 live timer threads sitting idle almost all the time.
After: 2 checkpoint-timer-%d threads total for the whole member, plus a checkpoint-dispatcher-%d cached pool that only grows while checkpoint bodies are actually being handed off. This is exactly what SharedCheckpointSchedulerTest#testTimerThreadCountStaysFlatAsPipelinesGrow asserts (500 leases scheduling a task keep getTimerPoolSize() <= 2).

1. Code Change Review

1.1 Core Logic Analysis

Runtime flow that reaches the changed code:

CheckpointManager.addCoordinator(...)                      [per-pipeline coordinator construction]
  -> new CheckpointCoordinator(manager, ...)
       -> this.scheduler = manager.leaseCheckpointScheduler(pipelineId)   // CheckpointCoordinator.java:229
            -> CheckpointManager.leaseCheckpointScheduler(pipelineId)
                 -> this.checkpointScheduler.lease(jobId, pipelineId)     // one lease per (jobId, pipelineId)

Periodic trigger path (steady state):
tryTriggerPendingCheckpoint() [synchronized(lock)]
  -> scheduleTriggerPendingCheckpoint(type, delay)
       -> scheduler.schedule(() -> tryTriggerPendingCheckpoint(type), delay, MILLISECONDS)  // PipelineCheckpointScheduler.schedule()
            -> new PipelineCheckpointTask(MDCTracer.tracing(command), outstanding::remove)
            -> parent.dispatchAfter(task, delay, unit)                    // SharedCheckpointScheduler: timer.schedule(() -> submit(task), ...)
                 [checkpoint-timer-%d fires after `delay`]
                 -> dispatcher.execute(task)                              // hand-off only, no blocking work here
                      [checkpoint-dispatcher-%d actually runs task.run() -> body.run() -> tryTriggerPendingCheckpoint(...)]

Timeout watchdog path:
triggerCheckpoint(...) -> pendingCheckpoint.setCheckpointTimeOutFuture(scheduler.schedule(() -> { ... }, timeout, MILLISECONDS))  // CheckpointCoordinator.java:972-989
  [re-checks pendingCheckpoints.get(id) != null && !isFullyAcknowledged() before firing CHECKPOINT_EXPIRED]

Reset / master-failover path:
restoreCoordinator(alreadyStarted)                                        // CheckpointCoordinator.java:664
  -> cleanPendingCheckpoint(CHECKPOINT_COORDINATOR_RESET)
       -> scheduler.cancelAll()                                           // CheckpointCoordinator.java:1196 (was: scheduler.shutdownNow() + rebuild pool)
  -> shutdown = false                                                     // CheckpointCoordinator.java:673, same instance resumes scheduling on the SAME lease

Precise change: CheckpointCoordinator.scheduler goes from a mutable, self-owned ScheduledExecutorService (rebuilt on every cleanPendingCheckpoint()/reset via Executors.newScheduledThreadPool(2, ...), CheckpointCoordinator.java:227-238 and :1200-1211 on dev) to a final transient PipelineCheckpointScheduler obtained once at construction (CheckpointCoordinator.java:155,229) and never rebuilt; cleanPendingCheckpoint() now calls scheduler.cancelAll() (:1196) instead of shutdownNow() + reconstruction. Three new classes carry the mechanism: SharedCheckpointScheduler (member-wide, two real pools), PipelineCheckpointScheduler (per-pipeline lease, tracks its own outstanding ConcurrentHashMap.newKeySet()), PipelineCheckpointTask (a hand-rolled ScheduledFuture that crosses both pools and re-checks its own AtomicBoolean cancelled at run time because the timer-side future is already complete by the time the body runs on the dispatch pool).

Key findings from tracing the invariants that must survive this refactor:

Single-pending-checkpoint guard is untouched. tryTriggerPendingCheckpoint() still enters synchronized (lock) and checks isCompleted() || isShutdown() before doing anything (verified at CheckpointCoordinator.java:774), and the pendingCounter.get() > 0 short-circuit at :791 is unmodified. Since lock is a per-coordinator (per-pipeline) instance field, moving the timer to a shared pool cannot introduce cross-pipeline contention on this guard — every dispatch thread that happens to run pipeline A's body only ever contends with other threads running pipeline A's body.
Master-failover restore/reuse is correctly preserved. I traced restoreCoordinator() (:664) -> cleanPendingCheckpoint(CHECKPOINT_COORDINATOR_RESET) -> scheduler.cancelAll() (:1196) -> shutdown = false (:673) and confirmed the SAME CheckpointCoordinator instance (and therefore the same lease) is reused after a reset, exactly matching the old semantics where the executor was destroyed and rebuilt in place. This is directly exercised by SharedCheckpointSchedulerTest#testLeaseIsReusableAfterCancelAll.
Timeout watchdog firing after a stale cancellation is harmless by construction. The watchdog body re-checks pendingCheckpoints.get(pendingCheckpoint.getCheckpointId()) != null (CheckpointCoordinator.java:977-981) before acting, and pendingCheckpoints.clear() happens inside the same synchronized(lock) block as scheduler.cancelAll() in cleanPendingCheckpoint(). So even in the narrow race described below, a stale watchdog/trigger firing after a reset is a guaranteed no-op.
PipelineCheckpointTask cross-pool cancellation has a narrow, provably-benign race (see Issue 1) — worth calling out explicitly for an engine/checkpoint change even though it is not exploitable given the guards above.
MDC context propagation is equivalent, not just claimed. The old code wrapped the whole executor once via MDCTracer.tracing(scheduler); the new code wraps each command at schedule time via MDCTracer.tracing(command) inside PipelineCheckpointScheduler.schedule(). I checked MDCTracer.tracing(Runnable) -> tracing(MDCContext.current(), delegate) and MDCRunnable.run() (activates the context, runs delegate, closes it) — since scheduleTriggerPendingCheckpoint() is itself invoked recursively from inside an already-active MDCRunnable (the previous scheduled body), MDCContext.current() correctly resolves to this pipeline's context at every re-arm, not just the first one. Equivalent to the old behavior.
The claim about createCoordinatorExecutor() is accurate. I verified CoordinatorService.java:287-298: it really is a ThreadPoolExecutor over a SynchronousQueue with ThreadPoolStatus.RejectionCountingHandler, i.e. a bounded hand-off queue that can reject under saturation — this is a correct, verified justification for keeping checkpoint dispatch off that pool.
Only one CheckpointManager/SeaTunnelEngineContext construction path exists in production (SeaTunnelServer.java:155 builds the context once; CheckpointManager's constructor resolves engineContext.getCheckpointScheduler() which is a non-null final field set unconditionally in SeaTunnelEngineContext's private constructor) — the sharedScheduler != null ? sharedScheduler : new SharedCheckpointScheduler() fallback in CheckpointManager is reachable only when engineContext itself is a test double (e.g. a Mockito mock returning null), never in production. Confirmed by grepping the whole tree for SharedCheckpointScheduler/PipelineCheckpointScheduler/PipelineCheckpointTask usages — exactly the 8 files in this diff reference them, no other call site was missed.
Lifecycle wiring is complete: SeaTunnelEngineContext.close() now calls checkpointScheduler.close() before stateStores.close() (new), and SeaTunnelServer.java:256 calls engineContext.close() on shutdown — so the shared pools are torn down exactly once, at member shutdown, not per-pipeline.

1.2 Compatibility Impact

Classification: Fully compatible.

No config option added/removed/renamed, no default changed (verified: no new Option/OptionRule entries anywhere in the diff).
No checkpoint state, serialization format, or storage schema is touched — ProtoStuffSerializer, PendingCheckpoint, and the checkpoint storage layer are untouched by this diff. A checkpoint/savepoint written by a pre-PR member restores identically on a post-PR member and vice versa, since nothing about what gets persisted changed, only how the in-memory timer is implemented.
The only externally observable difference is thread naming in stack dumps/profiles (checkpoint-coordinator-<pipelineId>/<jobId> -> checkpoint-timer-%d / checkpoint-dispatcher-%d), which I confirmed via git grep has no other reference anywhere in the codebase (docs, tests, tooling) that depends on the old name.
Per-pipeline checkpoint isolation is preserved: PipelineCheckpointScheduler.outstanding is a private, per-lease ConcurrentHashMap.newKeySet(), so cancelAll() on one pipeline cannot affect a sibling's outstanding tasks — directly proven by SharedCheckpointSchedulerTest#testCancelAllOnOneLeaseLeavesSiblingPipelineRunning.

1.3 Performance / Side-Effect Analysis

Positive, and the intended fix: timer-thread count drops from O(active pipelines) to a constant 2 per member. This is a real reduction in idle-thread memory/scheduling overhead on members hosting many small pipelines.
New shared resource, worth flagging for a financial-grade review: the dispatch pool is Executors.newCachedThreadPool(...) (SharedCheckpointScheduler constructor) — unbounded. Today this is safe because the dispatched body is documented and structurally verified to be non-blocking (the heavy barrier work is off-loaded again via thenApplyAsync(..., executorService); the dispatch-pool thread's job is just to invoke tryTriggerPendingCheckpoint's synchronous, per-pipeline-locked portion and return). But because this pool is now shared across every pipeline on the member (previously each pipeline had its own worst-case-bounded 2-thread pool), a future change that makes the dispatched body block for any reason (e.g. a slow synchronous call added inside the synchronized(lock) section) would no longer be bounded per pipeline — it could grow the shared pool by one thread per concurrently-blocked pipeline. See Issue 2.
No new locks are introduced; lock remains a per-coordinator field, so no new cross-pipeline contention.
RejectedExecutionException from either pool during shutdown is caught and the task is cancelled rather than lost (SharedCheckpointScheduler.dispatchAfter/submit), so no silent task leak on shutdown.

1.4 Error Handling and Logging

Rejection during shutdown is logged at debug (SharedCheckpointScheduler.java, dispatchAfter/submit) — appropriate level, since this is an expected event during normal shutdown, not an error condition.
No sensitive data is logged anywhere in the new code (thread names, job/pipeline ids only).
See Issue 1 for the one correctness-adjacent note in this area (not an error-handling gap, a scheduling race that is already absorbed by existing state guards).

2. Code Quality Assessment

2.1 Coding Standards

All three new classes (SharedCheckpointScheduler, PipelineCheckpointScheduler, PipelineCheckpointTask) carry clear class-level Javadoc explaining ownership, lifetime, and the two-pool split rationale. Non-trivial methods (lease, dispatchAfter, schedule, cancelAll, attachTimerFuture) all have Javadoc explaining the "why", not just restating the signature — e.g. attachTimerFuture's note about losing the race against cancel(). This meets the bar for new core methods/fields; no missing-Javadoc issue to raise.
CheckpointManager.checkpointScheduler and SeaTunnelEngineContext.checkpointScheduler fields both have Javadoc explaining ownership and why a fallback exists. Good.

2.2 Test Coverage and Test Stability

Coverage: SharedCheckpointSchedulerTest (new, 8 tests) directly proves every invariant claimed in the PR description: body runs on dispatch thread not timer thread; cancelAll() is pipeline-scoped; cancel-before-fire and cancel-after-fire both stop the body; a lease is reusable after cancelAll() (the master-failover path); finished tasks self-remove (no accumulation); schedule-after-close returns an already-cancelled future; and the acceptance criterion from the linked issue (500 leases keep the timer pool at 2 threads) is asserted directly rather than inferred. CheckpointCoordinatorTest is updated so all pre-existing coordinator tests continue to exercise real lease/cancel behavior instead of a manager mock returning null for leaseCheckpointScheduler (which would otherwise NPE on cleanup) — I confirmed every new CheckpointCoordinator(...) call site in that file (10 of them) is now backed either by the new mockCheckpointManager() helper or by a real SeaTunnelEngineContext obtained from a live test server (TestCheckpointManager, which was never a bare Mockito mock to begin with).

Test stability rating: Stable, with one Low-risk note:

SharedCheckpointSchedulerTest.testCancelBeforeTimerFiresStopsTheBody (asserting ran.get()==false after Thread.sleep(800) following a 500ms-delay cancel) uses Thread.sleep where the rest of the suite consistently uses CountDownLatch/await. Because this is a negative assertion guarded by an upfront synchronous cancel() call, the only failure mode is a false failure under extreme host contention (the sleep window being too short relative to a slow test runner), never a false pass that hides a real regression — so I rate this Low risk rather than High, but it's still the one place in an otherwise very clean, Thread.sleep-free suite. See Issue 3.
All other 7 tests are deterministic: CountDownLatch/await(10s)-based, fresh scheduler per test (@BeforeEach/@AfterEach), no shared mutable state, no order dependence, no floating-point comparisons.
SupportMultiTableSinkWriter contract check: not applicable — this PR touches no SinkWriter/Sink code.

2.3 Documentation Updates

None needed, and none made. Confirmed via git grep across docs/en and docs/zh that no page references the old checkpoint-coordinator-<pipelineId>/<jobId> thread-naming pattern or any checkpoint-scheduler-specific config, so there is nothing to update. No new engine Option was introduced (a deliberate, well-reasoned choice per the PR description, to avoid a user-facing config contract for something that should not need tuning) — I agree with that call for this change's scope.

3. Architectural Soundness

3.1 Precise fix vs. workaround vs. long-term solution

This is the long-term/root-cause fix, not a workaround: the actual defect was "the timer pool's lifetime and cardinality is tied to the coordinator's lifetime and count," and this PR fixes exactly that by relocating pool ownership to the member (SeaTunnelEngineContext) while keeping cancellation scoped correctly via the lease abstraction. It does not paper over the symptom (e.g. by capping pipeline count or adding a config knob) — it removes the structural cause of thread growth.

3.2 Maintainability

The three-class split (SharedCheckpointScheduler / PipelineCheckpointScheduler / PipelineCheckpointTask) cleanly separates "member-wide pool ownership," "per-pipeline cancellation scope," and "cross-pool task lifecycle," which is easier to reason about than a single class trying to do all three. The explicit rationale comments (why not the coordinator executor, why a constant not a config) will save the next reader real investigation time.

3.3 Extensibility

The lease pattern (SharedCheckpointScheduler.lease(jobId, pipelineId)) generalizes cleanly if another per-pipeline periodic concern ever needs the same shared-timer/elastic-dispatch treatment; nothing here is checkpoint-specific except the class names and Javadoc.

3.4 Historical-Version Compatibility

No checkpoint/savepoint format, protocol, or serialization change — a job checkpointed on a pre-PR member restores identically on a post-PR member. No config default changed, so upgrading a running cluster to a build containing this PR requires no operator action. This is an internal-implementation-only change from the perspective of any persisted state or historical job.

4. Issue Summary

Number	Issue	Location	Severity
1	`PipelineCheckpointTask` has a narrow, currently-benign race: a task already dequeued by the dispatch pool right as `cancelAll()` fires can still run its body once, because the cancellation check is a single unsynchronized read at the top of `run()`	`seatunnel-engine/seatunnel-engine-server/.../checkpoint/scheduler/PipelineCheckpointTask.java` (`run()`/`cancel()`)	Low
2	`checkpoint-dispatcher` pool is an unbounded `newCachedThreadPool`, now shared by every pipeline on the member instead of isolated per pipeline; safe today only because the dispatched body is non-blocking by design	`seatunnel-engine/seatunnel-engine-server/.../checkpoint/scheduler/SharedCheckpointScheduler.java` (constructor)	Low
3	One new test uses `Thread.sleep(800)` for a negative assertion instead of the `CountDownLatch`/Awaitility style used by the rest of the suite; false-failure risk only, not a false-pass risk	`seatunnel-engine/seatunnel-engine-server/.../checkpoint/scheduler/SharedCheckpointSchedulerTest.java` (`testCancelBeforeTimerFiresStopsTheBody`)	Low

All three are non-blocking observations from a deep engine/checkpoint-level review; none of them represent an active correctness, compatibility, or data-loss risk given the guards already in place elsewhere in CheckpointCoordinator.

5. Merge Recommendation

Conclusion: Ready to merge

Blockers — must be fixed

None.

Recommended fixes — non-blocking
Add a short code comment on PipelineCheckpointTask.run()/cancel() documenting that the residual cross-pool cancellation race (Issue 1) is intentionally tolerated because every caller re-validates coordinator/checkpoint state before acting, so a future contributor does not "fix" it in a way that changes the cancellation contract.
Consider a maximumPoolSize bound (with a sane default) on the checkpoint-dispatcher pool as defense-in-depth (Issue 2), since it is now a member-wide shared resource rather than an isolated per-pipeline pool; document the tradeoff either way.
Replace the Thread.sleep(800) in testCancelBeforeTimerFiresStopsTheBody (Issue 3) with an Awaitility-based negative-assertion idiom for consistency with the rest of the suite.

Overall assessment: This is an excellent first contribution — a real, well-diagnosed production concern (unbounded per-pipeline timer-thread growth), fixed at the correct root cause, with a genuinely thoughtful design (splitting timer-fire from body-execution across two pools specifically so a slow checkpoint never delays another pipeline's timer) and a test suite that proves the actual claims (thread-count-stays-flat, pipeline-scoped cancellation, reuse-after-reset) rather than just re-asserting output. I traced every invariant that matters for an engine/checkpoint change — the single-pending-checkpoint guard, master-failover restore/reuse, and MDC context propagation — and all of them are correctly preserved. Welcome to Apache SeaTunnel, and thank you for the very clear PR description and the honesty about the two deliberate design tradeoffs — that made this review considerably easier to do thoroughly. I'll take another look once CI (currently still queued on the fork) has run.

nzw921rx

Thank you for your wonderful speech. I have a question. What happens when the API calls savepoint and the PipeCheckpointScheduler only triggers?

DanielLeens · 2026-09-09T00:35:13Z

@nzw921rx Thanks — good question, and worth spelling out since it's exactly the kind of interaction an engine/checkpoint reviewer should check on this PR.

Short answer: the savepoint API path does not go through the periodic re-arm loop (scheduleTriggerPendingCheckpoint / tryTriggerPendingCheckpoint) at all, so there's no conflict with PipelineCheckpointScheduler's periodic-trigger role. It only touches the shared scheduler for the same single purpose every checkpoint type uses it for: arming that checkpoint's timeout watchdog.

Tracing it on the current head (85efedd):

CheckpointCoordinator.startSavepoint() (CheckpointCoordinator.java:824) busy-waits under synchronized (lock) while pendingCounter.get() > 0 (:840) — i.e. it waits for any in-flight checkpoint (periodic or otherwise) on this pipeline to drain — then calls createPendingCheckpoint(..., SAVEPOINT_TYPE) and startTriggerPendingCheckpoint(savepoint) directly (:847-848).
That's a completely separate code path from tryTriggerPendingCheckpoint() (the one that calls scheduler.schedule(...) for periodic re-arming at :523 and the pendingCounter > 0 back-off reschedule at line ~792 on dev's equivalent). A savepoint never enters that method, so it never asks the scheduler to re-arm a periodic trigger.
The only place a savepoint touches scheduler at all is inside startTriggerPendingCheckpoint() (:901), which every checkpoint type shares: it calls scheduler.schedule(...) once (:973) purely to arm the checkpoint-timeout watchdog for that checkpoint id.

So in concrete terms: if you call the savepoint API while a periodic checkpoint is already pending on that pipeline, the savepoint call blocks (sleeping 500ms at a time) until pendingCounter drops to 0, then fires immediately and synchronously — it doesn't wait in the scheduler's timer queue the way a periodic re-arm does. Once it fires, its timeout watchdog is scheduled on the exact same per-pipeline lease a periodic checkpoint's watchdog would use, but since pendingCounter (untouched by this PR) already guarantees only one pending checkpoint per pipeline at a time, there's no scenario where a savepoint's watchdog and a periodic checkpoint's watchdog are both live for the same pipeline simultaneously. That invariant is exactly what I traced and confirmed in my review above, and this PR doesn't change any of it — it only changes where the timer pool lives, not when/whether a checkpoint of any type gets triggered.

Vivek1106-04 · 2026-09-10T04:31:47Z

@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:

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.
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.
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.

nzw921rx · 2026-09-10T04:56:52Z

@Vivek1106-04 Thank you for your reply. Version 3.0 is about to be released, and I will carefully review it after completion. Currently, I think the direction of this PR is very good. My suggestion is that you can incorporate the graphics and logic in the video into the MD of the PR. This will help me better understand your design concept🚀

Vivek1106-04 · 2026-09-10T05:47:19Z

@nzw921rx Added a How it works section to the description with the flow diagrams, as suggested. Six diagrams, rendered inline with mermaid so there is nothing external to open: pool ownership before and after, what is member-scoped vs. pipeline-scoped, the timer/dispatch split, savepoint vs. periodic trigger, failover lease reuse, and task cancellation across the two pools.

No rush on the review, and good luck with the 3.0 release.

DanielLeens

Thanks for the update — I re-reviewed the latest head from scratch rather than just the delta, since a checkpoint-scheduling change in the engine deserves a full pass every time new commits land, not just a diff check.

What Problem Does This PR Solve?

Every CheckpointCoordinator built its own two-thread scheduled pool (CheckpointCoordinator.java:227-238 on dev), and CheckpointManager builds one coordinator per pipeline. Timer threads therefore grew linearly with the number of active checkpoint-enabled pipelines on a member: a member running 500 pipelines held up to 1000 mostly-sleeping timer threads, purely to re-arm a periodic trigger and arm a timeout watchdog.

This PR introduces one SharedCheckpointScheduler per member (owned by SeaTunnelEngineContext, closed with it) backed by two pools: a tiny fixed checkpoint-timer pool that only fires and hands off, and an elastic checkpoint-dispatcher pool that actually runs the body, so a slow checkpoint body never delays another pipeline's timer. Each CheckpointCoordinator now leases a PipelineCheckpointScheduler from CheckpointManager instead of owning a pool, and the lease is reusable after cancelAll(), so a coordinator restored on master failover keeps working without rebuilding a pool.

Example: a member running 500 checkpoint-enabled pipelines goes from up to 1000 live per-pipeline timer threads to 2 shared checkpoint-timer-%d threads plus an elastic checkpoint-dispatcher-%d pool that only grows while checkpoint bodies are actually being handed off. This is exactly what SharedCheckpointSchedulerTest#testTimerThreadCountStaysFlatAsPipelinesGrow asserts.

1. Code Change Review

1.1 Core Logic Analysis

What changed since my last review, verified precisely: I diffed my previously-reviewed head against the current one commit by commit. Two new commits landed:

85efedd — swaps PipelineCheckpointTask's CompletableFuture import from java.util.concurrent.CompletableFuture to org.apache.seatunnel.engine.common.utils.concurrent.CompletableFuture. I checked this shim class: it overrides cancel, isCancelled, isDone, get, get(timeout, unit), complete, completeExceptionally as plain delegating calls to super (the JDK class) with identical semantics, so this is behavior-neutral. I also confirmed it is the established convention in this exact package — CheckpointCoordinator.java, CheckpointManager.java, PendingCheckpoint.java, StateStoreCheckpointIDCounter.java, StandaloneCheckpointIDCounter.java all already import the same shim. This is a legitimate consistency fix, not a functional change.
fd2841fd7 — adds checkpoint-timer- and checkpoint-dispatcher- as whitelisted system-thread prefixes in SeaTunnelContainer#isSystemThread (e2e test infra). This is a real CI-driven fix: I confirmed it directly addresses the exact failure the author described (CheckpointRestoreWithStopIT, SavepointRestoreIT and other e2e jobs failing the "no leaked threads" check because the shared pools now correctly outlive an individual job and are only torn down at member shutdown, not job completion). The fix is precise and minimal, matching the actual architectural change in this PR — it doesn't touch anything else.

I re-verified via git diff that CheckpointCoordinator.java, CheckpointManager.java, PipelineCheckpointScheduler.java, SeaTunnelEngineContext.java, CheckpointCoordinatorTest.java, and SharedCheckpointSchedulerTest.java are byte-for-byte identical to the version I reviewed and approved previously, so every invariant I traced then still holds on the current head. I re-walked them directly against the current source rather than trusting my earlier notes:

Runtime flow that reaches the changed code (unchanged from before, reconfirmed):

CheckpointManager.addCoordinator(...) -> new CheckpointCoordinator(...)
  -> scheduler = manager.leaseCheckpointScheduler(pipelineId)  // one lease per (jobId, pipelineId)

Periodic trigger (steady state):
tryTriggerPendingCheckpoint() [synchronized(lock)]
  -> scheduleTriggerPendingCheckpoint(type, delay)
       -> scheduler.schedule(...) -> PipelineCheckpointTask -> parent.dispatchAfter(...)
            [checkpoint-timer-%d fires after delay] -> dispatcher.execute(task) [hand-off only]
                 [checkpoint-dispatcher-%d runs task.run() -> body.run() -> tryTriggerPendingCheckpoint(...)]

Timeout watchdog: triggerCheckpoint(...) -> scheduler.schedule(..., timeout, ...) [re-checks pendingCheckpoints.get(id)!=null before firing CHECKPOINT_EXPIRED]

Master-failover reset: restoreCoordinator(alreadyStarted) -> cleanPendingCheckpoint(CHECKPOINT_COORDINATOR_RESET) -> scheduler.cancelAll() -> shutdown = false [SAME instance, SAME lease resumes]

Key findings, all reconfirmed directly against the current head:

Single-pending-checkpoint guard is untouched. tryTriggerPendingCheckpoint() still enters synchronized (lock) and checks isCompleted() || isShutdown() before doing anything, and the pendingCounter.get() > 0 short-circuit is unmodified. lock is a per-coordinator field, so sharing the timer cannot introduce cross-pipeline contention on this guard.
Master-failover restore/reuse is correctly preserved. restoreCoordinator() -> cleanPendingCheckpoint(CHECKPOINT_COORDINATOR_RESET) -> scheduler.cancelAll() -> shutdown = false reuses the SAME CheckpointCoordinator instance and therefore the SAME lease after a reset, exactly matching the old semantics. Directly exercised by SharedCheckpointSchedulerTest#testLeaseIsReusableAfterCancelAll.
No permanent scheduler wedge from one pipeline's failure — I traced this specifically given the task brief's emphasis on it. SharedCheckpointScheduler.dispatchAfter schedules () -> submit(task) on the timer pool, and submit() itself wraps dispatcher.execute(task) in a try/catch for RejectedExecutionException, so the lambda running on a timer thread can never throw uncaught. PipelineCheckpointTask.run() (which can rethrow a Throwable from a misbehaving checkpoint body) only ever executes on a dispatcher-pool thread, never a timer thread. A dispatcher-pool thread dying from an uncaught exception is ordinary, self-healing ThreadPoolExecutor behavior (the pool spawns a fresh worker on the next submission) — it cannot wedge the shared timer that every other pipeline depends on. This confirms one pipeline's misbehaving checkpoint body cannot silently stall or drop another pipeline's periodic trigger.
Savepoint vs. periodic-trigger interaction — this was asked by @nzw921rx after my last review, and I re-verified my own answer against the current head rather than just restating it. startSavepoint() busy-waits under the per-pipeline lock while pendingCounter.get() > 0, then calls createPendingCheckpoint(...) and startTriggerPendingCheckpoint(savepoint) directly — it never enters tryTriggerPendingCheckpoint(), so it never asks the scheduler to re-arm a periodic trigger. The only place a savepoint touches scheduler at all is inside startTriggerPendingCheckpoint(), shared by every checkpoint type, to arm that checkpoint's timeout watchdog. Since pendingCounter (untouched by this PR) already guarantees only one pending checkpoint per pipeline at a time, a savepoint's watchdog and a periodic checkpoint's watchdog are never both live for the same pipeline simultaneously. Confirmed correct.
MDC context propagation is equivalent, not just claimed — each command is wrapped via MDCTracer.tracing(command) at schedule time inside PipelineCheckpointScheduler.schedule(), and since re-arming is invoked recursively from inside an already-active MDCRunnable, MDCContext.current() resolves correctly at every re-arm.
Lifecycle wiring is complete and unchanged: SeaTunnelEngineContext.close() calls checkpointScheduler.close(), called once at member shutdown via SeaTunnelServer.java.

1.2 Compatibility Impact

Classification: Fully compatible. No config option added/removed/renamed, no default changed. No checkpoint state, serialization format, or storage schema touched — a checkpoint/savepoint written by a pre-PR member restores identically on a post-PR member and vice versa. The only externally observable difference is thread naming in stack dumps/profiles, and the e2e thread-leak whitelist fix in the newest commit is exactly the adjustment needed to keep test infrastructure honest about that naming change. Per-pipeline checkpoint isolation is preserved (PipelineCheckpointScheduler.outstanding is a private per-lease set).

One coordination note worth surfacing for the merge decision rather than a code defect in this PR: the PR description itself flags that #12149 and #12096 are open against the same constructor and the same cleanPendingCheckpoint() lines. gh pr view currently reports this PR as MERGEABLE against dev, so there is no live textual conflict today, but whichever of these three PRs merges last will need a careful rebase against the others' changes to that exact method. Worth a maintainer decision on sequencing before merge, not a blocker on this PR's own correctness.

1.3 Performance / Side-Effect Analysis

Timer-thread count drops from O(active pipelines) to a constant 2 per member — the intended fix, and a real reduction in idle-thread overhead.
The dispatch pool (Executors.newCachedThreadPool) is unbounded and now shared across every pipeline on the member rather than isolated per pipeline. Safe today because the dispatched body is non-blocking by design (heavy barrier work already offloaded via thenApplyAsync), but this remains a structural change in blast radius if a future change ever makes the dispatched body block (see carried-over Issue 2 below).
RejectedExecutionException from either pool during shutdown is caught and the task cancelled rather than lost, so no silent task leak on shutdown.

1.4 Error Handling and Logging

Rejection during shutdown is logged at debug, an appropriate level since it's expected during normal shutdown. No sensitive data logged. See the carried-over issues below for the one scheduling-race note (already absorbed by existing state guards, not a correctness gap).

Re-verification of previously-raised issues against the current head (all three were raised by me in my prior review; the author agreed with all three in a follow-up comment and said they planned to address them in follow-up commits on this branch — I checked the actual diff, and none of the three have been pushed yet, so I'm carrying them forward as still-open, still non-blocking, Low-severity items):

Issue 1 (carryover, still open): PipelineCheckpointTask cross-pool cancellation race. run() checks cancelled.get() once at the top; a task already dequeued by the dispatch pool right as cancelAll() fires can still run its body once. This remains provably benign — every caller (periodic trigger, watchdog) re-validates coordinator/checkpoint state under the same lock before acting — but the author agreed to add a documenting comment on run()/cancel() so a future contributor doesn't "fix" this in a way that changes the cancellation contract. Not yet in the diff.
Issue 2 (carryover, still open): unbounded checkpoint-dispatcher pool. Executors.newCachedThreadPool(...) in SharedCheckpointScheduler's constructor has no maximumPoolSize bound, and it is now a member-wide shared resource. The author agreed to add a bound as defense-in-depth. Not yet in the diff.
Issue 3 (carryover, still open): Thread.sleep(800) in SharedCheckpointSchedulerTest#testCancelBeforeTimerFiresStopsTheBody. Still the one Thread.sleep-based negative assertion in an otherwise Awaitility/CountDownLatch-based suite. False-failure risk only under host contention, not a false-pass risk that would hide a real regression. The author agreed to switch it to an Awaitility-based idiom. Not yet in the diff.

None of these three block merge on their own — I want to be clear I'm not escalating their severity, just accurately reporting that the code doesn't yet reflect the author's stated intent, so nobody assumes they're done.

2. Code Quality Assessment

2.1 Coding Standards

All three new classes (SharedCheckpointScheduler, PipelineCheckpointScheduler, PipelineCheckpointTask) carry clear class-level Javadoc explaining ownership, lifetime, and the two-pool split rationale, and non-trivial methods (lease, dispatchAfter, schedule, cancelAll, attachTimerFuture) all explain the "why". No missing-Javadoc issue to raise. CheckpointManager.checkpointScheduler and SeaTunnelEngineContext.checkpointScheduler fields both have Javadoc explaining ownership and the test-double fallback. The new SeaTunnelContainer whitelist entries carry an explanatory comment about why these threads are member-scoped rather than job-scoped — good practice for an easy-to-misread test-infra change.

2.2 Test Coverage and Test Stability

SharedCheckpointSchedulerTest (8 tests) directly proves every invariant claimed: body runs on dispatch thread not timer thread; cancelAll() is pipeline-scoped; cancel-before-fire and cancel-after-fire both stop the body; a lease is reusable after cancelAll() (the master-failover path); finished tasks self-remove; schedule-after-close returns an already-cancelled future; and the 500-lease/2-timer-thread acceptance criterion is asserted directly. CheckpointCoordinatorTest correctly threads real lease/cancel behavior through all new CheckpointCoordinator(...) call sites instead of relying on a manager mock that would otherwise NPE on leaseCheckpointScheduler.

Test stability rating: Stable, with the one Low-risk carryover noted above (Issue 3, Thread.sleep(800)) — unchanged assessment from my last review since the test file hasn't been touched.

2.3 Documentation Updates

None needed, none made — correctly reasoned in the PR description: no new engine Option, and a git grep across docs/en/docs/zh confirms nothing references the old per-pipeline thread-naming pattern. I'd also call out that the author's "How it works" section added to the PR description since my last review (six Mermaid diagrams covering pool ownership, the timer/dispatch split, the savepoint interaction, failover lease reuse, and cross-pool cancellation) is unusually good reviewer-facing documentation — I checked each diagram's claims against the actual source and they're accurate, not just illustrative.

3. Architectural Soundness

3.1 Elegance of the Solution

Precise fix, not a workaround: the actual defect was "the timer pool's lifetime and cardinality is tied to the coordinator's lifetime and count," and this PR fixes exactly that by relocating pool ownership to the member while keeping cancellation scoped correctly via the lease abstraction.

3.2 Maintainability

The three-class split cleanly separates "member-wide pool ownership," "per-pipeline cancellation scope," and "cross-pool task lifecycle." The explicit rationale comments (why not the coordinator executor, why a constant not a config) will save the next reader real investigation time.

3.3 Extensibility

The lease pattern (SharedCheckpointScheduler.lease(jobId, pipelineId)) generalizes cleanly if another per-pipeline periodic concern ever needs the same shared-timer/elastic-dispatch treatment.

3.4 Historical-Version Compatibility

No checkpoint/savepoint format, protocol, or serialization change — a job checkpointed on a pre-PR member restores identically on a post-PR member. No config default changed, so upgrading a running cluster to a build containing this PR requires no operator action.

4. Issue Summary

Number	Issue	Location	Severity	Raised by another reviewer
1	Carryover: `PipelineCheckpointTask` cross-pool cancellation race is benign but undocumented as intentional	`seatunnel-engine/seatunnel-engine-server/.../checkpoint/scheduler/PipelineCheckpointTask.java` (`run()`/`cancel()`)	Low	No (mine, still open)
2	Carryover: unbounded `checkpoint-dispatcher` pool is now a member-wide shared resource	`seatunnel-engine/seatunnel-engine-server/.../checkpoint/scheduler/SharedCheckpointScheduler.java` (constructor)	Low	No (mine, still open)
3	Carryover: `Thread.sleep(800)` in one negative-assertion test, inconsistent with the rest of the suite	`seatunnel-engine/seatunnel-engine-server/.../checkpoint/scheduler/SharedCheckpointSchedulerTest.java` (`testCancelBeforeTimerFiresStopsTheBody`)	Low	No (mine, still open)

5. Merge Recommendation

Conclusion: Ready to merge after fixes

Blockers — must be fixed
- CI is currently failing on the head commit (fd2841fd7) and must go green before merge. I pulled the fork's actual job logs (the apache-side Build check is only a pointer to the fork run) and found four failing jobs: all-connectors-it-6 (DatabendCDCSinkIT, container-level error), all-connectors-it-1 (NebulaGraphIT), and transform-v2-it-part-1 on both JDK 8 and JDK 11 (TestFilterRowKindIT#testFilterRowKindMultiTable, a data-assertion failure on the Flink-engine test container). I traced all four against this PR's actual diff (Zeta checkpoint-scheduler internals plus one e2e thread-whitelist entry) and none of them touch code paths this PR modifies — Databend/NebulaGraph are unrelated connector containers, and TestFilterRowKindIT exercises Flink's own translation layer and the Filter transform, neither of which this PR touches. Given the PR is also 35 commits behind dev and diverged, my recommendation is to sync with dev and get one clean CI run before merging, rather than assuming these are environmental based on code-path analysis alone.
Recommended fixes — non-blocking
- Issues 1-3 above (carried over from my last review): document the intentional cancellation race, bound the dispatcher pool's maximumPoolSize, and replace the one Thread.sleep(800) with an Awaitility-based idiom. The author already agreed with all three in a follow-up comment; they just haven't landed in the diff yet.
- Worth a maintainer decision on sequencing against #12149 and #12096, which touch the same cleanPendingCheckpoint() method, before merging any of the three.

Overall assessment: The core design is sound and I found nothing new to walk back from my previous approval — I re-traced every invariant that matters for an engine/checkpoint change (single-pending-checkpoint guard, master-failover restore/reuse, MDC propagation, and specifically the "can one pipeline wedge the shared scheduler" and "does a savepoint fight the periodic trigger" questions this task calls out) and all of them hold on the current head. The two commits since my last review are both good: a real consistency fix and a precise, correctly-scoped CI fix for a leak-check assertion that needed to catch up with the new member-scoped thread lifetime. The only reason I'm not re-approving outright this round is that CI is genuinely red on the commit I'm reviewing, and I'd rather see a clean run than assert from code-reading alone that four e2e job failures are unrelated. Once CI is green on a synced head, and ideally once the three carried-over Low-severity nits land, this is very close to done — nice work continuing to sweat the details here.

nzw921rx · 2026-09-15T14:24:28Z

+     * Timer threads only dispatch, so a small fixed pool is enough regardless of pipeline count.
+     * Two threads keep one slow dispatch hand-off from stalling the other timers.
+     */
+    private static final int TIMER_THREAD_NUM = 2;


I think it can be configured here, otherwise the more tasks there will be a lot of backlog, and the limited space for users to adjust. We should give users this right

Good question — worth separating the two pools here, since they behave very differently under load.

TIMER_THREAD_NUM = 2 (this line) is not a size backlog could ever pressure: ScheduledThreadPoolExecutor's internal delay queue is unbounded by JDK design, and these timer threads do nothing but a single dispatcher.execute(task) hand-off (SharedCheckpointScheduler.java:105-114) — no blocking work ever runs on them. Two threads exist only so one slow hand-off can't stall another pipeline's timer momentarily; they are never a throughput bottleneck regardless of pipeline count.

The pool that actually does the work is dispatcher (lines 76-81), and it's Executors.newCachedThreadPool() — also effectively unbounded (maximumPoolSize = Integer.MAX_VALUE, backed by a SynchronousQueue), so it grows threads on demand instead of queuing or backlogging. So today there genuinely is no user-adjustable "queue size" — but there's also no queue to size: both pools degrade by growing (threads) or queuing indefinitely (timer), never by throttling or rejecting under load.

The real, honest tradeoff here (the opposite direction from what you're worried about) is that dispatcher being unbounded means a member running many pipelines with slow checkpoint bodies could grow an unbounded number of dispatch threads. That's already tracked as a known Low-severity, non-blocking follow-up from my 2026-09-10 review (Issue #2: bound dispatcher's maximumPoolSize and make it configurable). I agree that's worth doing, but I don't think it should block this PR — "grow threads" is a strictly safer default than "reject/queue when saturated" would be, and the author already agreed to take it as a fast-follow.

@nzw921rx Late reply from me on this one, and the code has moved since, so here is where it stands.

The dispatcher is no longer unbounded. Since b0bc24603 it is a ThreadPoolExecutor with core equal to maximum at max(8, availableProcessors * 2), an unbounded LinkedBlockingQueue, and allowCoreThreadTimeOut(true) - so it grows on demand, reaps idle threads, is bounded in thread count, and never rejects. The reason it cannot use a bounded queue is that a rejection here is a dropped checkpoint trigger or timeout watchdog, which is a correctness problem rather than a resource one, so the bound is paid for in delay instead.

On making it configurable, which was your actual ask: I measured it rather than argued it. #12418 adds a scheduling-delay benchmark, and the comparison is on #12339. The short version is that scheduling delay is flat at about 10 us from 1 to 500 pipelines here against a knee between 100 and 500 for the per-pipeline model, and flat through 31% dispatch utilization - but past saturation it collapses to a p50 of 461 ms, so the pool size does matter in a regime the default interval does not reach.

Both @DanielLeens and @SEZ9 read that as: no checkpoint.scheduler-dispatch-thread-num option in this STIP, and scheduling-delay observability ahead of any tuning knob, since the metric is what would tell an operator they are approaching the regime where the option would help. I have taken that as the decision for this PR. If you want the option anyway, I would rather add it on the back of that metric than on its own, and I am happy to do both as a follow-up.

nzw921rx · 2026-09-15T14:26:03Z

+        try {
+            dispatcher.execute(task);
+        } catch (RejectedExecutionException e) {
+            log.debug("Shared checkpoint dispatcher is shut down, dropping checkpoint task", e);


Should users feel rejected here?

What if it fails here? Is the task status corresponding to it a failure or a silent no longer triggering checkpoint?

No — and the code here specifically guards against that reading. Both catch (RejectedExecutionException e) blocks in this file (the timer-side one around line 108-113, and this submit() one at line 119-122) are only reachable after close() has begun: closed = true then timer.shutdownNow() / dispatcher.shutdownNow() (SharedCheckpointScheduler.java:137-139). The class's own inline comments say this explicitly — "Only reachable once close() has begun; the coordinator is stopping anyway" / "Shared checkpoint dispatcher is shut down." Neither pool has a bounded queue that could saturate under normal load (see my reply on the line-59 thread about TIMER_THREAD_NUM), so there is no live-traffic path that reaches RejectedExecutionException at all — only an intentional-shutdown path.

So "rejected" here really means "the member (or this scheduler instance) is already shutting down," which is why it's logged at debug rather than warn/error — it's an expected, non-actionable event during teardown, not a user-facing failure.

Continuing from my reply above: in this specific path, "fails" means "the scheduler is already closing," and task.cancel(false) (line 111/121) marks the PipelineCheckpointTask cancelled and completes its internal future via cancel() (PipelineCheckpointTask.java:82-93) — genuinely cancelled, not exceptionally-failed. That's the correct terminal state for "we stopped because we're shutting down," and nothing downstream treats it as a checkpoint failure, because by the time close() runs, the owning CheckpointCoordinator/pipeline is also being torn down — there's no live consumer left to silently miss a checkpoint.

There IS a real "silent no longer triggering checkpoint" risk in this design, but it's not in this rejection path — it's in the checkpoint body itself. tryTriggerPendingCheckpoint() (CheckpointCoordinator.java:722) is not wrapped in try/catch, and it's the same method that re-arms the next periodic trigger (scheduleTriggerPendingCheckpoint(...) at line 802) before returning. If it throws for an unrelated reason (bug, NPE, etc.) before reaching that line, the periodic loop breaks and checkpointing silently stops for that one pipeline.

I checked whether this PR introduces or worsens that, and it doesn't — I diffed against the pre-PR code on dev, where CheckpointCoordinator called scheduler.schedule(runnable, ...) directly on a real ScheduledThreadPoolExecutor. That path is actually more silent than today's: ScheduledThreadPoolExecutor.schedule() wraps the runnable in an internal future-backed task whose run() catches Throwable and stores it via setException(), with zero stderr output, since the returned future was never queried. Today's dispatcher.execute(task) runs PipelineCheckpointTask as a plain Runnable (not future-wrapped), so an uncaught exception now propagates to the worker thread's default uncaught-exception handler, which prints a stack trace to stderr — strictly more visible than before, even though it's still not routed through SeaTunnel's own logger or any alert.

This is a real, pre-existing gap — not something this PR needs to fix, and not something it makes worse — but I think it's worth a dedicated follow-up: wrap the scheduled body (or PipelineCheckpointTask.run()) with a catch that logs at WARN/ERROR through SeaTunnel's logger, since checkpoint triggering going silent is exactly the kind of failure an operator needs visibility into. I would not treat it as a blocker for this PR specifically, since it predates it and isn't made worse by it, but it's a legitimate item for a follow-up issue.

@nzw921rx To answer both of your questions on this thread directly, with the code as it stands today.

Can a user hit this rejection? No. Both RejectedExecutionException catches in this class are reachable only after close() has begun, which happens when the member's engine context shuts down. The dispatch pool is bounded in threads but backed by an unbounded queue and never rejects under load, and the timer's delay queue is unbounded by JDK design, so there is no live-traffic path into either catch. That is why they log at debug: it means "this member is stopping", not "a checkpoint was refused".

Is the task then failed, or silently no longer triggering? Neither, in that path. task.cancel(false) puts the PipelineCheckpointTask in a genuinely cancelled terminal state, and the owning coordinator is being torn down at the same time, so there is no live consumer left to miss a checkpoint.

Your underlying worry - work disappearing without anyone being told - was the right thing to ask about, and it does have a real instance elsewhere, which I have now fixed in da0cbbf52. The checkpoint timeout watchdog used to run the whole expiry inline on a dispatch thread, including an IMap retry loop and a wait on the coordinator lock. It now hands that to the coordinator executor. That executor is bounded over a SynchronousQueue and does abort when saturated, and there a dropped task would be exactly the silent failure you are describing - the checkpoint would stay pending with nothing left to fail it - so the rejection is caught and the expiry runs inline instead, with a WARN rather than a debug line. Covered by CheckpointCoordinatorTest#testCheckpointExpiryHandlingNeverRunsOnTheDispatchThread.

The separate gap @DanielLeens identified on this thread - tryTriggerPendingCheckpoint not being wrapped in try/catch, so an unrelated throw could break the periodic re-arm - predates this PR and is not made worse by it. I agree it deserves a logged catch and will file it as a follow-up rather than widen this diff.

Filed as #12442, as promised above - an unexpected throw in tryTriggerPendingCheckpoint leaving a pipeline with no armed timer, logged nowhere through SeaTunnel's own logger. Pre-existing on dev and not made worse by this PR, so it is tracked rather than fixed here.

nzw921rx · 2026-09-15T14:34:44Z

+        return new PipelineCheckpointScheduler(this, jobId, pipelineId);
+    }
+
+    /**


How can we ensure the completeness of his lifecycle here? For example, can there only be one jobId+pipelineId?

Good catch on the ambiguity — SharedCheckpointScheduler itself keeps no registry at all, so there is no code-level assertion that only one PipelineCheckpointScheduler can exist per (jobId, pipelineId). lease(jobId, pipelineId) (lines 95-97) just returns a brand-new, unregistered object on every call; uniqueness is not enforced by this class.

What actually makes this safe in practice is the calling discipline one layer up, which I traced end-to-end:

CheckpointCoordinator.scheduler is a final field, assigned exactly once in the constructor (CheckpointCoordinator.java:155,229) via manager.leaseCheckpointScheduler(pipelineId).

CheckpointManager builds exactly one CheckpointCoordinator per pipelineId in its coordinatorMap (CheckpointManager.java:132-181, keyed by CheckpointCoordinator::getPipelineId), so exactly one lease is created per pipeline for that CheckpointManager's lifetime.

Critically, on master-failover reset the SAME CheckpointCoordinator instance is reused: cleanPendingCheckpoint(CHECKPOINT_COORDINATOR_RESET) calls scheduler.cancelAll() (CheckpointCoordinator.java:1196) on the existing lease and flips shutdown = false — it does not call leaseCheckpointScheduler() again. So restart/failover doesn't create a second lease for the same (jobId, pipelineId); it reuses the first one.

So there's no "replace vs. leak vs. exception" branch to worry about, because re-registration never happens on the paths I found. A second lease for the same (jobId, pipelineId) would only appear if a brand-new CheckpointManager were constructed for the same pipeline while an old one was still alive — which would itself be a bug in the job-restart lifecycle outside this PR's scope, not something SharedCheckpointScheduler needs to guard against. And because leases hold no shared state (each just has its own local outstanding set, PipelineCheckpointScheduler.java:45), even a hypothetical duplicate lease wouldn't corrupt the shared pools — it would just be two independent cancellation domains, a correctness smell rather than a resource leak, since nothing is centrally registered to go stale.

If you'd like a hard guarantee here (a defensive check-and-throw on duplicate lease, or an explicit registry with dispose()), I'd treat that as a legitimate hardening ask — but I don't see a live code path today that would actually trigger the failure mode you're describing.

@nzw921rx On lifecycle and uniqueness: @DanielLeens traced this correctly, and I can confirm it against the current head.

SharedCheckpointScheduler deliberately keeps no registry. lease(jobId, pipelineId) returns a new object each call and registers nothing, which is the reason dropping a lease when its job ends leaks nothing - there is no central map to go stale.

Uniqueness comes from the layer above, not from this class: CheckpointCoordinator.scheduler is final and assigned once in the constructor, and CheckpointManager builds exactly one coordinator per pipelineId in its coordinatorMap. On a master-failover reset the same coordinator instance is reused - cleanPendingCheckpoint(CHECKPOINT_COORDINATOR_RESET) calls scheduler.cancelAll() on the existing lease and leaves the lease usable - so a reset does not create a second lease for the same (jobId, pipelineId).

A duplicate lease would therefore require a second CheckpointManager for the same pipeline while the first is still alive, which would be a job-restart lifecycle bug outside this class. Even then the blast radius is small, because leases share no state: each holds only its own outstanding set, so two leases would be two independent cancellation domains rather than a corrupted shared one.

SharedCheckpointSchedulerTest covers the parts that matter here: cancellation scoped to one lease leaving other leases untouched, a lease still usable after cancelAll, and cancellation working both before the timer fires and after the body has been dispatched. If you want a hard guarantee on top - a check-and-throw on a duplicate lease, or an explicit registry with dispose() - I am happy to add it, but I would rather not introduce a registry whose only purpose is to be a thing that can leak.

nzw921rx · 2026-09-15T14:40:48Z

I have another question. If thread scheduling slows down due to IO or a certain jobId, the CK of subsequent tasks will be delayed. Do we have any better solutions?

nzw921rx · 2026-09-15T14:41:34Z

CI error, please rebase upstream/dev

nzw921rx

Have you considered excessive solutions on your end? For example, clearly defining it as experimental ability? I am concerned about the scheduling efficiency and latency of large-scale tasks, and if users upgrade, there will be uncertainty involved. I am very worried.

nzw921rx · 2026-09-15T15:56:40Z

Thank you for such a detailed design. I think it should comply with the STIP principle. You can refer to the following link to initiate STIP🔥

https://seatunnel.apache.org/community/contribution_guide/STIP

DanielLeens · 2026-09-15T16:15:54Z

Replying to your latest round of comments (review comment, and three issue-level comments from today) together, since they're all design/process-level rather than line-specific.

On "excessive solutions" / marking this experimental (review comment)

Have you considered excessive solutions on your end? For example, clearly defining it as experimental ability?

I don't think an experimental flag fits this particular change, for a structural reason rather than a dismissal of the concern: this PR has no new Option, no config default change, and no checkpoint/savepoint format or protocol change — I verified that directly in my 2026-09-10 review (no new config surface, git grep across docs/en/docs/zh found nothing referencing the old per-pipeline thread-naming pattern to update). The old per-coordinator pool code path is fully removed, not kept behind a switch, so there's no clean on/off seam to gate as "experimental" — it's either correct for everyone or it isn't. If the community wants a staged rollout for a change like this, I'd suggest that's better done via the normal RC/soak cycle than an in-product experimental toggle, since the architecture doesn't leave a natural place to put one.

That said, I do share the underlying worry about validating this carefully before it reaches everyone, which is exactly why my CHANGES_REQUESTED review is still blocking on a clean CI run rather than approving from code-reading alone (see CI note below).

On scheduling efficiency / large-scale latency (review comment) and head-of-line blocking (issue comment)

I am concerned about the scheduling efficiency and latency of large-scale tasks
If thread scheduling slows down due to IO or a certain jobId, the CK of subsequent tasks will be delayed. Do we have any better solutions?

This is exactly the tradeoff the timer/dispatch split is built to avoid for the "arm the next trigger" step. The shared checkpoint-timer pool only ever runs () -> submit(task) (SharedCheckpointScheduler.java:107) — a single non-blocking dispatcher.execute(task) hand-off. It never runs the actual checkpoint body, so a slow or blocked pipeline cannot stall another pipeline's timer from firing on schedule. The real work (the checkpoint body, which can call into storage/RPC) runs on dispatcher, and dispatcher is unbounded (newCachedThreadPool), so job A's slow IO occupies its own dedicated thread there rather than contending for a shared, limited worker with job B's checkpoint body.

That said, it is still a shared JVM-wide resource in a broader sense: if enough pipelines are simultaneously blocked on slow IO at very large scale, you could see member-wide resource pressure (thread count, context-switching, memory) even though no single mutex or bounded queue is shared between jobs. That's the same tradeoff already tracked as Issue #2 (Low, non-blocking) in my CHANGES_REQUESTED review — bounding and making dispatcher's pool size configurable is the natural follow-up. So: for the direct "job A blocks job B's trigger" scenario, the design already avoids it structurally today; for member-wide pressure at extreme scale, that's the same known, already-flagged carryover rather than a new gap.

On CI / rebase (issue comment)

CI error, please rebase upstream/dev

Agreed, and this matches the actual blocker in my 2026-09-10 CHANGES_REQUESTED review: CI was red on the head commit for reasons I traced back to being unrelated to this PR's own diff, and the branch was already well behind dev at that point. The head commit is still the same one I reviewed then, so this is still open — rebasing onto current dev and getting one clean CI run is the right next step before this can move forward.

On STIP (issue comment)

I think it should comply with the STIP principle... You can refer to the following link to initiate STIP

Reasonable suggestion given the scope — this does touch core, node-wide checkpoint-scheduling internals, so going through STIP for broader design visibility and consensus makes sense to me. That said, whether to require it is a process call for the author and the PMC/maintainers to make, not something I can gate as a reviewer.

Vivek1106-04 · 2026-09-16T03:50:01Z

@nzw921rx Thanks for the review. Answering your three points.

1. Rebased

Rebased onto current dev and force-pushed. Head is now 7310dd7a2. Locally: spotless:check clean, full -am build of seatunnel-engine-server and seatunnel-e2e-common succeeds, and CheckpointCoordinatorTest (16) plus SharedCheckpointSchedulerTest (8) pass. CI is running on the new head.

2. "If thread scheduling slows down due to IO or a certain jobId, the CK of subsequent tasks will be delayed"

This is the right thing to check, and it is the reason the scheduler is split into two pools rather than one.

A timer thread never runs checkpoint work. In SharedCheckpointScheduler, dispatchAfter schedules () -> submit(task) on the timer, and submit does nothing but dispatcher.execute(task). The dispatcher is a cached pool, so execute neither blocks nor queues: it hands the task to an idle thread or creates one. The timer thread's work per firing is a constant-time, non-blocking hand-off.

So a checkpoint that blocks on IO or a slow RPC occupies a dispatcher thread, not a timer thread. The next periodic trigger for every other pipeline still fires on schedule, and so does the next one for the slow pipeline itself.

Only two things are ever scheduled, both in CheckpointCoordinator:

the periodic re-arm, scheduleTriggerPendingCheckpoint -> tryTriggerPendingCheckpoint
the checkpoint-timeout watchdog, armed inside startTriggerPendingCheckpoint

Both of those bodies run on the dispatcher.

Worth noting this is stricter than what dev does today. On dev each coordinator builds Executors.newScheduledThreadPool(2) and runs the bodies on those same two threads. A checkpoint body that blocks there consumes one of that pipeline's own two timer threads, and a blocked trigger plus a blocked watchdog can occupy both, delaying that pipeline's own next tick. After this PR that self-interference is gone as well, because the timing threads are no longer the ones doing the work.

The honest caveat: the dispatch pool is unbounded. If N pipelines all block at once, up to N dispatcher threads exist for as long as they block, and idle ones are reaped after the cached pool's 60s keep-alive. That worst case is no worse than dev's fixed two-threads-per-pipeline, and the steady state improves from 2 * pipelines threads to 2 timer threads per member plus whatever is actively running.

I deliberately did not bound the dispatch pool. A bounded pool over a SynchronousQueue rejects when saturated, and a rejected task here is a silently dropped checkpoint trigger or a silently dropped timeout watchdog, which is worse than an extra thread. Bounding it properly needs a rejection policy, and that is a design question rather than something to decide inside this PR.

3. Experimental flag

@DanielLeens already covered this in #12165 (comment), and I agree with their reading, so this is just me confirming it as the author rather than a new argument.

I do not think a flag fits, for a structural reason. The PR adds no new Option, changes no default, and changes no checkpoint or savepoint format or protocol. The behavior change is entirely "which thread runs the timer". There is no user-visible surface for an experimental flag to gate, so the only way to make it switchable would be to keep both the old per-coordinator pool and the new shared one alive behind a branch, which doubles the code under test for no configuration benefit.

If the concern is that the scheduling change should be reversible in the field, the cleaner form of that is the STIP discussion below rather than a config toggle.

4. STIP

Agreed. This touches node-wide checkpoint scheduling internals in the engine, which fits the "engine-level change" bar. I will open [STIP-40][Improve][Zeta] covering the design, the alternatives, the unbounded-dispatch-pool tradeoff above, and the test plan, and post to [email protected] first. This PR will reference the STIP number once it is open.

Vivek1106-04 · 2026-09-16T03:55:08Z

@nzw921rx Opened the STIP as requested: [STIP-40][Improve][Zeta] Member-wide shared checkpoint trigger scheduling - #12339

It covers Background, Motivation, Goals, Non-Goals, Design, Compatibility, Test Plan, Alternatives and References, and carries the thread-ownership and dispatch diagrams.

Two things I flagged there rather than smoothed over, since they are the parts that need the list's opinion:

Bounding the dispatch pool. [Improve][Zeta] Share checkpoint trigger scheduling across active pipelines #12122 asks for a shared bounded scheduler. The timer pool is bounded; the dispatch pool currently is not. Bounding it properly means picking a rejection policy, and every candidate has a cost: caller-runs pushes checkpoint work back onto a timer thread and reintroduces the head-of-line blocking this design removes, an unbounded queue trades threads for latency and memory, and a bounded queue needs agreement on what a member should do when it cannot trigger a checkpoint on time. This is the main open question in the proposal.
Timer pool size as a constant vs. an Option. No strong view from me; I will follow the list.

Two asks:

I do not have permission to add the design label to the issue. Could a committer add it so it shows up in the STIP list?
I will post the proposal to [email protected] next, linking the issue.

Treating this PR as the candidate implementation for STIP-40, not as an accepted design. Happy to hold it until the STIP discussion concludes.

DanielLeens · 2026-10-23T07:28:13Z

Thanks @nzw921rx for dispatching the benchmark on top of #12896 and attaching the raw data, and thanks @Vivek1106-04 for syncing. I read the three comments against the unchanged head 771fce22127f51c0419f0633e8c95419a4b2493a (static read only; I did not run or dispatch anything). This is a thread follow-up, not a new full review, because no commit has been pushed since the sync merge. My source verdict is unchanged.

What the numbers say (median of 2 alternating runs on the same worker, CheckpointScheduling.periodicTriggerDelay, lower is better):

pipelineNum	Baseline (us/op)	Candidate (us/op)	Candidate CV
10	91.967	135.108	12.42% (baseline 1.35%)
100	86.367	124.745	7.17% (baseline 2.95%)
500	106.186	122.286	3.59% (baseline 4.88%)

My reading (an inference from the posted table, not a measurement of mine): the candidate adds roughly 16 to 43 microseconds of trigger delay per due trigger, which is consistent with the extra timer-to-dispatcher hop this design introduces. That is tiny next to a checkpoint interval measured in seconds, and the PR's goal is the checkpoint-timer thread count, not trigger latency, so I do not read this as a functional regression. But it is a real, repeatable direction in the data, so it should be explained in the PR rather than left as a bare table. The 10-pipeline row also has a candidate CV of 12.42% with only 2 runs, so that row is not a stable estimate yet.

The RejectedExecutionException in the second comment: the stack shows checkpoint-dispatcher-N running PipelineCheckpointTask.run() and then failing in CheckpointCoordinator.createPendingCheckpoint (supplyAsync) because the coordinator executor was already Terminated. That is a dispatch body racing the end-of-trial teardown, not a trigger that was lost mid-measurement (the same iteration reports 100 of 100 due triggers measured with 0 skipped). It is also exactly the behavior of my carried-over Issue 2: PipelineCheckpointTask.java:100 rethrows after completing the future, so the error ends the dispatch worker and surfaces only through the default uncaught-exception handler on stderr. It stays Low severity, but this log is concrete evidence for it.

Concrete next steps for @Vivek1106-04:

Post a short interpretation under the benchmark table: the per-trigger delay delta, why it comes from the extra hop, and that it is acceptable against checkpoint intervals. If you can, add a thread-count or memory data point for the same pipelineNum values, since that is the benefit this PR claims.
Ask for or run more than 2 rounds for the pipelineNum=10 row, so the 12.42% CV can be judged.
Fix Issue 2: log the failing body with pipeline and job context inside PipelineCheckpointTask.run() and do not rethrow, since completion already carries the exception. That also removes the stderr noise in the benchmark output. If you prefer to keep it as is, please say so and I will keep it as a Low note.
Refresh the PR description (still says "elastic dispatch pool"; the code has a bounded pool).

CI: the apache-side Build check is red; per my 10-22 note the earlier fork failure was SplitClusterFaultToleranceIT.testStreamJobCancelResolvesWhenWorkerCrashesBeforeCancelAck, which relates to the separate cancellation fix #12311 and not to this diff. I have not re-diagnosed the run for the new head; please post the fork run link for 771fce221 once it finishes. Status evidence: head 771fce22127f51c0419f0633e8c95419a4b2493a, base dev, compare diverged (ahead 9, behind 12), mergeable state blocked. A committer's final approval is still needed; I cannot give it.

Vivek1106-04 · 2026-10-24T07:42:12Z

@nzw921rx thanks for running the benchmark on top of #12896 and attaching the raw data. @DanielLeens thanks for the read. I went through the raw data and then traced the slowdown down to the line that causes it. Short version: the whole difference comes from the timer -> dispatcher hand-off, which I'm keeping deliberately. Details and evidence below.

What the raw data shows

The run used 1 fork x 5 iterations x 10 s, which is about 100 samples per iteration. In those JSON files there are two separate effects:

A steady shift. If you drop the samples above 500 us, the candidate is +28-35 us at P=10/100 and +10-15 us at P=500. That shows up in every run, on both JDKs. The candidate is flat across P (about 110-120 us), while dev rises at P=500.
Rare 1-4 ms samples. There are 22 of them in the candidate (out of about 6,000) and 1 in dev. These cause all of the extra CV. Without them the candidate's per-iteration CV is 0.7-2.9%, the same as dev. In the 12.42% row (P=10, JDK 11), one iteration with a 4.2 ms sample moves the mean of its 100 samples by about 40 us. More rounds would not fix that CV; the spikes would just land in different iterations.

Where the time goes

I rebuilt exactly the two commits you compared (4dcfa28cc and 771fce221), added a throwaway per-stage tracer (System.nanoTime at each step of the trigger path, plus /proc/thread-self/schedstat for the threads that carry it), and ran them on fork runners in the same alternating order. The tracer patches and workflow are on bench-diag-12165. The tracer adds the same overhead to both sides.

Run 40061505634 (JDK 11 + 17, P=10/100/500) reproduced both effects:

Timer lateness (due -> timer thread runs) is the same on both sides: about 80 us at p50.
The steady difference is the hop itself: timer fires -> dispatcher thread starts the body is 22-25 us at p50. Lock wait and createPendingCheckpoint are the same on both sides.
During every candidate spike the dispatch pool was idle (8 threads, 0 active, 0 queued). So it is not thread creation, not a backlog, and not other pipelines.
The spikes that aren't safepoints (those hit both sides) are run-queue wait: the dispatcher thread was runnable but had no CPU. The kernel's run_delay for that thread matches the spike, for example 1853 us of run-queue wait in a 1868 us hand-off.

A scheduler trace (bpftrace on sched_wakeup/sched_switch, run 40064392859) showed what held the CPU. That run's latencies are inflated by the tracing itself, so I only use it for this. For slow dispatcher wakes, the thread that ran just before the dispatcher on its CPU was most often the JMH worker (40 cases), which busy-spins from about 1 ms before each due time until pendingCounter moves. Next most common were the JIT compiler threads. For dev's coordinator threads the JMH worker shows up once. On a 4-vCPU runner there is often no idle CPU at the moment the dispatcher wakes, so it waits for the running thread's time slice to end, about 0.5-3 ms, which is the spike range.

To confirm the cause rather than just correlate it, run 40071856278 adds a third build: 771fce221 with only timer.schedule(() -> submit(task), ...) changed to timer.schedule(task, ...). Same shared scheduler and same two timer threads, just no hand-off. The three builds alternated on one runner, two rounds each (us/op; spikes are samples over 500 us):

JDK	pipelineNum	dev	PR	PR without hand-off
11	10	96.0-97.0, 0 spikes	124.7-138.7, 1-2	95.8-96.5, 0
11	100	92.6, 0	131.6-137.6, 3-6	93.8-94.1, 0
17	10	87.5-89.5, 0	126.0-138.6, 1-6	85.8-89.7, 0-1
17	100	83.6-83.9, 0	118.4-122.1, 2-5	84.0-84.4, 0

Over all ~4,000 triggers per trial in the trace, p99.9 is 0.2-0.4 ms for dev and for the no-hand-off build (one no-hand-off trial: 1.4 ms), and 0.65-2.1 ms for the PR. The same results on both runner CPUs this run got (EPYC 7763 and 9V74). So the shared timer itself costs nothing, and the whole delta is the hand-off.

Why I'm keeping the hand-off

The hand-off is there so timer threads never run checkpoint bodies. Those bodies take the coordinator lock, and on dev startSavepoint holds that lock across its sleep loop (#12441). The expiry path also does IMap work. If the 2 shared timer threads ran those bodies, one slow coordinator would delay timers for every pipeline on the member. That is the coupling Issue 1 is about, so the hand-off makes the worst case smaller. The cost is one extra thread wake per trigger: about 25 us, plus an occasional wait of up to one scheduler time slice when every CPU is busy. Against checkpoint intervals in seconds that is negligible, and it does not grow with the number of pipelines. I'd rather keep that trade than move lock-taking work onto the timer threads. If you'd prefer the periodic trigger to run on the timer thread once #12511 has taken the savepoint wait off the lock, the change is the one line above, and I can do that as a follow-up.

The benefit, measured in the same traces

From the safepoint logs of the JDK 17 run (-Xlog:safepoint, whole JVM run), dev's time-to-safepoint grows with its timer threads (2 per pipeline), while the PR stays flat:

pipelineNum	reach safepoint p50, dev / PR	total pause p50, dev / PR	max pause, dev / PR
10	10.0 / 8.6 us	1.35 / 1.33 ms	7.4-8.4 / 6.8-7.4 ms
100	17.6-18.2 / 7.8-8.1 us	1.74-1.87 / 1.50-1.53 ms	8.9-9.4 / 7.2-9.3 ms
500	56-57 / 8.4-8.6 us	3.27-3.34 / 1.76-1.80 ms	33.6-37.3 / 6.6-6.9 ms

The scheduler thread count is 1,000 on dev at P=500, and on the PR it is 2 timer threads plus at most max(8, 2 x cores) dispatchers (8 on these runners), regardless of P.

The `RejectedExecutionException`

This is not a lost trigger. The fixture's teardown calls coordinatorExecutor.shutdownNow() before manager.cancelCheckpoint(...), so a trigger that was already handed to a dispatcher in that gap hits a terminated executor. dev has the same race, but there the body runs inside its own ScheduledThreadPoolExecutor, which stores the exception in the future and prints nothing. On this PR it shows up only because PipelineCheckpointTask.run() rethrows. That is @DanielLeens's Issue 2. The same iteration reports 100 of 100 due triggers measured, 0 skipped. I'll fix Issue 2 in the next push: log with job/pipeline context and don't rethrow, since completion already carries the error. That also removes the stderr noise.

CI on `771fce221`

Fork run 39773641731: the only failure is unit-test (17, ubuntu-latest), SavePointTest#testSavePointButJobGoingToFail (the job stays RUNNING instead of FAILED within 120 s). JDK 11 and both Windows unit-test legs passed. That is the SubPlan restore-wait race described in #12441 and fixed by #12511 / #12623, not this diff. Kept separate from the benchmark question, as you asked.

Next push

Issue 2 fix with a test for a throwing body
PR description refresh: bounded dispatch pool, current test counts, and a short "cost of the hand-off" section with the table above
New CI link on the new head

…atch thread PipelineCheckpointTask.run() rethrew a body failure after completing its future exceptionally. The rethrow ended the dispatch worker and reported the error only through the default uncaught-exception handler on stderr, outside the engine log. The failure is already carried by the task's future, so log it with the job and pipeline id and keep the worker.

DanielLeens

Thanks @Vivek1106-04 for the new commit and for the very thorough benchmark investigation. The no-hand-off control build, the run-queue (run_delay) evidence and the safepoint table answer my 2026-10-23 questions well. I accept that the extra ~25 us per trigger comes from the timer-to-dispatcher hand-off, which is a deliberate trade-off, and that the benefit is the flat thread count and time-to-safepoint. This is a full re-read at head 8766ee4dbc71. Since my last review at 771fce2212 there is one new commit, 8766ee4dbc ("Log a failing checkpoint task instead of killing its dispatch thread"). It fixes my previous Issue 2, and the PR description now describes the bounded pool and the cost of the hand-off, which closes my previous Issue 4. Thank you!

While re-checking the branch against current dev, though, I found one new blocker. It did not exist at my last full review: it comes from #12448, which merged into dev after that review and after your last sync (details in Issue 1).

What Problem Does This PR Solve?

User pain point: every CheckpointCoordinator built its own two-thread scheduled pool, and CheckpointManager builds one coordinator per pipeline, so the number of checkpoint-timer threads grew linearly with active pipelines.
Fix approach: one member-wide SharedCheckpointScheduler owned by SeaTunnelEngineContext. It has two checkpoint-timer threads that only fire, and a bounded checkpoint-dispatcher pool that runs the body. Each coordinator leases a PipelineCheckpointScheduler, so cancelAll() stays pipeline-scoped.
One-sentence summary: replace N per-pipeline timer pools with one shared timer plus a bounded dispatch pool, with no config, checkpoint-format or default-behaviour change.

1. Code Change Review

1.1 Core Logic Analysis

New commit 8766ee4dbc:

// PipelineCheckpointTask.java:99-115, before
} catch (Throwable t) {
    completion.completeExceptionally(t);
    throw t;                     // ended the dispatch worker, error only on stderr
}
// after
} catch (Throwable t) {
    log.error("Checkpoint task failed for job {} pipeline {}", jobId, pipelineId, t);
    completion.completeExceptionally(t);
}

This now matches what dev effectively did: ScheduledThreadPoolExecutor's FutureTask also captured the throwable and kept the worker alive. The PR adds a log line with job/pipeline context through the engine logger. jobId / pipelineId are passed from PipelineCheckpointScheduler.schedule, and the lease already holds both, so the change is small. The new testFailingBodyCompletesExceptionallyAndKeepsItsDispatchThread asserts both that the exception is propagated through the future (assertSame(failure, thrown.getCause())) and that the worker thread survives. If the throw t were restored, the worker would reach TERMINATED and the test would fail, so it can catch the regression.

Semantic interaction with dev (#12448): the branch is diverged (ahead 10, behind 17). GitHub reports no textual conflict, and my trial merge also has none. But #12448 (df94fb07db) added CheckpointCoordinatorScheduledTriggerTest, which builds a coordinator on a plain Mockito.mock(CheckpointManager.class):

// dev: CheckpointCoordinatorScheduledTriggerTest.java:44, 77, 87
CheckpointManager checkpointManager = Mockito.mock(CheckpointManager.class);
...
coordinator.scheduleTriggerPendingCheckpoint(CheckpointType.CHECKPOINT_TYPE, 0);
...
coordinator.cancelCheckpoint();

On this branch the coordinator's constructor does this.scheduler = manager.leaseCheckpointScheduler(pipelineId) (CheckpointCoordinator.java:242). A plain mock returns null for it, so scheduleTriggerPendingCheckpoint hits scheduler.schedule(...) (CheckpointCoordinator.java:572) and throws NullPointerException before the test can observe the coordinator failure. The finally block's cancelCheckpoint() then reaches scheduler.cancelAll() (line 1283) and throws again. This is exactly the mocked-manager case that your own CheckpointCoordinatorTest helper (mockCheckpointManager(), lines 103-110) was written for. The fork CI runs only the branch head, which does not contain #12448, so it cannot show this. It would show up on the merged result (merge queue or dev).

The production side of #12448 merges cleanly in the trial merge: its try/catch lambda body is still passed to scheduler.schedule(...), which has the same (Runnable, long, TimeUnit) signature on the lease. So only the test needs adapting.

Runtime path (unchanged apart from the logging change):

SeaTunnelServer -> SeaTunnelEngineContext (owns SharedCheckpointScheduler)
  -> CheckpointManager.leaseCheckpointScheduler(pipelineId) -> CheckpointCoordinator.scheduler
  -> scheduleTriggerPendingCheckpoint: lease.schedule -> checkpoint-timer fires
     -> hand-off to checkpoint-dispatcher -> PipelineCheckpointTask.run (cancel check, body, log on failure)
  -> coordinator close/reset: lease.cancelAll() (pipeline-scoped); context close: both pools shut down

1.2 Compatibility Impact

Fully compatible. No option, default, checkpoint/savepoint format or storage schema is touched. The observable difference is thread naming, and an ERROR log now appears where dev stayed silent for a failing scheduled body.

1.3 Performance / Side-Effect Analysis

Same as before, now backed by your measurements: about 25 us of hand-off per trigger, plus an occasional wait of up to one scheduler time slice when every CPU is busy. In exchange the thread count and time-to-safepoint stay flat across pipelines. The new log fires only when a body throws, and a failing trigger does not re-arm itself, so there is no log loop.

1.4 Error Handling and Logging

Issue 1: the PR breaks dev's new CheckpointCoordinatorScheduledTriggerTest once merged (semantic merge conflict with #12448)

Location: CheckpointCoordinator.java:242 and :572 (this PR) versus seatunnel-engine/seatunnel-engine-server/src/test/java/org/apache/seatunnel/engine/server/checkpoint/CheckpointCoordinatorScheduledTriggerTest.java:44-87 (on dev)
Problem: the plain CheckpointManager mock returns a null lease, so the test fails with an NPE instead of reaching its FAILED assertion.
Risk: the merged result has a failing engine unit test. The merge queue would reject it, or dev would go red, and the branch's own CI cannot detect it.
Best improvement: merge the latest dev, then give that test a real lease the same way CheckpointCoordinatorTest.mockCheckpointManager() does, for example Mockito.when(checkpointManager.leaseCheckpointScheduler(Mockito.anyInt())).thenAnswer(...) backed by a test-scoped SharedCheckpointScheduler that is closed in finally. Keep every assertion of the #12448 test as it is. With the lease in place, the test also checks that #12448's wrapped body still fails the coordinator through the shared dispatcher.
Severity: High
Raised by another reviewer: No

Issue 2: residual dispatch-pool coupling through savepoint lock-holding (carryover, tracked)

Location: CheckpointCoordinator.java startSavepoint versus tryTriggerPendingCheckpoint's synchronized (lock)
Problem / Risk: enough pipelines savepointing at the same time can occupy the bounded dispatch pool and delay other pipelines' triggers. The delay is bounded and there is no data loss.
Best improvement: tracked in #12441 / #12511.
Severity: Low
Raised by another reviewer: Yes, related to @nzw921rx's head-of-line concern.

Issue 3: no guard on the thenAcceptAsync hop if the coordinator executor is ever bounded (carryover)

Location: CheckpointCoordinator.java startTriggerPendingCheckpoint
Problem / Risk: not reachable today. This is a note for whoever bounds the coordinator executor later.
Severity: Low
Raised by another reviewer: No

Issue 4: complete the future before logging (new, optional)

Location: PipelineCheckpointTask.java:109-110
Problem: log.error(...) runs before completion.completeExceptionally(t). If logging itself throws (for example under OutOfMemoryError, the case where t is most likely an Error), the future is never completed and a waiter on get() would hang.
Best improvement: swap the two lines, so the future is completed first and the log line comes after.
Severity: Low
Raised by another reviewer: No

2. Code Quality Assessment

2.1 Coding Standards

Good. The new fields are self-explanatory and the comment in run() explains why the rethrow was removed.

2.2 Test Coverage and Test Stability

SharedCheckpointSchedulerTest now covers the throwing-body case and CheckpointCoordinatorTest is unchanged. The new test waits on a thread-state condition with Awaitility (10 s bound) and has no fixed sleeps. The pool's 60 s keep-alive cannot turn a live idle worker into TERMINATED within that window. Stability rating: Stable. The dev test from Issue 1 needs adapting as described.

2.3 Documentation Updates

No user-facing option or behaviour change, so no docs are needed. The PR description has been refreshed.

3. Architectural Soundness

3.1 Elegance

Long-term solution. Leasing at the CheckpointManager/Coordinator boundary keeps pipeline-scoped cancellation and gives an orderly member shutdown.

3.2 Maintainability

Good. Lifecycle, cancellation races and the hand-off trade-off are documented in the code and the description.

3.3 Extensibility

Moving the periodic trigger back onto the timer thread after #12511 is a one-line change, as you noted.

3.4 Historical-Version Compatibility

No state or format impact. A rolling upgrade only changes thread names.

4. Issue Summary

Number	Issue	Location	Severity
1	Merged result breaks `dev`'s `CheckpointCoordinatorScheduledTriggerTest` (null lease from plain mock)	`CheckpointCoordinator.java:242`, `:572`; `CheckpointCoordinatorScheduledTriggerTest.java:44-87`	High
2	Dispatch-pool coupling via savepoint lock (tracked #12441)	`CheckpointCoordinator.java` `startSavepoint`	Low
3	No guard if coordinator executor is bounded later	`CheckpointCoordinator.java`	Low
4	Complete the future before logging	`PipelineCheckpointTask.java:109-110`	Low

5. Merge Recommendation

Conclusion: Ready to merge after fixes

Blockers (must fix)

Issue 1: merge the latest dev (it brings in #12448 and #12387) and give CheckpointCoordinatorScheduledTriggerTest a real scheduler lease, then get a green fork Build on that new head.

CI: the fork Build run for this head (Vivek1106-04/seatunnel run 40075717724) is still in progress. It has no failed job yet, and one all-connectors-it-3 (11) leg was cancelled. Because the head does not contain #12448, a green result here does not cover Issue 1.

Recommended fixes (non-blocking): Issue 4 (a two-line swap). Issues 2 and 3 stay as tracked notes.

Overall assessment: the core design has been solid for several rounds, and this round's fix and benchmark analysis are excellent. The one remaining item is an integration gap with a test that landed on dev after your last sync, and it should be a small test-only change. A committer's final approval is still needed for the merge itself.

…ler lease CheckpointCoordinatorScheduledTriggerTest (apache#12448) builds the coordinator on a plain CheckpointManager mock, which returns a null scheduler lease on this branch. Back the lease with a test-scoped SharedCheckpointScheduler, closed in finally, and keep the test's assertions unchanged.

If logging itself throws, for example under OutOfMemoryError, the future would never complete and a waiter on get() would hang.

Vivek1106-04 · 2026-10-25T04:24:44Z

Thanks @DanielLeens, good catch on the semantic conflict with #12448. Both items are addressed at head 29ab7519b:

Issue 1 (blocker): merged the latest dev (feedd2506, brings in #12448 and #12387). Before the fix, CheckpointCoordinatorScheduledTriggerTest failed on the merged tree exactly as you described: the plain mock returned a null lease, and the NPE from cancelCheckpoint() in finally hid the one from scheduleTriggerPendingCheckpoint. f6224fc3b makes the mock hand out a lease from a test-scoped SharedCheckpointScheduler, which is closed in finally, the same way CheckpointCoordinatorTest.mockCheckpointManager() does. All of the #12448 assertions are unchanged, and the test now passes. So #12448's wrapped trigger body still fails the coordinator (FAILED plus handleCheckpointError(1, false)) when it runs through the shared dispatcher. No other test on dev builds a coordinator on a bare manager mock.

Issue 4: 29ab7519b completes the future before logging.

Local, JDK 17: CheckpointCoordinatorScheduledTriggerTest 1/1, CheckpointCoordinatorTest 18/18, SharedCheckpointSchedulerTest 11/11, StateTransitionCleanupTest 13/13; spotless:apply and module verify are clean.

Fork Build on the new head (still running, I will follow up with the result): https://github.com/Vivek1106-04/seatunnel/actions/runs/40161631838

Issues 2 and 3 stay as tracked notes (#12441 / #12511).

DanielLeens

Thanks for the quick turnaround, @Vivek1106-04! Both items from my last review are addressed. I re-read the new head 29ab7519b483 (a dev merge plus two small commits) and re-traced the changed paths. This is a code-reading review only; nothing was executed locally.

What Problem Does This PR Solve?

User pain point: every CheckpointCoordinator created its own two-thread scheduled pool, and CheckpointManager creates one coordinator per pipeline, so checkpoint-timer threads grew linearly with active pipelines.
Fix approach: one member-wide SharedCheckpointScheduler (owned by SeaTunnelEngineContext) with two checkpoint-timer threads that only fire, and a bounded checkpoint-dispatcher pool that runs the body. Each coordinator leases a PipelineCheckpointScheduler, so cancelAll() stays pipeline-scoped.
One-sentence summary: replace N per-pipeline timer pools with one shared timer plus a bounded dispatch pool, with no config, checkpoint-format or default-behaviour change.

1. Code Change Review

1.1 Core Logic Analysis

Changes since my last review (8766ee4dbc71):

f6224fc3b (my Issue 1, was High): CheckpointCoordinatorScheduledTriggerTest (added to dev by #12448) now stubs leaseCheckpointScheduler(anyInt()) to return a lease from a test-scoped SharedCheckpointScheduler, closed in finally. All #12448 assertions are unchanged, so the test again proves that a throwing periodic trigger body fails the coordinator (FAILED plus handleCheckpointError(1, false)) when it runs through the shared dispatcher.
29ab7519b (my Issue 4): in PipelineCheckpointTask.run() the order is now completion.completeExceptionally(t) first, then log.error(...). If logging itself throws (for example under OOM), waiters on the future are no longer left hanging, and onSettled still runs from finally.
A merge of dev (feedd2506) that brings in #12448 and #12387 with no textual conflict.

Runtime path (unchanged):

SeaTunnelServer -> SeaTunnelEngineContext (owns SharedCheckpointScheduler)
  -> CheckpointManager.leaseCheckpointScheduler(pipelineId) -> CheckpointCoordinator.scheduler
  -> scheduleTriggerPendingCheckpoint -> lease.schedule -> checkpoint-timer fires
     -> hand-off to checkpoint-dispatcher -> PipelineCheckpointTask.run (cancel check, body, complete future, log on failure)
  -> coordinator close/reset: lease.cancelAll() (pipeline-scoped); context close: both pools shut down

I re-checked PipelineCheckpointTask.run() and cancel() at head: the cancel flag is read before the body, the late-body race is documented and safe because both scheduled bodies re-validate state under the coordinator lock, and cancel() is idempotent via CAS.

1.2 Compatibility Impact

Fully compatible. No Option, default, checkpoint/savepoint format, storage or RPC change. The visible differences are thread names and an ERROR log line where dev stayed silent for a failing scheduled body.

1.3 Performance / Side-Effect Analysis

Unchanged from the last round: roughly 25 us extra per trigger from the timer-to-dispatcher hand-off (documented with your benchmark data), in exchange for a flat thread count and flat time-to-safepoint. The log line only fires when a body throws, and a failing trigger does not re-arm itself, so there is no log loop.

1.4 Error Handling and Logging

No new issues. Remaining items are the same tracked notes:

Issue 1 (carryover, tracked): dispatch-pool coupling through savepoint lock-holding

Location: CheckpointCoordinator.java startSavepoint versus tryTriggerPendingCheckpoint's synchronized (lock)
Problem / Risk: many pipelines savepointing at once can occupy the bounded dispatch pool and delay other pipelines' triggers. The delay is bounded and there is no data loss.
Best fix: tracked in #12441 / #12511.
Severity: Low
Raised by another reviewer: Yes (related to @nzw921rx's head-of-line concern)

Issue 2 (carryover): no guard on the thenAcceptAsync hop if the coordinator executor is ever bounded

Location: CheckpointCoordinator.java startTriggerPendingCheckpoint
Problem / Risk: not reachable today; a note for whoever bounds that executor later.
Severity: Low
Raised by another reviewer: No

2. Code Quality Assessment

2.1 Coding Standards

Good. The new classes carry thorough Javadoc on lifecycle, cancellation and the deliberate best-effort race. Formatting is clean.

2.2 Test Coverage and Test Stability

Rating: Stable. SharedCheckpointSchedulerTest covers lease isolation, cancellation, a failing body keeping the worker alive, and shutdown. The repaired CheckpointCoordinatorScheduledTriggerTest now uses a real lease, so removing the lease wiring turns it red with the NPE you reproduced. CheckpointCoordinatorTest uses a real lease through mockCheckpointManager(). Tests close their scheduler in finally; I saw no sleeps used as assertions and no fixed ports in the tests I read.

2.3 Documentation Updates

No user-facing option or behaviour change, so no docs are required. The PR description documents the bounded pool and the hand-off cost.

3. Architectural Soundness

3.1 Elegance

Long-term solution: one owner for timers, a lease abstraction that keeps per-pipeline cancel semantics, and a clean separation of "fire" from "run".

3.2 Maintainability

Clear boundaries (SharedCheckpointScheduler / PipelineCheckpointScheduler / PipelineCheckpointTask) with the cancellation contract documented where it matters.

3.3 Extensibility

Pool sizes live in one class; the open note is only about bounded coordinator executors (Issue 2).

3.4 Historical-Version Compatibility

No impact on existing jobs, checkpoint data or savepoints. Rolling upgrades are unaffected because nothing crosses the wire.

4. Issue Summary

Number	Issue	Location	Severity
1	Dispatch-pool coupling via savepoint lock (tracked #12441 / #12511)	`CheckpointCoordinator.java` `startSavepoint`	Low
2	No guard if the coordinator executor is bounded later	`CheckpointCoordinator.java`	Low

5. Merge Recommendation

Conclusion: Ready to merge

Blockers (must fix): none. My previous blocker (semantic conflict with #12448) is resolved.

Recommended fixes (non-blocking): Issues 1 and 2 stay as tracked notes.

CI: the real build runs in the fork: Build run https://github.com/Vivek1106-04/seatunnel/actions/runs/40161631838 is success for head 29ab7519b483, and this head contains #12448, so it also covers the merged-test interaction I raised. The apache-side PR signal run is action_required, which is only the first-run approval gate. The branch is 3 commits behind dev with no conflicts, so no further sync is needed.

Overall assessment: the design has been solid for several rounds, and both of this round's fixes are exactly what was asked. I have no remaining source-level objections. @nzw921rx's earlier concerns (savepoint interaction, scale/latency, upgrade uncertainty) are answered by the savepoint-lock note, the published benchmark data and the no-format-change statement; whether to label the feature experimental is a maintainer call. A committer's final approval is still needed for the merge itself.

DanielLeens

Thanks for fixing the merged trigger-test fixture and completing the task future before logging. I re-read the entire current implementation and every changed test, including the original timer/dispatch design. The current-head fork Build is now green, so the earlier pending/red CI guidance is superseded. Sorry for missing the stale expiry-handoff boundary in the previous full review. Issue 5 below is an omission carried over from that implementation, not a newly introduced problem in your two latest fixes.

What problem this PR solves

User pain: Each active pipeline previously owned its own two-thread checkpoint timer pool, increasing member thread count as pipeline count grows.
Fix approach: Share member-owned timing and bounded dispatch pools while giving every coordinator a reusable pipeline-local cancellation lease.
One-line summary: Reduce checkpoint timer thread growth while retaining per-pipeline scheduling and cleanup.

Example: Five hundred active pipelines share two timer threads plus a host-sized dispatch pool instead of owning five hundred separate scheduled executors. Cancelling one pipeline cancels its timers without stopping another pipeline's lease.

1. Code-change review

1.1 Core logic analysis

All ten files (+1101/-43), including the complete three new scheduler classes, coordinator/manager/context changes, container thread classification, scheduled-trigger integration test and both entire changed test bodies, were read. The complete target-relative diff was read at 29ab7519b483425e30faffb2bacc4043a7fc11a9, not just the latest commit. A member-level resource-ownership change; source/checkpoint/barrier contracts require special attention at asynchronous recovery boundaries.

Before / after:

Before: each CheckpointCoordinator creates a two-thread ScheduledThreadPoolExecutor; cleanup shuts it down and creates another.
After: EngineContext owns SharedCheckpointScheduler -> coordinator leases PipelineCheckpointScheduler -> timer fires -> bounded dispatcher executes PipelineCheckpointTask.
Trigger barrier continuation: thenAccept -> thenAcceptAsync(executorService).
Timeout: dispatch-side pending/ACK check -> expireCheckpoint -> coordinator-executor runnable calling handleCoordinatorError, with inline rejection fallback.
Task failure: completeExceptionally first, then contextual ERROR log, no rethrow; finally unregister task.

Complete relevant execution chain:

SeaTunnelServer builds SeaTunnelEngineContext -> member-scoped SharedCheckpointScheduler
 -> JobMaster.initCheckPointManager -> CheckpointManager stores shared scheduler
 -> coordinator construction -> leaseCheckpointScheduler(pipelineId), one final lease per coordinator
Periodic: scheduleTriggerPendingCheckpoint (570) wraps failures -> lease.schedule -> capture MDC / outstanding set -> timer.schedule handoff
 -> dispatcher -> PipelineCheckpointTask.run (97): cancellation check -> body -> completed future -> unregister
 -> tryTriggerPendingCheckpoint (785): readiness/interval/min-pause -> lock -> shutdown/pending/schema-change guards
 -> createPendingCheckpoint -> async barrier continuation (1004) on coordinator executor -> triggerCheckpoint RPCs -> ACK/complete/storage
 -> arm watchdog (1073) on lease -> pending identity lookup + notFullyAcknowledged (1078) -> expireCheckpoint (363)
 -> coordinator executor -> handleCoordinatorError (376): current coordinator future/status -> fail manager -> cleanPendingCheckpoint
Savepoint: startSavepoint takes the existing pipeline lock and waits for pending work; direct trigger still shares watchdog lease, not periodic re-arm policy
Reset: CheckpointManager.restoreCoordinator (300) reuses coordinator -> restoreCoordinator (726) replaces future/error state -> clean old checkpoints and lease.cancelAll -> resume same lease
Terminal: lease.cancelAll affects one pipeline; server shutdown ends coordinator services, then EngineContext.close shuts shared timer/dispatcher before state stores.

Key findings:

The normal checkpoint path reaches the shared scheduler; barriers/storage remain on the coordinator executor. Savepoints retain their direct trigger path and current serialization contract.
The latest dev trigger test now receives a real scoped lease and closes its scheduler; #12448's FAILED/handleCheckpointError assertions remain intact. Previous Issue 1 is closed.
Task failure completes the future before logging and keeps its worker alive; previous Issue 4 is closed. MDC is captured per scheduled body, not assumed from a member-wide thread name.
The bounded dispatcher uses an unbounded queue, so saturation delays rather than rejects. Existing savepoint lock-holding can occupy all dispatch capacity; this remains the explicitly accepted follow-up, not a latency bound.
A watchdog expiry that has already left the lease is no longer cancelled by cancelAll. Its executor runnable has neither the expired PendingCheckpoint nor a coordinator generation, so it can act on a subsequently reset coordinator (Issue 5).

Community context: +1 to @nzw921rx's lifecycle, saturation and benchmark questions: leases are cancellation domains rather than a global registry; the manager owns one coordinator per pipeline; bounded worker count does not bound queue latency. I agree with @SEZ9's recorded acceptance of the documented savepoint-contention follow-up (#12441/#12511), not a claim that it is implemented here. The author's measured hand-off cost and no-hand-off control explain the latency tradeoff; these are author/maintainer measurements, not runs I performed. The posted benchmark rejection stack was against a terminated fixture executor; the current trigger wrapper and contextual task logging must be evaluated separately from benchmark throughput. @SEZ9's comments state no further source changes on the older head; the current head is independently re-read here. There is no current write-capable approval I can replace or dismiss.

1.2 Compatibility impact

Compatible for public configuration, checkpoint/savepoint formats and protocols. Existing user options/defaults, barrier payloads and persistent state layouts are unchanged. Observable thread ownership/names and failure logging change. Resource isolation changes from per-pipeline timers to shared dispatch capacity, which is the explicit design tradeoff. No checkpoint migration is needed. Repair the stale reset action in Issue 5 before relying on recovery compatibility.

1.3 Performance / side effects

The production cost and ownership implications are described below. Two member timer threads replace pipeline-scaled pools; dispatch workers are bounded by max(8, 2*availableProcessors), idle timeout 60 seconds. Each handoff adds scheduling/allocation overhead; reported tens-of-microseconds medians and occasional run-queue tails are not zero overhead. Queue memory and latency remain unbounded under sustained saturation; this is acknowledged rather than hidden behind the thread bound. Timers only submit work. Barrier/expiry distributed-map work normally hops to the coordinator executor; rejected expiry runs inline with WARN to avoid silently losing failure handling. Per-lease outstanding tasks remove themselves on completion/cancel, timer remove-on-cancel is enabled, and member close shuts both pools. No extra network request or checkpoint serialization step is added. Savepoint lock contention remains a cross-pipeline coupling until the separate fix lands.

1.4 Error handling and logging

The existing exception and logging boundaries remain in place. The scheduled trigger wrapper from dev still transitions the coordinator to FAILED on an unexpected throw. PipelineCheckpointTask completes its future before logging with job/pipeline context, catches task failures and unregisters in finally. Shutdown-only submission rejection cancels the task rather than claiming success. Expiry rejection is logged and handled inline. The missing checkpoint/generation context in the asynchronous expiry action is the correctness issue below; a current coordinatorFuture.isDone check alone does not establish that an old timeout belongs to that future.

Issue 5: Fence an already-dispatched expiry against coordinator reset (High, blocking; carryover omission).

Location: seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/checkpoint/CheckpointCoordinator.java:363-372, :1078-1086, :726-735.
Description: The watchdog validates the old pending checkpoint on the shared dispatch thread, then expireCheckpoint submits a parameterless runnable that later calls handleCoordinatorError. restoreCoordinator reuses the same coordinator, replaces checkpointCoordinatorFuture/error state and clears old pending checkpoints. cancelAll only owns PipelineCheckpointTasks; after the watchdog has submitted this separate coordinator-executor runnable and returned, cancelling that lease cannot cancel or invalidate the submitted expiry. If reset occurs before that runnable executes, it fails the new coordinator future and cleans its new state for an old checkpoint. handleCoordinatorError checks the current future, not the generation that timed out.
Risk: A delayed old watchdog can fail a recovered pipeline and abort newly started checkpoint work. This is a concrete permitted interleaving at the new asynchronous handoff, not a claim that the green CI run reproduced it. It is distinct from the accepted savepoint dispatch-capacity tradeoff.
Best improvement: Carry the expired checkpoint identity plus a coordinator generation/future token into the submitted action. Revalidate ownership under a reset-compatible fence at the point where failure mutates coordinator state; a check only before executor submission is insufficient. Preserve the inline rejection fallback with the same guard. Add a deterministic controlled-executor test: let the old watchdog hand off, hold its action, reset/restart the coordinator, release the old action and assert the new generation remains healthy; retain a test proving a current-generation genuine expiry still fails it.
Already raised by another reviewer: No identical point in the reviewed comments. This asynchronous expiry hop existed in the previous reviewed implementation; apologies for missing it then.

Issue 6: Wait for task settlement rather than a worker thread state (Medium, non-blocking test risk).

Location: seatunnel-engine/seatunnel-engine-server/src/test/java/org/apache/seatunnel/engine/server/checkpoint/scheduler/SharedCheckpointSchedulerTest.java:298-313; PipelineCheckpointTask.java:109-114.
Description: The future is now completed before log.error and before finally removes the outstanding task. The test then waits for the worker to be non-RUNNABLE and immediately asserts outstandingCount==0. A worker can be BLOCKED or WAITING inside the logging path while the task is still outstanding; non-RUNNABLE does not mean run() reached finally. This can produce a timing-dependent failure, and it is not a reliable proof that the worker has returned to the dispatch queue.
Risk: Intermittent test failure under logging contention; the production completion-before-log ordering is correct and should remain.
Best improvement: Await outstandingCount==0 directly, as testCompletedTasksDoNotAccumulate already does. To prove worker survival, submit a latch-controlled follow-up task to that worker or otherwise observe completed dispatch work, rather than infer it from a transient Thread.State.
Already raised by another reviewer: No identical point. The completion-before-log fix makes the already weak thread-state proxy especially visible; no production ordering rollback is requested.

Previous Issues 2/3 (Low, unchanged): savepoint lock-holding/shared-capacity coupling remains tracked in #12441/#12511; future changes to coordinator executor policy must preserve failure handling for the asynchronous continuation. Neither replaces Issue 5's recovery-generation guard.

2. Code-quality assessment

2.1 Code conventions

New scheduler files carry ASF headers and detailed lifecycle/cancellation/ownership comments; methods/fields have meaningful contract documentation. Existing coordinator/test classes are extended rather than duplicating the lifecycle fixture. The scoped whitespace check produced no entries.

2.2 Test coverage and stability

Stability rating: Risk present, non-blocking (Issue 6). No local build, unit/integration test, container or service was run. The 325-line scheduler test covers dispatch thread selection, sibling cancellation isolation, cancellation before/after handoff, 500-pipeline timer bounds, reusable leases, settlement, close, dispatch bounds, spare-capacity progress and failed-body futures. Most use latches/atomics/Awaitility and release blocked bodies in finally; the fixed sleep has been replaced by a negative during window. CheckpointCoordinatorTest:443/507 exercises the already-completed-future asynchronous trigger and expiry executor ownership. The merged ScheduledTriggerTest:45-92 now uses a scoped real lease and retains dev's assertions. Issue 6 identifies the remaining weak settlement signal. The current tests do not hold an already-submitted expiry across reset, which is the missing correctness regression in Issue 5. The high-core dispatch-bound test scales workers with availableProcessors, an explicit resource cost rather than an environment-independent constant.

Exact current-head fork evidence at 29ab7519b483425e30faffb2bacc4043a7fc11a9: Build 40161631838: completed/success. Apache required-check and review gates remain independent of the fork run. Queued/in-progress/cancelled is not success; the earlier-head result does not transfer to this head.

2.3 Documentation

No new user option or checkpoint format requires bilingual configuration docs. The class Javadocs and PR description must retain the measured handoff cost, bounded-worker/unbounded-queue tradeoff and accepted savepoint contention. Avoid claiming unrelated pipeline execution is always unaffected or that savepoint duration is a guaranteed upper latency bound.

3. Architectural soundness

3.1 Solution quality

Member ownership and pipeline leases are a coherent long-term resource design. Keep timers free of lock-taking bodies; removing the handoff solely to improve benchmark medians would reintroduce timer-wide blocking.

3.2 Maintainability

Shared ownership and reusable cancellation domains reduce pool churn. Expiry actions need explicit lifetime identity when they cross a second executor; comments about safe late lease bodies do not cover a separate stale action after that body has returned.

3.3 Extensibility

Scheduling-delay/queue observability is a useful follow-up before tuning knobs. Additional control paths should carry the same generation/ownership contract instead of relying on lease cancellation after handoff.

3.4 Historical compatibility

No wire/state migration is needed. Existing savepoint/min-pause/pending guards remain. Recovery correctness must include stale asynchronous actions on reused coordinators, as requested in Issue 5.

4. Findings summary

No.	Finding	Location	Severity
5 (carryover omission)	Old expiry action can fail a reset coordinator	CheckpointCoordinator.java:363; :726; :1078	High
6	Worker state is not task settlement	SharedCheckpointSchedulerTest.java:298-313	Medium, non-blocking
2/3 (previous)	Accepted contention / executor-policy follow-ups	Coordinator savepoint/continuation	Low

5. Merge conclusion

Conclusion: Can merge after fixes

Blocking items: Issue 5: bind deferred expiry to its checkpoint/coordinator generation and add the controlled-reset regression. Current-head CI is green, but does not cover this interleaving. Effective write-capable maintainer approval remains required.
Non-blocking recommendations: Issue 6: replace the thread-state settlement proxy. Keep the accepted savepoint/observability work bounded to its follow-up and do not remove the timer handoff to chase medians.

The latest fixture fix and completion-before-log change address the previous round, and the exact-head Build succeeds. The remaining correctness request is a narrow stale-action guard at the expiry handoff; it does not reopen the accepted shared-thread design or invalidate the benchmark measurements.

Compare against dev: diverged, ahead 13 / behind 8, merge conflict=False. Drift is recorded independently; it does not itself justify a sync-first comment without an unrelated CI failure.

Daniel has READ permission. This COMMENT is a technical assessment, not an effective write-level approval. A write-capable maintainer must confirm repository gates and supply the effective approval.

Local evidence: exact-head fetch/checkout on seatunnel-review-12165, complete diff and source call-chain reads cited above. No runtime verification was claimed.

…dinator The timeout watchdog hands the expiry to the coordinator executor, and restoreCoordinator() can reuse the coordinator before that action runs. The action carried no identity, so it failed the restored generation's future and cleaned its new pending state for an old checkpoint. The expiry now carries the expired PendingCheckpoint and the coordinator future that was current when it was triggered. Its ownership check and FAILED transition hold a dedicated generationLock, which also guards the future swap in restoreCoordinator(), so a stale expiry is dropped. The inline rejection fallback takes the same path. handleCheckpointError() runs after the lock is released because it takes the pipeline monitor, which a restoring pipeline holds while calling restoreCoordinator().

…rker The failing-body test inferred that the worker had returned to the queue from it leaving RUNNABLE, but the task completes its future before it logs and unregisters, so the worker can be blocked in the logger while the task is still outstanding. Wait for outstandingCount() to reach zero first, then for the worker to park in the queue's timed poll.

Vivek1106-04 · 2026-10-27T06:55:30Z

@DanielLeens thanks for the full re-read, and for catching the expiry hand-off. You are right: once the watchdog has handed the parameterless runnable to the coordinator executor, cancelAll() cannot reach it, and handleCoordinatorError only checks the current future. Both items are addressed at head a932e27a7.

Issue 5 (stale expiry fails a restored coordinator) - 3cc4dea45

startTriggerPendingCheckpoint reads checkpointCoordinatorFuture once, while its caller still holds lock and before the checkpoint exists. That future is the generation token. The watchdog now calls expireCheckpoint(pendingCheckpoint, owner).
The submitted action, and the inline rejection fallback, both run failExpiredCheckpoint(expired, owner). Under a new generationLock it checks owner == checkpointCoordinatorFuture, !owner.isDone() and pendingCheckpoints.get(id) == expired. Only if all three hold does it set FAILED, complete owner and clean the pending state. Otherwise it logs at INFO and drops the expiry.
restoreCoordinator swaps errorByPhysicalVertex / checkpointCoordinatorFuture and writes RUNNING under the same generationLock. So the expiry either fails the old generation before the swap, or sees the new future and is dropped. checkpointCoordinatorFuture is now volatile, because the swap and the token read no longer share a monitor.
Why a new monitor instead of lock: startSavepoint holds lock in its sleep-poll until pendingCounter drops, and that can be the checkpoint that is expiring. Taking lock here would deadlock. Taking generationLock does not: cleanPendingCheckpoint sets shutdown = true before it takes lock, which ends that poll, same as today.
checkpointManager.handleCheckpointError(...) runs after generationLock is released. It takes the SubPlan monitor, and a restoring pipeline holds that monitor when it calls restoreCoordinator, so calling it under the lock would invert the order. Because of this, the expiry path now cleans pending state before notifying the manager instead of after. The old comment's reason for that order (do not kill the pool before the manager is notified) no longer applies: cleanPendingCheckpoint only calls cancelAll(), which cancels timers with cancel(false) and never interrupts. I checked the sync consumers of the futures completed under the lock. SubPlan and JobMaster block with join()/get(), and the pending-checkpoint handlers are *Async, so no callback takes the pipeline monitor under generationLock.
The other handleCoordinatorError callers are unchanged. They run synchronously in the generation that raised the error, not as a deferred action.

Tests in CheckpointCoordinatorTest:

testExpiryHeldAcrossCoordinatorRestoreDoesNotFailTheRestoredGeneration: a single-thread coordinator executor is held by a latch, so the expiry queues behind it. The test then calls restoreCoordinator(false) and puts the same PendingCheckpoint back into pendingCheckpoints, so only the generation check can drop it. Then it releases the latch and asserts the restored future is not done and handleCheckpointError is never called.
testExpiryOfTheCurrentGenerationFailsTheCoordinator: the same expiry without a restore still completes the future FAILED and calls handleCheckpointError(1, false). The existing testCheckpointExpiryHandlingNeverRunsOnTheDispatchThread still covers the end-to-end watchdog path.

Red runs, made by changing only the production code:

With the action reverted to the old unconditional handleCoordinatorError(CHECKPOINT_EXPIRED, null), the held-expiry test fails: an expiry of the generation before the restore must not fail the restored one ==> expected: <false> but was: <true>.
With only the owner check removed (pending-identity check kept), it fails the same way. So the token is what does the work, not the map lookup.

Issue 6 (thread state is not task settlement) - a932e27a7

testFailingBodyCompletesExceptionallyAndKeepsItsDispatchThread now waits for lease.outstandingCount() == 0 first, so the task has passed its finally. Then it waits for the worker to reach TIMED_WAITING or TERMINATED and asserts TIMED_WAITING. After finally, the only timed wait on that path is the dispatcher queue's poll(keepAlive), because allowCoreThreadTimeOut is on and the worker stays parked there for the 60 s idle timeout. A worker blocked in the logger can no longer pass, and a worker killed by a rethrow ends TERMINATED. The production completion-before-log order is unchanged.

Verification (JDK 17, local)

CheckpointCoordinatorTest, SharedCheckpointSchedulerTest, CheckpointCoordinatorScheduledTriggerTest, StateTransitionCleanupTest: 45 tests, 0 failures.
All org.apache.seatunnel.engine.server.checkpoint.** tests plus SavePointTest and CheckpointErrorRestoreEndTest: 69 run, 0 failures, 2 skipped (existing @Disabled).
The two new tests plus the changed scheduler test, 20 consecutive runs: 20/20 green.
spotless:apply and -DskipTests verify on seatunnel-engine-server: clean.

Fork Build for a932e27a7: https://github.com/Vivek1106-04/seatunnel/actions/runs/40463705801 (running). I will follow up with the result.

DanielLeens

Thank you @Vivek1106-04 for the fast, precise turnaround. The head is now a932e27a7 (commits 3cc4dea45 and a932e27a7 on top of 29ab7519b). I re-read both commits in full, the complete failExpiredCheckpoint / restoreCoordinator / cleanPendingCheckpoint / startSavepoint paths at this head, and the new tests. This is a static review; nothing was built or run locally, and the runtime evidence is the fork CI.

Status of my earlier points (verified in the head code):

Issue 5 (an already-dispatched expiry could fail a reset coordinator): resolved. The generation token is read in startTriggerPendingCheckpoint (CheckpointCoordinator.java:1060, while the caller holds lock), carried into expireCheckpoint(expired, owner) (:375), and re-validated in failExpiredCheckpoint (:403-424) under generationLock against owner == checkpointCoordinatorFuture, !owner.isDone() and the identity of the pending checkpoint. restoreCoordinator swaps the future under the same monitor (:780-784), so the expiry either fails the old generation first or sees the new one and is dropped. The inline-rejection fallback runs the same guarded action. Thank you, and sorry again for missing this on earlier rounds.
Issue 6 (thread state is not task settlement): resolved. The test now waits for lease.outstandingCount() == 0 first, then for TIMED_WAITING/TERMINATED, and asserts TIMED_WAITING.
Issues 2/3 (savepoint contention and executor-policy notes): unchanged, still tracked in #12441/#12511.

What Problem Does This PR Solve?

Pain point: every CheckpointCoordinator owned a two-thread scheduled pool, so timer threads grew linearly with active pipelines.
Fix approach: one member-wide SharedCheckpointScheduler (two timer threads plus a bounded dispatch pool) and a per-pipeline cancellation lease for each coordinator.
Summary: replace N per-pipeline timer pools with shared timers and a bounded dispatcher, with no config, wire or state-format change.

1. Code Change Review

1.1 Core Logic Analysis

trigger (under lock): owner = checkpointCoordinatorFuture            [generation token]
  -> pending checkpoint created -> watchdog armed on the lease (1143)
watchdog (dispatch thread): pending still tracked && !fullyAcknowledged
  -> expireCheckpoint(expired, owner) -> coordinator executor (or inline if rejected)
  -> failExpiredCheckpoint: synchronized(generationLock)
        owner current? not done? pendingCheckpoints.get(id) == expired?
        yes: errorByPhysicalVertex, updateStatus(FAILED), owner.complete(FAILED), cleanPendingCheckpoint
        no : INFO log, drop
     then (outside the lock) checkpointManager.handleCheckpointError(pipelineId, false)
restoreCoordinator: synchronized(generationLock){ new error ref, new future, RUNNING } -> clean (RESET) -> resume

I checked the points that matter for this kind of change:

Lock order: generationLock -> this (updateStatus) -> lock (clean). Nothing that holds lock or the coordinator monitor ever waits for generationLock (completePendingCheckpoint and tryTrigger only read the volatile field), so I find no cycle. startSavepoint can hold lock in its sleep-poll, but cleanPendingCheckpoint sets shutdown = true before taking lock, which ends that poll (:983-986). handleCheckpointError is correctly called after the lock is released, because it takes the SubPlan monitor.
Reordering: the expiry path now cleans pending state before notifying the manager. The old comment's reason (do not kill the pool first) no longer applies, since cleanPendingCheckpoint only calls scheduler.cancelAll() (:1352).
Side benefit: the pending-identity check also drops an expiry for a checkpoint that completed between the watchdog decision and execution, which the old unconditional handleCoordinatorError would have failed.
Reset semantics: PendingCheckpoint.abortCheckpoint completes with null for CHECKPOINT_COORDINATOR_RESET, so aborting old pending checkpoints during a restore does not trigger the error handler on the new generation.

1.2 Compatibility Impact

Fully compatible. No Option, default, checkpoint/savepoint format, storage layout or RPC change. Visible differences are thread names and an ERROR log line for a failing scheduled body.

1.3 Performance / Side-effects

Unchanged from the last round (timer-to-dispatcher hop, bounded workers, unbounded queue by design). The new generationLock is only taken on expiry and on restore, so there is no hot-path cost.

1.4 Error handling and logging

Issue 1 - Medium (non-blocking, pre-existing on dev, same class as Issue 5): other deferred actions still use handleCoordinatorError without a generation token.

Location: CheckpointCoordinator.java:1071-1079 and :1109-1115 (completion/barrier-trigger failure handlers in the thenAcceptAsync continuation, including allOf(...).get() at :1107), and the scheduled-trigger wrapper at :620-640.
Problem: these run later on the coordinator executor or a dispatch thread and call handleCoordinatorError, which checks only the current checkpointCoordinatorFuture. A barrier RPC that fails because a member went away, observed after restoreCoordinator swapped the future, would fail the restored generation in the same way Issue 5 described. Your PR description is right that synchronous callers are unaffected; these are the asynchronous ones.
Risk: a narrow recovery race that exists on dev today, so not introduced here and not a blocker for this PR.
Best improvement: a follow-up that threads the same owner token through these handlers (a small private overload of handleCoordinatorError(owner, reason, e) that applies the generationLock check). Worth filing next to #12441.
Severity: Medium. Raised by another reviewer: No.

Issue 2 - Low: generationLock now spans distributed IMap I/O.

Location: CheckpointCoordinator.java:406-423 (updateStatus retries against runningJobStateIMap, cleanPendingCheckpoint removes IMap keys and notifies the monitor) while restoreCoordinator (:780) waits on the same lock, typically holding the SubPlan monitor.
Problem/Risk: a slow IMap during an expiry delays a concurrent restore by the retry budget. Before this change restoreCoordinator only waited on the coordinator monitor for updateStatus, so this is a small extension, not a new hazard class. No test exercises expiry racing a concurrent savepoint, which is the interaction the lock-choice comment relies on.
Best improvement: keep as is; optionally add one controlled test that holds the lock in a savepoint-style poll while an expiry runs, to pin the "shutdown ends the poll" argument.
Severity: Low. Raised by another reviewer: No.

Issue 3 - Low: new tests reach private members by reflection.

Location: CheckpointCoordinatorTest.java helpers expire(...) (invokes private expireCheckpoint(PendingCheckpoint, CompletableFuture)), coordinatorFuture(...), trackPendingCheckpoint(...) (read private fields).
Problem: the signature of a private method is now part of the test contract. This matches the existing style of the class, so I treat it as a nit.
Best improvement: make expireCheckpoint package-private with @VisibleForTesting, as scheduleTriggerPendingCheckpoint already is.
Severity: Low. Raised by another reviewer: No.

2. Code Quality

2.1 Coding standards

The new generationLock field, expireCheckpoint and failExpiredCheckpoint carry Javadoc that states why it is not lock and why handleCheckpointError runs outside the lock. checkpointCoordinatorFuture is now volatile for the token read. ASF headers on the new scheduler files were already in place.

2.2 Test coverage and stability

Rating: Stable. testExpiryHeldAcrossCoordinatorRestoreDoesNotFailTheRestoredGeneration holds the coordinator executor with a latch, restores the coordinator, re-tracks the same pending checkpoint so only the generation check can drop the expiry, and asserts the restored future is not done and handleCheckpointError is never called; removing the owner condition makes it red (you demonstrated this). testExpiryOfTheCurrentGenerationFailsTheCoordinator guards against over-fencing. testFailingBodyCompletesExceptionallyAndKeepsItsDispatchThread is now settlement-based (SharedCheckpointSchedulerTest.java:307-322): after outstandingCount() == 0, the only timed wait on the worker's path is the dispatcher's keep-alive poll, so TIMED_WAITING is a sound signal. No sleeps as assertions, bounded Awaitility, executors shut down in finally. No existing assertion was weakened.

CI on a932e27a7 (fork run 40463705801): unit-test (ubuntu 11/17, windows 11/17), engine-v2-it (11/17), Code style, License header and 80+ other jobs are green. One job failed, rocketmq-connector-it (17), and all-connectors-it-4 (11) was cancelled. This PR touches no connector code, so I read the RocketMQ failure as unrelated (a connector IT already known to be flaky); a job-level rerun is the cheapest confirmation. The Apache side is BLOCKED on required review; the branch is 32 commits behind dev and ahead 15, with no conflicts reported.

2.3 Documentation

No user-facing option or format change, so no docs are required; the Javadoc and PR description carry the tradeoffs.

3. Architectural Soundness

3.1 Fix quality

Member-owned timers, bounded dispatch and per-pipeline leases remain a coherent long-term design, and the generation token is the right shape of fence: cheap, local, and scoped to the one asynchronous hop that outlives lease cancellation.

3.2 Maintainability

Clear boundaries between scheduler, lease and task, with the cancellation contract documented where it matters.

3.3 Extensibility

Issue 1 is the natural next step: make the token a general part of deferred coordinator error handling.

3.4 Historical compatibility

No impact on existing jobs, checkpoints or savepoints; nothing crosses the wire.

4. Issue Summary

#	Issue	Location	Severity
1	Other deferred `handleCoordinatorError` callers lack a generation token (pre-existing)	`CheckpointCoordinator.java:1071-1115`, `:620-640`	Medium (non-blocking)
2	`generationLock` spans IMap I/O; no savepoint-interaction test	`CheckpointCoordinator.java:406-423`	Low
3	Tests invoke a private method by reflection	`CheckpointCoordinatorTest.java` `expire(...)`	Low

5. Merge Recommendation

Conclusion: Ready to merge

Blockers: none on the source side. Both items from my last review are fixed and verified. Remaining gates: CI on this head (one unrelated connector IT to rerun) and a write-capable maintainer approval, which my account cannot provide.

Recommended non-blocking fixes: Issue 1 as a follow-up PR, Issues 2 and 3 if you touch these files again. Thank you for the very careful engineering on this one.

DanielLeens · 2026-10-28T15:56:47Z

Thanks for the contribution. The CI fact cited in my previous conclusion has changed; this follow-up updates that gate on the unchanged reviewed head, without issuing another full source review.

Gate snapshot (2026-10-28T15:56:42+00:00), head a932e27a757519fe74a26cb2920966cad853aead: Build is FAILURE.

The blocking jobs are Run / all-connectors-it-4 (11, ubuntu-latest) cancelled; Run / rocketmq-connector-it (17, ubuntu-latest) failure.

The failed job list is observed, but its causal error is not established by the available summary. Please inspect the linked failed step/report before choosing a source change; no verified upstream sync-only cure is established in this follow-up.

My previous source conclusion remains clear for the reviewed scope. CI is not being passed by inference, and this follow-up is not an effective approval.

Comparison recorded in this scan: base dev, diverged, ahead 15 / behind 49, no merge conflict. Branch drift is metadata here and is not an additional sync request.

Vivek1106-04 · 2026-10-28T16:35:49Z

Thanks @DanielLeens for the re-review and for confirming Issues 5 and 6.

CI on a932e27a7 (fork run 40463705801): I checked the logs for both non-green jobs. Neither one is a test failure:

rocketmq-connector-it (17): every test that ran passed. The job ended with The runner has received a shutdown signal ... The operation was canceled. (08:23:53Z). The runner was stopped mid-job.
all-connectors-it-4 (11): every reported module passed (the last was EdgeSocketCheckpointRestoreIT, 6/6, 0 failures). The job was cancelled at 09:25:45Z, 150 minutes after it started at 06:55:25Z, so it hit the job time limit.

I re-ran the failed jobs (attempt 2) and will post the result here.

Issue 1 (other deferred handleCoordinatorError callers without a generation token): agreed. I will follow up with a separate issue and PR. The PR will route the asynchronous barrier-trigger/completion failure handlers and the scheduled-trigger wrapper through an owner-checked overload under generationLock, with controlled-reset regression tests like the expiry ones. That keeps this PR's scope unchanged. I will also pick up Issue 3 (expireCheckpoint package-private with @VisibleForTesting) and the savepoint-poll interaction test from Issue 2 there, because both touch the same code.

Vivek1106-04 · 2026-10-28T17:37:48Z

@DanielLeens Issue 1 is now tracked in #13052. The fix, together with Issues 2 and 3, is ready on top of this branch. I will open its PR once this one merges, so its diff only contains that change.

SEZ9 · 2026-10-29T02:22:19Z

Thanks @Vivek1106-04 for digging into the logs on run 40463705801. Your read of both non-green jobs (runner shutdown mid-job for rocketmq-connector-it (17), job time limit for all-connectors-it-4 (11), with no reported test failures in either) is consistent with the summary I had, which only showed the cancelled/failure states without a causal error. I am not going to treat that as a pass by inference, though, so the gate on a932e27a7 stays open until the re-run lands.

Keeping this PR at its current head while that is resolved is the right call; please do not push anything unless the re-run points at a real source problem.

Concrete asks:

When attempt 2 finishes, post the link to the re-run here together with the outcome of those two jobs. If either job goes non-green again, please include the failing step/report rather than just the job status so we can tell infrastructure noise from a real regression.
Confirm the head is still a932e27a7 at that point (or call out any new commit explicitly), so I know whether a re-read is needed or only the CI gate update.

On #13052: agreed with the sequencing. Tracking the remaining handleCoordinatorError callers there, along with the Issues 2/3 items you mentioned, and opening that PR only after this one merges keeps this diff unchanged, which is what I want for the final pass.

DanielLeens · 2026-10-29T17:14:05Z

Thanks @Vivek1106-04 for tracing the runner shutdown/time limit and tracking the deferred generation-owner work in #13052. I verified that issue is open and the current-head Build is now successful. The prior failed/cancelled-job guidance is superseded. Keep this PR's reviewed scope unchanged; the planned follow-up owns the remaining optional callers/savepoint interaction coverage. My technical source conclusion stays clear; effective maintainer approval is separate.

Checked 2026-10-29T17:13:58+00:00; head a932e27a757519fe74a26cb2920966cad853aead. Signal pull request update = SUCCESS; Build = SUCCESS

Comparison: base dev, head a932e27a757519fe74a26cb2920966cad853aead, diverged, ahead 15 / behind 61; GitHub mergeable state dirty. GitHub reports real merge conflicts; the author must sync the latest base and resolve them before the resulting head can be checked and re-reviewed.

Daniel has READ repository permission. This is a discussion/CI clarification, not a new complete source review, an effective APPROVED event or dismissal of historical requested changes. A write-capable maintainer must perform any required final approval/dismissal. Thanks for keeping the follow-up focused.

SEZ9 · 2026-10-30T03:19:08Z

Thanks @Vivek1106-04. Tracking Issue 1 separately and bundling Issues 2 and 3 into the same follow-up PR, opened once this one merges, keeps this PR's reviewed scope unchanged, which works for me.

Nothing new from my side beyond my last comment: once the branch is synced with the latest base and the conflicts are resolved, please leave a note here so the new head can be checked. Final approval rests with a write-capable maintainer.

DanielLeens · 2026-10-30T03:30:59Z

Thanks @SEZ9. I agree with keeping the follow-up scope separate. The current reviewed head is source-clear and Build-green, but GitHub still reports actual merge conflicts. The contributor must sync the latest base and resolve those conflicts while preserving both sides, then obtain CI on the resulting new head for complete re-review. Final effective approval belongs to a write-capable maintainer.

Checked 2026-10-30T03:30:37.151775+00:00; head a932e27a757519fe74a26cb2920966cad853aead. Signal pull request update=SUCCESS; Build=SUCCESS. Base dev, diverged, ahead 15/behind 64; merge dirty.

This is an unchanged-head discussion/CI clarification, not another full source review or an effective approval. Daniel has READ repository permission.

SEZ9 · 2026-10-31T04:20:35Z

Thanks @Vivek1106-04 for tracing the runner shutdown and the job time limit on the two non-green jobs in run 40463705801 on a932e27a7, and for filing the follow-up issue for the deferred generation-owner work. I have not inspected those job logs myself, but the re-run on head a932e27a757519fe74a26cb2920966cad853aead is green: Signal pull request update and Build are both SUCCESS, so the earlier failed/cancelled-job guidance is superseded.

Keeping the owner-checked handleCoordinatorError overload, the package-private expireCheckpoint change, and the savepoint-poll interaction test in a separate PR opened after this one merges sounds right. No further source changes are needed in this PR for the reviewed scope.

One remaining item: GitHub reports the branch as diverged from dev (ahead 15 / behind 64) with mergeable state dirty, i.e. real merge conflicts. Please sync the latest dev, resolve the conflicts, and push the resulting head so CI can run on it. Once that head is green I will re-check it; final approval rests with a write-capable maintainer.

…-checkpoint-scheduler # Conflicts: # seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/checkpoint/CheckpointCoordinator.java # seatunnel-engine/seatunnel-engine-server/src/test/java/org/apache/seatunnel/engine/server/checkpoint/CheckpointCoordinatorTest.java

Vivek1106-04 · 2026-10-31T04:48:46Z

@SEZ9 @DanielLeens I merged the latest dev (3ac2d1cec) into the branch. The new head is b48e21019, and GitHub now reports the PR as mergeable.

The conflicts came from #12042 (idle reader state after a master switch), which touched the same two files:

CheckpointCoordinator: an import block conflict only. I kept both sides (PipelineCheckpointScheduler from this PR, and ExecutionState / TaskGroupLocation from dev). No code bodies conflicted.
CheckpointCoordinatorTest: two additions in the same spot (this PR's shared-scheduler test helpers, and dev's new testRunningRestoreCompletesCheckpointWithoutFinishedReaderAcknowledgements). I kept both. Dev's test built its coordinator on a plain Mockito.mock(CheckpointManager.class), which returns a null scheduler lease on this branch. That is the same semantic conflict as the earlier [Bug][Zeta] Surface periodic checkpoint trigger failures #12448 one, so the test now uses the existing mockCheckpointManager() helper. Nothing else in the test changed.

Local, JDK 17, org.apache.seatunnel.engine.server.checkpoint.** and dag.physical.**: 109 run, 0 failures, 0 errors, 2 skipped (pre-existing @Disabled). This includes CheckpointCoordinatorTest 21/21, with dev's new idle-reader test passing, and SharedCheckpointSchedulerTest 11/11. spotless:check is clean.

Fork Build for b48e21019: https://github.com/Vivek1106-04/seatunnel/actions/runs/41170040575 (running). I will post the result here.

DanielLeens

Thank you @Vivek1106-04 for syncing with dev and for describing the conflict resolution so precisely. I re-reviewed the current head b48e2101972853a13914547d659e55b0e64f6ffb from scratch, not just the merge. My previous round (a932e27a7) already found no source blockers, and nothing in this merge reopens that. Both items I raised earlier (the stale-expiry generation fence and the thread-state test) are still resolved at this head. This is a static review; nothing was built or run locally, and runtime evidence comes from the fork CI.

What Problem Does This PR Solve?

User pain point: every CheckpointCoordinator owned a two-thread scheduled pool, so timer threads grew as 2 x active checkpoint-enabled pipelines (500 pipelines: up to 1000 mostly idle threads).
Fix approach: one member-wide SharedCheckpointScheduler (owned by SeaTunnelEngineContext): 2 timer threads that only fire and hand off, plus a bounded dispatch pool that runs the body. Each coordinator leases a PipelineCheckpointScheduler that tracks only its own tasks, so cancelAll() is per pipeline and the lease survives a master-failover reset.
One-sentence summary: replaces N per-pipeline timer pools with one shared timer plus a bounded dispatcher, with no config, wire or checkpoint-format change.

1. Code Change Review

1.1 Core Logic Analysis

What is new since my last review. The head is a merge of dev (3ac2d1cec6a, which is also the merge base, so the effective PR diff is clean). I diffed the engine main sources against my previously reviewed head: the only change is the import block plus the restoreCoordinator(alreadyStarted) idle-reader bookkeeping that arrived from #12042. That block is dev code and does not touch the scheduler. The PR's own main-source diff is byte-identical to what I reviewed before.

CheckpointManager.<init> -> engineContext.getCheckpointScheduler()            CheckpointManager.java (ctor)
  -> leaseCheckpointScheduler(pipelineId) -> SharedCheckpointScheduler.lease    per (jobId, pipelineId)
tryTriggerPendingCheckpoint (under lock) -> scheduler.schedule(...)
  -> timer thread: dispatchAfter -> submit -> dispatcher.execute(task)           SharedCheckpointScheduler.java
  -> dispatch thread: PipelineCheckpointTask.run() -> body (re-validates state under coordinator lock)
trigger body -> startTriggerPendingCheckpoint: owner = checkpointCoordinatorFuture (generation token)
  -> thenAcceptAsync(..., executorService)                                       barrier wait off the dispatcher
  -> watchdog armed on the lease -> expireCheckpoint(expired, owner) -> failExpiredCheckpoint
restoreCoordinator: synchronized(generationLock){ new future, RUNNING } -> clean(RESET) -> scheduler.cancelAll()

I re-verified the merged result on these points:

Interplay with #12042: the new finishedGroups read and closedIdleTask rebuild run in restoreCoordinator after the generationLock block and before notifyCompleted. They use runningJobStateIMap only, take no scheduler lease and no new lock, so they cannot interact with the lease or the generation fence. Lock order stays generationLock -> coordinator monitor (updateStatus) -> lock (clean).
Cancellation contract: PipelineCheckpointTask.cancel is a CAS, cancels the timer future and the completion once, and attachTimerFuture closes the schedule-vs-cancel window. A body already on the dispatcher can still run once; both bodies re-validate under the coordinator's own state, and the stale expiry is additionally fenced by the generation token. Correct as documented.
Lifecycle: SharedCheckpointScheduler.close() runs from SeaTunnelEngineContext.close(); a closed scheduler cancels tasks instead of throwing, and daemon threads cannot pin the JVM. cleanPendingCheckpoint now only calls cancelAll(), so no pool is rebuilt on every reset.
Restore: the lease is reusable after cancelAll(), so the failover path (restoreCoordinator -> clean(RESET) -> trigger again) re-arms normally.

1.2 Compatibility Impact

Fully compatible. No Option, default, checkpoint or savepoint format, state-store layout or RPC change. The only visible differences are thread names (checkpoint-timer-*, checkpoint-dispatcher-*, replacing checkpoint-coordinator-<p>/<j>) and an ERROR log for a failing scheduled body. The E2E thread-leak whitelist was extended accordingly.

1.3 Performance / Side-effects

One extra timer-to-dispatcher hop per trigger (microseconds against a seconds-scale checkpoint interval). Thread count is capped at 2 + max(8, 2 x cores), idle dispatchers are reaped after 60 s. The unbounded queue is deliberate (a rejection would mean a dropped trigger or watchdog), and the one path that can hold a dispatcher for long (startSavepoint holding lock in its poll) is documented in the class Javadoc and tracked in #12441. For the 10-billion-row-per-job and trillion-per-day baseline this change adds no per-record cost; it only touches per-checkpoint control-plane scheduling, so I see no OOM, leak or backpressure amplification path.

1.4 Error handling and logging

No new blocking issue. Two nits and the carried-over follow-up:

Issue 1 - Low: dead fallback creates an unmanaged scheduler.

Location: CheckpointManager.java constructor (sharedScheduler != null ? sharedScheduler : new SharedCheckpointScheduler()).
Problem: engineContext is dereferenced a few lines later (engineContext.getStateStores()), so a null context already fails before this fallback matters. If a context ever returned a null scheduler, the private instance would never be closed and would leave two timer threads plus dispatcher threads behind per job.
Best improvement: drop the fallback (the context constructor always creates one), or keep it only if you add a close path.
Severity: Low. Raised by another reviewer: No.

Issue 2 - Low: a plain Mockito.mock(CheckpointManager.class) yields a null lease.

Location: CheckpointCoordinatorTest.java (~line 1705, the parallelism-remap test) still uses a bare mock, while every other construction site now uses mockCheckpointManager().
Problem: it passes today because that test only calls restoreTaskState, but any future use of the scheduler in that path would NPE with a confusing message.
Best improvement: use mockCheckpointManager() there too.
Severity: Low. Raised by another reviewer: No.

Carried over, non-blocking, agreed with @SEZ9 to ship as a follow-up PR: the owner-checked handleCoordinatorError overload for the other deferred callers, the package-private expireCheckpoint, and the savepoint-poll interaction test. Scope stays as reviewed.

2. Code Quality

2.1 Coding standards

New classes carry ASF headers and Javadoc that explain why (two-pool split, unbounded queue, best-effort cancellation, generationLock vs lock). @Slf4j is used consistently; the shim CompletableFuture convention is followed.

2.2 Test coverage and stability

Rating: Stable. SharedCheckpointSchedulerTest covers flat timer thread count at 500 leases, per-lease cancellation, cancel-after-dispatch, failing-body survival and the dispatcher bound, with Awaitility instead of sleeps; CheckpointCoordinatorTest covers the stale-expiry fence in both directions (old generation dropped, current generation still fails). The merge only adjusted dev's new idle-reader test to build its coordinator via mockCheckpointManager(), which keeps its assertions intact. No existing assertion was weakened.

CI on b48e21019 (fork run 41170040575): 95 of 96 jobs succeeded, including unit-test on ubuntu 11/17 and windows 11 and the engine ITs. The single failure, unit-test (17, windows-latest), ended after 36 seconds in the "run all modules unit test (Windows)" step with 403 Forbidden while downloading the Maven wrapper (ClassNotFoundException: MavenWrapperMain), so no test executed on that leg. That is runner infrastructure, not this PR. A job-level rerun (gh run rerun 41170040575 --job 124446368937) is the cheapest confirmation; I would still like to see that leg green before merge. The Apache-side Build pointer had not reported when I checked, and the Apache side shows BLOCKED on required review only (mergeable, no conflicts).

2.3 Documentation

No user-facing option or format change, so no docs/en or docs/zh update is needed. Trade-offs are in Javadoc and the PR description.

3. Architectural Soundness

Fix quality: this fixes the root cause (pool-per-pipeline), not a symptom. Member-owned threads, per-pipeline cancellation scope and a generation fence on the one async hop that outlives lease cancellation is a coherent design.
Maintainability: clear scheduler / lease / task boundaries with the cancellation contract documented where it matters.
Extensibility: the follow-up generation-owner work extends the same token naturally.
Historical compatibility: nothing crosses the wire or touches persisted state, so rolling upgrades, 2.6 -> 3.x style upgrades and savepoint/checkpoint restores are unaffected.

4. Issue Summary

#	Issue	Location	Severity
1	Dead fallback creates an unmanaged `SharedCheckpointScheduler`	`CheckpointManager.java` ctor	Low
2	One test still builds a bare `CheckpointManager` mock (null lease)	`CheckpointCoordinatorTest.java` ~L1705	Low
-	Follow-up: owner-checked error overload, package-private `expireCheckpoint`, savepoint-poll test (agreed, separate PR)	`CheckpointCoordinator.java`	Medium, non-blocking

5. Merge Recommendation

Conclusion: Ready to merge

Blockers: none on the source side. Remaining gates are mechanical: a rerun of the one infrastructure-failed Windows 17 job, the Apache Build result for this head, and approval from a write-capable maintainer (my account cannot provide that).

Recommended non-blocking fixes: Issues 1 and 2 (or fold them into the agreed follow-up PR).

Overall assessment: a careful, well-documented and well-tested change; the merge with #12042 was resolved correctly and preserved both sides. Thank you for the thorough work on this one.

Vivek1106-04 · 2026-11-03T05:18:08Z

@SEZ9 @DanielLeens Here is the fork Build result for b48e21019 that I said I would post.

The first attempt of run 41170040575 failed on one job only: unit-test (17, windows-latest). That job got 403 Forbidden while downloading the Maven wrapper, so no tests ran on it, as @DanielLeens found. I reran that job. The run now reports success on attempt 2, head b48e2101972853a13914547d659e55b0e64f6ffb. The head has not changed.

The two Low nits from the last review are the dead fallback scheduler and the bare CheckpointManager mock in CheckpointCoordinatorTest. I will fold them into the agreed follow-up PR (#13052) so this PR's reviewed scope stays the same. If you would rather have them here, tell me and I will push them.

DanielLeens

Thanks @Vivek1106-04 for rerunning the Windows leg and posting the result. I confirmed it on the unchanged head b48e2101972853a13914547d659e55b0e64f6ffb: fork run 41170040575 is success on attempt 2, with all 96 jobs either success (85) or skipped (11), so the 403 Maven-wrapper download on unit-test (17, windows-latest) was indeed runner infrastructure.

Folding the two Low nits (the dead new SharedCheckpointScheduler() fallback in the CheckpointManager constructor and the bare CheckpointManager mock in CheckpointCoordinatorTest) into the agreed follow-up #13052 is fine with me; keeping this head unchanged avoids another review and CI round for no behavioural gain.

With CI green and no source blockers, I am approving this head. The Apache-side check rollup for this SHA is still empty, so a write-capable maintainer needs to reconcile the required Build check and give the final approval for the merge gate. Thanks again for the careful work on this one.

SEZ9 · 2026-11-04T02:25:05Z

Thanks @Vivek1106-04 for posting the fork result for b48e21019. The 403 Forbidden on the Maven wrapper download in unit-test (17, windows-latest) was runner infrastructure, not the change, so the attempt 2 success of run 41170040575 on the unchanged head b48e2101972853a13914547d659e55b0e64f6ffb is what I needed to see.

On your question: please keep the two Low nits (the dead fallback scheduler and the bare CheckpointManager mock in CheckpointCoordinatorTest) in the agreed follow-up PR. No need to push them here; the reviewed scope of this PR stays as it is at b48e21019.

Nothing further is needed from you on this one. A write-capable maintainer still needs to reconcile the required Build check and give the final approval for the merge gate.

DanielLeens · 2026-11-05T00:36:35Z

Thanks @SEZ9 for confirming the same-head retry result and the agreed scope. My technical approval stands. Keep the two low-priority scheduler/test nits in the agreed follow-up #13052; no additional commit is needed here for them. The fork result and Apache required-check adoption are separate, and a write-capable maintainer still owns the effective approval and merge.

This is an unchanged-head discussion reply, not another complete source review. Daniel has READ permission; technical approval and GitHub's effective maintainer gate are separate.

github-actions Bot added the Zeta label Sep 7, 2026

Vivek1106-04 force-pushed the improve-12122-shared-checkpoint-scheduler branch from 0521939 to 0032ddc Compare September 7, 2026 06:12

DanielLeens approved these changes Sep 7, 2026

View reviewed changes

github-actions Bot added the reviewed label Sep 7, 2026

davidzollo added the First-time contributor First-time contributor label Sep 8, 2026

nzw921rx reviewed Sep 8, 2026

View reviewed changes

This was referenced Sep 9, 2026

[Improve][Zeta] Apply the finished-jobs page before the per-job metrics and DAG lookups #12130

Merged

[Bug][E2E][Flink] KuduIT hangs until the 90-minute job limit: the JobManager stops responding right after deploying the Kudu source task #12132

Closed

jjj-n mentioned this pull request Sep 10, 2026

[Improve][Zeta] Separate JobMaster lifetime waits from the coordinator executor before bounding max-thread-num #12117

Open

Vivek1106-04 requested review from DanielLeens and nzw921rx September 10, 2026 04:37

github-actions Bot added the e2e label Sep 10, 2026

DanielLeens suggested changes Sep 10, 2026

View reviewed changes

github-actions Bot removed the reviewed label Sep 10, 2026

jjj-n mentioned this pull request Sep 11, 2026

[Improve][Zeta] Isolate coordinator admission from job lifecycle execution #12265

Open

5 tasks done

nzw921rx reviewed Sep 15, 2026

View reviewed changes

nzw921rx added the STIP SeaTunnel Improvement Proposal label Sep 15, 2026

Vivek1106-04 force-pushed the improve-12122-shared-checkpoint-scheduler branch from fd2841f to 7310dd7 Compare September 16, 2026 03:47

Vivek1106-04 mentioned this pull request Sep 16, 2026

[STIP-40][Improve][Zeta] Member-wide shared checkpoint trigger scheduling #12339

Open

DanielLeens reviewed Oct 24, 2026

View reviewed changes

Vivek1106-04 added 3 commits October 25, 2026 09:49

Merge branch 'dev' into improve-12122-shared-checkpoint-scheduler

feedd25

[Fix][Zeta] Complete a failing checkpoint task's future before logging

29ab751

If logging itself throws, for example under OutOfMemoryError, the future would never complete and a waiter on get() would hang.

DanielLeens reviewed Oct 25, 2026

View reviewed changes

Vivek1106-04 added 2 commits October 27, 2026 12:22

DanielLeens reviewed Oct 27, 2026

View reviewed changes

Vivek1106-04 mentioned this pull request Oct 28, 2026

[Bug][Zeta] A stale asynchronous checkpoint error can fail a restored checkpoint coordinator #13052

Open

DanielLeens reviewed Oct 31, 2026

View reviewed changes

DanielLeens approved these changes Nov 3, 2026

View reviewed changes

Conversation

Vivek1106-04 commented Sep 7, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Purpose of this pull request

How it works

1. Before and after

2. Who owns what

3. The timer only fires; the dispatch pool does the work

4. Savepoints don't queue behind the periodic timer

5. Master failover reuses the lease

6. Cancelling a task that has already been handed off

Shutdown

Timeout watchdog: expiry runs off the dispatch thread

Cost of the hand-off

What this PR does not change: savepoint lock holding (#12441)

Does this PR introduce any user-facing change?

How was this patch tested?

Rebase note

Check list

Uh oh!

DanielLeens left a comment

Choose a reason for hiding this comment

What Problem Does This PR Solve?

Example: Before vs After

1. Code Change Review

1.1 Core Logic Analysis

1.2 Compatibility Impact

1.3 Performance / Side-Effect Analysis

1.4 Error Handling and Logging

2. Code Quality Assessment

2.1 Coding Standards

2.2 Test Coverage and Test Stability

2.3 Documentation Updates

3. Architectural Soundness

3.1 Precise fix vs. workaround vs. long-term solution

3.2 Maintainability

3.3 Extensibility

3.4 Historical-Version Compatibility

4. Issue Summary

5. Merge Recommendation

Conclusion: Ready to merge

Uh oh!

nzw921rx left a comment

Choose a reason for hiding this comment

Uh oh!

DanielLeens commented Sep 9, 2026

Uh oh!

Vivek1106-04 commented Sep 10, 2026

Uh oh!

nzw921rx commented Sep 10, 2026

Uh oh!

Vivek1106-04 commented Sep 10, 2026

Uh oh!

DanielLeens left a comment

Choose a reason for hiding this comment

What Problem Does This PR Solve?

1. Code Change Review

1.1 Core Logic Analysis

1.2 Compatibility Impact

1.3 Performance / Side-Effect Analysis

1.4 Error Handling and Logging

2. Code Quality Assessment

2.1 Coding Standards

2.2 Test Coverage and Test Stability

2.3 Documentation Updates

3. Architectural Soundness

3.1 Elegance of the Solution

3.2 Maintainability

3.3 Extensibility

3.4 Historical-Version Compatibility

4. Issue Summary

5. Merge Recommendation

Conclusion: Ready to merge after fixes

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Vivek1106-04 commented Sep 7, 2026 •

edited

Loading

The `RejectedExecutionException`

CI on `771fce221`