ztison commented on code in PR #27921:
URL: https://github.com/apache/flink/pull/27921#discussion_r3145349638
##########
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;
+ if (!activeCheckpointTriggerEnabled) {
+ return;
+ }
Review Comment:
This should never happen and if it happens something is broken. So I would
rather use `Preconditions.checkState(activeCheckpointTriggerEnabled)` and move
the check somewhere else, e.g. to `fireActiveCheckpointTrigger` to be sure
that we never trigger checkpoint if feature is not enabled.
--
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]
