All three items are pushed.

• Heap: e8a8450b0, 4g, matching CheckpointingTimeBenchmark.
• README: 556de1f92, reverted. I had missed that the benchmark documentation
  moved to docs/en/engines/zeta/benchmark.md; that file is unchanged too.
• benchmarks.yml: 556de1f92, CheckpointSchedulingBenchmark is now in the
  benchmarks choice list next to CheckpointingTimeBenchmark.
  tools/benchmarks/test_run_benchmarks.py still passes, 7/7.

One judgement call I did not make on my own: I did not add it to
tools/benchmarks/suites/benchmarks_core.txt. The benchmark has 8 parameter
configurations (pipelineNum 1/10/100/500 x triggerBodyMicros 0/500) at
roughly 4 minutes each, so nightly inclusion costs about 32 minutes per Java
version against the 240-minute budget, and the suite file selects by class or
method, so the parameter matrix cannot be narrowed there.

If you want it tracked nightly the options are to take that cost, or to cut the
defaults in the benchmark itself to a smaller matrix and keep the full sweep for
manual dispatch. That is your call on the nightly budget - say which and I will
push it.

I am running the full sweep on dev now and will post the numbers here, then
the identical sweep against #12165 so both scheduling models are measured on the
same settings.

Full sweep on this branch, replacing the smoke numbers in the description.

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

Errors on the means are 0.003 to 0.086 us. The benchmark resolves the thing it
was built to resolve: scheduling delay is flat from 1 to 100 pipelines and then
has a knee, reaching 6.8x the single-pipeline figure at 500, while the thread
count tracks the pipeline count exactly as newScheduledThreadPool(2, ...) per
coordinator implies.

The max column is left out on purpose. It is one sample out of several million
and is noisy here (333 us to 4096 us across these rows), so it is not a
statistic worth reading.

I ran the same benchmark with a second arm for the #12165 shared scheduler, and
posted that comparison, a dispatch-saturation sweep, and a trace of what the
dispatch body actually does on
#12339 (comment) rather
than here, since it is about that design rather than about this benchmark. The
short version relevant to this PR is that the benchmark separates the two
scheduling models cleanly, including locating a regime where the shared model is
four orders of magnitude worse, which was the point of building it.

These are single-machine numbers on JDK 17, while the workflow matrix is Java 8
and 11. Now that the benchmark is in the benchmarks.yml dispatch list it can
be run on CI hardware against either branch if you want the figures confirmed
there.

Thanks both for keeping this moving.

@Vivek1106-04, the full sweep is useful and it actually lines up with Issue 1 from my earlier review: triggerBodyMicros (0 vs 500) barely moves the numbers at any pipeline count, which is exactly what you'd expect if the measured task almost never has to wait behind the periodic body. The knee at 500 pipelines looks like thread/CPU contention from ~1000 live threads in one JVM rather than scheduling-queue pressure, so I'd read this sweep as "per-pool wake-up latency under near-zero load," not yet as a load-bearing comparison for #12165 - which is really the same thing @nzw921rx is getting at below.

@nzw921rx, I think you're right that the harness measures thread wake-up delay rather than queueing under the engine's real trigger pattern, and that this limits how far the numbers can be trusted as a stand-in for the actual scheduling model comparison - that's the same gap I flagged as Issue 1 (non-blocking) in my review.

On routing through the real path: there's already a testing hook at that boundary - CheckpointCoordinator#scheduleTriggerPendingCheckpoint(CheckpointType, long) (CheckpointCoordinator.java:547-553) is @VisibleForTesting protected. The catch is that it schedules tryTriggerPendingCheckpoint(checkpointType), which on the happy path calls createPendingCheckpoint(...) and startTriggerPendingCheckpoint(...) (CheckpointCoordinator.java:816-824) - i.e. it builds a real PendingCheckpoint, writes to checkpoint storage, and sends real RPC operations out to task locations. Driving the benchmark through that method means constructing a real (or fully mocked, as CheckpointCoordinatorTest does with Mockito) CheckpointManager, execution plan and task locations, so the sample would then include storage and RPC noise on top of scheduling delay, which is a different and heavier thing to measure than what this PR set out to isolate.

A cheaper middle ground that gets at the same question without that machinery: my Issue 1 write-up already proposed arming both a trigger-like task and a watchdog-like task per pool with a duty cycle long enough that both of a pool's 2 workers can be occupied, so the measured task genuinely has to queue behind them - closer to how tryTriggerPendingCheckpoint's own re-scheduling (scheduleTriggerPendingCheckpoint(checkpointType, 500L) when pendingCounter.get() > 0, line ~816) can pile a second timer onto the same pool. That would show whether pool width matters under real contention, still as a standalone harness.

Could you say which of these you're looking for: exercising tryTriggerPendingCheckpoint's full side effects (storage + RPC, higher fidelity but a much bigger, more intrusive change), or a queuing-faithful standalone harness along those lines? Either is a reasonable ask, but they're different asks with different cost, and it'd help @Vivek1106-04 to know which one unblocks this PR versus which is really scoped to the #12165 discussion.

Thanks both for keeping this moving.

@Vivek1106-04, the full sweep is useful and it actually lines up with Issue 1 from my earlier review: triggerBodyMicros (0 vs 500) barely moves the numbers at any pipeline count, which is exactly what you'd expect if the measured task almost never has to wait behind the periodic body. The knee at 500 pipelines looks like thread/CPU contention from ~1000 live threads in one JVM rather than scheduling-queue pressure, so I'd read this sweep as "per-pool wake-up latency under near-zero load," not yet as a load-bearing comparison for #12165 - which is really the same thing @nzw921rx is getting at below.

@nzw921rx, I think you're right that the harness measures thread wake-up delay rather than queueing under the engine's real trigger pattern, and that this limits how far the numbers can be trusted as a stand-in for the actual scheduling model comparison - that's the same gap I flagged as Issue 1 (non-blocking) in my review.

On routing through the real path: there's already a testing hook at that boundary - CheckpointCoordinator#scheduleTriggerPendingCheckpoint(CheckpointType, long) (CheckpointCoordinator.java:547-553) is @VisibleForTesting protected. The catch is that it schedules tryTriggerPendingCheckpoint(checkpointType), which on the happy path calls createPendingCheckpoint(...) and startTriggerPendingCheckpoint(...) (CheckpointCoordinator.java:816-824) - i.e. it builds a real PendingCheckpoint, writes to checkpoint storage, and sends real RPC operations out to task locations. Driving the benchmark through that method means constructing a real (or fully mocked, as CheckpointCoordinatorTest does with Mockito) CheckpointManager, execution plan and task locations, so the sample would then include storage and RPC noise on top of scheduling delay, which is a different and heavier thing to measure than what this PR set out to isolate.

A cheaper middle ground that gets at the same question without that machinery: my Issue 1 write-up already proposed arming both a trigger-like task and a watchdog-like task per pool with a duty cycle long enough that both of a pool's 2 workers can be occupied, so the measured task genuinely has to queue behind them - closer to how tryTriggerPendingCheckpoint's own re-scheduling (scheduleTriggerPendingCheckpoint(checkpointType, 500L) when pendingCounter.get() > 0, line ~816) can pile a second timer onto the same pool. That would show whether pool width matters under real contention, still as a standalone harness.

Could you say which of these you're looking for: exercising tryTriggerPendingCheckpoint's full side effects (storage + RPC, higher fidelity but a much bigger, more intrusive change), or a queuing-faithful standalone harness along those lines? Either is a reasonable ask, but they're different asks with different cost, and it'd help @Vivek1106-04 to know which one unblocks this PR versus which is really scoped to the #12165 discussion.

@DanielLeens Thank you for your reply
Real RPC/IO can cause distortion in test scheduling, but existing custom thread pools cannot test the engine's true methods. I provide the following ideas:
Add new construction methods to eliminate existing RPC behaviors and mockups such as IMAP

Design:

Checkpoint Scheduling Benchmark Design

Goal

Measure the scheduling delay of the real CheckpointCoordinator path without starting SeaTunnel or Hazelcast nodes.

The benchmark must not override tryTriggerPendingCheckpoint.

Measurement Boundary

Start:
scheduleTriggerPendingCheckpoint()

Measured:
scheduler queueing
→ scheduler wake-up
→ shared dispatch, when applicable
→ tryTriggerPendingCheckpoint()
→ state checks
→ coordinator lock
→ submission of createPendingCheckpoint to executorService

End:
executorService.execute() receives the checkpoint creation task

The remaining checkpoint path continues outside the measured region:

CheckpointID allocation
→ PendingCheckpoint creation
→ barrier RPC
→ task ACK
→ checkpoint storage
→ IMap state update
→ checkpoint completion

CheckpointCoordinator Constructor

The production constructor delegates to a testing constructor that accepts the real scheduler.

public CheckpointCoordinator(
        CheckpointManager manager,
        CheckpointStorage checkpointStorage,
        CheckpointConfig checkpointConfig,
        long jobId,
        CheckpointPlan plan,
        CheckpointIDCounter checkpointIdCounter,
        PipelineState pipelineState,
        ExecutorService executorService,
        IMap<Object, Object> runningJobStateIMap,
        boolean isStartWithSavePoint,
        CheckpointMonitorService checkpointMonitorService) {

    this(
            manager,
            checkpointStorage,
            checkpointConfig,
            jobId,
            plan,
            checkpointIdCounter,
            pipelineState,
            executorService,
            runningJobStateIMap,
            isStartWithSavePoint,
            checkpointMonitorService,
            manager.leaseCheckpointScheduler(plan.getPipelineId()));
}

@VisibleForTesting
protected CheckpointCoordinator(
        CheckpointManager manager,
        CheckpointStorage checkpointStorage,
        CheckpointConfig checkpointConfig,
        long jobId,
        CheckpointPlan plan,
        CheckpointIDCounter checkpointIdCounter,
        PipelineState pipelineState,
        ExecutorService executorService,
        IMap<Object, Object> runningJobStateIMap,
        boolean isStartWithSavePoint,
        CheckpointMonitorService checkpointMonitorService,
        PipelineCheckpointScheduler scheduler) {

    // Keep the existing initialization unchanged.
    this.executorService = executorService;
    this.checkpointManager = manager;
    this.checkpointStorage = checkpointStorage;
    this.checkpointIdCounter = checkpointIdCounter;
    this.runningJobStateIMap = runningJobStateIMap;
    this.checkpointMonitorService = checkpointMonitorService;
    this.scheduler = scheduler;

    // Initialize the remaining existing fields normally.
}

On dev, the same testing constructor accepts the real per-pipeline ScheduledExecutorService. The two revisions are compiled separately by the benchmark workflow.

Scheduling Probe

Use the constructor-injected ExecutorService as the measurement probe.

final class ProbingExecutorService extends AbstractExecutorService {

    private final ExecutorService delegate;
    private final AtomicReference<CountDownLatch> nextProbe =
            new AtomicReference<>();

    ProbingExecutorService(ExecutorService delegate) {
        this.delegate = delegate;
    }

    void arm(CountDownLatch probe) {
        nextProbe.set(probe);
    }

    @Override
    public void execute(Runnable command) {
        CountDownLatch probe = nextProbe.getAndSet(null);

        if (probe != null) {
            probe.countDown();
        }

        delegate.execute(command);
    }

    // Delegate shutdown and lifecycle methods to delegate.
}

The probe is armed only after coordinator initialization and task readiness have completed.

Benchmark Coordinator

The benchmark subclass only exposes the protected scheduling method.

final class BenchmarkCheckpointCoordinator
        extends CheckpointCoordinator {

    void scheduleForBenchmark() {
        scheduleTriggerPendingCheckpoint(
                CheckpointType.CHECKPOINT_TYPE,
                0L);
    }
}

It must not override:

tryTriggerPendingCheckpoint(...)

Therefore all existing checks, locking and pending-checkpoint behavior remain active.

Fixture

The Fixture uses real scheduling implementations and local in-memory replacements for external services.

Scheduler                 real
CheckpointCoordinator     real
tryTriggerPendingCheckpoint real
ExecutorService           real delegate with a submission probe
CheckpointIDCounter       in-memory AtomicLong
IMap                      ConcurrentHashMap-backed adapter
CheckpointStorage         in-memory implementation
RPC                       successful local fake
Task ACK                  successful local callback
SeaTunnel nodes           not started
Hazelcast cluster         not started

The memory-backed IMap must support the operations used by CheckpointCoordinator:

get()
set()
compute()
remove()

Example:

ConcurrentMap<Object, Object> state =
        new ConcurrentHashMap<>();

IMap<Object, Object> imap = mock(IMap.class);

when(imap.get(any()))
        .thenAnswer(invocation ->
                state.get(invocation.getArgument(0)));

doAnswer(invocation -> {
            state.put(
                    invocation.getArgument(0),
                    invocation.getArgument(1));
            return null;
        })
        .when(imap)
        .set(any(), any());

when(imap.compute(any(), any()))
        .thenAnswer(invocation -> {
            Object key = invocation.getArgument(0);
            BiFunction<Object, Object, Object> function =
                    invocation.getArgument(1);
            return state.compute(key, function);
        });

when(imap.remove(any()))
        .thenAnswer(invocation ->
                state.remove(invocation.getArgument(0)));

The RPC fake returns a completed InvocationFuture. After a barrier is sent, it feeds successful TaskAcknowledgeOperation instances back to the coordinator so that pendingCounter returns to zero.

The Fixture must report all tasks as READY_START and wait for initialization to finish before arming the scheduling probe.

Benchmark Invocation

CountDownLatch triggerSubmitted = new CountDownLatch(1);

probingExecutor.arm(triggerSubmitted);
coordinator.scheduleForBenchmark();
triggerSubmitted.await();

After the measured invocation, wait for the mocked ACK and checkpoint completion outside the measured region:

fixture.awaitCheckpointCompleted();

This prevents one sample from leaking pending checkpoint state into the next sample.

Comparison

The existing workflow compiles and runs each revision separately:

A1 = dev
B1 = candidate PR
B2 = candidate PR
A2 = dev

dev uses its real per-pipeline scheduler. The candidate PR uses its real shared scheduler. Both revisions keep the same Fixture behavior, workload, parameters and measurement boundary.

@Vivek1106-04 We look forward to your reply

@nzw921rx @DanielLeens On the question of which direction this harness should take: I
would take the queuing-faithful standalone harness, and I can push it here.

Taking the two options as Daniel framed them.

Driving the real CheckpointCoordinator#scheduleTriggerPendingCheckpoint. The hook
is there and is already @VisibleForTesting protected, so reaching it is not the
problem. The problem is what it pulls in: tryTriggerPendingCheckpoint on the happy path
calls createPendingCheckpoint and startTriggerPendingCheckpoint, so the sample would
include checkpoint-id allocation, a real PendingCheckpoint, checkpoint storage writes
and CheckpointBarrierTriggerOperation RPCs to task locations, on top of a mocked
CheckpointManager, execution plan and task locations of the kind CheckpointCoordinatorTest
builds with Mockito.

Two reasons I do not think that answers the question this benchmark was built for.
First, the measurement stops being scheduling delay: barrier plus storage dominates, and
the numbers would move with storage and RPC behaviour rather than with the scheduling
model. Second, we already have that measurement - CheckpointingTimeBenchmark is
barrier-ACK dominated, and its CheckpointBenchmarkTrigger reflects straight into
createPendingCheckpoint and startTriggerPendingCheckpoint. Routing this benchmark
through the same machinery would mostly reproduce it at P=1.

The queuing-faithful harness. @nzw921rx's objection is that the current state arms
one trigger-like task per pool and so measures thread wake-up latency rather than
queueing, and that is correct as written. Daniel's Issue 1 proposal fixes it directly:
arm both a trigger-like task and a watchdog-like task per pool, with a duty cycle long
enough that both of a pool's two workers can be occupied, so the measured task actually
has to queue behind them. That is faithful to the engine's real pattern - the trigger
re-arms itself at scheduleTriggerPendingCheckpoint(checkpointType, 500L) while a
checkpoint is pending, and the timeout watchdog is armed per checkpoint, so two timers
against a two-thread pool is what a busy pipeline actually looks like - and it keeps the
measured quantity as scheduling delay.

It also fixes the reading of the current sweep, which is Daniel's other point: the knee
at P=500 in what I posted is thread and CPU contention from roughly 1000 live threads in
one JVM, not queue pressure, and the harness as it stands cannot separate those. With
both timers armed it can.

So unless you want the full-fidelity version for its own sake, I will:

1. add the watchdog-like task alongside the trigger-like one, with a configurable duty
   cycle, so a measured task can genuinely queue behind both workers of a pool;
2. re-run the sweep and replace the numbers in the description, including the numbers I
   posted for [Improve][Zeta] Share checkpoint trigger scheduling across active pipelines #12165, since those were taken on the current harness and the P=500 knee
   in them should be read as thread contention rather than queueing;
3. say in the class Javadoc what the harness does and does not model, so the next reader
   does not have to rediscover it.

If you would rather have the real-engine-path version, or both arms, say so and I will
do that instead - I would just rather not put a number in front of reviewers without
being able to say which of the two it measures.

Separately, this PR's earlier three items remain addressed at this head: 4g heap, README
reverted, and the class registered in the benchmarks.yml choice list. The
benchmarks_core.txt decision is still yours; my view is to leave it out until the
matrix is settled, since eight parameter configurations at roughly four minutes each is
about 32 minutes per Java version against the 240-minute nightly budget.

@nzw921rx @DanielLeens On the question of which direction this harness should take: I would take the queuing-faithful standalone harness, and I can push it here.

Taking the two options as Daniel framed them.

Driving the real CheckpointCoordinator#scheduleTriggerPendingCheckpoint. The hook is there and is already @VisibleForTesting protected, so reaching it is not the problem. The problem is what it pulls in: tryTriggerPendingCheckpoint on the happy path calls createPendingCheckpoint and startTriggerPendingCheckpoint, so the sample would include checkpoint-id allocation, a real PendingCheckpoint, checkpoint storage writes and CheckpointBarrierTriggerOperation RPCs to task locations, on top of a mocked CheckpointManager, execution plan and task locations of the kind CheckpointCoordinatorTest builds with Mockito.

Two reasons I do not think that answers the question this benchmark was built for. First, the measurement stops being scheduling delay: barrier plus storage dominates, and the numbers would move with storage and RPC behaviour rather than with the scheduling model. Second, we already have that measurement - CheckpointingTimeBenchmark is barrier-ACK dominated, and its CheckpointBenchmarkTrigger reflects straight into createPendingCheckpoint and startTriggerPendingCheckpoint. Routing this benchmark through the same machinery would mostly reproduce it at P=1.

The queuing-faithful harness. @nzw921rx's objection is that the current state arms one trigger-like task per pool and so measures thread wake-up latency rather than queueing, and that is correct as written. Daniel's Issue 1 proposal fixes it directly: arm both a trigger-like task and a watchdog-like task per pool, with a duty cycle long enough that both of a pool's two workers can be occupied, so the measured task actually has to queue behind them. That is faithful to the engine's real pattern - the trigger re-arms itself at scheduleTriggerPendingCheckpoint(checkpointType, 500L) while a checkpoint is pending, and the timeout watchdog is armed per checkpoint, so two timers against a two-thread pool is what a busy pipeline actually looks like - and it keeps the measured quantity as scheduling delay.

It also fixes the reading of the current sweep, which is Daniel's other point: the knee at P=500 in what I posted is thread and CPU contention from roughly 1000 live threads in one JVM, not queue pressure, and the harness as it stands cannot separate those. With both timers armed it can.

So unless you want the full-fidelity version for its own sake, I will:

1. add the watchdog-like task alongside the trigger-like one, with a configurable duty
   cycle, so a measured task can genuinely queue behind both workers of a pool;
2. re-run the sweep and replace the numbers in the description, including the numbers I
   posted for [Improve][Zeta] Share checkpoint trigger scheduling across active pipelines #12165, since those were taken on the current harness and the P=500 knee
   in them should be read as thread contention rather than queueing;
3. say in the class Javadoc what the harness does and does not model, so the next reader
   does not have to rediscover it.

If you would rather have the real-engine-path version, or both arms, say so and I will do that instead - I would just rather not put a number in front of reviewers without being able to say which of the two it measures.

Separately, this PR's earlier three items remain addressed at this head: 4g heap, README reverted, and the class registered in the benchmarks.yml choice list. The benchmarks_core.txt decision is still yours; my view is to leave it out until the matrix is settled, since eight parameter configurations at roughly four minutes each is about 32 minutes per Java version against the 240-minute nightly budget.

Thank you, please take a look at my design above.

@nzw921rx Our comments crossed - mine went up about three minutes after yours and was
written without having read this. Please take this one as my actual answer, because your
design changes my view.

You are right and my objection does not apply to your version. I argued against
routing through the real path on the grounds that the sample would be dominated by
storage and RPC. Your measurement boundary removes that: it starts at
scheduleTriggerPendingCheckpoint and ends when executorService receives the
checkpoint-creation task, so everything I was worried about - id allocation,
PendingCheckpoint construction, barrier RPC, ACK, storage, IMap - is outside the
measured region by construction. That is a better answer than the standalone harness I
was proposing, because it measures the real tryTriggerPendingCheckpoint including its
state checks and its synchronized (lock), which is exactly the part the two scheduling
models differ in. I am happy to build it.

Five things I checked against the code before replying. One of them I think removes work
from your proposal rather than adding it.

1. The testing constructor is not needed, and cannot work as written

dev and #12165 already share the identical 11-argument public constructor. The
difference is entirely inside it:

• dev builds its own pool there: this.scheduler = Executors.newScheduledThreadPool(...)
  (CheckpointCoordinator.java:239-250 on dev)
• [Improve][Zeta] Share checkpoint trigger scheduling across active pipelines #12165 leases instead: this.scheduler = manager.leaseCheckpointScheduler(pipelineId)

So a benchmark that constructs a coordinator through the existing public constructor gets
each revision's real scheduler automatically, with one source file that compiles on both.
A constructor taking PipelineCheckpointScheduler could not compile on dev at all -
that type does not exist there - so it would force two divergent benchmark sources, which
is the thing the A1/B1/B2/A2 comparison most needs to avoid.

It also keeps this PR test-only. Adding a constructor to CheckpointCoordinator makes a
benchmark PR touch production engine code, which I would rather not do without a
committer explicitly asking for it.

2. The interval guard will stop the probe firing after the first sample

This is the one that needs a decision from you, because it shapes the fixture.
tryTriggerPendingCheckpoint returns early, after re-arming itself, when the trigger is
inside the configured interval (CheckpointCoordinator.java:782-799):

long interval = currentTimestamp - latestTriggerTimestamp.get();
if (interval < coordinatorConfig.getCheckpointInterval()) {
    scheduleTriggerPendingCheckpoint(
            checkpointType, coordinatorConfig.getCheckpointInterval() - interval);
    return;
}

latestTriggerTimestamp is only set in createPendingCheckpoint
(CheckpointCoordinator.java:1126), so after the first successful trigger every
subsequent scheduleForBenchmark() inside the interval takes that branch and never
reaches executorService. The armed latch would then wait for the re-armed trigger
instead of the one the sample intended. There is a second guard right after it for
checkpoint.min-pause.

Three ways out, and I would like your call:

• pin checkpointInterval to the 10 ms minimum and checkpointMinPause to -1, which
  bounds sampling at roughly 100/s but keeps the production code path completely intact;
• reset latestTriggerTimestamp between samples, which means reaching into coordinator
  state - the kind of override you asked to avoid;
• treat the guarded path as the measured one deliberately, since the 500 ms re-arm at
  CheckpointCoordinator.java:841 is itself a real production behaviour.

I lean to the first: slow but honest, and JMH SampleTime copes with a low rate.

3. The single-shot probe can be tripped by the previous sample

arm() / getAndSet(null) fires on the next execute(), but the coordinator posts a lot
to that same executor - whenCompleteAsync, thenApplyAsync, thenAcceptAsync in
startTriggerPendingCheckpoint, and as of da0cbbf52 on #12165 the checkpoint-expiry
hop as well. If the previous sample's completion work is still draining, the probe closes
on that instead of on the trigger. awaitCheckpointCompleted() helps, but I would rather
the probe identify the submission than trust quiescence - simplest is for the fixture to
recognise the task it is waiting for, and to fail the sample loudly if it sees an
unexpected submission rather than silently recording the wrong number.

4. Mockito is not available in seatunnel-benchmarks, and these are main sources

The benchmarks live in src/main/java, so anything they need becomes a compile-scope
dependency of a shipped module. Mockito is not currently a dependency there. For the IMap
this is easy to avoid: a java.lang.reflect.Proxy over a ConcurrentHashMap covers
get / set / compute / remove in about twenty JDK-only lines. CheckpointManager
is a class rather than an interface, so it cannot be proxied - that one needs either
Mockito added to this module, or a real manager.

5. There is already a no-mock route to a real coordinator

SeaTunnelCheckpointEnvironmentContext starts in-JVM master and worker instances for
CheckpointingTimeBenchmark, so a real CheckpointManager, real IMap and loopback RPC
are reachable today without mocking anything. The cost is a heavier fixture per fork, and
the probe then needs a different seam, since the executor belongs to CoordinatorService
rather than to us.

What I would like from you

Point 1 removes the constructor change, and points 2 to 4 are fixture details I can
settle myself once you have answered point 2. The one genuine fork is whether you want
the no-cluster fixture you sketched, with a proxied IMap and Mockito added here for
CheckpointManager, or the mini-cluster route in point 5 that needs no fakes at all.

Say which, plus your preference on the interval guard, and I will push it. If you would
rather see it than discuss it, I can push the no-cluster version with the 10 ms interval
and the existing constructor, and we can argue with numbers in front of us.

@Vivek1106-04 Here are my views. If you have any different opinions, you can correct them

1. Thank you for pointing out this gap. I think this can actually be handled directly in your PR by modifying the JMH setup() method to initialize a CheckpointManager. Then the coordinator can obtain the real scheduler through:

this.scheduler = manager.leaseCheckpointScheduler(pipelineId);

2. Please take a look at the comments after my research🔥

3. I agree with your point.

4. Between the two approaches, I prefer building a real in-memory IMap implementation.

5. I do not think we need to start a real SeaTunnel engine.

@Vivek1106-04 @Vivek1106-04 I agree with the first option in point 2: use the minimum valid checkpointInterval of 10 ms and set checkpointMinPause to -1.

To prevent the measured coordinator's automatically re-armed trigger from leaking into the next invocation, we can expose a small coordinator-level test hook:

@VisibleForTesting
void cancelScheduledCheckpointTriggers() {
    // Cancel only this coordinator's outstanding trigger tasks.
}

This method must not shut down or recreate the scheduler:

• On dev, it should cancel the tracked ScheduledFuture instances belonging to this coordinator.
• On [Improve][Zeta] Share checkpoint trigger scheduling across active pipelines #12165, it can delegate to the pipeline-scoped scheduler lease cancellation.
• It must not affect background coordinators.

The invocation setup can then isolate checkpoint completion, timer cleanup, and the 10 ms interval from the measured sample:

@Setup(Level.Invocation)
public void prepareInvocation() throws Exception {
    awaitPreviousCheckpointCompleted();

    // Prevent the measured coordinator's automatic trigger
    // from racing with the next explicit benchmark trigger.
    coordinator.cancelScheduledCheckpointTriggers();

    // Outside the measured region.
    TimeUnit.MILLISECONDS.sleep(11);

    // Arm the task-specific probe only after the fixture is quiescent.
    probe.arm();
}

@Setup(Level.Invocation) is outside the timed benchmark operation, so the 11 ms wait is not included in the reported latency.

The measured region remains:

scheduleTriggerPendingCheckpoint
→ real scheduler
→ shared dispatch, when applicable
→ tryTriggerPendingCheckpoint
→ state checks
→ coordinator lock
→ submission to executorService

The previous checkpoint completion, automatic-trigger cleanup, and interval wait remain outside the measurement. Background coordinators continue running normally to preserve realistic multi-pipeline scheduling load.

Thanks both — this has converged into a genuinely good design, and I want to engage with the technical substance rather than just referee.

On the measurement boundary itself: I'm convinced. @nzw921rx's design (scheduleTriggerPendingCheckpoint() → real scheduler → tryTriggerPendingCheckpoint() → state checks → lock → submission to executorService, with everything from checkpoint-ID allocation onward left outside the sample) answers the question I asked in my last comment better than either of the two options I framed it as. It measures the real decision logic the two scheduling models actually differ in (the pendingCounter.get() > 0 re-arm at CheckpointCoordinator.java:816, the interval/min-pause guards just above it) without pulling in storage or RPC noise. Good outcome.

One thing I need to flag before this goes further: the "add a testing constructor that calls manager.leaseCheckpointScheduler(...)" piece won't compile on dev, and I don't think it's needed. I checked this in the current checkout rather than take either side's word for it:

• CheckpointManager.java has no leaseCheckpointScheduler method and there is no PipelineCheckpointScheduler type anywhere in this tree — that API only exists on [Improve][Zeta] Share checkpoint trigger scheduling across active pipelines #12165's branch.
• CheckpointCoordinator.java has exactly one public constructor (CheckpointCoordinator.java:212), and it already builds its own real scheduler inline (CheckpointCoordinator.java:239-250: Executors.newScheduledThreadPool(2, ...), then MDCTracer.tracing(scheduler)).

So @Vivek1106-04's point 1 is right as written: a benchmark that goes through the existing 11-arg constructor gets each revision's real scheduler for free — dev's pool here, #12165's leased scheduler there — with one source file that compiles against both checkouts. A constructor or setup path that calls leaseCheckpointScheduler directly is #12165-only and would force the two-source-files problem the A1/B1/B2/A2 comparison is specifically trying to avoid. @nzw921rx, given that, I'd drop the testing-constructor / explicit-lease idea from the design and lean on the constructor you already have — it gets you the same real-scheduler behavior with zero changes to CheckpointCoordinator.

On the interval guard (Vivek's point 2): pinning checkpointInterval to the 10 ms floor and checkpointMinPause to -1 is the right call, and I'm glad you both landed there — it keeps the actual production guard branches (CheckpointCoordinator.java:782-799 and the min-pause check just above) live and exercised on every sample instead of routing around them, at the cost of capping throughput to ~100 samples/sec, which SampleTime mode tolerates fine.

On cancelScheduledCheckpointTriggers() (nzw921rx, 13:44:16): the sketch undersells what this needs. It's not "expose a small hook over what's already there" — I checked, and there's nothing there to expose yet. scheduler.schedule(...) is only called twice in this file: once at line 1029 for the checkpoint-timeout future, whose result is captured and stored on the PendingCheckpoint object, and once at line 550 inside scheduleTriggerPendingCheckpoint itself, whose returned ScheduledFuture is discarded on the spot — nothing in CheckpointCoordinator keeps a handle to it. So "cancel only this coordinator's outstanding trigger tasks" requires adding a field to track that future (or a small set of them) to production CheckpointCoordinator, not just annotating an existing method. That's a small change, but it's real production surface, and it's the same tension @Vivek1106-04 already flagged about the constructor idea: I'd rather this PR stay test-only unless a committer is explicitly asked and agrees to the addition. Two ways to keep it test-only instead: (a) since you're already pinning the interval to 10 ms, just let the re-armed trigger fire and land on the next sample's queue rather than trying to suppress it, and have the fixture/probe recognize and discard an out-of-band submission (this is close to what Vivek proposed in his point 3 already — "fail the sample loudly if it sees an unexpected submission" is the same instinct in the other direction); or (b) accept the small field addition, but call it out explicitly in the PR description as a deliberate, minimal addition to CheckpointCoordinator rather than folding it in as if it were free.

Still open, and I don't think it needs me to resolve it: Vivek's point 5 in his 12:56:17 comment — no-cluster fixture (proxied IMap, plus adding Mockito to seatunnel-benchmarks to handle CheckpointManager, which is a concrete class and can't be proxied) versus the mini-cluster route (SeaTunnelCheckpointEnvironmentContext, the same one CheckpointingTimeBenchmark already uses — real CheckpointManager, real IMap, loopback RPC, no mocks, heavier fixture per fork). Mild preference from me toward the mini-cluster route only because it adds zero new test dependencies and zero fakes to reason about, but you two have done the legwork here and either is defensible — whichever you pick, please spell out the cost (Mockito-as-a-new-seatunnel-benchmarks-dependency in one case, fixture-per-fork overhead in the other) in the PR description so it's an explicit, reviewable trade-off rather than an implementation detail.

Nothing here blocks continuing the design conversation, and there's no new commit yet, so there's nothing to re-review on the code side. Once this lands as a diff I'll go through it against this design (measurement boundary, interval pinning, and whichever fixture route you pick) rather than re-litigate the discussion.

@nzw921rx @DanielLeens Before choosing between the two fixtures I built both and ran them, on dev and on #12165, because a few of the open questions could only be settled by running code. Short version: neither fixture works exactly as proposed, but a variant does, it needs no production change, and the same source runs on both revisions.

1. Measure the real periodic trigger instead of injecting one

Injecting a trigger per sample fights the interval guard in tryTriggerPendingCheckpoint. With a long interval the injected trigger is re-armed instead of run; with the 10 ms interval the coordinator re-arms itself every 10 ms and races the next sample, which is what cancelScheduledCheckpointTriggers() was for. That hook needs a new field on dev (the trigger's ScheduledFuture is discarded at CheckpointCoordinator.java:550) and a different implementation on #12165, so it would be production code on both sides.

The alternative needs neither. Let the coordinator run its own periodic trigger at its configured interval, and observe it: a sample is the time pendingCounter goes 0 -> 1, minus the previous trigger time plus the interval. A sample is kept only if the previous checkpoint completed before the next one was due, so every kept sample went through the normal path, including the lock and the executor submission, rather than the 500 ms pending re-arm. The field is read, never written. Nothing in CheckpointCoordinator changes.

2. The no-cluster fixture cannot sustain samples

pendingCounter is incremented synchronously on the trigger path and only comes back down when the checkpoint completes, or through cleanPendingCheckpoint, which sets shutdown = true. With no tasks, the checkpoint never completes, and from the second sample on every trigger takes the pendingCounter > 0 re-arm and never reaches the lock or the executor. So each sample's checkpoint has to really complete.

Faking the tasks entirely is also blocked: the fake network has to return an InvocationFuture, and in Hazelcast 5.1 that class is final with a package-private constructor that dereferences a live Invocation. A real member is needed to produce one.

3. The mini-cluster does not scale to the sweep

With SeaTunnelCheckpointEnvironmentContext and real submitted jobs (-XX:ActiveProcessorCount=4, 4g, as in the benchmark):

• P = 1 works: the real trigger is observed and checkpoints take about 11 ms.
• P = 100 does not: 361 s from submission to all jobs RUNNING, then zero completed checkpoints in a 60 s window, with TaskAcknowledgeOperation got rejected before execution due to not starting within the operation-call-timeout.

P = 500 is out of reach on the benchmark's own CPU budget.

4. What works: a real member, real coordinators, fake tasks

One real SeaTunnel member provides the real checkpoint storage, IMap, monitor service and engine context, so on #12165 the coordinators use the real member-wide scheduler. A real CheckpointManager builds real coordinators through the existing constructor. The only fake is an anonymous subclass overriding the existing protected sendOperationToMemberNode: it answers each barrier through the public acknowledgeTask, and returns a real completed future from a local no-op operation. No Mockito, no new dependency, no change to engine code.

P = 500 as 500 jobs of one pipeline, 10 s interval, starts staggered across one interval, 60 s window:

5. Two things the runs forced on the design

Pipelines must be spread across jobs. The same 500 pipelines as one job averaged 12 s per checkpoint, and no sample was usable. As 500 jobs it was 5.6 ms. I have not proven the cause; my suspicion is that every completion updates one per-job entry in the checkpoint overview IMap. Either way the benchmark should model many jobs, which is also what "500 active pipelines" in #12122 means in practice. The interval is set from the same constraint: at P = 500, a 1 s interval is 500 real checkpoint completions per second and storage cannot keep up, so the sweep uses a fixed 10 s.

Absolute delay numbers must come from CI. On macOS every timed wait fires about 4-5 ms late, and a plain ScheduledThreadPoolExecutor shows the same, so locally the delay distribution sits on the OS timer floor and hides a microsecond-scale difference. The thread counts and the checkpoint completion figures above are valid locally; the delay comparison will come from the Linux runners through the benchmarks.yml entry.

Costs, stated up front

• Two internal fields are read by reflection: pendingCounter on the coordinator, and barrier on CheckpointBarrierTriggerOperation, which has no getter. Read-only, but a rename breaks the benchmark, and I would add a check that fails loudly if either field is missing.
• Fake tasks acknowledge immediately, so the numbers are a best case for the completion side. The scheduling side is fully real.
• The fake encodes two pieces of the task protocol, barrier acknowledgement and READY_START reporting, which would need updating if those change.

Next

Unless you see a problem with this, I will replace the standalone harness in this PR with this fixture: sweep P over 1 / 10 / 100 / 500 as jobs of one pipeline, report scheduler thread count and the scheduling delay distribution, fail the run if kept samples drop below a threshold rather than report a number from the re-arm path, and keep the benchmark in src/main with no new dependencies. checkpointSingleInput stays untouched.

@Vivek1106-04 Thank you for clarifying. I have a question

How will the calculated periodic-trigger delay be reported as a JMH metric?

JMH measures the execution time of the @Benchmark method. If the method only waits for the next periodic trigger, the result may include the remaining portion of the 10-second interval. Returning the calculated delay also does not make that value the SampleTime score.

Please clarify the exact @Setup(Level.Invocation) and @Benchmark boundaries, or how the calculated actualTriggerTime - expectedTriggerTime values will be exported as the benchmark distribution.

How is it compatible with the existing reporting of CV, ERROR, and SCORE?

Thanks for actually building and running both fixtures instead of settling this by argument — the P=500 table is the evidence that should decide it, and "real member, real coordinators, fake tasks only at the network boundary" is the right shape: no production code changes on either revision, and the two costs you flagged up front (reflection field renames, the two-message fake protocol) are both acceptable and worth a loud failure check as you said.

I don't have anything to add on top of @nzw921rx's open question about how the periodic-trigger delay actually surfaces as a JMH @Benchmark measurement — that's the one piece I'd also want nailed down before this lands, since it's the actual metric the sweep is reporting. Once that's answered I'm happy to do a final pass on the pushed fixture.

One small ask for whenever you do push it: could the two costs (the two reflected fields, and the fake's assumption about which two messages it needs to answer) get a short comment at their read sites in the benchmark source, not just in this thread? That's the part most likely to silently break on an unrelated engine-internals change later.

@nzw921rx Good question. As I described it, the @Benchmark method would have waited for the trigger, so the score would have included most of the 10 s interval. The fix is to make the timed region start at the moment the trigger is due. JMH's own timer then measures actual - expected, and the metric is a normal JMH primary metric.

Boundaries

State (Scope.Benchmark): the real member, the P coordinators, and for each coordinator the time its last trigger was observed (lastTriggerNanos).

@Setup(Level.Invocation), not timed:

1. Pick the coordinator whose next trigger is due soonest, where expected = lastTriggerNanos + interval. The starts are spread across the interval, so at P = 500 with a 10 s interval one trigger is due every 20 ms.
2. Check that its previous checkpoint has completed (pendingCounter == 0). If it has not, this trigger would take the 500 ms pending re-arm, so skip this coordinator for this cycle and count the skip.
3. Park until about 1 ms before expected, then spin until System.nanoTime() >= expected. Parking alone would put the OS timer slack into the start of the timed region, so the last millisecond is a spin.
4. Check the trigger has not fired yet. It cannot fire before expected (see the note below), so this only catches a setup that overran. Count it as a skip too.

@Benchmark, timed: spin on that coordinator's pendingCounter until it goes 0 -> 1, record the time in lastTriggerNanos, and return.

So JMH's timed region runs from expected to "trigger observed". That is the scheduling delay plus the trigger's decision logic up to the executor submission, which is the boundary agreed earlier. Nothing is computed afterwards or returned for JMH to interpret: the delay is the invocation time.

@TearDown(Level.Iteration) fails the run if skips go over a fixed share of invocations, so a run where most triggers took the re-arm path produces an error, not a number.

Notes on precision:

• expected is slightly early. pendingCounter is incremented just before the coordinator calls schedule(..., interval) for the next trigger, so the real deadline is a few microseconds after lastTriggerNanos + interval. Each sample therefore also includes the end of the previous trigger body. That code is the same on dev and [Improve][Zeta] Share checkpoint trigger scheduling across active pipelines #12165, so it adds a small, equal offset to both.
• Both timestamps come from the same spin loop on the same thread, and the setup overhead JMH leaves outside the timed region is in the tens of nanoseconds. That is small against a delay measured in microseconds, which is why Level.Invocation is acceptable here even though it is usually not recommended.
• If two coordinators are due within one sample, the second one's trigger happens while the first is being measured, so it has no fresh lastTriggerNanos. It is resynchronised on its next trigger and not sampled for that cycle. With 20 ms between due times this should be rare, and it is counted with the other skips.
• The spinning thread uses one core for roughly 1 ms out of every 20 ms, which is about 5 % of one of the 4 CPUs. It is the same on both revisions.

Reporting: SCORE and ERROR work unchanged, CV does not

With Mode.SampleTime, SCORE is the mean invocation time and ERROR is JMH's confidence interval. save_jmh_result.py reads both from primaryMetric as it does for every other benchmark, and p50 / p99 / max are in scorePercentiles in the same JSON.

CV is the gap. For sample modes JMH writes the raw data as rawDataHistogram (per fork, per iteration, [value, count] pairs), not rawData. jmh_metrics() only flattens rawData, so for any SampleTime benchmark it gets an empty list, sample_standard_deviation returns 0.0, and the report shows CV = 0 %. No existing benchmark in seatunnel-benchmarks uses SampleTime, which is why this has not shown up.

Two ways to handle it:

• (a) Keep SampleTime and teach save_jmh_result.py to read rawDataHistogram. Compute each iteration's mean from its histogram and take the CV across those, so it means the same as the CV of the other modes (variation between iterations, not the spread of individual delays, which is expected to be wide). That is about ten lines plus a case in test_save_jmh_result.py. It changes shared tooling, so I would call it out in the PR description.
• (b) Use Mode.AverageTime with the same boundaries. SCORE, ERROR and CV all work with no tool change, but the percentiles are lost, and for a scheduler the tail is the part that matters.

I prefer (a). If you would rather keep the tooling out of this PR, I will use (b) here and do (a) in a follow-up.

@DanielLeens Yes. The two reflected fields (pendingCounter and CheckpointBarrierTriggerOperation.barrier) and the two messages the fake task answers (barrier acknowledgement and READY_START) will each get a comment at the place they are read. The comment will say what engine behaviour it depends on and that the fixture fails at setup if the field is missing.

Thanks @Vivek1106-04, this answers the piece I was waiting on. The boundary you describe -- @Setup(Level.Invocation) parks/spins up to expected untimed, and the timed @Benchmark region only starts there and ends when pendingCounter flips -- means the measured value really is scheduling delay plus the trigger's own decision logic up to executor submission, not delay-plus-wait-for-the-next-trigger. That's the boundary this benchmark needs to be trustworthy, so I'm satisfied on that front.

On the SampleTime-vs-AverageTime call: I'd lean toward (a) too (teach save_jmh_result.py to read rawDataHistogram so CV keeps meaning "variation between iterations") since losing the tail percentiles on a scheduler benchmark seems like the wrong trade-off, but I don't think it's blocking either way -- (b) as a stopgap with (a) as a fast follow-up is fine if you'd rather keep the tooling change out of this PR's description.

Appreciate you committing to comments at both read sites (the two reflected fields, and the fake's two-message assumption) explaining what engine behavior each depends on and that the fixture fails loudly at setup if it's missing -- that was the main thing I wanted nailed down given how easy this would be to silently break on an unrelated engine-internals change. Once that's pushed alongside the CV fix (or the AverageTime fallback), I'll do the final pass on the fixture.

@Vivek1106-04 Two small follow-ups on the fixture design, nothing blocking:

1. Your note that expected is slightly early (because pendingCounter is incremented just before the schedule(..., interval) call, so each sample also includes the end of the previous trigger body, equally on both revisions) is fine with me, but please put it as a comment next to the spin loop so nobody later "fixes" it.

2. For the iteration-level teardown skip guard: what fixed share of invocations do you have in mind, and could the skip count (split into "previous checkpoint still pending" vs "setup overran") be printed per iteration even when the run passes? That keeps a borderline run visible instead of silently accepted.

Otherwise nothing to add beyond my last comment; I'll do the final pass once the fixture is pushed.

The fixture is pushed (afd311519, be57ed0ae) and the PR description is rewritten to match it.

@SEZ9

1. The note is at the timed spin loop in CheckpointSchedulingFixture#awaitTrigger: why expected is a few microseconds early, that the offset is identical on every scheduler compared, and not to move the start later.

2. The limit is 10% of due triggers per iteration. Skips are printed every iteration, pass or fail, split by cause. Running it made me split one level further than you asked, because the causes turned out to mean different things:

   # measured 100 of 101 due triggers; skipped 0 with a checkpoint already pending before the due time,
   #   0 that came due while another trigger was being measured, 1 where parking overran the due time,
   #   0 that ran before their estimated due time
   

   "Setup overran" is now three counters: came due during another sample, parking overshot, and ran before the estimate (which would mean the phase estimate is wrong).

@nzw921rx The boundaries are the ones in my last comment: the untimed @Setup(Level.Invocation) ends at the due time, and the timed @Benchmark ends when pendingCounter goes from 0 to 1. For CV I went with (a): save_jmh_result.py now reads rawDataHistogram. Its new test failed before the fix. On the local sweep it now reports CV from 3.4% to 17%, plus 81% for dev at 500 pipelines, where the macOS distribution splits between two timer behaviours. Before the fix it reported 0% for every point.

@DanielLeens Both reflected fields and both faked task messages have a comment at their read site. A missing field fails at setup with the class and field named, which shouldNameTheEngineFieldWhenItNoLongerExists covers.

Three things the runs changed, compared with what I described earlier:

• Only a fixed subset of coordinators is measured. Measuring every coordinator, a run on [Improve][Zeta] Share checkpoint trigger scheduling across active pipelines #12165 at 10 pipelines skipped 62 of 250 due triggers, all as "came due while another trigger was being measured". Phases drift as delays vary, so due times bunch up. Dropping the second of two near-simultaneous triggers biases the result towards short delays exactly under contention, which is the case this benchmark exists for. So probes are now every k-th coordinator, started at least 100 ms apart, and the rest are load. Probe triggers still collide with the load, and which coordinators are probes is fixed by index, so the samples carry no selection. After the change, that same run skipped 2 of 305.
• IMap persistence is off. I first reused the storage benchmarks' member. Its write-through MapStore puts a file write on a partition thread for every engine IMap update, and a thread dump showed checkpoint id allocation blocked there for over 600 ms. The fixture keeps that member's isolated checkpoint storage but disables the MapStore.
• The interval is pipelineNum * 20 ms, floored at 200 ms, not a flat 10 s. Every point of the sweep then sees the same trigger rate and storage load. 500 pipelines is still 10 s, and the floor keeps 1 pipeline from sending triggers down the 500 ms pending re-arm path.

Local sweep, dev against #12165 at c773cc091, the same benchmark code on both, full table in the description: all four pipeline counts pass on both, with at most 2 skips per point, and scheduler threads at 500 pipelines are 995 against 10. I have left the delay numbers out on purpose: on macOS every timed wait fires about 5 ms late, so the local distribution sits on that floor. The delay comparison needs the Linux runners.

@nzw921rx @DanielLeens @SEZ9 ready for the final pass.

@DanielLeens Thank you for the approval and for checking the source and CI yourself instead of relying on my summary. I have pushed all three of your low-severity items in 550c22568:

1. The printed thread count was a lower bound. Trial setup now resyncs every coordinator, not only the probes, before it prints. Every coordinator has therefore triggered and armed its scheduler by the time the threads are counted, so the line gives the full count. Iteration resyncs still cover only the probes. The thread counts in the PR description's local table were taken before this change, and the description now says they are lower bounds.
2. The teardown test scanned every thread in the JVM. The fixture now records the checkpoint-* threads that already exist when its setup starts, and counts only threads that appear after that. The teardown test starts a checkpoint-foreign thread before the fixture and checks that the test passes while that thread is still alive. A scheduler thread left over from another test class therefore can no longer fail it.
3. Two constants had the same name. The benchmark's constant is renamed to MIN_MEASURABLE_INTERVAL_MILLIS, and its Javadoc now says how it differs from the fixture's MIN_CHECKPOINT_INTERVAL_MILLIS, which is the lowest interval SeaTunnel accepts.

The test count in the description is corrected as well.

@nzw921rx When you have time, could you look at the current head again? Your concern that the benchmark did not exercise the real engine path shaped the rewrite. The benchmark now runs real CheckpointCoordinators through tryTriggerPendingCheckpoint and fakes only the network boundary, and the engine has no production code changes. If anything is still missing, please tell me and I will address it.

Thanks to @nzw921rx, @SEZ9 and @DanielLeens for reviewing this so carefully over several rounds. The benchmark is much stronger because of your feedback than the first version I opened.

Thanks @Vivek1106-04 — 550c22568 closes out the three low-severity items from my last pass:

1. Thread count no longer a lower bound. Resyncing every coordinator, not just the probes, in trial setup before printing means the count reflects all armed schedulers at pipelineNum. Keeping the iteration resync probe-only is the right call since it stays outside the timed region. Marking the PR description's local table as lower bounds is the honest fix for numbers taken before this change.
2. Teardown test scoped to the fixture's own threads. Snapshotting pre-existing checkpoint-* threads and asserting against a live checkpoint-foreign thread makes this a real regression test rather than a rename; a leftover scheduler from another test class can't fail it anymore.
3. Constant naming. MIN_MEASURABLE_INTERVAL_MILLIS with a Javadoc contrasting it against the fixture's MIN_CHECKPOINT_INTERVAL_MILLIS removes the ambiguity.

Nothing further from me on these.

I see @nzw921rx has left a couple of new comments on the current head. I'll leave those between the two of you — please respond on the thread (or push a follow-up if you agree) so we have a clear final state before this is merged.

@nzw921rx Agreed, moved BenchmarkReflection, CheckpointSchedulingFixture and FakeTaskCheckpointManager into org.apache.seatunnel.benchmark.checkpoint. This mirrors CheckpointStorageBenchmark with storage/checkpoint/CheckpointStorageBenchmarkFixture. CheckpointSchedulingBenchmark stays in the root package with the other @Benchmark classes, so the benchmarks.yml selector is unchanged. The fixture tests moved with it into checkpoint/CheckpointSchedulingFixtureTest, and CheckpointSchedulingBenchmarkTest keeps the annotation check. 7/7 pass locally.

The fork count is now 3 as well; details in the inline thread.

Hi @Vivek1106-04, following up after the merge, I ran the benchmark on Linux: https://github.com/apache/seatunnel/actions/runs/36820452487.

The run already uses three 10-second warmup iterations, but some measurements still trend downward. For example, with one pipeline on JDK 8, the first fork’s measurement scores are 161.505, 163.602, 143.061, 142.223 and 131.410 μs/op.

Could you please investigate whether warmup is sufficient and what causes this trend? Please also run the complete benchmark sweep locally, including all parameter combinations and forks, and share the commands, environment details and full per-iteration results.

As a follow-up, we should also consider a fixed-checkpoint-interval scenario and longer measurement runs to evaluate increasing scheduling load and obtain enough samples for tail percentiles. This would help strengthen the baseline for evaluating the shared scheduler.

@nzw921rx Thanks for running it, and sorry for the slow reply. I'm on it now and will start the local runs within the next hour.

A first look at the per-iteration data from run 36820452487: only the pipelineNum = 1 point is still trending down during measurement; 10, 100 and 500 flatten by about the first measured iteration. My working suspicion is the 200 ms interval floor. At one pipeline the member sees 5 triggers/s, against 50/s at 10, 100 and 500, so the trigger path gets about 10x less warmup and half the samples per iteration. That would also explain the CV shrinking as the pipeline count grows. This is not verified yet.

Plan:

1. Confirm or rule this out on Linux (JMH -prof comp, longer warmup at pipelineNum = 1). macOS timed waits fire about 5 ms late, so local runs will be in a Linux container.
2. Fix the warmup in a follow-up PR against dev.
3. Run the full sweep (all parameters, 3 forks, JDK 8 and 11) and post the commands, environment details and full per-iteration results here.

I'll take the fixed-interval scenario and longer runs for tail percentiles as a separate follow-up.

@nzw921rx Thanks for running it on Linux. You were right: the warmup is not sufficient. Below are the cause, the evidence, a proposed fix, and the full sweep you asked for. I have not opened a PR for the fix yet; there is a question for you at the end.

What the score contains

The measured window runs from the time a trigger is due until its body has incremented pendingCounter. To split it, I ran a diagnostic build (fork branch bench-diag-12418, never proposed) that timestamps every trigger at its due time, at the start of its body, and at the increment. JDK 17, dev, 30 x 10 s iterations, p50 in microseconds:

Both parts fall for the first two minutes. "Due -> body start" is not only the OS wake-up: it also covers ScheduledThreadPoolExecutor's queue and the MDCContext wrapper around every scheduled task, which compile late for the same reason.

Why it trends: the JIT compiles by invocation count

The trigger path runs 5 to 50 times a second, so it is compiled by how often it ran, not by how long the fork has run. From -XX:+PrintCompilation on the same runners, JDK 11 and 17 alike:

• One pipeline: tryTriggerPendingCheckpoint and createPendingCheckpoint are interpreted until about 80 s of uptime (C1 tier 3) and never reach C2 within a 300 s trial. The ScheduledThreadPoolExecutor path reaches C2 at about 100 s.
• Ten pipelines or more: the same methods reach C1 at 7 to 11 s and C2 at 110 to 130 s.
• CI warms up for 30 s and then measures from roughly 30 to 100 s of uptime, so it records the curve on its way down.

A causal check: with -XX:TieredStopAtLevel=1 (C1 only), the one-pipeline body drops from 68 to 23 us in a single step at 54 s, exactly when those methods compile, and stays flat afterwards.

The one-pipeline point is hit hardest because of the 200 ms interval floor. The class Javadoc says every point sees the same trigger rate, but at one pipeline the member triggers 5 checkpoints a second where every other point triggers 50. Its code warms up ten times slower. Even once settled, its body runs at a lower tier (about 21 us against 6.6 us at ten pipelines), and its trigger wakes up on an otherwise idle 4-vCPU VM, which costs about 25 us more (about 95 against 70 us). That is why one pipeline scores higher than ten.

Why CV shrinks as pipelines grow: three separate effects

Decomposing your run's CV (computed, like the report, from the 15 iteration means):

1. Drift, at one pipeline: every fork falls 2 to 7% per iteration. This is the warmup problem.
2. Sampling noise from a heavy tail, e.g. JDK 8 at 10 and 100 pipelines: the CV there is fully explained by sampling (single samples spread 24 to 35% around the mean, and one 2 ms sample moves a 100-sample mean by about 20%). No drift.
3. Fork-to-fork offsets of 1 to 7%: each JVM compiles this code a little differently.

So CV is not monotonic in the pipeline count; the one-pipeline point is worst because it has the drift, 50 samples per iteration instead of 100, and the noisiest wake-ups.

Proposed fix

On branch fix-12418-benchmark-warmup, commit 1914eabf6, benchmark code and tests only:

• Forks run with -XX:CompileThresholdScaling=0.05. (0.01 compiles so much at once in the first iteration that more than 10% of triggers come due unmeasured, and the skip limit fails the run.)
• Each trial lets its own coordinators trigger 1,000 checkpoints, for at least 60 s, before JMH's warmup. Counting triggers rather than seconds gives the one-pipeline point the same compiled code as the others: about 200 s there, 60 s elsewhere.
• The Javadoc documents the one-pipeline point as an idle-member point, to compare across revisions rather than against the other pipeline counts.

A first attempt, warming up on throwaway coordinators and cancelling them, did not work: cancelling them made the JIT deoptimise tryTriggerPendingCheckpoint right before the measurement started.

Full sweep, dev against the fix

Exact command of tools/benchmarks/run_benchmarks.sh, all pipeline counts, 3 forks, 3 warmup and 5 measurement iterations of 10 s. dev and the fix ran on the same runner, in both orders, on JDK 11 and 17: 12 forks per side and point. Run: https://github.com/Vivek1106-04/seatunnel/actions/runs/38295747109

No run of the fix hit the skip limit; one earlier dev run did, at 100 pipelines on JDK 11 (19 triggers came due while another was being measured).

What remains, and is not a warmup effect: at one pipeline, iteration-to-iteration variation stays around 9 to 10% on both sides, because that trigger's wake-up latency on an idle VM wanders by tens of microseconds over minutes. Forks still settle 1 to 5% apart. The per-iteration median is much steadier than the mean (CV 2 to 4% at ten pipelines or more with the fix), because the mean carries the occasional millisecond sample; reporting it would be a separate change to save_jmh_result.py.

Commands and environment

# Build, from each checkout
./mvnw -B -q -Pbenchmark -pl seatunnel-benchmarks -am -DskipTests -Dskip.spotless=true package
# Run, same as tools/benchmarks/run_benchmarks.sh
java -jar seatunnel-benchmarks/target/benchmarks.jar CheckpointSchedulingBenchmark -f 3 -wi 3 -i 5 -foe true -rf json -rff <side>.json

• GitHub-hosted ubuntu-24.04, Linux 6.17.0-1022-azure x86_64, AMD EPYC 7763, 4 vCPU
• Temurin 11.0.32.1 and 17.0.20.1. JDK 8 is no longer in the comparison because dev now builds for Java 11, and benchmarks.yml runs 11 and 17.
• dev = b3898b62d; fix = 738f372e4 on the throwaway verification branch, identical benchmark code to 1914eabf6
• The fork JVM options are the benchmark's own @Fork settings.

These ran on GitHub's runners in my fork rather than on my machine: on macOS every timed wait fires about 5 ms late, and Docker Desktop's Linux VM inherits that (one pipeline measured about 2.5 ms there against about 150 us on the runners), so a local run cannot show a microsecond-scale trend.

Before I open a PR

Two choices in the proposed fix change how this benchmark runs, so I would like your view first:

1. -XX:CompileThresholdScaling=0.05 changes the JIT's thresholds inside the benchmark. The alternative is to keep JVM defaults and warm up for about 150 s per trial instead. In the 300 s trajectories above that is where ten pipelines stops drifting, though I have not run it as a full sweep, and it leaves the one-pipeline body at a lower compilation tier than the other points. Which would you prefer?
2. Cost: the trigger-count warmup adds about 19 minutes per JDK to the CheckpointSchedulingBenchmark sweep in benchmarks.yml (about 200 s per fork at one pipeline, 60 s per fork elsewhere).

The one-pipeline point is kept and documented as an idle-member point. If you would rather drop it and start the sweep at ten pipelines, that removes the noisiest point and about half of the added time.

If this direction works for you, I will open it as a PR, or file an issue first if you prefer that. The fixed-interval scenario and longer measurement runs for tail percentiles stay a separate follow-up, which I will open as an issue once this is settled.