This is an automated email from the ASF dual-hosted git repository.
pnowojski pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new 9661ab549bc [FLINK-36753][runtime]Adaptive Scheduler actively triggers
a Checkpoint after all resources are ready
9661ab549bc is described below
commit 9661ab549bc0371087e4be222bce7f109210fdd9
Author: Samrat002 <[email protected]>
AuthorDate: Sun Apr 12 22:59:06 2026 +0530
[FLINK-36753][runtime]Adaptive Scheduler actively triggers a Checkpoint
after all resources are ready
---
.../generated/all_jobmanager_section.html | 6 +
.../generated/expert_scheduling_section.html | 6 +
.../generated/job_manager_configuration.html | 6 +
.../flink/configuration/JobManagerOptions.java | 18 +
.../runtime/checkpoint/CheckpointCoordinator.java | 22 +
.../scheduler/adaptive/AdaptiveScheduler.java | 16 +-
.../adaptive/DefaultStateTransitionManager.java | 18 +
.../runtime/scheduler/adaptive/Executing.java | 117 ++++-
.../scheduler/adaptive/StateTransitionManager.java | 7 +
.../CheckpointCoordinatorTestingUtils.java | 10 +-
.../DefaultStateTransitionManagerTest.java | 70 +++
.../runtime/scheduler/adaptive/ExecutingTest.java | 584 ++++++++++++++++++++-
.../ActiveCheckpointToRescaleITCase.java | 198 +++++++
13 files changed, 1068 insertions(+), 10 deletions(-)
diff --git a/docs/layouts/shortcodes/generated/all_jobmanager_section.html
b/docs/layouts/shortcodes/generated/all_jobmanager_section.html
index 49ccf35c8da..5d01a4a4885 100644
--- a/docs/layouts/shortcodes/generated/all_jobmanager_section.html
+++ b/docs/layouts/shortcodes/generated/all_jobmanager_section.html
@@ -50,6 +50,12 @@
<td>Boolean</td>
<td>This parameter defines whether the adaptive scheduler
prioritizes using the minimum number of <code
class="highlighter-rouge">TaskManagers</code> when scheduling tasks.<br />Note,
this parameter is suitable if <code
class="highlighter-rouge">execution.state-recovery.from-local</code> is not
enabled. More details about this configuration are available at <a
href="https://issues.apache.org/jira/browse/FLINK-33977";>FLINK-33977</a>.</td>
</tr>
+ <tr>
+
<td><h5>jobmanager.adaptive-scheduler.rescale-trigger.active-checkpoint.enabled</h5></td>
+ <td style="word-wrap: break-word;">true</td>
+ <td>Boolean</td>
+ <td>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 the configured
minimum pause between checkpoints and will not fire if a checkpoint is already
in progress or being triggered.</td>
+ </tr>
<tr>
<td><h5>jobmanager.adaptive-scheduler.rescale-trigger.max-checkpoint-failures</h5></td>
<td style="word-wrap: break-word;">2</td>
diff --git a/docs/layouts/shortcodes/generated/expert_scheduling_section.html
b/docs/layouts/shortcodes/generated/expert_scheduling_section.html
index 524ef524d15..d8ffc2ca3ee 100644
--- a/docs/layouts/shortcodes/generated/expert_scheduling_section.html
+++ b/docs/layouts/shortcodes/generated/expert_scheduling_section.html
@@ -104,6 +104,12 @@
<td>Boolean</td>
<td>This parameter defines whether the adaptive scheduler
prioritizes using the minimum number of <code
class="highlighter-rouge">TaskManagers</code> when scheduling tasks.<br />Note,
this parameter is suitable if <code
class="highlighter-rouge">execution.state-recovery.from-local</code> is not
enabled. More details about this configuration are available at <a
href="https://issues.apache.org/jira/browse/FLINK-33977";>FLINK-33977</a>.</td>
</tr>
+ <tr>
+
<td><h5>jobmanager.adaptive-scheduler.rescale-trigger.active-checkpoint.enabled</h5></td>
+ <td style="word-wrap: break-word;">true</td>
+ <td>Boolean</td>
+ <td>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 the configured
minimum pause between checkpoints and will not fire if a checkpoint is already
in progress or being triggered.</td>
+ </tr>
<tr>
<td><h5>jobmanager.adaptive-scheduler.rescale-trigger.max-checkpoint-failures</h5></td>
<td style="word-wrap: break-word;">2</td>
diff --git a/docs/layouts/shortcodes/generated/job_manager_configuration.html
b/docs/layouts/shortcodes/generated/job_manager_configuration.html
index d197b268780..75aa1a1d9c8 100644
--- a/docs/layouts/shortcodes/generated/job_manager_configuration.html
+++ b/docs/layouts/shortcodes/generated/job_manager_configuration.html
@@ -50,6 +50,12 @@
<td>Boolean</td>
<td>This parameter defines whether the adaptive scheduler
prioritizes using the minimum number of <code
class="highlighter-rouge">TaskManagers</code> when scheduling tasks.<br />Note,
this parameter is suitable if <code
class="highlighter-rouge">execution.state-recovery.from-local</code> is not
enabled. More details about this configuration are available at <a
href="https://issues.apache.org/jira/browse/FLINK-33977";>FLINK-33977</a>.</td>
</tr>
+ <tr>
+
<td><h5>jobmanager.adaptive-scheduler.rescale-trigger.active-checkpoint.enabled</h5></td>
+ <td style="word-wrap: break-word;">true</td>
+ <td>Boolean</td>
+ <td>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 the configured
minimum pause between checkpoints and will not fire if a checkpoint is already
in progress or being triggered.</td>
+ </tr>
<tr>
<td><h5>jobmanager.adaptive-scheduler.rescale-trigger.max-checkpoint-failures</h5></td>
<td style="word-wrap: break-word;">2</td>
diff --git
a/flink-core/src/main/java/org/apache/flink/configuration/JobManagerOptions.java
b/flink-core/src/main/java/org/apache/flink/configuration/JobManagerOptions.java
index 065de7e63d3..a185ce22f67 100644
---
a/flink-core/src/main/java/org/apache/flink/configuration/JobManagerOptions.java
+++
b/flink-core/src/main/java/org/apache/flink/configuration/JobManagerOptions.java
@@ -745,6 +745,24 @@ public class JobManagerOptions {
.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 the configured minimum pause between checkpoints and "
+ + "will not fire if a
checkpoint is already in progress or being triggered.")
+ .build());
+
/**
* @deprecated Use {@link
*
JobManagerOptions#SCHEDULER_SUBMISSION_RESOURCE_STABILIZATION_TIMEOUT}.
diff --git
a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinator.java
b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinator.java
index 2a86a566b1f..7750543be41 100644
---
a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinator.java
+++
b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinator.java
@@ -61,6 +61,7 @@ import javax.annotation.Nullable;
import javax.annotation.concurrent.GuardedBy;
import java.io.IOException;
+import java.time.Duration;
import java.util.ArrayDeque;
import java.util.ArrayList;
import java.util.Collection;
@@ -1999,6 +2000,27 @@ public class CheckpointCoordinator {
return checkpointTimeout;
}
+ /**
+ * Returns the remaining {@link Duration} until {@code
minPauseBetweenCheckpoints} is satisfied
+ * for a new active-trigger checkpoint, computed from the time elapsed
since the last completed
+ * checkpoint (or from the coordinator clock's epoch when no checkpoint
has completed yet —
+ * which is normally far in the past in production). {@link Duration#ZERO}
means the trigger can
+ * fire immediately.
+ *
+ * <p>Returns {@link Optional#empty()} as a fallback if a checkpoint is
already in flight
+ * (triggering or pending), in which case no active trigger should be
scheduled.
+ */
+ public Optional<Duration> getActiveCheckpointTriggerDelay() {
+ synchronized (lock) {
+ if (isTriggering || !pendingCheckpoints.isEmpty()) {
+ return Optional.empty();
+ }
+ final long elapsed = clock.relativeTimeMillis() -
lastCheckpointCompletionRelativeTime;
+ final long remaining = minPauseBetweenCheckpoints - elapsed;
+ return Optional.of(remaining > 0 ? Duration.ofMillis(remaining) :
Duration.ZERO);
+ }
+ }
+
/**
* @deprecated use {@link #getNumQueuedRequests()}
*/
diff --git
a/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/AdaptiveScheduler.java
b/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/AdaptiveScheduler.java
index 70c5b62ce90..b75f0d09f29 100644
---
a/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/AdaptiveScheduler.java
+++
b/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/AdaptiveScheduler.java
@@ -314,7 +314,9 @@ public class AdaptiveScheduler
SCHEDULER_RESCALE_TRIGGER_MAX_DELAY,
maximumDelayForRescaleTriggerDefault),
rescaleOnFailedCheckpointsCount,
-
configuration.get(WebOptions.MAX_ADAPTIVE_SCHEDULER_RESCALE_HISTORY_SIZE));
+
configuration.get(WebOptions.MAX_ADAPTIVE_SCHEDULER_RESCALE_HISTORY_SIZE),
+ configuration.get(
+
JobManagerOptions.SCHEDULER_RESCALE_TRIGGER_ACTIVE_CHECKPOINT_ENABLED));
}
private final SchedulerExecutionMode executionMode;
@@ -326,6 +328,7 @@ public class AdaptiveScheduler
private final Duration maximumDelayForTriggeringRescale;
private final int rescaleOnFailedCheckpointCount;
private final int rescaleHistoryMax;
+ private final boolean activeCheckpointTriggerEnabled;
private Settings(
SchedulerExecutionMode executionMode,
@@ -336,7 +339,8 @@ public class AdaptiveScheduler
Duration executingResourceStabilizationTimeout,
Duration maximumDelayForTriggeringRescale,
int rescaleOnFailedCheckpointCount,
- int rescaleHistoryMax) {
+ int rescaleHistoryMax,
+ boolean activeCheckpointTriggerEnabled) {
this.executionMode = executionMode;
this.submissionResourceWaitTimeout = submissionResourceWaitTimeout;
this.submissionResourceStabilizationTimeout =
submissionResourceStabilizationTimeout;
@@ -346,6 +350,7 @@ public class AdaptiveScheduler
this.maximumDelayForTriggeringRescale =
maximumDelayForTriggeringRescale;
this.rescaleOnFailedCheckpointCount =
rescaleOnFailedCheckpointCount;
this.rescaleHistoryMax = rescaleHistoryMax;
+ this.activeCheckpointTriggerEnabled =
activeCheckpointTriggerEnabled;
}
public SchedulerExecutionMode getExecutionMode() {
@@ -384,6 +389,10 @@ public class AdaptiveScheduler
return rescaleHistoryMax;
}
+ public boolean isActiveCheckpointTriggerEnabled() {
+ return activeCheckpointTriggerEnabled;
+ }
+
public JobRescaleConfigInfo toJobRescaleConfigInfo() {
return new JobRescaleConfigInfo(
rescaleHistoryMax,
@@ -1311,7 +1320,8 @@ public class AdaptiveScheduler
userCodeClassLoader,
failureCollection,
this::createExecutingStateTransitionManager,
- settings.getRescaleOnFailedCheckpointCount()));
+ settings.getRescaleOnFailedCheckpointCount(),
+ settings.isActiveCheckpointTriggerEnabled()));
}
private StateTransitionManager createExecutingStateTransitionManager(
diff --git
a/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/DefaultStateTransitionManager.java
b/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/DefaultStateTransitionManager.java
index 87f810ae784..a7945e36166 100644
---
a/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/DefaultStateTransitionManager.java
+++
b/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/DefaultStateTransitionManager.java
@@ -164,6 +164,21 @@ public class DefaultStateTransitionManager implements
RescaleContext, StateTrans
progressToPhase(new Stabilized(clock, this, firstChangeEventTimestamp,
maxTriggerDelay));
}
+ /**
+ * Requests the active checkpoint trigger from the context. Called from
within phase lifecycle
+ * methods:
+ *
+ * <ul>
+ * <li>On entering {@link Stabilizing} (to overlap checkpoint with the
stabilization wait)
+ * <li>On each {@link Stabilizing#onChange} event (retry if a previous
trigger was skipped)
+ * <li>On entering {@link Stabilized} (fallback if no checkpoint
completed during
+ * stabilization)
+ * </ul>
+ */
+ private void requestActiveCheckpointTrigger() {
+ transitionContext.requestActiveCheckpointTrigger();
+ }
+
private void triggerTransitionToSubsequentState() {
progressToPhase(new Transitioning(clock, this));
transitionContext.transitionToSubsequentState();
@@ -362,6 +377,7 @@ public class DefaultStateTransitionManager implements
RescaleContext, StateTrans
resourceStabilizationTimeout);
scheduleTransitionEvaluation();
+ context().requestActiveCheckpointTrigger();
}
@Override
@@ -370,6 +386,7 @@ public class DefaultStateTransitionManager implements
RescaleContext, StateTrans
// event was already handled by a onTrigger callback with a no-op
onChangeEventTimestamp = now();
scheduleTransitionEvaluation();
+ context().requestActiveCheckpointTrigger();
}
@Override
@@ -427,6 +444,7 @@ public class DefaultStateTransitionManager implements
RescaleContext, StateTrans
},
firstChangeEventTimestamp,
maxTriggerDelay);
+ context().requestActiveCheckpointTrigger();
}
@Override
diff --git
a/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/Executing.java
b/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/Executing.java
index deae52856dc..d0ecf284b79 100644
---
a/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/Executing.java
+++
b/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/Executing.java
@@ -18,9 +18,11 @@
package org.apache.flink.runtime.scheduler.adaptive;
+import org.apache.flink.api.common.JobID;
import org.apache.flink.api.common.JobStatus;
import org.apache.flink.core.execution.SavepointFormatType;
import org.apache.flink.runtime.JobException;
+import org.apache.flink.runtime.checkpoint.CheckpointCoordinator;
import org.apache.flink.runtime.checkpoint.CheckpointScheduling;
import org.apache.flink.runtime.checkpoint.CheckpointStatsListener;
import org.apache.flink.runtime.checkpoint.CompletedCheckpoint;
@@ -40,6 +42,7 @@ import
org.apache.flink.runtime.scheduler.exceptionhistory.ExceptionHistoryEntry
import
org.apache.flink.runtime.scheduler.stopwithsavepoint.StopWithSavepointTerminationManager;
import org.apache.flink.util.ExceptionUtils;
import org.apache.flink.util.Preconditions;
+import org.apache.flink.util.concurrent.FutureUtils;
import org.slf4j.Logger;
@@ -64,6 +67,9 @@ class Executing extends StateWithExecutionGraph
private final StateTransitionManager stateTransitionManager;
private final int rescaleOnFailedCheckpointCount;
+ private final boolean activeCheckpointTriggerEnabled;
+ // only modifiable from the main thread
+ private boolean activeCheckpointTriggerScheduled;
// null indicates that there was no change event observed, yet
@Nullable private AtomicInteger failedCheckpointCountdown;
@@ -77,7 +83,8 @@ class Executing extends StateWithExecutionGraph
List<ExceptionHistoryEntry> failureCollection,
Function<StateTransitionManager.Context, StateTransitionManager>
stateTransitionManagerFactory,
- int rescaleOnFailedCheckpointCount) {
+ int rescaleOnFailedCheckpointCount,
+ boolean activeCheckpointTriggerEnabled) {
super(
context,
executionGraph,
@@ -96,6 +103,7 @@ class Executing extends StateWithExecutionGraph
rescaleOnFailedCheckpointCount > 0,
"The rescaleOnFailedCheckpointCount should be larger than 0.");
this.rescaleOnFailedCheckpointCount = rescaleOnFailedCheckpointCount;
+ this.activeCheckpointTriggerEnabled = activeCheckpointTriggerEnabled;
this.failedCheckpointCountdown = null;
recordRescaleForJobIntoExecuting(logger, context);
@@ -182,6 +190,105 @@ class Executing extends StateWithExecutionGraph
return context.runIfState(this, callback, delay);
}
+ @Override
+ public void requestActiveCheckpointTrigger() {
+ if (!activeCheckpointTriggerEnabled) {
+ return;
+ }
+ final CheckpointCoordinator checkpointCoordinator =
+ getExecutionGraph().getCheckpointCoordinator();
+ if (checkpointCoordinator == null
+ || !checkpointCoordinator.isPeriodicCheckpointingConfigured())
{
+ getLogger()
+ .debug(
+ "Skipping active checkpoint trigger for rescale:
checkpointing not configured.");
+ return;
+ }
+
+ final Optional<Duration> triggerDelay =
+ checkpointCoordinator.getActiveCheckpointTriggerDelay();
+ if (triggerDelay.isEmpty()) {
+ getLogger()
+ .debug(
+ "Skipping active checkpoint trigger for rescale:
checkpoint already in progress.");
+ return;
+ }
+ scheduleActiveCheckpointTriggerRetry(triggerDelay.get());
+ }
+
+ private void scheduleActiveCheckpointTriggerRetry(Duration delay) {
+ if (activeCheckpointTriggerScheduled) {
+ return;
+ }
+ activeCheckpointTriggerScheduled = true;
+ if (!delay.isZero()) {
+ getLogger()
+ .debug(
+ "Min pause not satisfied, scheduling active
checkpoint trigger retry in {} ms.",
+ delay.toMillis());
+ }
+ context.runIfState(this, this::tryFireActiveCheckpointAfterRetry,
delay);
+ }
+
+ private void tryFireActiveCheckpointAfterRetry() {
+ activeCheckpointTriggerScheduled = false;
+
+ // Parallelism is the only guard re-evaluated here: it can change
between the request
+ // and the scheduled fire (e.g. resources changed again, or the
parallelism was reverted
+ // back to the current value while we waited for min-pause). The null
check and
+ // periodic-checkpoint config are invariants validated at request time.
+ if (!parallelismChanged()) {
+ getLogger()
+ .debug("Active checkpoint trigger for rescale dropped:
parallelism unchanged.");
+ return;
+ }
+ final CheckpointCoordinator checkpointCoordinator =
+ Preconditions.checkNotNull(
+ getExecutionGraph().getCheckpointCoordinator(),
+ "Checkpoint coordinator was non-null when the trigger
was scheduled; "
+ + "an Executing state never drops its
coordinator.");
+ final Optional<Duration> triggerDelay =
+ checkpointCoordinator.getActiveCheckpointTriggerDelay();
+ if (triggerDelay.isEmpty()) {
+ getLogger()
+ .debug(
+ "Active checkpoint trigger for rescale dropped:
checkpoint already in progress after retry.");
+ } else if (triggerDelay.get().isZero()) {
+ fireActiveCheckpointTrigger(checkpointCoordinator);
+ } else {
+ getLogger()
+ .debug(
+ "Active checkpoint trigger for rescale silently
dropped: a periodic checkpoint completed while the trigger was scheduled.");
+ }
+ }
+
+ private void fireActiveCheckpointTrigger(CheckpointCoordinator
checkpointCoordinator) {
+ Preconditions.checkState(
+ activeCheckpointTriggerEnabled,
+ "Active checkpoint trigger fired while the feature is
disabled.");
+ final JobID jobId = getExecutionGraph().getJobID();
+ getLogger().info("Actively triggering checkpoint to expedite
rescaling, job {}.", jobId);
+ // isPeriodic=false: min-pause is enforced above via
getActiveCheckpointTriggerDelay.
+ FutureUtils.assertNoException(
+ checkpointCoordinator
+ .triggerCheckpoint(false)
+ .handle(
+ (completedCheckpoint, throwable) -> {
+ if (throwable != null) {
+ getLogger()
+ .warn(
+ "Active checkpoint
trigger for rescale failed.",
+ throwable);
+ } else {
+ getLogger()
+ .info(
+ "Active checkpoint for
rescale completed successfully: {}.",
+
completedCheckpoint.getCheckpointID());
+ }
+ return null;
+ }));
+ }
+
@Override
public void transitionToSubsequentState() {
Optional<VertexParallelism> availableVertexParallelism =
@@ -399,6 +506,7 @@ class Executing extends StateWithExecutionGraph
private final Function<StateTransitionManager.Context,
StateTransitionManager>
stateTransitionManagerFactory;
private final int rescaleOnFailedCheckpointCount;
+ private final boolean activeCheckpointTriggerEnabled;
Factory(
ExecutionGraph executionGraph,
@@ -410,7 +518,8 @@ class Executing extends StateWithExecutionGraph
List<ExceptionHistoryEntry> failureCollection,
Function<StateTransitionManager.Context,
StateTransitionManager>
stateTransitionManagerFactory,
- int rescaleOnFailedCheckpointCount) {
+ int rescaleOnFailedCheckpointCount,
+ boolean activeCheckpointTriggerEnabled) {
this.context = context;
this.log = log;
this.executionGraph = executionGraph;
@@ -420,6 +529,7 @@ class Executing extends StateWithExecutionGraph
this.failureCollection = failureCollection;
this.stateTransitionManagerFactory = stateTransitionManagerFactory;
this.rescaleOnFailedCheckpointCount =
rescaleOnFailedCheckpointCount;
+ this.activeCheckpointTriggerEnabled =
activeCheckpointTriggerEnabled;
}
public Class<Executing> getStateClass() {
@@ -436,7 +546,8 @@ class Executing extends StateWithExecutionGraph
userCodeClassLoader,
failureCollection,
stateTransitionManagerFactory,
- rescaleOnFailedCheckpointCount);
+ rescaleOnFailedCheckpointCount,
+ activeCheckpointTriggerEnabled);
}
}
}
diff --git
a/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/StateTransitionManager.java
b/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/StateTransitionManager.java
index 98229a9afd3..5b2ca1b928d 100644
---
a/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/StateTransitionManager.java
+++
b/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/StateTransitionManager.java
@@ -89,5 +89,12 @@ public interface StateTransitionManager {
* @return the {@link JobID} of the job
*/
JobID getJobId();
+
+ /**
+ * Requests the context to actively trigger a checkpoint to expedite
rescaling. The
+ * implementation decides whether to actually trigger based on its own
guard conditions.
+ * Multiple calls are safe; guards prevent redundant triggers.
+ */
+ default void requestActiveCheckpointTrigger() {}
}
}
diff --git
a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinatorTestingUtils.java
b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinatorTestingUtils.java
index 0a505ab07f2..5493e05667c 100644
---
a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinatorTestingUtils.java
+++
b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinatorTestingUtils.java
@@ -63,6 +63,7 @@ import org.apache.flink.runtime.testtasks.NoOpInvokable;
import org.apache.flink.runtime.testutils.CommonTestUtils;
import org.apache.flink.util.InstantiationUtil;
import org.apache.flink.util.Preconditions;
+import org.apache.flink.util.clock.Clock;
import org.apache.flink.util.clock.SystemClock;
import org.apache.flink.util.concurrent.Executors;
import org.apache.flink.util.concurrent.ManuallyTriggeredScheduledExecutor;
@@ -791,6 +792,8 @@ public class CheckpointCoordinatorTestingUtils {
VertexFinishedStateChecker>
vertexFinishedStateCheckerFactory =
VertexFinishedStateChecker::new;
+ private Clock clock = SystemClock.getInstance();
+
public CheckpointCoordinatorBuilder
setCheckpointCoordinatorConfiguration(
CheckpointCoordinatorConfiguration
checkpointCoordinatorConfiguration) {
this.checkpointCoordinatorConfiguration =
checkpointCoordinatorConfiguration;
@@ -870,6 +873,11 @@ public class CheckpointCoordinatorTestingUtils {
return this;
}
+ public CheckpointCoordinatorBuilder setClock(Clock clock) {
+ this.clock = clock;
+ return this;
+ }
+
public CheckpointCoordinator build(ScheduledExecutorService
executorService)
throws Exception {
return build(
@@ -899,7 +907,7 @@ public class CheckpointCoordinatorTestingUtils {
timer,
failureManager,
checkpointPlanCalculator,
- SystemClock.getInstance(),
+ clock,
checkpointStatsTracker,
vertexFinishedStateCheckerFactory);
}
diff --git
a/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/DefaultStateTransitionManagerTest.java
b/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/DefaultStateTransitionManagerTest.java
index cf6051a90b3..8453f5c5a6f 100644
---
a/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/DefaultStateTransitionManagerTest.java
+++
b/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/DefaultStateTransitionManagerTest.java
@@ -400,6 +400,66 @@ class DefaultStateTransitionManagerTest {
assertThat(testInstance.getPhase()).isInstanceOf(Transitioning.class);
}
+ @Test
+ void testActiveCheckpointTriggerCalledOnEnteringStabilizing() {
+ final TestingStateTransitionManagerContext ctx =
+ TestingStateTransitionManagerContext.stableContext();
+ ctx.withSufficientResources();
+ final DefaultStateTransitionManager testInstance =
+ ctx.createTestInstanceThatPassedCooldownPhase();
+
+ assertThat(testInstance.getPhase()).isInstanceOf(Idling.class);
+ assertThat(ctx.getActiveCheckpointTriggerCount())
+ .as("Neither Cooldown nor Idling should fire the active
trigger")
+ .isZero();
+
+ testInstance.onChange(true);
+
+ assertThat(testInstance.getPhase()).isInstanceOf(Stabilizing.class);
+ assertThat(ctx.getActiveCheckpointTriggerCount())
+ .as("Entering Stabilizing must fire the active trigger exactly
once")
+ .isEqualTo(1);
+ }
+
+ @Test
+ void testActiveCheckpointTriggerCalledOnChangeInStabilizing() {
+ final TestingStateTransitionManagerContext ctx =
+ TestingStateTransitionManagerContext.stableContext();
+ ctx.withSufficientResources();
+ final DefaultStateTransitionManager testInstance =
+ ctx.createTestInstanceThatPassedCooldownPhase();
+
+ testInstance.onChange(true);
+ assertThat(testInstance.getPhase()).isInstanceOf(Stabilizing.class);
+
+ testInstance.onChange(true);
+
+ assertThat(testInstance.getPhase()).isInstanceOf(Stabilizing.class);
+ assertThat(ctx.getActiveCheckpointTriggerCount())
+ .as(
+ "Two onChange calls. One entering Stabilizing, one
onChange while in Stabilizing — must fire the trigger twice")
+ .isEqualTo(2);
+ }
+
+ @Test
+ void testActiveCheckpointTriggerCalledOnEnteringStabilized() {
+ final TestingStateTransitionManagerContext ctx =
+ TestingStateTransitionManagerContext.stableContext();
+ ctx.withSufficientResources();
+ final DefaultStateTransitionManager testInstance =
+ ctx.createTestInstanceThatPassedCooldownPhase();
+
+ testInstance.onChange(true);
+ assertThat(testInstance.getPhase()).isInstanceOf(Stabilizing.class);
+
+ ctx.passResourceStabilizationTimeout();
+
+ assertThat(testInstance.getPhase()).isInstanceOf(Stabilized.class);
+ assertThat(ctx.getActiveCheckpointTriggerCount())
+ .as("Entering Stabilizing then Stabilized must fire the
trigger twice in total")
+ .isEqualTo(2);
+ }
+
private static void changeWithoutPhaseMove(
TestingStateTransitionManagerContext ctx,
DefaultStateTransitionManager testInstance,
@@ -460,6 +520,7 @@ class DefaultStateTransitionManagerTest {
// internal state used for assertions
private final AtomicBoolean transitionTriggered = new AtomicBoolean();
+ private int activeCheckpointTriggerCount = 0;
private final SortedMap<Instant, List<ScheduledTask<Object>>>
scheduledTasks =
new TreeMap<>();
@@ -537,6 +598,11 @@ class DefaultStateTransitionManagerTest {
transitionTriggered.set(true);
}
+ @Override
+ public void requestActiveCheckpointTrigger() {
+ activeCheckpointTriggerCount++;
+ }
+
@Override
public ScheduledFuture<?> scheduleOperation(Runnable callback,
Duration delay) {
final Instant triggerTime =
@@ -703,5 +769,9 @@ class DefaultStateTransitionManagerTest {
public void clearStateTransition() {
transitionTriggered.set(false);
}
+
+ public int getActiveCheckpointTriggerCount() {
+ return activeCheckpointTriggerCount;
+ }
}
}
diff --git
a/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/ExecutingTest.java
b/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/ExecutingTest.java
index b9ff24d7c58..0d3f76d3aed 100644
---
a/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/ExecutingTest.java
+++
b/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/ExecutingTest.java
@@ -29,6 +29,7 @@ import
org.apache.flink.runtime.checkpoint.CheckpointCoordinatorTestingUtils;
import org.apache.flink.runtime.checkpoint.CheckpointScheduling;
import org.apache.flink.runtime.client.JobExecutionException;
import org.apache.flink.runtime.concurrent.ComponentMainThreadExecutor;
+import
org.apache.flink.runtime.concurrent.ManuallyTriggeredComponentMainThreadExecutor;
import org.apache.flink.runtime.deployment.TaskDeploymentDescriptorFactory;
import org.apache.flink.runtime.execution.ExecutionState;
import org.apache.flink.runtime.executiongraph.AccessExecutionJobVertex;
@@ -57,6 +58,7 @@ import
org.apache.flink.runtime.jobgraph.IntermediateDataSetID;
import org.apache.flink.runtime.jobgraph.IntermediateResultPartitionID;
import org.apache.flink.runtime.jobgraph.JobVertex;
import org.apache.flink.runtime.jobgraph.JobVertexID;
+import
org.apache.flink.runtime.jobgraph.tasks.CheckpointCoordinatorConfiguration;
import org.apache.flink.runtime.metrics.groups.UnregisteredMetricGroups;
import org.apache.flink.runtime.operators.coordination.CoordinatorStoreImpl;
import org.apache.flink.runtime.scheduler.DefaultVertexParallelismInfo;
@@ -75,6 +77,8 @@ import
org.apache.flink.runtime.taskmanager.TaskExecutionState;
import org.apache.flink.testutils.TestingUtils;
import org.apache.flink.testutils.executor.TestExecutorExtension;
import org.apache.flink.util.Preconditions;
+import org.apache.flink.util.clock.ManualClock;
+import org.apache.flink.util.concurrent.ManuallyTriggeredScheduledExecutor;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.RegisterExtension;
@@ -160,7 +164,8 @@ class ExecutingTest {
ClassLoader.getSystemClassLoader(),
new ArrayList<>(),
(context) -> TestingStateTransitionManager.withNoOp(),
- 1);
+ 1,
+ false);
assertThat(mockExecutionVertex.isDeployCalled()).isFalse();
}
}
@@ -186,7 +191,8 @@ class ExecutingTest {
ClassLoader.getSystemClassLoader(),
new ArrayList<>(),
context ->
TestingStateTransitionManager.withNoOp(),
- 1);
+ 1,
+ false);
}
})
.isInstanceOf(IllegalStateException.class);
@@ -556,6 +562,564 @@ class ExecutingTest {
}
}
+ @Test
+ void testActiveCheckpointTriggerRetryIsNoopAfterStopWithSavepoint() throws
Exception {
+ try (MockExecutingContext ctx = new MockExecutingContext()) {
+ MockExecutionJobVertex mejv = new
MockExecutionJobVertex(MockExecutionVertex::new);
+
+ ManuallyTriggeredScheduledExecutor checkpointTimer =
+ new ManuallyTriggeredScheduledExecutor();
+ ManualClock clock = new ManualClock();
+ CheckpointCoordinatorConfiguration coordConfig =
+ new CheckpointCoordinatorConfiguration
+
.CheckpointCoordinatorConfigurationBuilder()
+ .setCheckpointInterval(10_000L)
+ .setMinPauseBetweenCheckpoints(10_000L)
+ .setMaxConcurrentCheckpoints(Integer.MAX_VALUE)
+ .build();
+ CheckpointCoordinator coordinator =
+ new
CheckpointCoordinatorTestingUtils.CheckpointCoordinatorBuilder()
+ .setCheckpointCoordinatorConfiguration(coordConfig)
+ .setTimer(checkpointTimer)
+ .setClock(clock)
+ .build(EXECUTOR_EXTENSION.getExecutor());
+ StateTrackingMockExecutionGraph graph =
+ new StateTrackingMockExecutionGraph() {
+ @Nullable
+ @Override
+ public CheckpointCoordinator
getCheckpointCoordinator() {
+ return coordinator;
+ }
+
+ @Override
+ public Iterable<ExecutionJobVertex>
getVerticesTopologically() {
+ return Collections.singletonList(mejv);
+ }
+ };
+ ctx.setVertexParallelism(
+ new VertexParallelism(
+ graph.getAllVertices().values().stream()
+ .collect(
+ Collectors.toMap(
+
AccessExecutionJobVertex::getJobVertexId,
+ v -> v.getParallelism() +
1))));
+
+ Executing exec =
+ new ExecutingStateBuilder()
+ .setExecutionGraph(graph)
+ .setActiveCheckpointTriggerEnabled(true)
+ .build(ctx);
+ exec.requestActiveCheckpointTrigger();
+ ManuallyTriggeredComponentMainThreadExecutor ctxExecutor =
+ (ManuallyTriggeredComponentMainThreadExecutor)
ctx.getMainThreadExecutor();
+ assertThat(ctxExecutor.getActiveNonPeriodicScheduledTask())
+ .as("A retry should be scheduled while min-pause is not
satisfied")
+ .isNotEmpty();
+
+ ctx.setExpectStopWithSavepoint(assertNonNull());
+ exec.stopWithSavepoint("file:///tmp/target", true,
SavepointFormatType.CANONICAL);
+
+ ctxExecutor.triggerAll();
+ checkpointTimer.triggerAll();
+ int pendingAfterSavepoint =
coordinator.getNumberOfPendingCheckpoints();
+
+ // Trigger the previously scheduled retry; it must be a no-op
because the state
+ // transitioned to StopWithSavepoint (runIfState gates the action
on
+ // hadStateTransition).
+ ctxExecutor.triggerNonPeriodicScheduledTasks();
+ checkpointTimer.triggerAll();
+ assertThat(coordinator.getNumberOfPendingCheckpoints())
+ .as("Active checkpoint trigger retry must not fire after
stopWithSavepoint")
+ .isEqualTo(pendingAfterSavepoint);
+ }
+ }
+
+ @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);
+
+ StateTrackingMockExecutionGraph graph =
+ new StateTrackingMockExecutionGraph() {
+ @Nullable
+ @Override
+ public CheckpointCoordinator
getCheckpointCoordinator() {
+ return null;
+ }
+
+ @Override
+ public Iterable<ExecutionJobVertex>
getVerticesTopologically() {
+ return Collections.singletonList(mejv);
+ }
+ };
+ // Set parallelism change so the only blocking guard is the null
coordinator check;
+ // if the null check is ever removed from
requestActiveCheckpointTrigger this test
+ // will NPE on `cc.isPeriodicCheckpointingConfigured()` and fail
loudly.
+ ctx.setVertexParallelism(
+ new VertexParallelism(
+ graph.getAllVertices().values().stream()
+ .collect(
+ Collectors.toMap(
+
AccessExecutionJobVertex::getJobVertexId,
+ v -> v.getParallelism() +
1))));
+
+ Executing exec =
+ new ExecutingStateBuilder()
+ .setExecutionGraph(graph)
+ .setActiveCheckpointTriggerEnabled(true)
+ .build(ctx);
+ ManuallyTriggeredComponentMainThreadExecutor ctxExecutor =
+ (ManuallyTriggeredComponentMainThreadExecutor)
ctx.getMainThreadExecutor();
+ int baseline =
ctxExecutor.getActiveNonPeriodicScheduledTask().size();
+ exec.requestActiveCheckpointTrigger();
+ assertThat(ctxExecutor.getActiveNonPeriodicScheduledTask())
+ .as("No checkpoint trigger should be scheduled when
coordinator is null")
+ .hasSize(baseline);
+ }
+ }
+
+ @Test
+ void
testActiveCheckpointTriggerSkipsWhenPeriodicCheckpointingNotConfigured() throws
Exception {
+ try (MockExecutingContext ctx = new MockExecutingContext()) {
+ MockExecutionJobVertex mejv = new
MockExecutionJobVertex(MockExecutionVertex::new);
+
+ ManuallyTriggeredScheduledExecutor checkpointTimer =
+ new ManuallyTriggeredScheduledExecutor();
+ CheckpointCoordinator coordinator =
+ new
CheckpointCoordinatorTestingUtils.CheckpointCoordinatorBuilder()
+ .setCheckpointCoordinatorConfiguration(
+ new CheckpointCoordinatorConfiguration
+
.CheckpointCoordinatorConfigurationBuilder()
+
.setCheckpointInterval(Long.MAX_VALUE)
+ .build())
+ .setTimer(checkpointTimer)
+ .build(EXECUTOR_EXTENSION.getExecutor());
+ StateTrackingMockExecutionGraph graph =
+ new StateTrackingMockExecutionGraph() {
+ @Nullable
+ @Override
+ public CheckpointCoordinator
getCheckpointCoordinator() {
+ return coordinator;
+ }
+
+ @Override
+ public Iterable<ExecutionJobVertex>
getVerticesTopologically() {
+ return Collections.singletonList(mejv);
+ }
+ };
+ // Set parallelism change so the only blocking guard is
periodic-checkpointing-not-
+ // configured. If the periodic check is removed, the request will
schedule a retry —
+ // which the assertion below catches.
+ ctx.setVertexParallelism(
+ new VertexParallelism(
+ graph.getAllVertices().values().stream()
+ .collect(
+ Collectors.toMap(
+
AccessExecutionJobVertex::getJobVertexId,
+ v -> v.getParallelism() +
1))));
+
+ Executing exec =
+ new ExecutingStateBuilder()
+ .setExecutionGraph(graph)
+ .setActiveCheckpointTriggerEnabled(true)
+ .build(ctx);
+
+
assertThat(coordinator.isPeriodicCheckpointingConfigured()).isFalse();
+ ManuallyTriggeredComponentMainThreadExecutor ctxExecutor =
+ (ManuallyTriggeredComponentMainThreadExecutor)
ctx.getMainThreadExecutor();
+ int baseline =
ctxExecutor.getActiveNonPeriodicScheduledTask().size();
+ exec.requestActiveCheckpointTrigger();
+ assertThat(ctxExecutor.getActiveNonPeriodicScheduledTask())
+ .as(
+ "No retry should be scheduled when periodic
checkpointing is not configured")
+ .hasSize(baseline);
+ checkpointTimer.triggerAll();
+ assertThat(coordinator.getNumberOfPendingCheckpoints())
+ .as(
+ "No checkpoint should be triggered when periodic
checkpointing is not configured")
+ .isEqualTo(0);
+ }
+ }
+
+ @Test
+ void testActiveCheckpointTriggerSkipsWhenParallelismUnchanged() throws
Exception {
+ try (MockExecutingContext ctx = new MockExecutingContext()) {
+ MockExecutionJobVertex mejv = new
MockExecutionJobVertex(MockExecutionVertex::new);
+
+ ManuallyTriggeredScheduledExecutor checkpointTimer =
+ new ManuallyTriggeredScheduledExecutor();
+ CheckpointCoordinator coordinator =
+ new
CheckpointCoordinatorTestingUtils.CheckpointCoordinatorBuilder()
+ .setTimer(checkpointTimer)
+ .build(EXECUTOR_EXTENSION.getExecutor());
+ StateTrackingMockExecutionGraph graph =
+ new StateTrackingMockExecutionGraph() {
+ @Nullable
+ @Override
+ public CheckpointCoordinator
getCheckpointCoordinator() {
+ return coordinator;
+ }
+
+ @Override
+ public Iterable<ExecutionJobVertex>
getVerticesTopologically() {
+ return Collections.singletonList(mejv);
+ }
+ };
+ // Parallelism is intentionally UNCHANGED — the parallelism guard
in
+ // tryFireActiveCheckpointAfterRetry must drop the trigger before
it fires.
+ ctx.setVertexParallelism(
+ new VertexParallelism(
+ graph.getAllVertices().values().stream()
+ .collect(
+ Collectors.toMap(
+
AccessExecutionJobVertex::getJobVertexId,
+
AccessExecutionJobVertex::getParallelism))));
+
+ Executing exec =
+ new ExecutingStateBuilder()
+ .setExecutionGraph(graph)
+ .setActiveCheckpointTriggerEnabled(true)
+ .build(ctx);
+ ManuallyTriggeredComponentMainThreadExecutor ctxExecutor =
+ (ManuallyTriggeredComponentMainThreadExecutor)
ctx.getMainThreadExecutor();
+ exec.requestActiveCheckpointTrigger();
+ // The request path schedules a retry (parallelism check lives in
the retry, not in
+ // the request). Run the scheduled retry to exercise the
parallelism check; without
+ // this drain the test would pass for the wrong reason (retry
never ran).
+ ctxExecutor.triggerNonPeriodicScheduledTasks();
+ checkpointTimer.triggerAll();
+ assertThat(coordinator.getNumberOfPendingCheckpoints())
+ .as("No checkpoint should be triggered when parallelism is
unchanged")
+ .isEqualTo(0);
+ }
+ }
+
+ @Test
+ void testActiveCheckpointTriggerSkipsWhenCheckpointInProgress() throws
Exception {
+ try (MockExecutingContext ctx = new MockExecutingContext()) {
+ MockExecutionJobVertex mejv = new
MockExecutionJobVertex(MockExecutionVertex::new);
+
+ ManuallyTriggeredScheduledExecutor checkpointTimer =
+ new ManuallyTriggeredScheduledExecutor();
+ CheckpointCoordinator coordinator =
+ new
CheckpointCoordinatorTestingUtils.CheckpointCoordinatorBuilder()
+ .setTimer(checkpointTimer)
+ .build(EXECUTOR_EXTENSION.getExecutor());
+ StateTrackingMockExecutionGraph graph =
+ new StateTrackingMockExecutionGraph() {
+ @Nullable
+ @Override
+ public CheckpointCoordinator
getCheckpointCoordinator() {
+ return coordinator;
+ }
+
+ @Override
+ public Iterable<ExecutionJobVertex>
getVerticesTopologically() {
+ return Collections.singletonList(mejv);
+ }
+ };
+
+ // Parallelism +1 and periodic configured so the only blocker is
the in-progress
+ // check at request time.
+ ctx.setVertexParallelism(
+ new VertexParallelism(
+ graph.getAllVertices().values().stream()
+ .collect(
+ Collectors.toMap(
+
AccessExecutionJobVertex::getJobVertexId,
+ v -> v.getParallelism() +
1))));
+
+ Executing exec =
+ new ExecutingStateBuilder()
+ .setExecutionGraph(graph)
+ .setActiveCheckpointTriggerEnabled(true)
+ .build(ctx);
+ coordinator.triggerCheckpoint(false);
+ checkpointTimer.triggerAll();
+
+ int pendingBefore = coordinator.getNumberOfPendingCheckpoints();
+ assertThat(pendingBefore).isGreaterThan(0);
+
+ ManuallyTriggeredComponentMainThreadExecutor ctxExecutor =
+ (ManuallyTriggeredComponentMainThreadExecutor)
ctx.getMainThreadExecutor();
+ int scheduledBefore =
ctxExecutor.getActiveNonPeriodicScheduledTask().size();
+ exec.requestActiveCheckpointTrigger();
+ // The in-progress check runs in the request path and must
short-circuit before
+ // scheduling a retry — otherwise the test passes for the wrong
reason (retry would
+ // also see in-progress and drop).
+ assertThat(ctxExecutor.getActiveNonPeriodicScheduledTask())
+ .as(
+ "No retry should be scheduled when a checkpoint is
already in progress at request time")
+ .hasSize(scheduledBefore);
+ checkpointTimer.triggerAll();
+ assertThat(coordinator.getNumberOfPendingCheckpoints())
+ .as(
+ "No additional checkpoint should be triggered when
a checkpoint is already in progress")
+ .isEqualTo(pendingBefore);
+ }
+ }
+
+ @Test
+ void testActiveCheckpointTriggerFiresWhenAllGuardsPass() throws Exception {
+ try (MockExecutingContext ctx = new MockExecutingContext()) {
+ MockExecutionJobVertex mejv = new
MockExecutionJobVertex(MockExecutionVertex::new);
+
+ ManuallyTriggeredScheduledExecutor checkpointTimer =
+ new ManuallyTriggeredScheduledExecutor();
+ CheckpointCoordinator coordinator =
+ new
CheckpointCoordinatorTestingUtils.CheckpointCoordinatorBuilder()
+ .setTimer(checkpointTimer)
+ .build(EXECUTOR_EXTENSION.getExecutor());
+ StateTrackingMockExecutionGraph graph =
+ new StateTrackingMockExecutionGraph() {
+ @Nullable
+ @Override
+ public CheckpointCoordinator
getCheckpointCoordinator() {
+ return coordinator;
+ }
+
+ @Override
+ public Iterable<ExecutionJobVertex>
getVerticesTopologically() {
+ return Collections.singletonList(mejv);
+ }
+ };
+ ctx.setVertexParallelism(
+ new VertexParallelism(
+ graph.getAllVertices().values().stream()
+ .collect(
+ Collectors.toMap(
+
AccessExecutionJobVertex::getJobVertexId,
+ v -> v.getParallelism() +
1))));
+
+ Executing exec =
+ new ExecutingStateBuilder()
+ .setExecutionGraph(graph)
+ .setActiveCheckpointTriggerEnabled(true)
+ .build(ctx);
+
+
assertThat(coordinator.getNumberOfPendingCheckpoints()).isEqualTo(0);
+
+ ManuallyTriggeredComponentMainThreadExecutor ctxExecutor =
+ (ManuallyTriggeredComponentMainThreadExecutor)
ctx.getMainThreadExecutor();
+ exec.requestActiveCheckpointTrigger();
+ ctxExecutor.triggerNonPeriodicScheduledTasks();
+ checkpointTimer.triggerAll();
+
assertThat(coordinator.getNumberOfPendingCheckpoints()).isGreaterThan(0);
+ }
+ }
+
+ @Test
+ void testActiveCheckpointTriggerRespectsMinPauseBetweenCheckpoints()
throws Exception {
+ try (MockExecutingContext ctx = new MockExecutingContext()) {
+ MockExecutionJobVertex mejv = new
MockExecutionJobVertex(MockExecutionVertex::new);
+
+ ManuallyTriggeredScheduledExecutor checkpointTimer =
+ new ManuallyTriggeredScheduledExecutor();
+ ManualClock clock = new ManualClock();
+ CheckpointCoordinatorConfiguration coordConfig =
+ new CheckpointCoordinatorConfiguration
+
.CheckpointCoordinatorConfigurationBuilder()
+ .setCheckpointInterval(10_000L)
+ .setMinPauseBetweenCheckpoints(10_000L)
+ .setMaxConcurrentCheckpoints(Integer.MAX_VALUE)
+ .build();
+
+ CheckpointCoordinator coordinator =
+ new
CheckpointCoordinatorTestingUtils.CheckpointCoordinatorBuilder()
+ .setCheckpointCoordinatorConfiguration(coordConfig)
+ .setTimer(checkpointTimer)
+ .setClock(clock)
+ .build(EXECUTOR_EXTENSION.getExecutor());
+
+ StateTrackingMockExecutionGraph graph =
+ new StateTrackingMockExecutionGraph() {
+ @Nullable
+ @Override
+ public CheckpointCoordinator
getCheckpointCoordinator() {
+ return coordinator;
+ }
+
+ @Override
+ public Iterable<ExecutionJobVertex>
getVerticesTopologically() {
+ return Collections.singletonList(mejv);
+ }
+ };
+ ctx.setVertexParallelism(
+ new VertexParallelism(
+ graph.getAllVertices().values().stream()
+ .collect(
+ Collectors.toMap(
+
AccessExecutionJobVertex::getJobVertexId,
+ v -> v.getParallelism() +
1))));
+
+ Executing exec =
+ new ExecutingStateBuilder()
+ .setExecutionGraph(graph)
+ .setActiveCheckpointTriggerEnabled(true)
+ .build(ctx);
+ ManuallyTriggeredComponentMainThreadExecutor ctxExecutor =
+ (ManuallyTriggeredComponentMainThreadExecutor)
ctx.getMainThreadExecutor();
+ exec.requestActiveCheckpointTrigger();
+ ctxExecutor.triggerNonPeriodicScheduledTasks();
+ checkpointTimer.triggerAll();
+
assertThat(coordinator.getNumberOfPendingCheckpoints()).isEqualTo(0);
+ clock.advanceTime(10_000L,
java.util.concurrent.TimeUnit.MILLISECONDS);
+ exec.requestActiveCheckpointTrigger();
+ ctxExecutor.triggerNonPeriodicScheduledTasks();
+ checkpointTimer.triggerAll();
+
assertThat(coordinator.getNumberOfPendingCheckpoints()).isEqualTo(1);
+ }
+ }
+
+ @Test
+ void testActiveCheckpointTriggerDeduplicatesScheduledRetries() throws
Exception {
+ try (MockExecutingContext ctx = new MockExecutingContext()) {
+ MockExecutionJobVertex mejv = new
MockExecutionJobVertex(MockExecutionVertex::new);
+
+ ManuallyTriggeredScheduledExecutor checkpointTimer =
+ new ManuallyTriggeredScheduledExecutor();
+ ManualClock clock = new ManualClock();
+ CheckpointCoordinatorConfiguration coordConfig =
+ new CheckpointCoordinatorConfiguration
+
.CheckpointCoordinatorConfigurationBuilder()
+ .setCheckpointInterval(10_000L)
+ .setMinPauseBetweenCheckpoints(10_000L)
+ .setMaxConcurrentCheckpoints(Integer.MAX_VALUE)
+ .build();
+
+ CheckpointCoordinator coordinator =
+ new
CheckpointCoordinatorTestingUtils.CheckpointCoordinatorBuilder()
+ .setCheckpointCoordinatorConfiguration(coordConfig)
+ .setTimer(checkpointTimer)
+ .setClock(clock)
+ .build(EXECUTOR_EXTENSION.getExecutor());
+
+ StateTrackingMockExecutionGraph graph =
+ new StateTrackingMockExecutionGraph() {
+ @Nullable
+ @Override
+ public CheckpointCoordinator
getCheckpointCoordinator() {
+ return coordinator;
+ }
+
+ @Override
+ public Iterable<ExecutionJobVertex>
getVerticesTopologically() {
+ return Collections.singletonList(mejv);
+ }
+ };
+ ctx.setVertexParallelism(
+ new VertexParallelism(
+ graph.getAllVertices().values().stream()
+ .collect(
+ Collectors.toMap(
+
AccessExecutionJobVertex::getJobVertexId,
+ v -> v.getParallelism() +
1))));
+
+ Executing exec =
+ new ExecutingStateBuilder()
+ .setExecutionGraph(graph)
+ .setActiveCheckpointTriggerEnabled(true)
+ .build(ctx);
+ ManuallyTriggeredComponentMainThreadExecutor ctxExecutor =
+ (ManuallyTriggeredComponentMainThreadExecutor)
ctx.getMainThreadExecutor();
+ int baseline =
ctxExecutor.getActiveNonPeriodicScheduledTask().size();
+
+ // Call requestActiveCheckpointTrigger multiple times while
min-pause is not satisfied.
+ // Only one delayed retry should be scheduled (dedup via
+ // activeCheckpointTriggerScheduled
+ // flag).
+ exec.requestActiveCheckpointTrigger();
+ exec.requestActiveCheckpointTrigger();
+ exec.requestActiveCheckpointTrigger();
+
+
assertThat(ctxExecutor.getActiveNonPeriodicScheduledTask()).hasSize(baseline +
1);
+
+ // Advance clock past the min-pause and fire the single scheduled
retry.
+ clock.advanceTime(10_000L,
java.util.concurrent.TimeUnit.MILLISECONDS);
+ ctxExecutor.triggerNonPeriodicScheduledTasks();
+ checkpointTimer.triggerAll();
+
assertThat(coordinator.getNumberOfPendingCheckpoints()).isEqualTo(1);
+ }
+ }
+
+ @Test
+ void testActiveCheckpointTriggerHandlesFailureGracefully() throws
Exception {
+ try (MockExecutingContext ctx = new MockExecutingContext()) {
+ MockExecutionJobVertex mejv = new
MockExecutionJobVertex(MockExecutionVertex::new);
+ ManuallyTriggeredScheduledExecutor checkpointTimer =
+ new ManuallyTriggeredScheduledExecutor();
+ CheckpointCoordinator coordinator =
+ new
CheckpointCoordinatorTestingUtils.CheckpointCoordinatorBuilder()
+ .setTimer(checkpointTimer)
+ .build(EXECUTOR_EXTENSION.getExecutor());
+ coordinator.shutdown();
+
+ StateTrackingMockExecutionGraph graph =
+ new StateTrackingMockExecutionGraph() {
+ @Nullable
+ @Override
+ public CheckpointCoordinator
getCheckpointCoordinator() {
+ return coordinator;
+ }
+
+ @Override
+ public Iterable<ExecutionJobVertex>
getVerticesTopologically() {
+ return Collections.singletonList(mejv);
+ }
+ };
+ ctx.setVertexParallelism(
+ new VertexParallelism(
+ graph.getAllVertices().values().stream()
+ .collect(
+ Collectors.toMap(
+
AccessExecutionJobVertex::getJobVertexId,
+ v -> v.getParallelism() +
1))));
+
+ Executing exec =
+ new ExecutingStateBuilder()
+ .setExecutionGraph(graph)
+ .setActiveCheckpointTriggerEnabled(true)
+ .build(ctx);
+ ManuallyTriggeredComponentMainThreadExecutor ctxExecutor =
+ (ManuallyTriggeredComponentMainThreadExecutor)
ctx.getMainThreadExecutor();
+ ctxExecutor.triggerNonPeriodicScheduledTasks();
+ int baseline =
ctxExecutor.getActiveNonPeriodicScheduledTask().size();
+ exec.requestActiveCheckpointTrigger();
+ ctxExecutor.triggerNonPeriodicScheduledTasks();
+ checkpointTimer.triggerAll();
+
+
assertThat(coordinator.getNumberOfPendingCheckpoints()).isEqualTo(0);
+
assertThat(ctxExecutor.getActiveNonPeriodicScheduledTask()).hasSize(baseline);
+ }
+ }
+
@Test
void testJobInformationMethods() throws Exception {
try (MockExecutingContext ctx = new MockExecutingContext()) {
@@ -691,6 +1255,7 @@ class ExecutingTest {
private Function<StateTransitionManager.Context,
StateTransitionManager>
stateTransitionManagerFactory = context ->
TestingStateTransitionManager.withNoOp();
private int rescaleOnFailedCheckpointCount = 1;
+ private boolean activeCheckpointTriggerEnabled = false;
private ExecutingStateBuilder() throws JobException,
JobExecutionException {
operatorCoordinatorHandler = new
TestingOperatorCoordinatorHandler();
@@ -720,6 +1285,12 @@ class ExecutingTest {
return this;
}
+ public ExecutingStateBuilder setActiveCheckpointTriggerEnabled(
+ boolean activeCheckpointTriggerEnabled) {
+ this.activeCheckpointTriggerEnabled =
activeCheckpointTriggerEnabled;
+ return this;
+ }
+
private Executing build(MockExecutingContext ctx) {
executionGraph.transitionToRunning();
@@ -733,7 +1304,8 @@ class ExecutingTest {
ClassLoader.getSystemClassLoader(),
new ArrayList<>(),
stateTransitionManagerFactory::apply,
- rescaleOnFailedCheckpointCount);
+ rescaleOnFailedCheckpointCount,
+ activeCheckpointTriggerEnabled);
} finally {
Preconditions.checkState(
!ctx.hadStateTransition,
@@ -1029,6 +1601,12 @@ class ExecutingTest {
public Iterable<ExecutionJobVertex> getVerticesTopologically() {
return getVerticesTopologicallySupplier.get();
}
+
+ @Nullable
+ @Override
+ public CheckpointCoordinator getCheckpointCoordinator() {
+ return null;
+ }
}
private static class FinishingMockExecutionGraph extends
StateTrackingMockExecutionGraph {
diff --git
a/flink-tests/src/test/java/org/apache/flink/test/scheduling/ActiveCheckpointToRescaleITCase.java
b/flink-tests/src/test/java/org/apache/flink/test/scheduling/ActiveCheckpointToRescaleITCase.java
new file mode 100644
index 00000000000..af6456ffc1c
--- /dev/null
+++
b/flink-tests/src/test/java/org/apache/flink/test/scheduling/ActiveCheckpointToRescaleITCase.java
@@ -0,0 +1,198 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.test.scheduling;
+
+import org.apache.flink.api.common.JobID;
+import org.apache.flink.client.program.rest.RestClusterClient;
+import org.apache.flink.configuration.CheckpointingOptions;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.configuration.JobManagerOptions;
+import org.apache.flink.configuration.WebOptions;
+import org.apache.flink.runtime.jobgraph.JobGraph;
+import org.apache.flink.runtime.jobgraph.JobResourceRequirements;
+import org.apache.flink.runtime.jobgraph.JobVertex;
+import org.apache.flink.runtime.jobgraph.JobVertexID;
+import org.apache.flink.runtime.minicluster.MiniCluster;
+import org.apache.flink.runtime.testutils.MiniClusterResourceConfiguration;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.streaming.api.functions.sink.v2.DiscardingSink;
+import org.apache.flink.test.junit5.InjectClusterClient;
+import org.apache.flink.test.junit5.InjectMiniCluster;
+import org.apache.flink.test.junit5.MiniClusterExtension;
+import org.apache.flink.testutils.TestingUtils;
+import org.apache.flink.util.TestLoggerExtension;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.Timeout;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.junit.jupiter.api.extension.RegisterExtension;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.time.Duration;
+import java.util.Iterator;
+
+import static
org.apache.flink.test.scheduling.UpdateJobResourceRequirementsITCase.waitForAvailableSlots;
+import static
org.apache.flink.test.scheduling.UpdateJobResourceRequirementsITCase.waitForRunningTasks;
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * Tests that the Adaptive Scheduler actively triggers a checkpoint to
expedite a desired rescale,
+ * controlled by {@link
JobManagerOptions#SCHEDULER_RESCALE_TRIGGER_ACTIVE_CHECKPOINT_ENABLED}.
+ *
+ * <p>The cluster is configured so that no periodic checkpoint can complete
during a test (the
+ * checkpointing interval is effectively infinite). Consequently the only
checkpoint that can be
+ * completed - and therefore the only event that can drive a rescale of a
checkpointing job - is the
+ * one actively triggered by the scheduler. With the feature disabled the
rescale never happens and
+ * the test times out.
+ */
+@ExtendWith(TestLoggerExtension.class)
+class ActiveCheckpointToRescaleITCase {
+
+ private static final Logger LOG =
+ LoggerFactory.getLogger(ActiveCheckpointToRescaleITCase.class);
+ private static final int NUMBER_OF_SLOTS = 4;
+ private static final int BEFORE_RESCALE_PARALLELISM = NUMBER_OF_SLOTS;
+ private static final int AFTER_RESCALE_PARALLELISM = NUMBER_OF_SLOTS - 1;
+ private static final Duration REQUIREMENT_UPDATE_TO_CHECKPOINT_GAP =
Duration.ofSeconds(2);
+
+ @RegisterExtension
+ private static final MiniClusterExtension MINI_CLUSTER_EXTENSION =
+ new MiniClusterExtension(
+ new MiniClusterResourceConfiguration.Builder()
+ .setConfiguration(createConfiguration())
+ .setNumberSlotsPerTaskManager(NUMBER_OF_SLOTS)
+ .build());
+
+ private static Configuration createConfiguration() {
+ final Configuration configuration = new Configuration();
+ configuration.set(JobManagerOptions.SCHEDULER,
JobManagerOptions.SchedulerType.Adaptive);
+ configuration.set(WebOptions.REFRESH_INTERVAL, Duration.ofMillis(50L));
+ configuration.set(JobManagerOptions.SLOT_IDLE_TIMEOUT,
Duration.ofMillis(50L));
+ configuration.set(
+ CheckpointingOptions.CHECKPOINTING_INTERVAL,
TestingUtils.infiniteDuration());
+ configuration.set(
+ JobManagerOptions.SCHEDULER_RESCALE_TRIGGER_MAX_DELAY,
+ TestingUtils.infiniteDuration());
+
+ configuration.set(
+
JobManagerOptions.SCHEDULER_EXECUTING_COOLDOWN_AFTER_RESCALING, Duration.ZERO);
+
+ // the feature under test
+ configuration.set(
+
JobManagerOptions.SCHEDULER_RESCALE_TRIGGER_ACTIVE_CHECKPOINT_ENABLED, true);
+
+ return configuration;
+ }
+
+ @Test
+ @Timeout(120000)
+ void testRescaleWithActiveCheckpointTrigger(
+ @InjectMiniCluster MiniCluster miniCluster,
+ @InjectClusterClient RestClusterClient<?> restClusterClient)
+ throws Exception {
+ final Configuration config = new Configuration();
+
+ final StreamExecutionEnvironment env =
+ StreamExecutionEnvironment.getExecutionEnvironment(config);
+ env.setParallelism(BEFORE_RESCALE_PARALLELISM);
+ env.enableCheckpointing(TestingUtils.infiniteDuration().toMillis());
+ env.fromSequence(0, Integer.MAX_VALUE).sinkTo(new DiscardingSink<>());
+
+ final JobGraph jobGraph = env.getStreamGraph().getJobGraph();
+ System.out.println(
+ "DIAGNOSTIC effective checkpoint interval (ms) = "
+ + jobGraph.getCheckpointingSettings()
+ .getCheckpointCoordinatorConfiguration()
+ .getCheckpointInterval());
+ final Iterator<JobVertex> jobVertexIterator =
jobGraph.getVertices().iterator();
+ assertThat(jobVertexIterator.hasNext()).isTrue();
+ final JobVertexID jobVertexId = jobVertexIterator.next().getID();
+
+ final JobResourceRequirements jobResourceRequirements =
+ JobResourceRequirements.newBuilder()
+ .setParallelismForJobVertex(jobVertexId, 1,
AFTER_RESCALE_PARALLELISM)
+ .build();
+
+ restClusterClient.submitJob(jobGraph).join();
+
+ final JobID jobId = jobGraph.getJobID();
+ try {
+ LOG.info(
+ "Waiting for job {} to reach parallelism of {} for vertex
{}.",
+ jobId,
+ BEFORE_RESCALE_PARALLELISM,
+ jobVertexId);
+ waitForRunningTasks(restClusterClient, jobId,
BEFORE_RESCALE_PARALLELISM);
+
+ LOG.info(
+ "Updating job {} resource requirements: parallelism {} ->
{}.",
+ jobId,
+ BEFORE_RESCALE_PARALLELISM,
+ AFTER_RESCALE_PARALLELISM);
+ restClusterClient.updateJobResourceRequirements(jobId,
jobResourceRequirements).join();
+ LOG.info(
+ "Waiting for job {} to rescale to parallelism {} via
active checkpoint trigger.",
+ jobId,
+ AFTER_RESCALE_PARALLELISM);
+ waitForRunningTasks(restClusterClient, jobId,
AFTER_RESCALE_PARALLELISM);
+ final int expectedFreeSlotCount = NUMBER_OF_SLOTS -
AFTER_RESCALE_PARALLELISM;
+ LOG.info(
+ "Waiting for {} slot(s) to become available after scale
down.",
+ expectedFreeSlotCount);
+ waitForAvailableSlots(restClusterClient, expectedFreeSlotCount);
+ } finally {
+ restClusterClient.cancel(jobGraph.getJobID()).join();
+ }
+ }
+
+ @Test
+ @Timeout(12000)
+ void testNoRescaleWithoutCheckpointingConfigured(
+ @InjectMiniCluster MiniCluster miniCluster,
+ @InjectClusterClient RestClusterClient<?> restClusterClient)
+ throws Exception {
+ final Configuration config = new Configuration();
+ final StreamExecutionEnvironment env =
+ StreamExecutionEnvironment.getExecutionEnvironment(config);
+ env.setParallelism(BEFORE_RESCALE_PARALLELISM);
+ env.fromSequence(0, Integer.MAX_VALUE).sinkTo(new DiscardingSink<>());
+
+ final JobGraph jobGraph = env.getStreamGraph().getJobGraph();
+ final Iterator<JobVertex> jobVertexIterator =
jobGraph.getVertices().iterator();
+ assertThat(jobVertexIterator.hasNext()).isTrue();
+ final JobVertexID jobVertexId = jobVertexIterator.next().getID();
+
+ final JobResourceRequirements jobResourceRequirements =
+ JobResourceRequirements.newBuilder()
+ .setParallelismForJobVertex(jobVertexId, 1,
AFTER_RESCALE_PARALLELISM)
+ .build();
+ restClusterClient.submitJob(jobGraph).join();
+ final JobID jobId = jobGraph.getJobID();
+ try {
+ waitForRunningTasks(restClusterClient, jobId,
BEFORE_RESCALE_PARALLELISM);
+ restClusterClient.updateJobResourceRequirements(jobId,
jobResourceRequirements).join();
+ Thread.sleep(REQUIREMENT_UPDATE_TO_CHECKPOINT_GAP.toMillis());
+ waitForRunningTasks(restClusterClient, jobId,
BEFORE_RESCALE_PARALLELISM);
+ LOG.info("Verified: job {} did not rescale without checkpointing
configured.", jobId);
+ } finally {
+ restClusterClient.cancel(jobGraph.getJobID()).join();
+ }
+ }
+}
