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

@nzw921rx

Member-wide shared checkpoint trigger scheduling for Zeta

Improvement issue: #12122.
Candidate implementation: PR #12165.

This proposal is raised at the request of @nzw921rx on PR #12165, so that the scheduling model is agreed on the dev list before the implementation is considered for merge. Nothing in the linked PR should be read as an accepted design.

Background

CheckpointCoordinator constructs its own Executors.newScheduledThreadPool(2, ...) in its constructor, and CheckpointManager constructs one coordinator per entry of the checkpoint plan map, that is, one per pipeline. Timer threads therefore scale linearly with the number of checkpoint-enabled pipelines on a member. A member running 500 such pipelines holds up to 1000 threads that are asleep nearly all of the time.

The pool is only ever used as a timer. There are exactly two scheduling call sites in CheckpointCoordinator:

re-arming the periodic trigger, in scheduleTriggerPendingCheckpoint, which runs tryTriggerPendingCheckpoint
arming the per-checkpoint timeout watchdog, inside startTriggerPendingCheckpoint

The heavy barrier work already runs on the injected executorService via thenApplyAsync, not on the scheduled pool. That is the property that makes a shared timer viable at all.

cleanPendingCheckpoint currently calls shutdownNow() on the per-coordinator pool and constructs a replacement, which means the pool's lifecycle is bound to a single pipeline's cleanup.

Two details are worth stating because they bound the size of the actual problem:

Core workers are created on demand; there is no prestartAllCoreThreads anywhere in seatunnel-engine. Periodic triggering begins from allTaskReady, and restoreCoordinator handles restored pipelines. A queued, never-started pipeline therefore holds the executor object but no threads.
The accurate bound today is "up to two worker threads per pipeline whose scheduler has actually been used", not "two per submitted job".

Motivation

An operator consolidating many small streaming jobs onto one Zeta member should not pay two sleeping OS threads per checkpoint-enabled pipeline. The cost is not CPU; it is thread-count pressure, stack memory, and the scheduler noise of a large pool of idle timers, all of which grow with a number the operator controls only indirectly.

After this change, the steady-state timer cost of a member becomes a small constant independent of pipeline count, while per-pipeline cancellation semantics and the single-pending-checkpoint invariant stay exactly as they are today.

This is a scalability change to thread ownership. It is not a regression report and not a change to checkpoint semantics.

Goals

Replace the per-coordinator scheduled pool with a member-wide timer whose thread count does not grow with the number of active pipelines.
Keep cancellation scoped per pipeline, keyed by (jobId, pipelineId), so one pipeline's cleanup cannot touch another's outstanding timers.
Isolate blocking checkpoint or RPC work so that a slow pipeline delays neither another pipeline's timers nor its own next tick.
Preserve the single-pending-checkpoint guard and the existing locking discipline in the trigger path.
Keep the coordinator usable across a master-failover reset without rebuilding any thread pool.
Introduce no new Option, no default change, and no checkpoint, savepoint, or protocol format change.

Non-Goals

Changing checkpoint triggering semantics, barrier alignment, checkpoint storage, or savepoint behavior.
Bounding or reshaping the coordinator executor. That is separate work, tracked in [Improve][Zeta] Separate JobMaster lifetime waits from the coordinator executor before bounding max-thread-num #12117 and [Improve][Zeta] Budget promoted cooperative workers in TaskExecutionService #12121, and this proposal deliberately stays independent of it.
Making the checkpoint interval, timeout, or any scheduling parameter user-tunable beyond what exists today.
Flink or Spark translation layers. This is a Zeta engine change only.
Cross-member or cluster-wide scheduling. The unit of sharing is one member.

Design

1. Ownership

One SharedCheckpointScheduler per member, owned by SeaTunnelEngineContext and closed with it. It is deliberately not a static singleton, because several members run inside one JVM during unit and E2E tests, and a static instance would let one member's shutdown tear down another's timers.

Each CheckpointCoordinator obtains a PipelineCheckpointScheduler lease instead of constructing a pool. The lease is identified by (jobId, pipelineId) and holds only its own outstanding tasks. Nothing is registered back on the shared object, so dropping a lease when a job ends leaks nothing.

cleanPendingCheckpoint cancels this pipeline's outstanding tasks rather than shutting down a pool. The lease stays usable afterwards, so a coordinator restored after a master-failover reset can schedule again immediately.

graph TD
    subgraph Today["Today: pool per pipeline"]
        C1[Coordinator p1] --> P1["pool: 2 threads"]
        C2[Coordinator p2] --> P2["pool: 2 threads"]
        CN[Coordinator pN] --> PN["pool: 2 threads"]
    end

    subgraph Proposed["Proposed: leases on member threads"]
        L1[Coordinator p1] --> S["SharedCheckpointScheduler<br/>owned by SeaTunnelEngineContext"]
        L2[Coordinator p2] --> S
        LN[Coordinator pN] --> S
    end

2. Split timing from execution

Timing and execution are handled by two different pools, and this is the part that most deserves review.

A fixed, small timer pool does nothing but hand a due task to a dispatch pool. The checkpoint body itself never runs on a timer thread. The hand-off is a constant-time execute that never blocks the timer thread: it reuses an idle dispatch thread, creates one up to the bound, or enqueues.

The consequence is that a checkpoint that blocks on IO or a slow RPC occupies a dispatch thread, not a timer thread. Every other pipeline's next trigger still fires on schedule, and so does the slow pipeline's own next tick.

This is stricter than today's behavior, not merely equivalent. Today a coordinator's two threads serve both timing and bodies, so a blocked body consumes one of that pipeline's own timer threads, and a blocked trigger together with a blocked watchdog can occupy both.

sequenceDiagram
    autonumber
    participant Lease as PipelineCheckpointScheduler
    participant Timer as Timer pool (fixed, small)
    participant Disp as Dispatch pool (bounded, on demand)
    participant Coord as CheckpointCoordinator body

    Lease->>Timer: schedule(task, delay)
    Note over Timer: sleeps until due
    Timer->>Disp: execute(task)
    Note over Timer: returns immediately,<br/>free for the next pipeline
    Disp->>Coord: run trigger or timeout watchdog
    Note over Coord: may block on RPC,<br/>holds a dispatch thread only

3. Cancellation across two pools

Because a task crosses two pools, cancellation has to work both before the timer fires and after the body has been handed to the dispatch pool. The task therefore re-checks its own cancellation flag at run time rather than relying on the timer future, which is already complete once the body is running. The timer uses setRemoveOnCancelPolicy(true), since coordinators cancel the timeout watchdog on every acknowledged checkpoint and cancelled entries would otherwise sit in the shared queue until their delay elapsed.

4. Invariants

The single-pending-checkpoint guard is untouched. The trigger path stays inside synchronized (lock) and the pending-counter guard still short-circuits a redundant trigger. Coordinators already had two threads able to enter that path concurrently, so sharing the timer introduces no new race: additional dispatch threads contend on the same lock and bail through the same guard.

5. Bounding the dispatch pool

Issue #12122 asks for a "shared, bounded scheduler". Both pools are now bounded, and the shape of the dispatch bound is the part worth reviewing.

The constraint that rules out the obvious approach: a bounded pool over a SynchronousQueue rejects when saturated, and a rejected task here is a dropped checkpoint trigger or a dropped timeout watchdog. That is a correctness problem, not a resource problem, so no bound that can reject is acceptable.

The candidate policies and their costs:

Caller-runs. Pushes checkpoint work back onto a timer thread, reintroducing exactly the head-of-line blocking this design exists to remove. Rejected.
Bounded queue with an explicit failure path. Needs agreement on what a member should do when it cannot trigger a checkpoint on time, which is a larger contract than this change should invent.
Unbounded queue with a bounded thread count. Trades an unbounded thread count for delay under saturation. Chosen.

The implementation is a ThreadPoolExecutor with core equal to maximum, an unbounded queue, and allowCoreThreadTimeOut(true). Core equals maximum because a pool backed by an unbounded queue never grows past its core size; allowing core threads to time out restores the on-demand growth and idle reaping that the previous cached pool provided. The bound is max(8, availableProcessors * 2), so it scales with the host rather than with the pipeline count.

The residual cost is honest and worth stating: if every dispatch thread is blocked at the same moment, further checkpoint bodies queue rather than starting immediately. That is a delay, not a drop, and past saturation the queue grows for as long as the overload lasts. The benchmark in #12418 puts a number on where that begins: scheduling delay is flat at about 10 us up to 31% dispatch utilization and reaches a p50 of 461 ms at 125%.

Correction. This section previously said a dispatched body is always short-lived, because the heavy barrier work runs on the coordinator's own executor via thenApplyAsync. That was too strong, and two paths have since been checked against the source rather than assumed.

The timeout watchdog ran its expiry handling inline on the dispatch thread. That handling drives distributed IMap reads and writes with retries through updateStatus, and takes the coordinator lock through cleanPendingCheckpoint. It now hands that work to the coordinator executor and keeps only the pending-checkpoint lookup on the dispatch thread. Because that executor is bounded over a SynchronousQueue and aborts when saturated, and a dropped expiry would leave the checkpoint pending with nothing left to fail it, a rejection falls back to running inline.
startSavepoint holds the coordinator lock across a 500 ms sleep-poll until the in-flight checkpoint completes, and tryTriggerPendingCheckpoint takes that same lock on a dispatch thread. A periodic trigger firing inside a savepoint window therefore blocks for the length of that savepoint, and CheckpointManager.triggerSavePoints fans out over every coordinator of a job at once. This is a new cross-pipeline coupling: with per-pipeline pools such a pipeline could block only its own two threads. Moving the wait off the lock is tracked as [Bug][Zeta] startSavepoint holds the coordinator lock across its sleep-poll, blocking concurrent checkpoint triggers #12441 rather than done here, since it changes dev's savepoint path rather than this design.

In steady state, with the watchdog hop in place, what remains on a dispatch thread is atomics, timestamp comparisons, logging and a map put. At the default checkpoint.interval of 300000 ms, dispatch utilization is effectively zero and the saturation regime above is not reachable; the savepoint path is the one mechanism that can hold a dispatch thread for seconds, and reaching simultaneous occupancy of the whole pool through it has not been demonstrated.

Feedback on this choice is still welcome, particularly from anyone who has seen checkpoint bodies block for long periods in practice.

A secondary question I have no strong view on: the timer pool size is a constant rather than an Option. Adding one would create a user-facing config contract for something that arguably should not need tuning, and would require matching docs/en and docs/zh updates. I will follow the list.

Compatibility and Migration

No incompatible change is proposed.

No new or renamed Option, and no default value change.
No checkpoint, savepoint, or wire format change; no SPI or public API change.
No entry in docs/en/introduction/concepts/incompatible-changes.md is required.
The only externally observable difference is thread naming. Per-coordinator threads named checkpoint-coordinator-<pipelineId>/<jobId> are replaced by member-wide pools. Anything scraping thread names, including the E2E thread-leak check, needs updating; the candidate PR updates that check.

Test Plan

Unit tests asserting that the member timer thread count stays flat as pipeline count grows, that cancellation is scoped to one lease and leaves other leases untouched, that a lease remains usable after cancelAll, and that cancellation works both before the timer fires and after the body is dispatched.
Existing CheckpointCoordinatorTest must pass unchanged in behavior, covering restore, savepoint, and cleanup paths.
E2E: the Zeta engine suites, including checkpoint restore and savepoint scenarios, plus the container thread-leak check with the new pool names accounted for.
Scalability check per [Improve][Zeta] Share checkpoint trigger scheduling across active pipelines #12122: benchmark active checkpoint-enabled pipelines separately from initialized or queued jobs, and confirm 500 active pipelines show a flat timer-thread count.
A test that saturates the dispatch pool and asserts the thread count stays within its bound.
A test that deliberately blocks pipelines in their checkpoint bodies and asserts another pipeline's trigger still runs, covering the design's central claim directly.
A test that expires a checkpoint and asserts the expiry handling does not run on a shared dispatch thread, covering the watchdog hop above.
A regression that fills the dispatch pool with coordinators whose lock is held by an in-progress savepoint, and asserts that an unrelated pipeline's trigger and timeout watchdog still run within a bound. It lands with the fix for [Bug][Zeta] startSavepoint holds the coordinator lock across its sleep-poll, blocking concurrent checkpoint triggers #12441, not in [Improve][Zeta] Share checkpoint trigger scheduling across active pipelines #12165: written red against the current blocking lock acquisition first, then made green by the fix. [Improve][Zeta] Share checkpoint trigger scheduling across active pipelines #12165 documents this coupling (see the Correction above) and does not claim to fix it.

Alternatives Considered

Dispatch checkpoint work onto the existing coordinator executor. Rejected for the scheduling itself. That executor is a bounded ThreadPoolExecutor over a SynchronousQueue with a rejection handler, so a saturated coordinator pool could silently drop a checkpoint trigger or timeout watchdog. It would also couple this change to the executor-bounding work in [Improve][Zeta] Separate JobMaster lifetime waits from the coordinator executor before bounding max-thread-num #12117 and [Improve][Zeta] Budget promoted cooperative workers in TaskExecutionService #12121. It is still the right destination for individual pieces of work that the watchdog or trigger would otherwise run inline, where a rejection can be caught and handled rather than lost: that is what the expiry hop above does.
One shared pool for both timing and bodies. Rejected. A blocking body would consume a shared timer thread and delay every other pipeline's tick, which is the exact failure mode [Improve][Zeta] Share checkpoint trigger scheduling across active pipelines #12122 asks to avoid.
Keep per-coordinator pools but shrink them to one thread. Halves the cost but does not change the linear growth, and makes self-interference worse, since a single thread serves both the trigger and the watchdog.
A static singleton scheduler. Rejected. Several members run in one JVM in tests and E2E, so one member's shutdown would tear down another's timers.
Leave as-is. Acceptable for small deployments and is the status quo; it simply does not address the consolidation case in [Improve][Zeta] Share checkpoint trigger scheduling across active pipelines #12122.

References

Issue #12122 - Share checkpoint trigger scheduling across active pipelines
PR #12165 - candidate implementation and review thread
Related executor-bounding work: #12117, #12121

good proposal🔥

please send email to [email protected]

title examples: [DISCUSS] STIP-number: xxxxxxx

Hi, I think we should support a configurable thread count for the following reasons:

Zeta supports a separated architecture. For example, in a deployment with 3 masters and N workers, checkpoint triggering may involve network communication. As the number of workers, jobs, and pipelines grows, it is difficult to guarantee that a fixed thread count will always be sufficient for large-scale data synchronization workloads.

Now that we are moving from per-pipeline CheckpointCoordinator schedulers to a shared SharedCheckpointScheduler, I think we should also provide users with a configuration option to tune the size of the shared thread pool.

A fixed default cannot guarantee that it will fit every deployment scale. Making it configurable would provide more flexibility and give operators a safer tuning path when resource contention appears in larger deployments.

The following is only a discussion that does not need to be addressed in this issue:

Should we also consider adding some simple metrics, such as queue size, active thread count, and scheduling delay?

Thanks for looking at this closely. I want to push back a bit, because I think the code puts the scaling concern in a different pool than the one the comment targets.

Where the network work actually runs

The shared scheduler deliberately does not carry the RPC path:

Timer threads only re-arm the trigger / timeout watchdog and hand the task to the dispatch pool. They never touch the network.
Dispatch threads run exactly two bodies. tryTriggerPendingCheckpoint takes the per-coordinator lock, does state checks, and calls createPendingCheckpoint -> triggerPendingCheckpoint, which is supplyAsync(..., executorService).thenApplyAsync(..., executorService). The timeout watchdog is a pendingCheckpoints lookup plus an error path. Both return quickly.
The barrier trigger and ACK handling run on executorService, the coordinator-service pool, via thenApplyAsync(this::triggerCheckpoint, executorService).

So when workers, jobs, and pipelines grow, the pressure lands on the coordinator-service pool, not on the shared timer/dispatch pools.

That pool is already configurable

engine.coordinator-service.core-thread-num (default 10) and max-thread-num (default Integer.MAX_VALUE) over a SynchronousQueue: it already grows without bound, and it is already the documented dial for exactly this scaling story. Adding checkpoint.scheduler-*-thread-num gives operators a second dial for one symptom, and the newer, more checkpoint-sounding name is the one they will reach for first -- while the pool that is actually saturated is the other one.

What the shared pools do today

Dispatch is max(8, availableProcessors * 2), threads created on demand, reaped after 60s idle, unbounded queue that never rejects (a rejection here would be a dropped trigger or a dropped watchdog). Undersizing it yields delay, not loss, and that delay is only reachable if all dispatch threads block at once -- which requires the dispatched bodies to block, and they do not.

The one real hole, which a knob would paper over

startTriggerPendingCheckpoint registers pendingCompletableFuture.thenAccept(...), and that body contains a blocking CompletableFuture.allOf(completableFutureArray).get(). Normally the pending future is still incomplete (its chain runs on executorService), so the body lands on a coordinator-service thread. But if it has already completed, thenAccept runs inline on the dispatch thread and blocks it on the barrier send. The fix for that is thenAcceptAsync(..., executorService), not a bigger dispatch pool. I would rather close that hole than add a dial that hides it.

What I propose instead -- invert the order of your two comments

Metrics first, option second, and only if the metrics show it:

Ship this PR with the current defaults.
Follow-up: the metrics from your second comment -- dispatch queue depth, active dispatch threads, and scheduling delay (actual fire time minus scheduled fire time). Scheduling delay is the discriminator: sustained nonzero means the pool is undersized; flat zero means the pool is irrelevant to whatever the operator is chasing.
If a real separated deployment then shows sustained delay, add the option with documentation naming the metric to tune it against.

Option names are permanent contracts here. I would rather not ship one for a hypothetical, and today an operator has no signal to tune it against -- which is precisely your second comment. A dial without a gauge is guessing, and guessing low silently delays triggers and watchdogs for every pipeline on the member.

If you still want it in this PR

I will add it, but as one option, not two: checkpoint.scheduler-dispatch-thread-num, default 0 meaning auto (max(8, availableProcessors * 2)), so no existing deployment changes behavior. No timer option -- timer threads only hand off, and there is no signal an operator could tune that against.

And a direct question: do you have a deployment where checkpoint triggers already drift under load? If there is a reproduction, that settles it and I will size from the measurement rather than from the argument above.

Thanks for the detailed breakdown, that clarifies where the load actually lands. Since the barrier trigger and ACK handling run on the coordinator-service pool, which is already tunable via engine.coordinator-service.core-thread-num / max-thread-num, I agree that adding a second checkpoint-specific dial without a signal to tune against is not ideal.

A few follow-ups:

The thenAccept path that can run the blocking CompletableFuture.allOf(completableFutureArray).get() inline on a dispatch thread sounds like it should be closed as part of this change. Will the thenAcceptAsync(..., executorService) fix go into the candidate PR?
If the option does end up in this PR, your compromise of a single checkpoint.scheduler-dispatch-thread-num with default 0 meaning auto works for me, since existing deployments keep the same behavior. Let's see what the dev list says.
Metrics (dispatch queue depth, active dispatch threads, scheduling delay) as a follow-up sounds good.

On your question about a deployment with drifting triggers: my concern comes from the separated-deployment scenario in general rather than a specific measured case, so I don't have a reproduction to offer. Could you also post the dev@ [DISCUSS] thread link here once it is sent, so the decision on the option is recorded alongside this issue?

Thanks — that settles the option question for now, so let me answer the three
follow-ups and record the dev@ discussion here.

1. The thenAccept fix — yes, it goes in this PR.

Confirmed against CheckpointCoordinator: startTriggerPendingCheckpoint
registers with thenAccept, so when the pending future is already complete the
body runs inline on the calling thread, and that body blocks on
CompletableFuture.allOf(completableFutureArray).get() for the whole barrier
ACK round-trip. The caller is tryTriggerPendingCheckpoint, which holds
synchronized (lock), so the block is taken while holding the coordinator lock.

This is pre-existing — today it stalls the coordinator's own timer thread. What
changes with a shared pool is the blast radius: the same stall would occupy a
member-wide dispatch thread. That is exactly the coupling this PR is meant to
remove, so closing it here rather than separately.

Fix is thenAcceptAsync(..., executorService), with a test that completes the
pending future before registration and asserts the body does not run on the
caller's thread.

2. The option — holding until the dev list converges.

Agreed on the shape if it lands: one checkpoint.scheduler-dispatch-thread-num,
default 0 meaning auto, so existing deployments are unchanged. Not adding it
speculatively.

3. Metrics — follow-up issue, not this PR.

Dispatch queue depth, active dispatch threads, and scheduling delay (scheduled
fire time vs actual, per pipeline). Scheduling delay is the one that would
justify or kill the option, so it should land before the option does, not after.

4. dev@ [DISCUSS] thread

https://lists.apache.org/thread/nxlg8d9hxy1jwj9pdhpyhc32ornnqrgj

Zhiwei Niu raised four implementation questions there — master switchover and
failover, schema evolution for CDC, single-pipeline failure inside a
multi-pipeline job, and how checkpointMinPause and the other checkpoint
options are handled. Answers are on the thread. The short version is that the
scheduler is only a delay primitive: it holds no job state, leases are keyed by
(jobId, pipelineId) so cancellation never crosses pipelines, and every policy
— interval, min-pause, checkpoint and schema-change timeouts, the pending-
checkpoint retry — stays in CheckpointCoordinator, unchanged.

One item from that thread lands here as work: I am running
MysqlCDCWithSchemaChangeIT against the branch to back the schema-evolution
answer with a result rather than an assertion, and will post it.

Ran MysqlCDCWithSchemaChangeIT against the branch, as promised above.

Tests run: 4, Failures: 0, Errors: 0, Skipped: 0  (276s)

All four Zeta cases: schema evolution, exactly-once schema evolution, multi-database same-table evolution, and the schema-change event type filter.

Across 96 schema-change checkpoint triggers there were no CHECKPOINT_EXPIRED events, no timed-out checkpoints, no skipped generic checkpoints, and no coordinator errors. So the extra queueing hop on a zero-delay schema-change trigger does not show up on this path.

Caveat worth stating: this is one local run on an unloaded host. It shows no regression, but it does not exercise a saturated dispatch pool, which is exactly what the scheduling-delay metric is for.

Separately, the thenAcceptAsync fix from point 1 above is now pushed to #12165 as 31a454f35, with a test that completes the pending checkpoint future before registration and asserts the body does not run on the registering thread.

Thank you for the update.

Regarding checkpoint.scheduler-dispatch-thread-num in SharedCheckpointScheduler, I agree that we should first determine whether this configuration is necessary based on evidence. After moving the callback containing the blocking CompletableFuture.allOf(completableFutureArray).get() call to executorService via thenAcceptAsync(...), most of my concerns have been addressed. If the dispatcher does not perform any I/O or RPC operations, this configuration does not seem necessary.
Regarding the new scheduling model, we now have a complete benchmarking framework. I think we should establish a dedicated benchmark for the new coordinator to evaluate the performance of SharedCheckpointScheduler. This could be submitted as a separate PR. It would help us evaluate the efficiency of the new SharedCheckpointScheduler and establish a complete performance baseline for it.

Currently, org.apache.seatunnel.benchmark.CheckpointingTimeBenchmark#checkpointSingleInput invokes createPendingCheckpoint through reflection, which bypasses the CheckpointScheduler. As a result, the scheduling and coordination efficiency of both the old and new implementations remains a blind spot in our benchmarks.

I suggest adding this benchmark so that we can use concrete evidence to determine whether the new implementation is better or worse than the previous one.

Agreed on both.

1. Dropping checkpoint.scheduler-dispatch-thread-num.

Not adding it in this PR. The dispatch runnables do no I/O and no RPC — they
allocate a checkpoint ID, build the PendingCheckpoint and hand off to the
coordinator's own executor — so with the blocking allOf(...).get() moved off
the dispatch thread by thenAcceptAsync, there is no longer a case I can point
to where the auto-sized pool is wrong. If the scheduling-delay metric later
shows dispatch queueing under real load, the option comes back with evidence
behind it.

2. Benchmark — agreed, separate PR. One design note so it measures the right thing.

You are right that checkpointSingleInput is blind to this. CheckpointBenchmarkTrigger
reflects straight into createPendingCheckpoint + startTriggerPendingCheckpoint,
so both the old ScheduledExecutorService and the new PipelineCheckpointScheduler
lease are out of the measured path entirely.

The thing to be careful about is that the obvious extension — route the bridge
through the scheduler and re-measure checkpointSingleInput — would still not
separate the implementations. That number is dominated by the barrier ACK
round-trip; the scheduler contributes one queue hop on top of it. At one
pipeline on an idle host, old and new will look the same, and that result would
be read as "no difference" when the benchmark simply cannot see the axis the
change is on.

The axis is pipeline count. On dev, each CheckpointCoordinator builds its
own pool:

this.scheduler = Executors.newScheduledThreadPool(2, ...);   // per (jobId, pipelineId)

so a member running P pipelines carries 2P timer threads. The shared scheduler
is TIMER_THREAD_NUM timer threads plus a bounded dispatch pool
(max(8, cores * 2)), independent of P. So what a benchmark should report, as
P sweeps 1 / 10 / 100 / 500:

scheduling delay — scheduled fire time vs actual fire time, per pipeline,
as a distribution (p50/p99/max), not a mean. This is the correctness-adjacent
number: if the shared pool ever delays a trigger past checkpointMinPause or
the timeout window, that shows up here and nowhere else.
thread count and thread CPU for the scheduling machinery, which is where
the shared model is expected to win and the per-coordinator model to degrade
linearly.
dispatch queue depth at steady state, which is the same signal the
follow-up metrics issue would expose at runtime — worth having both come from
one definition so the benchmark and production metric agree.

That shape is a latency-under-concurrency measurement, so Mode.SampleTime on
the fire-delay rather than Mode.AverageTime on completion, with the pipelines
as @State setup rather than the benchmarked call. I would keep
checkpointSingleInput exactly as it is alongside it — unchanged, so the
single-pipeline completion path stays comparable across the change — and add the
scheduler benchmark as a separate class rather than extending that one.

On sequencing: I will land the benchmark first, as its own PR against dev, so
the numbers exist before #12165 is judged on them. Running it on dev gives the
per-coordinator baseline; running the same benchmark on the #12165 branch gives
the comparison. If it shows the shared scheduler is worse on scheduling delay at
any P, that is a result worth having before the merge decision, not after.

You are right; my previous wording may not have been clear. checkpointSingleInput should remain unchanged, since it measures checkpoint completion time rather than scheduling efficiency. A dedicated benchmark should be added to evaluate the scheduling and coordination efficiency of CheckpointScheduler.

That matches what I built, so we are aligned. checkpointSingleInput is
untouched and the new benchmark is a separate class.

It is open as #12418.

CheckpointSchedulingBenchmark samples the delay between a checkpoint trigger
becoming due and its scheduling thread running it, as a distribution rather than
a mean (Mode.SampleTime, so p50/p99/max), sweeping pipelineNum 1 / 10 / 100 /
500. That is the axis the scheduling model is on: every CheckpointCoordinator
builds its own two-thread pool, so a member running P pipelines carries 2P
scheduler threads, while a shared pool is a fixed width whatever P is.

How long a trigger occupies its thread is a parameter (triggerBodyMicros)
rather than a fixed cost, since that is what decides whether a given pool width
is wide enough and it is not the same for every deployment.

Your three review points on that PR are addressed and pushed: the 4g heap, the
README revert, and registering the benchmark in benchmarks.yml.

Next I run the full sweep on dev for the baseline, then the identical sweep
against #12165, so the two scheduling models are compared on the same settings
rather than on argument.

Benchmark results, as promised. The headline is that the shared scheduler is
better where it was designed to help and indistinguishable elsewhere, but the
run also turned up a cliff I had not looked for, and it corrects something I
said earlier in this thread.

How this was measured

CheckpointSchedulingBenchmark from #12418, with a second arm added locally for
the #12165 scheduler. Both arms ran in the same JVM invocation on the same
branch, so this is a paired comparison rather than two runs stitched together.
The per-pipeline arm reproduces dev's construction directly -
newScheduledThreadPool(2, ...) per (jobId, pipelineId), same thread names,
same setRemoveOnCancelPolicy(true) - so it does not depend on branch code and
is valid on either base.

Mode.SampleTime, 3 forks, 3 warmup and 5 measurement iterations, JDK 17,
-Xms4g -Xmx4g -XX:ActiveProcessorCount=4, checkpointIntervalMillis=1000.
All figures us/op.

Pipeline count sweep

P	body us	per-pipeline p50	shared p50	per-pipeline p99	shared p99	mean ratio
1	0	7.12	10.74	11.82	15.33	0.66x
1	500	7.16	10.50	10.45	15.41	0.67x
10	0	7.42	9.90	10.78	13.49	0.75x
10	500	7.78	9.90	11.87	13.66	0.80x
100	0	9.62	9.95	15.74	14.42	0.97x
100	500	9.66	9.90	16.61	13.41	1.00x
500	0	45.70	9.95	81.15	14.83	4.61x
500	500	45.50	9.90	78.85	13.82	4.63x

Ratio above 1 means the shared scheduler is faster. Errors on all means are
between 0.002 and 0.061 us, so the differences are far outside the noise.

Three things this says:

The shared curve is flat. About 10 us p50 from 1 pipeline to 500. The
per-pipeline curve has a knee between 100 and 500 and reaches 45.7 us.
Crossover is at P=100, where the two are level. Below that the shared
scheduler is worse by roughly 3 us, which is the extra timer-to-dispatcher
hop. Against the default checkpoint.interval of 300000 ms that is three
microseconds added to a five minute timer. Real, measurable, and of no
operational consequence - but it is a regression and I would rather state it
than round it away.
Thread count behaves as the design claims. 509 live scheduler threads at
P=500 for the per-pipeline model, against at most 10 by construction for the
shared one (2 timer plus a dispatch bound of max(8, cores * 2) = 8 here),
independent of P.

The max column is omitted deliberately: it is a single sample out of several
million and is noisy in both arms (per-pipeline maxima ranged 1.2-13 ms, and one
shared sample at P=1 hit 63 ms). It is not a comparable statistic here.

A correction, and the cliff

In my first pass I concluded that triggerBodyMicros moved nothing in either
arm and read that as evidence the dispatch pool is never the constraint. That
conclusion was wrong, and I am glad I checked it before arguing from it.

Every configuration above sits at about 3% dispatch utilization:

P=500, interval=1000ms, body=500us
  -> 500 triggers/s x 500us = 0.25 thread-seconds/s against 8 threads = 3%

Body occupancy could not have mattered at that load. Saturating 8 dispatch
threads at P=500 needs body >= 16 ms. So I ran the regime that actually
matters:

body us	dispatch util	arm	mean	p50	p99
500	3%	per-pipeline	46.37	45.50	78.85
500	3%	shared	10.02	9.90	13.82
5000	31%	per-pipeline	50.84	51.07	83.84
5000	31%	shared	9.45	9.33	13.28
20000	125%	per-pipeline	49.39	49.98	83.20
20000	125%	shared	477564	460849	656188

At 31% utilization the shared scheduler still wins by 5.4x. Past saturation it
collapses: p50 goes from 9.3 us to 461 ms, a factor of about 49000. The
per-pipeline arm is flat across all three, because a dedicated two-thread pool
per pipeline never contends.

461 ms is not a bound. Past saturation the unbounded queue grows for as long as
the overload lasts; the 10 second iteration simply truncates the backlog. This
is the "delay rather than drop" tradeoff that was already argued explicitly on
#12165 when the pool was bounded - the benchmark now puts a number on where it
starts.

Is that cliff reachable? I traced the dispatch body

The cliff only matters if a real checkpoint trigger can occupy a dispatch thread
for tens of milliseconds. The body is tryTriggerPendingCheckpoint, and on the
#12165 branch everything expensive in it is already delegated:

createPendingCheckpoint allocates the checkpoint id through
supplyAsync(..., executorService), with a comment stating it must happen off
the lock because it talks to external services in HA mode and may block
triggerPendingCheckpoint chains thenApplyAsync(..., executorService)
startTriggerPendingCheckpoint is thenAcceptAsync(..., executorService)
since 31a454f35, and the barrier allOf(...).get() runs inside that async
body
completePendingCheckpoint is public synchronized, so it takes the this
monitor rather than lock; its checkpointStorage.storeCheckPoint IO does not
contend with the trigger

What is left on the dispatch thread is atomics, timestamp comparisons, logging
and a map put. Microseconds. At the production default interval of 300000 ms,
utilization is effectively zero and the cliff is not reachable in steady state.

There is one mechanism I found that can hold a dispatch thread far longer.
startSavepoint holds the coordinator lock across a sleep-poll:

synchronized (lock) {
    while (pendingCounter.get() > 0 && !shutdown) {
        Thread.sleep(500);
    }
    savepoint = createPendingCheckpoint(...);

tryTriggerPendingCheckpoint takes that same lock, so a periodic trigger that
fires during a savepoint window blocks on a dispatch thread until the in-flight
checkpoint completes - seconds under backpressure. CheckpointManager .triggerSavePoints() fans out over every coordinator in a parallelStream, so
one job suspend puts all of that job's pipelines into this state at once.

I want to be careful about how far that goes. The mechanism is verified in the
code. Reaching eight simultaneous occupancies is not demonstrated - it needs
periodic triggers to coincide with savepoint windows, which at a 300 s interval
is a narrow coincidence, though the 500 ms re-arm on a pending checkpoint widens
it. I have not reproduced it and I am not reporting it as a bug.

The part I am confident about is the structural one: this is a new
cross-pipeline coupling. On dev a savepointing pipeline can block only its
own two threads, and no other pipeline can be affected by it. With one shared
bounded pool, enough concurrently blocked pipelines can delay unrelated
pipelines' triggers. That is the same blast-radius concern as the thenAccept
path, relocated rather than eliminated.

Where this leaves the option

My earlier position was that with the blocking call moved off the dispatch
thread there was no case I could point to where the auto-sized pool is wrong.
That was too strong, and the saturation run is why.

I still would not add checkpoint.scheduler-dispatch-thread-num in this PR.
Nothing here shows the auto-sized pool being wrong in steady state, and the one
path that could saturate it is a coincidence I have not been able to
demonstrate.

What the data does change is the priority of the metric. I had it as a follow-up
behind the option. It should come first, because scheduling delay is exactly the
detector for approaching this cliff: it is flat at about 10 us at 31%
utilization and four orders of magnitude worse past 100%, so it separates the
safe regime from the unsafe one cleanly, and it is the evidence that would
justify the option if a real deployment ever needs it.

Caveats

One machine, JDK 17 local, while the CI benchmark matrix is Java 8 and 11.
ActiveProcessorCount=4 is pinned by the benchmark while the host has more
cores, so the dispatch bound under test is 8. These numbers establish the shape
and the crossover, not absolutes for any particular deployment.

Raw JSON for all three runs is available if it is useful. If you would rather
see this confirmed on CI hardware across the Java 8 and 11 matrix, #12418 adds
CheckpointSchedulingBenchmark to the workflow dispatch list, so it can be run
there against both branches.

Classification: D / Zeta checkpoint-scheduler design review.

Thank you for correcting the utilization conclusion and for separating the measured saturation result from the unproven production scenario. The benchmark establishes a useful result: the shared path is flat through the measured scale, but an overloaded bounded dispatcher turns its unbounded queue into potentially unbounded trigger delay.

I also re-checked the relevant code. On current dev, startSavepoint() holds the coordinator lock while it waits in its 500 ms sleep loop; #12165 sends periodic tryTriggerPendingCheckpoint() work through the member-wide bounded dispatcher before that work acquires the same lock. Therefore the cross-pipeline coupling you describe is a valid source-based design concern, even though it is not yet a reproduced incident. It means the claim that dispatcher bodies cannot block is not a sufficient merge argument by itself.

Decision for this STIP: do not add a scheduler-thread-count option now. Keep the proposal in review, and before #12165 is accepted, make the dispatcher contract explicit and prove it: shared dispatcher workers must not remain blocked waiting for a coordinator lock while a savepoint is pending. A non-blocking lock/reschedule design or moving the savepoint wait outside that lock are possible directions; the selected direction needs a regression that fills the dispatch capacity with savepoint-contended coordinators and demonstrates that an unrelated pipeline still receives its trigger/watchdog work. Scheduling-delay observability should be prioritized ahead of any future tuning option.

@Vivek1106-04

@Vivek1106-04 thanks for running both arms in the same JVM invocation - a paired comparison, with the per-pipeline arm reproducing dev's newScheduledThreadPool(2, ...) per (jobId, pipelineId) construction, makes these numbers straightforward to trust.

The sweep supports the design's claims: the shared p50 stays at roughly 10 us from P=1 to P=500, while the per-pipeline curve has its knee between P=100 and P=500 and reaches 45.7 us p50 / 81 us p99. The 509 live scheduler threads at P=500 against a bound of at most 10 for the shared model confirms the thread-count claim with data rather than by assertion.

The ~3 us cost below the P=100 crossover from the extra timer-to-dispatcher hop is acceptable to me. Please carry it into the STIP text as a stated trade-off rather than rounding it away - the framing you used here is what I want in the doc.

Nothing changes from my earlier reply: no scheduler-thread-count option in this STIP, scheduling-delay observability ahead of any tuning knob, and the remaining gate before the shared-scheduler change is accepted is making the dispatcher blocking contract explicit and proving it with the regression described there (dispatch capacity filled with savepoint-contended coordinators, with an unrelated pipeline still receiving its trigger/watchdog work on time). Ping me once that is pushed and I will take another pass.

@SEZ9

@SEZ9 Thanks for going through the paired run. Three things done, and one question I
would rather ask than assume.

1. The ~3 us is now in the STIP text, not rounded away. The cost below the P=100
crossover is stated as the extra timer-to-dispatcher hop, with the crossover named, in
the framing you asked for.

2. The dispatcher contract is corrected in the STIP body, as a marked Correction.
The section on bounding the dispatch pool previously said a dispatched body is always
short-lived because the barrier work runs on thenApplyAsync. That was too strong. It
now names both paths that were checked against the source rather than assumed:

The timeout watchdog ran its expiry inline on the dispatch thread - updateStatus
drives IMap reads and writes through a RetryUtils loop, and cleanPendingCheckpoint
takes the coordinator lock. That is fixed in [Improve][Zeta] Share checkpoint trigger scheduling across active pipelines #12165 at da0cbbf52: the dispatch thread
keeps the pending-checkpoint lookup, the expiry goes to the coordinator executor, and
because that executor aborts when saturated and a dropped expiry would leave the
checkpoint pending with nothing left to fail it, a rejection falls back to running
inline with a WARN.
startSavepoint holds the coordinator lock across its 500 ms sleep-poll while
tryTriggerPendingCheckpoint takes the same lock on a dispatch thread, with
triggerSavePoints fanning out over every coordinator of a job at once. Stated as a
new cross-pipeline coupling and as a tracked follow-up rather than something this
design already handles.

I also amended "Alternatives Considered" item 1 so the STIP is not self-contradictory:
dispatching the scheduling onto the coordinator executor stays rejected, but the
document now says it is the right destination for an individual piece of work where a
rejection can be caught rather than lost, which is what the expiry hop does. The new
regression is in the test plan.

3. The saturation figures are attached to the residual-cost paragraph, so the
"flat to 31%, four orders of magnitude past 100%" shape is in the design document rather
than only in this thread.

The question. @DanielLeens replied on #12165 a few hours after your comment here and
split the gate: he wants the watchdog hop in that PR, which is done, and is willing to
treat the savepoint lock-holding as a tracked follow-up with corrected docs, given the
simultaneous-occupancy case needs a coincidence I have not been able to demonstrate.
Your comment here keeps the regression - dispatch capacity filled with
savepoint-contended coordinators, unrelated pipeline still served - as a gate on
acceptance.

I have implemented the reading that is already actionable and left the other open rather
than pick between you. If the regression is still required before #12165 is accepted, I
will write it red against the current design and then fix with tryLock plus reschedule
inside the dispatch body, so the change stays inside this STIP's scheduler instead of
touching dev's savepoint path. Tell me which and I will start.

@Vivek1106-04

@Vivek1106-04 thanks for the detailed update. Your comment appears to have been cut off after "Your comment here keeps the" — could you post the rest of the question so I can answer it directly?

On the three items: the changes you describe (stating the ~3 µs hop cost with the crossover, marking the dispatcher-contract change as an explicit Correction, and moving the saturation figures into the residual-cost paragraph) sound like the right direction. I have not yet reviewed the updated STIP text or the change at da0cbbf52, so I'll confirm once I've read them rather than sign off on the description alone.

On the gate, my position from this thread is unchanged: before #12165 is accepted, the dispatcher contract needs to be explicit and proven, and the regression that fills the dispatch capacity with savepoint-contended coordinators and shows an unrelated pipeline still receives its trigger/watchdog work is part of that proof. If the startSavepoint lock-holding is going to be tracked as a follow-up, please link that follow-up from the STIP's Correction section and state clearly in the test plan where the regression lands, so I can evaluate whether the split is acceptable. Scheduling-delay observability stays ahead of any thread-count or tuning option.

Once I see the rest of your question and the updated text, I'll close out anything I've missed.

@SEZ9

@SEZ9 The comment is complete on GitHub; it ends with "Tell me which and I will start.", so I think the tail was lost on your side at a line wrap. The question, in one line:

DanielLeens accepted the savepoint lock-holding as a tracked follow-up on #12165, while you kept the savepoint-contention regression as a gate on acceptance; which reading stands?

Since then @DanielLeens has approved #12165 on c773cc091 (review). His position there: the savepoint-lock case is fine as a tracked follow-up, and the coupling is "intentionally left as a documented, tracked gap rather than something this PR's own test suite needs to reproduce."

So I am proposing the split you offered to evaluate, with both of your conditions met:

The Correction section links the follow-up. The startSavepoint bullet names [Bug][Zeta] startSavepoint holds the coordinator lock across its sleep-poll, blocking concurrent checkpoint triggers #12441 as where moving the wait off the lock is tracked.
The Test Plan states where the regression lands. New bullet: the regression that fills dispatch capacity with savepoint-contended coordinators and asserts an unrelated pipeline still receives its trigger and watchdog work lands with the [Bug][Zeta] startSavepoint holds the coordinator lock across its sleep-poll, blocking concurrent checkpoint triggers #12441 fix, written red against the current blocking lock acquisition first and then made green by the fix.

What #12165 does and does not claim, so the split is easy to judge:

It fixes the watchdog half of the coupling (da0cbbf52: CheckpointCoordinator#expireCheckpoint, covered by CheckpointCoordinatorTest#testCheckpointExpiryHandlingNeverRunsOnTheDispatchThread, verified red with the fix reverted).
It does not fix the savepoint half, and says so in the SharedCheckpointScheduler class Javadoc and in the STIP Correction, both pointing at [Bug][Zeta] startSavepoint holds the coordinator lock across its sleep-poll, blocking concurrent checkpoint triggers #12441.
The coupling is bounded by the length of a savepoint, not unbounded, and reaching simultaneous occupancy of the whole dispatch pool through it needs trigger and savepoint windows to coincide, which I have not been able to demonstrate.

If you still want the regression in #12165 rather than with #12441, say so and I will write it there, red first, with the non-blocking lock acquisition as the fix. That means new commits on an approved PR and another review round, which is why I am asking rather than doing it.

Scheduling-delay observability stays ahead of any thread-count or tuning option, as before.

Classification: D / Zeta shared checkpoint-scheduler design.

Decision: the split stands. #12165 may remain scoped to the shared scheduler and its watchdog dispatch correction; it must accurately document that it does not remove the pre-existing savepoint lock coupling and link #12441 as the owner of that work. The savepoint-contention regression belongs with #12441, where it must first fail against the current blocking behavior and then prove that an unrelated pipeline still receives trigger/watchdog work after the fix.

This is not an acceptance statement for #12165. Its current head c773cc091 is still BLOCKED with a failed Build check, so it is not ready to merge. Once its current CI state is resolved or explicitly attributed, it can be reviewed on its own documented scope; #12441 remains the required path for moving the savepoint wait off the coordinator lock. We will not add a scheduler-thread-count option before scheduling-delay observability is available.

@Vivek1106-04

@Vivek1106-04 You're right, the tail was cut off on my side — apologies for the extra round trip. Answering the one-line question directly: the split stands. The savepoint-lock coupling is a tracked follow-up, not a gate on #12165. I am not asking you to add the regression to #12165; it belongs with #12441.

Concretely, for #12165:

Keep the scope as you described: the watchdog half of the coupling via da0cbbf52 (CheckpointCoordinator#expireCheckpoint, covered by CheckpointCoordinatorTest#testCheckpointExpiryHandlingNeverRunsOnTheDispatchThread, verified red with the fix reverted). No new test commits needed on the approved head c773cc091 for the savepoint case.
The Correction section and the SharedCheckpointScheduler class Javadoc must state plainly that the PR does not remove the pre-existing savepoint lock coupling and link [Bug][Zeta] startSavepoint holds the coordinator lock across its sleep-poll, blocking concurrent checkpoint triggers #12441 as the owner of moving the wait off the lock — your proposed startSavepoint bullet does that.
The Test Plan bullet you drafted is the right wording: the regression that fills dispatch capacity with savepoint-contended coordinators and asserts an unrelated pipeline still receives its trigger and watchdog work lands with [Bug][Zeta] startSavepoint holds the coordinator lock across its sleep-poll, blocking concurrent checkpoint triggers #12441, written red against the current blocking lock acquisition first and then made green by the fix. Please carry that exact red-first requirement into [Bug][Zeta] startSavepoint holds the coordinator lock across its sleep-poll, blocking concurrent checkpoint triggers #12441 so it is not lost.

One thing to be clear about: agreeing the split is not an acceptance of #12165. The current head c773cc091 still shows a failed Build check, so it is not mergeable as it stands. Once that is resolved or explicitly attributed (flake, infra, unrelated), I will do the final pass on the documented scope only.

Your bounded-not-unbounded point on the coupling (needs trigger and savepoint windows to coincide) is a fair characterization and fine to include in the Correction text as stated, given you have not been able to demonstrate full pool occupancy through it.

Scheduling-delay observability stays ahead of any thread-count or tuning option, as before — no change there.

Remaining asks:

Confirm the Correction/Javadoc/Test Plan wording above is in the PR text, or push it if it is not yet.
Resolve or attribute the failed Build check on c773cc091.
Open or update [Bug][Zeta] startSavepoint holds the coordinator lock across its sleep-poll, blocking concurrent checkpoint triggers #12441 with the red-first regression requirement so the handoff is recorded there, not only in this thread.

Navigation Menu

Repository navigation

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

Member-wide shared checkpoint trigger scheduling for Zeta

Background

Motivation

Goals

Non-Goals

Design

1. Ownership

2. Split timing from execution

3. Cancellation across two pools

4. Invariants

5. Bounding the dispatch pool

Compatibility and Migration

Test Plan

Alternatives Considered

References

How this was measured

Pipeline count sweep

A correction, and the cliff

Is that cliff reachable? I traced the dispatch body

Where this leaves the option

Caveats

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

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

Description

Member-wide shared checkpoint trigger scheduling for Zeta

Background

Motivation

Goals

Non-Goals

Design

1. Ownership

2. Split timing from execution

3. Cancellation across two pools

4. Invariants

5. Bounding the dispatch pool

Compatibility and Migration

Test Plan

Alternatives Considered

References

Activity

nzw921rx commented on Sep 16, 2026

nzw921rx commented on Sep 20, 2026

nzw921rx commented on Sep 20, 2026

Vivek1106-04 commented on Sep 20, 2026

SEZ9 commented on Sep 21, 2026

Vivek1106-04 commented on Sep 21, 2026

Vivek1106-04 commented on Sep 21, 2026

nzw921rx commented on Sep 21, 2026

Vivek1106-04 commented on Sep 21, 2026

nzw921rx commented on Sep 21, 2026

Vivek1106-04 commented on Sep 21, 2026

Vivek1106-04 commented on Sep 21, 2026

How this was measured

Pipeline count sweep

A correction, and the cliff

Is that cliff reachable? I traced the dispatch body

Where this leaves the option

Caveats

DanielLeens commented on Sep 21, 2026

SEZ9 commented on Sep 22, 2026

Vivek1106-04 commented on Sep 22, 2026

SEZ9 commented on Sep 23, 2026

Vivek1106-04 commented on Sep 23, 2026

DanielLeens commented on Sep 23, 2026

SEZ9 commented on Sep 24, 2026

Metadata

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

Issue actions