ztison commented on code in PR #27921:
URL: https://github.com/apache/flink/pull/27921#discussion_r3116198947
##########
flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinator.java:
##########
@@ -1996,6 +1996,16 @@ public long getCheckpointTimeout() {
return checkpointTimeout;
}
+ /**
+ * Returns {@code true} if enough time has elapsed since the last
checkpoint completion to
+ * satisfy the configured {@code minPauseBetweenCheckpoints}. This can be
used by callers that
+ * trigger non-periodic checkpoints but still wish to respect the
min-pause constraint.
+ */
+ public boolean isMinPauseBetweenCheckpointsSatisfied() {
Review Comment:
@pnowojski Is this safe without lock?
##########
flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/Executing.java:
##########
@@ -182,6 +187,74 @@ public ScheduledFuture<?> scheduleOperation(Runnable
callback, Duration delay) {
return context.runIfState(this, callback, delay);
}
+ private static final Duration ACTIVE_CHECKPOINT_RETRY_DELAY =
Duration.ofSeconds(10);
Review Comment:
Do we really need this?
##########
flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/Executing.java:
##########
@@ -182,6 +187,74 @@ public ScheduledFuture<?> scheduleOperation(Runnable
callback, Duration delay) {
return context.runIfState(this, callback, delay);
}
+ private static final Duration ACTIVE_CHECKPOINT_RETRY_DELAY =
Duration.ofSeconds(10);
+
+ @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;
+ }
+
+ 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;
+ }
+
+ if (!checkpointCoordinator.isMinPauseBetweenCheckpointsSatisfied()) {
+ getLogger()
+ .debug(
+ "Skipping active checkpoint trigger for rescale:
min pause between checkpoints not satisfied, scheduling retry.");
+ context.runIfState(
+ this, this::requestActiveCheckpointTrigger,
ACTIVE_CHECKPOINT_RETRY_DELAY);
+ return;
+ }
+
+ getLogger().info("Actively triggering checkpoint to expedite
rescaling.");
+ FutureUtils.assertNoException(
+ checkpointCoordinator
+ .triggerCheckpoint(false)
+ .handle(
+ (completedCheckpoint, throwable) -> {
+ if (throwable != null) {
+ getLogger()
+ .warn(
+ "Active checkpoint
trigger for rescale failed, scheduling retry.",
+ throwable);
+ context.runIfState(
Review Comment:
Do we really want to introduce this endless cycle? I feel we should just log
it and give it up. If the checkpoint is failing there probably is different
issue with job and we shouldn't try it again and again without any retry cap.
##########
flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/Executing.java:
##########
@@ -182,6 +187,74 @@ public ScheduledFuture<?> scheduleOperation(Runnable
callback, Duration delay) {
return context.runIfState(this, callback, delay);
}
+ private static final Duration ACTIVE_CHECKPOINT_RETRY_DELAY =
Duration.ofSeconds(10);
+
+ @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;
+ }
+
+ 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;
+ }
+
+ if (!checkpointCoordinator.isMinPauseBetweenCheckpointsSatisfied()) {
Review Comment:
If you return a remaining time to the next checkpoint instead of boolean,
then you can directly use it in following call of `context.runIfState( this,
this::requestActiveCheckpointTrigger, remainingTimeToSatisfyMinPause)` and you
can get rid of hardcoded `ACTIVE_CHECKPOINT_RETRY_DELAY` .
##########
flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/Executing.java:
##########
@@ -182,6 +187,74 @@ public ScheduledFuture<?> scheduleOperation(Runnable
callback, Duration delay) {
return context.runIfState(this, callback, delay);
}
+ private static final Duration ACTIVE_CHECKPOINT_RETRY_DELAY =
Duration.ofSeconds(10);
+
+ @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;
+ }
+
+ 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;
+ }
+
+ if (!checkpointCoordinator.isMinPauseBetweenCheckpointsSatisfied()) {
+ getLogger()
+ .debug(
+ "Skipping active checkpoint trigger for rescale:
min pause between checkpoints not satisfied, scheduling retry.");
+ context.runIfState(
Review Comment:
If we get e.g. 10 `onChange` events then we will schedule this method 10
times. We should have some kind of deduplication.
--
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]
