This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a commit to branch dev
in repository https://gitbox.apache.org/repos/asf/seatunnel.git
The following commit(s) were added to refs/heads/dev by this push:
new 99ea16b9d6 [Test][Zeta] Warm up the checkpoint scheduling benchmark by
trigger count (#12896)
99ea16b9d6 is described below
commit 99ea16b9d6bdce452d4c4955581ccc7236a81bbd
Author: Gangavarapu Vivek <[email protected]>
AuthorDate: Thu Oct 22 05:07:34 2026 +0000
[Test][Zeta] Warm up the checkpoint scheduling benchmark by trigger count
(#12896)
---
.../benchmark/CheckpointSchedulingBenchmark.java | 44 +++++++++++++--
.../checkpoint/CheckpointSchedulingFixture.java | 64 +++++++++++++++++++++-
.../checkpoint/FakeTaskCheckpointManager.java | 8 ++-
.../CheckpointSchedulingBenchmarkTest.java | 34 ++++++++++++
.../CheckpointSchedulingFixtureTest.java | 42 ++++++++++++++
5 files changed, 182 insertions(+), 10 deletions(-)
diff --git
a/seatunnel-benchmarks/src/main/java/org/apache/seatunnel/benchmark/CheckpointSchedulingBenchmark.java
b/seatunnel-benchmarks/src/main/java/org/apache/seatunnel/benchmark/CheckpointSchedulingBenchmark.java
index 8e2b855310..7e11a3a18a 100644
---
a/seatunnel-benchmarks/src/main/java/org/apache/seatunnel/benchmark/CheckpointSchedulingBenchmark.java
+++
b/seatunnel-benchmarks/src/main/java/org/apache/seatunnel/benchmark/CheckpointSchedulingBenchmark.java
@@ -59,10 +59,25 @@ import java.util.concurrent.TimeUnit;
*
* <p>The checkpoint interval is {@code pipelineNum * triggerSpacingMillis},
so every point of the
* sweep sees the same rate of due triggers and the same checkpoint load on
storage, and only the
- * number of coordinators changes. It is floored at {@link
#MIN_MEASURABLE_INTERVAL_MILLIS}, so the
- * smallest pipeline counts sample less often. Each job has one pipeline: many
jobs is what "many
- * active pipelines" means on a member, and it keeps per-job checkpoint state
from becoming the
- * bottleneck.
+ * number of coordinators changes. It is floored at {@link
#MIN_MEASURABLE_INTERVAL_MILLIS}, which
+ * is why the sweep starts at ten pipelines: below that the floor binds, the
member triggers fewer
+ * checkpoints a second than at the other points, and the code on the trigger
path warms up that
+ * much slower. Each job has one pipeline: many jobs is what "many active
pipelines" means on a
+ * member, and it keeps per-job checkpoint state from becoming the bottleneck.
+ *
+ * <p>The trigger path runs only 50 times a second, and the JIT compiles by
invocation count, so
+ * with default thresholds that path is still being compiled minutes into a
fork, reaching C2 only
+ * after about two minutes. Every iteration in that window scores lower than
the one before. Two
+ * things keep that out of the measurement. The forks run with {@code
+ * -XX:CompileThresholdScaling=0.05}, so the path compiles after a twentieth
of the invocations;
+ * lower values compile so much at once that triggers come due unmeasured and
the skip limit fails
+ * the run. And each trial first lets its coordinators trigger {@link
#TRIGGER_PATH_WARMUP_TRIGGERS}
+ * checkpoints, a warmup counted in triggers rather than seconds, and runs for
at least {@link
+ * #MIN_WARMUP_SECONDS}. At 50 triggers a second the time is what binds; the
count keeps the warmup
+ * long enough if a lower trigger rate is added.
+ *
+ * <p>Scores taken before this warmup are not comparable with scores taken
after it, so a PR
+ * comparison needs the candidate branch to contain it as well as the baseline.
*/
@BenchmarkMode(Mode.SampleTime)
@OutputTimeUnit(TimeUnit.MICROSECONDS)
@@ -78,6 +93,7 @@ import java.util.concurrent.TimeUnit;
"-XX:+AlwaysPreTouch",
"-XX:+DisableExplicitGC",
"-XX:ActiveProcessorCount=4",
+ "-XX:CompileThresholdScaling=0.05",
"-Djava.net.preferIPv4Stack=true"
})
public class CheckpointSchedulingBenchmark extends BenchmarkBase {
@@ -90,6 +106,20 @@ public class CheckpointSchedulingBenchmark extends
BenchmarkBase {
*/
static final long MIN_MEASURABLE_INTERVAL_MILLIS = 200L;
+ /**
+ * Checkpoints each trial triggers before JMH's own warmup starts. With
the fork's lowered
+ * compile thresholds, the trigger path, the scheduler queue and the MDC
wrapper around each
+ * scheduled task stopped changing tier after about 1,000 triggers on
GitHub's Linux runners.
+ */
+ static final long TRIGGER_PATH_WARMUP_TRIGGERS = 1_000L;
+
+ /**
+ * Least time each trial runs its coordinators before JMH's own warmup. At
ten pipelines or more
+ * the trigger count above is reached in 20 seconds, while the rest of the
member is still
+ * compiling, and the first measured iteration then caught millisecond
outliers.
+ */
+ static final long MIN_WARMUP_SECONDS = 60L;
+
public static void main(String[] args) throws RunnerException {
Options options =
new OptionsBuilder()
@@ -111,7 +141,7 @@ public class CheckpointSchedulingBenchmark extends
BenchmarkBase {
@State(Scope.Thread)
public static class CoordinatorsState {
- @Param({"1", "10", "100", "500"})
+ @Param({"10", "100", "500"})
private int pipelineNum;
@Param({"20"})
@@ -126,7 +156,9 @@ public class CheckpointSchedulingBenchmark extends
BenchmarkBase {
pipelineNum,
Math.max(
MIN_MEASURABLE_INTERVAL_MILLIS,
- pipelineNum * triggerSpacingMillis));
+ pipelineNum * triggerSpacingMillis),
+ TRIGGER_PATH_WARMUP_TRIGGERS,
+ TimeUnit.SECONDS.toMillis(MIN_WARMUP_SECONDS));
fixture.setUp();
System.out.printf(
"# checkpoint scheduler threads for %d pipelines: %d;
measuring %d of them%n",
diff --git
a/seatunnel-benchmarks/src/main/java/org/apache/seatunnel/benchmark/checkpoint/CheckpointSchedulingFixture.java
b/seatunnel-benchmarks/src/main/java/org/apache/seatunnel/benchmark/checkpoint/CheckpointSchedulingFixture.java
index 2dc7cccb54..0ebe7b0153 100644
---
a/seatunnel-benchmarks/src/main/java/org/apache/seatunnel/benchmark/checkpoint/CheckpointSchedulingFixture.java
+++
b/seatunnel-benchmarks/src/main/java/org/apache/seatunnel/benchmark/checkpoint/CheckpointSchedulingFixture.java
@@ -41,6 +41,7 @@ import java.util.concurrent.SynchronousQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.atomic.AtomicReference;
import java.util.concurrent.locks.LockSupport;
import java.util.stream.Collectors;
@@ -127,9 +128,14 @@ public final class CheckpointSchedulingFixture {
private static final long EXECUTOR_KEEP_ALIVE_SECONDS = 60L;
private static final long SHUTDOWN_TIMEOUT_SECONDS = 30L;
+ private static final long WARMUP_POLL_NANOS =
TimeUnit.MILLISECONDS.toNanos(10);
+
private final int pipelineNum;
private final long intervalMillis;
private final long intervalNanos;
+ private final long warmupTriggers;
+ private final long minWarmupNanos;
+ private final AtomicLong triggeredCheckpoints = new AtomicLong();
private int probeStride;
private Set<Thread> preexistingSchedulerThreads = Collections.emptySet();
@@ -154,15 +160,29 @@ public final class CheckpointSchedulingFixture {
* @param intervalMillis checkpoint interval of every job
*/
public CheckpointSchedulingFixture(int pipelineNum, long intervalMillis) {
+ this(pipelineNum, intervalMillis, 0L, 0L);
+ }
+
+ /**
+ * @param pipelineNum number of jobs, each with one single-task pipeline
+ * @param intervalMillis checkpoint interval of every job
+ * @param warmupTriggers checkpoints the coordinators trigger before setup
returns, 0 for none
+ * @param minWarmupMillis least time the coordinators run before setup
returns, 0 for none
+ */
+ public CheckpointSchedulingFixture(
+ int pipelineNum, long intervalMillis, long warmupTriggers, long
minWarmupMillis) {
this.pipelineNum = pipelineNum;
this.intervalMillis = intervalMillis;
this.intervalNanos = TimeUnit.MILLISECONDS.toNanos(intervalMillis);
+ this.warmupTriggers = warmupTriggers;
+ this.minWarmupNanos = TimeUnit.MILLISECONDS.toNanos(minWarmupMillis);
}
/**
* Starts the member, creates the coordinators and starts their pipelines
staggered across one
- * interval, so due triggers arrive as a steady stream rather than a
burst. Returns once every
- * coordinator's trigger phase is known.
+ * interval, so due triggers arrive as a steady stream rather than a
burst. Then lets them
+ * trigger {@code warmupTriggers} checkpoints for at least {@code
minWarmupMillis}, and returns
+ * once every coordinator's trigger phase is known.
*/
public void setUp() throws Exception {
validateParameters();
@@ -175,11 +195,44 @@ public final class CheckpointSchedulingFixture {
coordinatorExecutor = createCoordinatorExecutor(engineConfig);
createManagers(server, engineConfig.getCheckpointConfig());
startStaggered();
+ awaitWarmupTriggers();
// Every coordinator, not only the probes: once each has triggered,
each has armed its
// scheduler, so countSchedulerThreads() sees the full thread cost of
pipelineNum pipelines.
resync(1);
}
+ /**
+ * Waits until the coordinators have triggered {@code warmupTriggers}
checkpoints. The JIT
+ * compiles the trigger path by how often it runs, not by how long the
fork has been running, so
+ * a warmup counted in triggers gives every pipeline count the same
compiled code to measure,
+ * where a warmup counted in seconds gives the lowest trigger rates the
least. Nothing is
+ * measured meanwhile, and the coordinators are the measured ones, so no
code path is exercised
+ * that the measurement does not also take. The minimum time covers the
rest of the member,
+ * whose compilation goes on for a while after the trigger path settles at
the higher rates.
+ */
+ private void awaitWarmupTriggers() {
+ long start = System.nanoTime();
+ long target = triggeredCheckpoints.get() + warmupTriggers;
+ long earliestEnd = start + minWarmupNanos;
+ // Twice the time the triggers take at the configured rate, plus the
pending re-arms.
+ long deadline =
+ Math.max(earliestEnd, start + 2 * warmupTriggers *
intervalNanos / pipelineNum)
+ + RESYNC_INTERVALS * intervalNanos
+ + PENDING_REARM_NANOS;
+ while (triggeredCheckpoints.get() < target || System.nanoTime() <
earliestEnd) {
+ checkFailure();
+ if (System.nanoTime() > deadline) {
+ throw new IllegalStateException(
+ String.format(
+ "%d pipelines triggered %d of %d warm-up
checkpoints in time",
+ pipelineNum,
+ warmupTriggers - (target -
triggeredCheckpoints.get()),
+ warmupTriggers));
+ }
+ parkUntil(System.nanoTime() + WARMUP_POLL_NANOS);
+ }
+ }
+
/** Resynchronises the coordinators that were taken out of the rotation,
then resets counts. */
public void beginIteration() {
resync(probeStride);
@@ -345,6 +398,10 @@ public final class CheckpointSchedulingFixture {
return sampled;
}
+ long getTriggeredCheckpoints() {
+ return triggeredCheckpoints.get();
+ }
+
/**
* Counts this fixture's live checkpoint scheduler threads. This is the
cost a shared scheduler
* exists to remove, so it is reported rather than derived.
@@ -393,7 +450,8 @@ public final class CheckpointSchedulingFixture {
runningJobState,
server.getEngineContext(),
server.getCheckpointMonitorService(),
- failure);
+ failure,
+ triggeredCheckpoints);
managers.add(manager);
pendingCounters[i] =
readPendingCounter(manager.getCheckpointCoordinator(PIPELINE_ID));
lastTriggerNanos[i] = NOT_SYNCED;
diff --git
a/seatunnel-benchmarks/src/main/java/org/apache/seatunnel/benchmark/checkpoint/FakeTaskCheckpointManager.java
b/seatunnel-benchmarks/src/main/java/org/apache/seatunnel/benchmark/checkpoint/FakeTaskCheckpointManager.java
index d57bed4a45..d3942b2b43 100644
---
a/seatunnel-benchmarks/src/main/java/org/apache/seatunnel/benchmark/checkpoint/FakeTaskCheckpointManager.java
+++
b/seatunnel-benchmarks/src/main/java/org/apache/seatunnel/benchmark/checkpoint/FakeTaskCheckpointManager.java
@@ -40,6 +40,7 @@ import
com.hazelcast.spi.impl.operationservice.impl.InvocationFuture;
import java.lang.reflect.Field;
import java.util.Collections;
import java.util.concurrent.ExecutorService;
+import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.atomic.AtomicReference;
/**
@@ -76,7 +77,9 @@ final class FakeTaskCheckpointManager extends
CheckpointManager {
private final NodeEngine nodeEngine;
private final CheckpointPlan plan;
private final AtomicReference<Throwable> failure;
+ private final AtomicLong triggeredCheckpoints;
+ /** @param triggeredCheckpoints incremented for every checkpoint barrier a
trigger sends */
FakeTaskCheckpointManager(
long jobId,
NodeEngine nodeEngine,
@@ -87,7 +90,8 @@ final class FakeTaskCheckpointManager extends
CheckpointManager {
IMap<Object, Object> runningJobStateIMap,
SeaTunnelEngineContext engineContext,
CheckpointMonitorService checkpointMonitorService,
- AtomicReference<Throwable> failure) {
+ AtomicReference<Throwable> failure,
+ AtomicLong triggeredCheckpoints) {
super(
jobId,
false,
@@ -106,6 +110,7 @@ final class FakeTaskCheckpointManager extends
CheckpointManager {
this.nodeEngine = nodeEngine;
this.plan = plan;
this.failure = failure;
+ this.triggeredCheckpoints = triggeredCheckpoints;
}
/**
@@ -123,6 +128,7 @@ final class FakeTaskCheckpointManager extends
CheckpointManager {
@Override
protected InvocationFuture<?> sendOperationToMemberNode(TaskOperation
operation) {
if (operation instanceof CheckpointBarrierTriggerOperation) {
+ triggeredCheckpoints.incrementAndGet();
acknowledgeBarrier(readBarrier((CheckpointBarrierTriggerOperation)
operation));
}
return NodeEngineUtil.sendOperationToMemberNode(
diff --git
a/seatunnel-benchmarks/src/test/java/org/apache/seatunnel/benchmark/CheckpointSchedulingBenchmarkTest.java
b/seatunnel-benchmarks/src/test/java/org/apache/seatunnel/benchmark/CheckpointSchedulingBenchmarkTest.java
index a9217bde95..01d80dfa85 100644
---
a/seatunnel-benchmarks/src/test/java/org/apache/seatunnel/benchmark/CheckpointSchedulingBenchmarkTest.java
+++
b/seatunnel-benchmarks/src/test/java/org/apache/seatunnel/benchmark/CheckpointSchedulingBenchmarkTest.java
@@ -19,13 +19,17 @@ package org.apache.seatunnel.benchmark;
import org.junit.jupiter.api.Test;
import org.openjdk.jmh.annotations.BenchmarkMode;
+import org.openjdk.jmh.annotations.Fork;
import org.openjdk.jmh.annotations.Mode;
import org.openjdk.jmh.annotations.OutputTimeUnit;
+import org.openjdk.jmh.annotations.Param;
import org.openjdk.jmh.annotations.Threads;
+import java.util.Arrays;
import java.util.concurrent.TimeUnit;
import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
class CheckpointSchedulingBenchmarkTest {
@@ -39,4 +43,34 @@ class CheckpointSchedulingBenchmarkTest {
CheckpointSchedulingBenchmark.class.getAnnotation(OutputTimeUnit.class).value());
assertEquals(1,
CheckpointSchedulingBenchmark.class.getAnnotation(Threads.class).value());
}
+
+ @Test
+ void shouldLowerJitThresholdsSoTheTriggerPathCompilesWithinWarmup() {
+ assertTrue(
+ Arrays.asList(
+ CheckpointSchedulingBenchmark.class
+ .getAnnotation(Fork.class)
+ .jvmArgsAppend())
+ .contains("-XX:CompileThresholdScaling=0.05"));
+ }
+
+ @Test
+ void shouldOnlySweepPipelineCountsTheIntervalFloorDoesNotSlowDown() throws
Exception {
+ String[] pipelineNums = paramValues("pipelineNum");
+ long triggerSpacingMillis =
Long.parseLong(paramValues("triggerSpacingMillis")[0]);
+
+ for (String pipelineNum : pipelineNums) {
+ assertTrue(
+ Integer.parseInt(pipelineNum) * triggerSpacingMillis
+ >=
CheckpointSchedulingBenchmark.MIN_MEASURABLE_INTERVAL_MILLIS,
+ "pipelineNum=" + pipelineNum + " would trigger below the
sweep's rate");
+ }
+ }
+
+ private static String[] paramValues(String field) throws
NoSuchFieldException {
+ return CheckpointSchedulingBenchmark.CoordinatorsState.class
+ .getDeclaredField(field)
+ .getAnnotation(Param.class)
+ .value();
+ }
}
diff --git
a/seatunnel-benchmarks/src/test/java/org/apache/seatunnel/benchmark/checkpoint/CheckpointSchedulingFixtureTest.java
b/seatunnel-benchmarks/src/test/java/org/apache/seatunnel/benchmark/checkpoint/CheckpointSchedulingFixtureTest.java
index c28b4786b2..36bbd517af 100644
---
a/seatunnel-benchmarks/src/test/java/org/apache/seatunnel/benchmark/checkpoint/CheckpointSchedulingFixtureTest.java
+++
b/seatunnel-benchmarks/src/test/java/org/apache/seatunnel/benchmark/checkpoint/CheckpointSchedulingFixtureTest.java
@@ -34,6 +34,8 @@ class CheckpointSchedulingFixtureTest {
private static final long CHECKPOINT_INTERVAL_MILLIS = 200L;
private static final int SAMPLE_COUNT = 20;
private static final long THREAD_STOP_TIMEOUT_NANOS =
TimeUnit.SECONDS.toNanos(30);
+ private static final long WARMUP_TRIGGERS = 40L;
+ private static final long MIN_WARMUP_MILLIS = 5_000L;
@Test
void shouldMeasureDueTriggersOfRealCoordinators() throws Exception {
@@ -61,6 +63,46 @@ class CheckpointSchedulingFixtureTest {
}
}
+ @Test
+ void shouldTriggerTheWarmupCheckpointsBeforeMeasuringAnything() throws
Exception {
+ CheckpointSchedulingFixture fixture =
+ new CheckpointSchedulingFixture(
+ PIPELINE_NUM, CHECKPOINT_INTERVAL_MILLIS,
WARMUP_TRIGGERS, 0L);
+ fixture.setUp();
+ try {
+ assertTrue(
+ fixture.getTriggeredCheckpoints() >= WARMUP_TRIGGERS,
+ "setup should return only after the warm-up checkpoints
were triggered");
+ // Phases are taken after the warm-up, so the first iteration
skips nothing for them.
+ fixture.beginIteration();
+ for (int i = 0; i < SAMPLE_COUNT; i++) {
+ fixture.awaitNextDueTrigger();
+ fixture.awaitTrigger();
+ }
+ assertEquals(SAMPLE_COUNT, fixture.getSampled());
+ fixture.endIteration();
+ } finally {
+ fixture.tearDown();
+ }
+ }
+
+ @Test
+ void shouldKeepWarmingUpForTheMinimumTimeOnceTheTriggersRan() throws
Exception {
+ // One trigger is done within a second; the minimum time is what setup
waits for.
+ CheckpointSchedulingFixture fixture =
+ new CheckpointSchedulingFixture(
+ PIPELINE_NUM, CHECKPOINT_INTERVAL_MILLIS, 1L,
MIN_WARMUP_MILLIS);
+ long start = System.nanoTime();
+ fixture.setUp();
+ try {
+ assertTrue(
+ System.nanoTime() - start >=
TimeUnit.MILLISECONDS.toNanos(MIN_WARMUP_MILLIS),
+ "setup should not return before the minimum warm-up time");
+ } finally {
+ fixture.tearDown();
+ }
+ }
+
@Test
void shouldStopEverySchedulerThreadOnTearDown() throws Exception {
// Stands in for a scheduler thread another test in this JVM has not
finished stopping.
