You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "XComp (via GitHub)" <gi...@apache.org> on 2026/05/11 12:28:04 UTC
Re: [PR] [FLINK-36753][runtime]Adaptive Scheduler actively triggers a Checkpoint after all resources are ready [flink]
XComp commented on code in PR #27921:
URL: https://github.com/apache/flink/pull/27921#discussion_r3218138993
##########
flink-core/src/main/java/org/apache/flink/configuration/JobManagerOptions.java:
##########
@@ -745,6 +745,25 @@ public InlineElement getDescription() {
.key()))
.build());
+ @Documentation.Section({
+ Documentation.Sections.EXPERT_SCHEDULING,
+ Documentation.Sections.ALL_JOB_MANAGER
+ })
+ public static final ConfigOption<Boolean> SCHEDULER_RESCALE_TRIGGER_ACTIVE_CHECKPOINT_ENABLED =
+ key("jobmanager.adaptive-scheduler.rescale-trigger.active-checkpoint.enabled")
+ .booleanType()
+ .defaultValue(true)
+ .withDescription(
+ Description.builder()
+ .text(
+ "When enabled, the Adaptive Scheduler actively triggers a checkpoint when resources change and rescaling is desired, "
+ + "rather than waiting for the next periodic checkpoint. "
+ + "This reduces rescaling latency, especially when checkpoint intervals are large. "
+ + "The active trigger respects %s and "
+ + "will not fire if a checkpoint is already in progress or being triggered.",
+ code("execution.checkpointing.min-pause"))
Review Comment:
```suggestion
code(CheckpointingOptions.MIN_PAUSE_BETWEEN_CHECKPOINTS.key()))
```
##########
flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/DefaultStateTransitionManager.java:
##########
@@ -164,6 +164,10 @@ private void progressToStabilized(Temporal firstChangeEventTimestamp) {
progressToPhase(new Stabilized(clock, this, firstChangeEventTimestamp, maxTriggerDelay));
}
+ void requestActiveCheckpointTrigger() {
Review Comment:
```suggestion
private void requestActiveCheckpointTrigger() {
```
we can reduce the access level because the `DefaultStateTransitionManager` packages its state implementations in the same class. No need to expose it on a package level.
##########
flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/Executing.java:
##########
@@ -182,6 +189,103 @@ public ScheduledFuture<?> scheduleOperation(Runnable callback, Duration delay) {
return context.runIfState(this, callback, delay);
}
+ @Override
+ public void requestActiveCheckpointTrigger() {
+ if (!activeCheckpointTriggerEnabled) {
+ return;
+ }
+
+ final CheckpointCoordinator checkpointCoordinator =
+ getExecutionGraph().getCheckpointCoordinator();
+ if (!shouldTriggerActiveCheckpoint(checkpointCoordinator)) {
+ return;
+ }
+
+ final long triggerDelayMillis = checkpointCoordinator.getActiveCheckpointTriggerDelay();
+ if (triggerDelayMillis < 0L) {
+ getLogger()
+ .debug(
+ "Skipping active checkpoint trigger for rescale: checkpoint already in progress.");
+ return;
+ }
+ if (triggerDelayMillis > 0L) {
Review Comment:
do we gain anything from this condition? We could just schedule the trigger in both cases. WDYT?
##########
flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/Executing.java:
##########
@@ -182,6 +189,103 @@ public ScheduledFuture<?> scheduleOperation(Runnable callback, Duration delay) {
return context.runIfState(this, callback, delay);
}
+ @Override
+ public void requestActiveCheckpointTrigger() {
+ if (!activeCheckpointTriggerEnabled) {
+ return;
+ }
+
+ final CheckpointCoordinator checkpointCoordinator =
+ getExecutionGraph().getCheckpointCoordinator();
+ if (!shouldTriggerActiveCheckpoint(checkpointCoordinator)) {
+ return;
+ }
+
+ final long triggerDelayMillis = checkpointCoordinator.getActiveCheckpointTriggerDelay();
+ if (triggerDelayMillis < 0L) {
+ getLogger()
+ .debug(
+ "Skipping active checkpoint trigger for rescale: checkpoint already in progress.");
+ return;
+ }
+ if (triggerDelayMillis > 0L) {
+ scheduleActiveCheckpointTriggerRetry(triggerDelayMillis);
+ return;
+ }
+ fireActiveCheckpointTrigger(checkpointCoordinator);
+ }
+
+ private boolean shouldTriggerActiveCheckpoint(
+ @Nullable CheckpointCoordinator checkpointCoordinator) {
+ if (checkpointCoordinator == null
+ || !checkpointCoordinator.isPeriodicCheckpointingConfigured()) {
+ getLogger()
+ .debug(
+ "Skipping active checkpoint trigger for rescale: checkpointing not configured.");
+ return false;
+ }
+ if (!parallelismChanged()) {
+ getLogger()
+ .debug(
+ "Skipping active checkpoint trigger for rescale: parallelism unchanged.");
+ return false;
+ }
+ return true;
+ }
+
+ private void scheduleActiveCheckpointTriggerRetry(long delayMillis) {
+ if (activeCheckpointTriggerScheduled) {
+ return;
+ }
+ activeCheckpointTriggerScheduled = true;
+ getLogger()
+ .debug(
+ "Min pause not satisfied, scheduling active checkpoint trigger retry in {} ms.",
+ delayMillis);
+ context.runIfState(
+ this, this::tryFireActiveCheckpointAfterRetry, Duration.ofMillis(delayMillis));
+ }
+
+ private void tryFireActiveCheckpointAfterRetry() {
+ activeCheckpointTriggerScheduled = false;
+ final CheckpointCoordinator checkpointCoordinator =
+ getExecutionGraph().getCheckpointCoordinator();
+ if (!shouldTriggerActiveCheckpoint(checkpointCoordinator)) {
+ return;
+ }
+ if (checkpointCoordinator.getActiveCheckpointTriggerDelay() == 0L) {
+ fireActiveCheckpointTrigger(checkpointCoordinator);
+ }
Review Comment:
should we add an else branch with a log message? Otherwise, the checkpoint trigger is silently dropped if another periodic checkpoint is completed while the trigger is scheduled. WDYT?
##########
flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/StateTransitionManager.java:
##########
@@ -89,5 +89,24 @@ interface Context extends RescaleContext {
* @return the {@link JobID} of the job
*/
JobID getJobId();
+
+ /**
+ * Requests the context to actively trigger a checkpoint to expedite rescaling. Called by
+ * the {@link DefaultStateTransitionManager} from within phase lifecycle methods:
Review Comment:
Although, it's an internal interface, we might not want to add documentation of implementations in the interface itself. How it's done in the end is an implementation detail, no?
That documentation is valid but might be better suited in the newly added `DefaultStateTransitionManager#requestActiveCheckpointTrigger`. WDYT?
##########
flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/Executing.java:
##########
@@ -64,6 +67,8 @@ class Executing extends StateWithExecutionGraph
private final StateTransitionManager stateTransitionManager;
private final int rescaleOnFailedCheckpointCount;
+ private final boolean activeCheckpointTriggerEnabled;
+ private boolean activeCheckpointTriggerScheduled;
Review Comment:
```suggestion
// only modifiable from the main thread
private boolean activeCheckpointTriggerScheduled;
```
to make the synchronization log more explicit
##########
flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/ExecutingTest.java:
##########
@@ -556,6 +562,452 @@ public CheckpointCoordinator getCheckpointCoordinator() {
}
}
+ @Test
+ void testActiveCheckpointTriggerSkipsWhenDisabled() throws Exception {
+ try (MockExecutingContext ctx = new MockExecutingContext()) {
+ final AtomicBoolean coordinatorAccessed = new AtomicBoolean(false);
+ CheckpointCoordinator coordinator =
+ new CheckpointCoordinatorTestingUtils.CheckpointCoordinatorBuilder()
+ .build(EXECUTOR_EXTENSION.getExecutor());
+ StateTrackingMockExecutionGraph graph =
+ new StateTrackingMockExecutionGraph() {
+ @Nullable
+ @Override
+ public CheckpointCoordinator getCheckpointCoordinator() {
+ coordinatorAccessed.set(true);
+ return coordinator;
+ }
+ };
+ Executing exec =
+ new ExecutingStateBuilder()
+ .setExecutionGraph(graph)
+ .setActiveCheckpointTriggerEnabled(false)
+ .build(ctx);
+
+ exec.requestActiveCheckpointTrigger();
+ assertThat(coordinatorAccessed.get()).isFalse();
+ }
+ }
+
+ @Test
+ void testActiveCheckpointTriggerSkipsWhenNoCoordinator() throws Exception {
+ try (MockExecutingContext ctx = new MockExecutingContext()) {
+ MockExecutionJobVertex mejv = new MockExecutionJobVertex(MockExecutionVertex::new);
+ // MockExecutionGraph.getCheckpointCoordinator() returns null
Review Comment:
That's an obvious one - the comment doesn't add much value, does it?
##########
flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/ExecutingTest.java:
##########
@@ -556,6 +562,452 @@ public CheckpointCoordinator getCheckpointCoordinator() {
}
}
+ @Test
+ void testActiveCheckpointTriggerSkipsWhenDisabled() throws Exception {
+ try (MockExecutingContext ctx = new MockExecutingContext()) {
+ final AtomicBoolean coordinatorAccessed = new AtomicBoolean(false);
+ CheckpointCoordinator coordinator =
+ new CheckpointCoordinatorTestingUtils.CheckpointCoordinatorBuilder()
+ .build(EXECUTOR_EXTENSION.getExecutor());
+ StateTrackingMockExecutionGraph graph =
+ new StateTrackingMockExecutionGraph() {
+ @Nullable
+ @Override
+ public CheckpointCoordinator getCheckpointCoordinator() {
+ coordinatorAccessed.set(true);
+ return coordinator;
+ }
+ };
+ Executing exec =
+ new ExecutingStateBuilder()
+ .setExecutionGraph(graph)
+ .setActiveCheckpointTriggerEnabled(false)
+ .build(ctx);
+
+ exec.requestActiveCheckpointTrigger();
+ assertThat(coordinatorAccessed.get()).isFalse();
+ }
+ }
+
+ @Test
+ void testActiveCheckpointTriggerSkipsWhenNoCoordinator() throws Exception {
+ try (MockExecutingContext ctx = new MockExecutingContext()) {
+ MockExecutionJobVertex mejv = new MockExecutionJobVertex(MockExecutionVertex::new);
Review Comment:
Why do we have this variable if we never verify its usage in the test?
##########
flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/StateTransitionManager.java:
##########
@@ -89,5 +89,24 @@ interface Context extends RescaleContext {
* @return the {@link JobID} of the job
*/
JobID getJobId();
+
+ /**
+ * Requests the context to actively trigger a checkpoint to expedite rescaling. Called by
+ * the {@link DefaultStateTransitionManager} from within phase lifecycle methods:
+ *
+ * <ul>
+ * <li>On entering {@link DefaultStateTransitionManager.Stabilizing} (to overlap
+ * checkpoint with the stabilization wait)
+ * <li>On each {@link DefaultStateTransitionManager.Stabilizing#onChange} event (retry if
+ * a previous trigger was skipped)
+ * <li>On entering {@link DefaultStateTransitionManager.Stabilized} (fallback if no
+ * checkpoint completed during stabilization)
+ * </ul>
+ *
+ * <p>The implementation decides whether to actually trigger based on its own guard
+ * conditions (e.g., checkpointing enabled, no checkpoint in progress, config flag).
+ * Multiple calls are safe; guards prevent redundant triggers.
Review Comment:
Isn't that an implementation detail that belongs to the `AdaptiveScheduler`? The `StateTransitionManager` only forwards the event to the `AdaptiveScheduler`. It doesn't control the checkpoint-related guardrails.
##########
flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinator.java:
##########
@@ -1999,6 +1999,21 @@ public long getCheckpointTimeout() {
return checkpointTimeout;
}
+ /**
+ * Returns {@code -1} if a checkpoint is already in flight, otherwise the remaining time (in ms)
+ * until {@code minPauseBetweenCheckpoints} is satisfied ({@code 0} = trigger now). All checks
+ * are made under the coordinator lock.
+ */
+ public long getActiveCheckpointTriggerDelay() {
Review Comment:
Wouldn't `Optional<Duration>` a more straightforward return type to serve a cleaner interface w/o needing to define the contract in JavaDoc? Or why did we decide to use `long` here?
##########
flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinator.java:
##########
@@ -1999,6 +1999,21 @@ public long getCheckpointTimeout() {
return checkpointTimeout;
}
+ /**
+ * Returns {@code -1} if a checkpoint is already in flight, otherwise the remaining time (in ms)
+ * until {@code minPauseBetweenCheckpoints} is satisfied ({@code 0} = trigger now). All checks
+ * are made under the coordinator lock.
+ */
+ public long getActiveCheckpointTriggerDelay() {
+ synchronized (lock) {
+ if (isTriggering || !pendingCheckpoints.isEmpty()) {
+ return -1L;
+ }
+ final long elapsed = clock.relativeTimeMillis() - lastCheckpointCompletionRelativeTime;
Review Comment:
Should we handle the case where no checkpoint was created (`lastCheckpointCompletionRelativeTime=0`), yet, explicitly and mention it in the JavaDoc?
##########
flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/ExecutingTest.java:
##########
@@ -556,6 +562,452 @@ public CheckpointCoordinator getCheckpointCoordinator() {
}
}
+ @Test
Review Comment:
Can we add a test case to check what happens if `stopWithSavepoint` is called concurrently?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org