pnowojski commented on code in PR #27921:
URL: https://github.com/apache/flink/pull/27921#discussion_r3077862845
##########
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)
Review Comment:
Is there a downside of using this option? If we expect this to be generally
positive change, and you disable it by default only as a pre-caution/for
backward compatibility, I would be actually fine setting it by default to
`true`.
##########
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() {
+ 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;
+ }
+
+ if (!parallelismChanged()) {
+ getLogger()
+ .debug(
+ "Skipping active checkpoint trigger for rescale:
parallelism unchanged.");
+ return;
+ }
+
+ if (checkpointCoordinator.getNumberOfPendingCheckpoints() > 0
+ || checkpointCoordinator.isTriggering()) {
+ getLogger()
+ .debug(
+ "Skipping active checkpoint trigger for rescale:
checkpoint already in progress.");
+ return;
+ }
+
+ getLogger().info("Actively triggering checkpoint to expedite
rescaling.");
+ checkpointCoordinator
+ .triggerCheckpoint(false)
+ .whenComplete(
+ (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());
+ }
+ });
Review Comment:
Shouldn't we wrap this with `FutureUtils.assertNoException`?
##########
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(
+ @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(Duration.ofHours(1).toMillis());
+ 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 {
+ 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);
Review Comment:
If I understand your test, it would still pass after 1h, after the regular
periodic checkpoint is triggered after 1h, even with your new option disabled,
right?
I think you should make sure that the timeout in `waitForRunningTasks`
`waitForAvailableSlots` (or CI timeout 4h) is longer than
`env.enableCheckpointing(Duration.ofHours(1).toMillis());`. So either, decrease
the timeout in waiting to < 30 minutes, or increase checkpointing interval to
24h (CI will be killed after 4h AFAIR).
##########
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:
Have you made sure that this test is failing without your change?
Also, I don't see this test enabling
`SCHEDULER_RESCALE_TRIGGER_ACTIVE_CHECKPOINT_ENABLED` anywhere? How is it
passing right now?
--
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]
