Re: [PR] [FLINK-36753][runtime]Adaptive Scheduler actively triggers a Checkpoint after all resources are ready [flink]

["XComp (via GitHub)" <gi...@apache.org>]

XComp commented on code in PR #27921:
URL: https://github.com/apache/flink/pull/27921#discussion_r3356270379


##########
flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/Executing.java:
##########
@@ -182,6 +190,106 @@ 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 (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.");
+        activeCheckpointTriggerScheduled = false;

Review Comment:
   Isn't that one obsolete? We're setting it already in the calling method. 🤔 



##########
flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinator.java:
##########
@@ -1999,6 +2000,29 @@ public long getCheckpointTimeout() {
         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.
+     *
+     * <p>All checks are made under the coordinator lock.

Review Comment:
   that's an implementation details that doesn't need to be exposed via JavaDoc



-- 
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