ztison commented on code in PR #27921:
URL: https://github.com/apache/flink/pull/27921#discussion_r3085724254
##########
flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/DefaultStateTransitionManager.java:
##########
@@ -158,10 +158,16 @@ private void progressToStabilizing(Temporal
firstChangeEventTimestamp) {
resourceStabilizationTimeout,
firstChangeEventTimestamp,
maxTriggerDelay));
+ transitionContext.requestActiveCheckpointTrigger();
Review Comment:
Why this call is needed here? ISn't it enough to call it in
`org.apache.flink.runtime.scheduler.adaptive.DefaultStateTransitionManager.Stabilizing#onChange`
?
##########
flink-core/src/main/java/org/apache/flink/configuration/JobManagerOptions.java:
##########
@@ -745,6 +745,24 @@ 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(false)
+ .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 trigger if a checkpoint is already in progress.",
+
text("execution.checkpointing.min-pause"))
Review Comment:
Might probably be:
`code(CheckpointingOptions.MIN_PAUSE_BETWEEN_CHECKPOINTS.key()))`
##########
flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/Executing.java:
##########
@@ -300,6 +309,69 @@ private void initializeFailedCheckpointCountdownIfUnset() {
}
}
+ /**
+ * Actively triggers a checkpoint to expedite rescaling. Without this, the
scheduler would
+ * passively wait for the next periodic checkpoint, which could delay
rescaling significantly
+ * when checkpoint intervals are large.
+ *
+ * <p>Guard conditions:
+ *
+ * <ul>
+ * <li>Checkpointing must be configured
+ * <li>Parallelism must have actually changed
+ * <li>No checkpoint must be currently in progress or being triggered
+ * </ul>
+ */
+ private void triggerCheckpointForRescale() {
Review Comment:
Wondering what is a benefit having this additional private method? We can
move its body directly into `requestActiveCheckpointTrigger`.
##########
flink-tests/src/test/java/org/apache/flink/test/scheduling/RescaleOnCheckpointITCase.java:
##########
@@ -175,4 +175,59 @@ void testRescaleOnCheckpoint(
restClusterClient.cancel(jobGraph.getJobID()).join();
}
}
+
+ @Test
+ void testRescaleWithActiveCheckpointTrigger(
Review Comment:
+1
##########
flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/DefaultStateTransitionManager.java:
##########
@@ -158,10 +158,16 @@ private void progressToStabilizing(Temporal
firstChangeEventTimestamp) {
resourceStabilizationTimeout,
firstChangeEventTimestamp,
maxTriggerDelay));
+ transitionContext.requestActiveCheckpointTrigger();
}
private void progressToStabilized(Temporal firstChangeEventTimestamp) {
progressToPhase(new Stabilized(clock, this, firstChangeEventTimestamp,
maxTriggerDelay));
+ transitionContext.requestActiveCheckpointTrigger();
Review Comment:
Now it looks the method is called on many places. Wondering if we could
control when it is called only in Phases. So moving this to `Stabilized` phase?
##########
flink-core/src/main/java/org/apache/flink/configuration/JobManagerOptions.java:
##########
@@ -745,6 +745,24 @@ 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(false)
+ .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 trigger if a checkpoint is already in progress.",
Review Comment:
Does it really respect the `min-pause`? What I see it only respect that it
doesn't trigger a new checkpoint when once is in progress.
##########
flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/StateTransitionManager.java:
##########
@@ -89,5 +89,14 @@ 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 when
+ * the {@link DefaultStateTransitionManager} enters a phase that is
ready to accept {@link
Review Comment:
Is it true? I see that the method is called on more places: entering
Stabilizing, entering Stabilized, and on each onChange event while in
Stabilizing
##########
flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/Executing.java:
##########
@@ -300,6 +309,69 @@ private void initializeFailedCheckpointCountdownIfUnset() {
}
}
+ /**
+ * Actively triggers a checkpoint to expedite rescaling. Without this, the
scheduler would
+ * passively wait for the next periodic checkpoint, which could delay
rescaling significantly
+ * when checkpoint intervals are large.
+ *
+ * <p>Guard conditions:
+ *
+ * <ul>
+ * <li>Checkpointing must be configured
+ * <li>Parallelism must have actually changed
+ * <li>No checkpoint must be currently in progress or being triggered
+ * </ul>
+ */
+ private void triggerCheckpointForRescale() {
Review Comment:
We need to cover all possible paths by tests.
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
