From 6a312269b69880ea9215dfc9ffb4b27d67a8b058 Mon Sep 17 00:00:00 2001 From: Prashant Bhardwaj Date: Thu, 21 May 2026 10:07:08 +0530 Subject: [PATCH] [FLINK-36512][runtime] Ignore non-rescale-relevant checkpoint failure reasons in adaptive rescale trigger Generated-by: Claude Code (Claude Opus 4.8) --- .../generated/all_jobmanager_section.html | 2 +- .../generated/expert_scheduling_section.html | 2 +- .../generated/job_manager_configuration.html | 2 +- .../configuration/JobManagerOptions.java | 4 +- .../checkpoint/CheckpointFailureManager.java | 51 +------ .../checkpoint/CheckpointFailureReason.java | 48 ++++++ .../checkpoint/CheckpointStatsListener.java | 11 +- .../checkpoint/CheckpointStatsTracker.java | 5 +- .../DefaultCheckpointStatsTracker.java | 6 +- .../checkpoint/FailedCheckpointStats.java | 21 ++- .../NoOpCheckpointStatsTracker.java | 3 +- .../scheduler/adaptive/AdaptiveScheduler.java | 5 +- .../runtime/scheduler/adaptive/Executing.java | 25 +++- .../checkpoint/CheckpointCoordinatorTest.java | 137 ++++++++++++++++++ .../DefaultCheckpointStatsTrackerTest.java | 6 +- .../checkpoint/FailedCheckpointStatsTest.java | 61 ++++++++ .../TestingCheckpointStatsTracker.java | 4 +- .../runtime/dispatcher/DispatcherTest.java | 4 +- .../adaptive/AdaptiveSchedulerTest.java | 4 +- .../scheduler/adaptive/ExecutingTest.java | 112 +++++++++++++- 20 files changed, 442 insertions(+), 71 deletions(-) diff --git a/docs/layouts/shortcodes/generated/all_jobmanager_section.html b/docs/layouts/shortcodes/generated/all_jobmanager_section.html index 5d01a4a4885a5d..82d5432a702c7c 100644 --- a/docs/layouts/shortcodes/generated/all_jobmanager_section.html +++ b/docs/layouts/shortcodes/generated/all_jobmanager_section.html @@ -60,7 +60,7 @@
jobmanager.adaptive-scheduler.rescale-trigger.max-checkpoint-failures
2 Integer - The number of consecutive failed checkpoints that will trigger rescaling even in the absence of a completed checkpoint. + The number of consecutive failed checkpoints that will trigger rescaling even in the absence of a completed checkpoint. Only failures that count against the tolerable checkpoint-failure threshold (e.g. IO errors, async failures, expirations, declines, finalization errors) advance this counter; other failures such as tasks not being fully running, coordinator shutdown, or checkpoint subsumption are ignored to avoid spurious rescales during startup or recovery.
jobmanager.adaptive-scheduler.rescale-trigger.max-delay
diff --git a/docs/layouts/shortcodes/generated/expert_scheduling_section.html b/docs/layouts/shortcodes/generated/expert_scheduling_section.html index d8ffc2ca3ee97d..5fc37cfa018c18 100644 --- a/docs/layouts/shortcodes/generated/expert_scheduling_section.html +++ b/docs/layouts/shortcodes/generated/expert_scheduling_section.html @@ -114,7 +114,7 @@
jobmanager.adaptive-scheduler.rescale-trigger.max-checkpoint-failures
2 Integer - The number of consecutive failed checkpoints that will trigger rescaling even in the absence of a completed checkpoint. + The number of consecutive failed checkpoints that will trigger rescaling even in the absence of a completed checkpoint. Only failures that count against the tolerable checkpoint-failure threshold (e.g. IO errors, async failures, expirations, declines, finalization errors) advance this counter; other failures such as tasks not being fully running, coordinator shutdown, or checkpoint subsumption are ignored to avoid spurious rescales during startup or recovery.
jobmanager.adaptive-scheduler.rescale-trigger.max-delay
diff --git a/docs/layouts/shortcodes/generated/job_manager_configuration.html b/docs/layouts/shortcodes/generated/job_manager_configuration.html index 75aa1a1d9c821a..e117f277154ea3 100644 --- a/docs/layouts/shortcodes/generated/job_manager_configuration.html +++ b/docs/layouts/shortcodes/generated/job_manager_configuration.html @@ -60,7 +60,7 @@
jobmanager.adaptive-scheduler.rescale-trigger.max-checkpoint-failures
2 Integer - The number of consecutive failed checkpoints that will trigger rescaling even in the absence of a completed checkpoint. + The number of consecutive failed checkpoints that will trigger rescaling even in the absence of a completed checkpoint. Only failures that count against the tolerable checkpoint-failure threshold (e.g. IO errors, async failures, expirations, declines, finalization errors) advance this counter; other failures such as tasks not being fully running, coordinator shutdown, or checkpoint subsumption are ignored to avoid spurious rescales during startup or recovery.
jobmanager.adaptive-scheduler.rescale-trigger.max-delay
diff --git a/flink-core/src/main/java/org/apache/flink/configuration/JobManagerOptions.java b/flink-core/src/main/java/org/apache/flink/configuration/JobManagerOptions.java index a185ce22f67773..5b33bf18d65ec7 100644 --- a/flink-core/src/main/java/org/apache/flink/configuration/JobManagerOptions.java +++ b/flink-core/src/main/java/org/apache/flink/configuration/JobManagerOptions.java @@ -724,7 +724,9 @@ public InlineElement getDescription() { .withDescription( Description.builder() .text( - "The number of consecutive failed checkpoints that will trigger rescaling even in the absence of a completed checkpoint.") + "The number of consecutive failed checkpoints that will trigger rescaling even in the absence of a completed checkpoint. " + + "Only failures that count against the tolerable checkpoint-failure threshold (e.g. IO errors, async failures, expirations, declines, finalization errors) advance this counter; " + + "other failures such as tasks not being fully running, coordinator shutdown, or checkpoint subsumption are ignored to avoid spurious rescales during startup or recovery.") .build()); @Documentation.Section({ diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointFailureManager.java b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointFailureManager.java index c154f0fb07b0d2..83f3df1d875b60 100644 --- a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointFailureManager.java +++ b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointFailureManager.java @@ -141,7 +141,8 @@ private void updateStatsAfterCheckpointFailed( statsTracker.reportFailedCheckpoint( pendingCheckpointStats.toFailedCheckpoint(failureTimestamp, exception)); } else { - statsTracker.reportFailedCheckpointsWithoutInProgress(); + statsTracker.reportFailedCheckpointsWithoutInProgress( + exception.getCheckpointFailureReason()); } } @@ -219,49 +220,11 @@ public void checkFailureCounter(CheckpointException exception, long checkpointId return; } - CheckpointFailureReason reason = exception.getCheckpointFailureReason(); - switch (reason) { - case PERIODIC_SCHEDULER_SHUTDOWN: - case TOO_MANY_CHECKPOINT_REQUESTS: - case MINIMUM_TIME_BETWEEN_CHECKPOINTS: - case NOT_ALL_REQUIRED_TASKS_RUNNING: - case CHECKPOINT_SUBSUMED: - case CHECKPOINT_COORDINATOR_SUSPEND: - case CHECKPOINT_COORDINATOR_SHUTDOWN: - case CHANNEL_STATE_SHARED_STREAM_EXCEPTION: - case JOB_FAILOVER_REGION: - // for compatibility purposes with user job behavior - case CHECKPOINT_DECLINED_TASK_NOT_READY: - case CHECKPOINT_DECLINED_TASK_CLOSING: - case CHECKPOINT_DECLINED_ON_CANCELLATION_BARRIER: - case CHECKPOINT_DECLINED_SUBSUMED: - case CHECKPOINT_DECLINED_INPUT_END_OF_STREAM: - - case TASK_FAILURE: - case TASK_CHECKPOINT_FAILURE: - case UNKNOWN_TASK_CHECKPOINT_NOTIFICATION_FAILURE: - // there are some edge cases shouldn't be counted as a failure, e.g. shutdown - case TRIGGER_CHECKPOINT_FAILURE: - case BLOCKING_OUTPUT_EXIST: - // ignore - break; - - case IO_EXCEPTION: - case CHECKPOINT_ASYNC_EXCEPTION: - case CHECKPOINT_DECLINED: - case CHECKPOINT_EXPIRED: - case FINALIZE_CHECKPOINT_FAILURE: - // we should make sure one checkpoint only be counted once - if (checkpointId == UNKNOWN_CHECKPOINT_ID - || countedCheckpointIds.add(checkpointId)) { - continuousFailureCounter.incrementAndGet(); - } - - break; - - default: - throw new FlinkRuntimeException( - "Unknown checkpoint failure reason : " + reason.name()); + if (exception.getCheckpointFailureReason().isCountedAgainstFailureThreshold()) { + // we should make sure one checkpoint only be counted once + if (checkpointId == UNKNOWN_CHECKPOINT_ID || countedCheckpointIds.add(checkpointId)) { + continuousFailureCounter.incrementAndGet(); + } } } diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointFailureReason.java b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointFailureReason.java index ee516815212d4c..8695296aeef1b9 100644 --- a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointFailureReason.java +++ b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointFailureReason.java @@ -18,6 +18,8 @@ package org.apache.flink.runtime.checkpoint; +import org.apache.flink.util.FlinkRuntimeException; + /** Various reasons why a checkpoint was failure. */ public enum CheckpointFailureReason { PERIODIC_SCHEDULER_SHUTDOWN(true, "Periodic checkpoint scheduler is shut down."), @@ -99,4 +101,50 @@ public String message() { public boolean isPreFlight() { return preFlight; } + + /** + * Whether a checkpoint failure with this reason counts toward the continuous checkpoint-failure + * threshold. This is the single source of truth shared by {@link + * CheckpointFailureManager#checkFailureCounter} (which fails the job once the tolerable failure + * number is exceeded) and the adaptive scheduler's rescale-on-failed-checkpoints countdown, so + * the two mechanisms cannot drift apart. Reasons such as tasks not yet running, coordinator + * shutdown, or checkpoint subsumption are not counted. + */ + public boolean isCountedAgainstFailureThreshold() { + switch (this) { + case PERIODIC_SCHEDULER_SHUTDOWN: + case TOO_MANY_CHECKPOINT_REQUESTS: + case MINIMUM_TIME_BETWEEN_CHECKPOINTS: + case NOT_ALL_REQUIRED_TASKS_RUNNING: + case CHECKPOINT_SUBSUMED: + case CHECKPOINT_COORDINATOR_SUSPEND: + case CHECKPOINT_COORDINATOR_SHUTDOWN: + case CHANNEL_STATE_SHARED_STREAM_EXCEPTION: + case JOB_FAILOVER_REGION: + // for compatibility purposes with user job behavior + case CHECKPOINT_DECLINED_TASK_NOT_READY: + case CHECKPOINT_DECLINED_TASK_CLOSING: + case CHECKPOINT_DECLINED_ON_CANCELLATION_BARRIER: + case CHECKPOINT_DECLINED_SUBSUMED: + case CHECKPOINT_DECLINED_INPUT_END_OF_STREAM: + + case TASK_FAILURE: + case TASK_CHECKPOINT_FAILURE: + case UNKNOWN_TASK_CHECKPOINT_NOTIFICATION_FAILURE: + // there are some edge cases shouldn't be counted as a failure, e.g. shutdown + case TRIGGER_CHECKPOINT_FAILURE: + case BLOCKING_OUTPUT_EXIST: + return false; + + case IO_EXCEPTION: + case CHECKPOINT_ASYNC_EXCEPTION: + case CHECKPOINT_DECLINED: + case CHECKPOINT_EXPIRED: + case FINALIZE_CHECKPOINT_FAILURE: + return true; + + default: + throw new FlinkRuntimeException("Unknown checkpoint failure reason : " + name()); + } + } } diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointStatsListener.java b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointStatsListener.java index 752e0881e716ac..62f3f1ca13b3c6 100644 --- a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointStatsListener.java +++ b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointStatsListener.java @@ -18,6 +18,8 @@ package org.apache.flink.runtime.checkpoint; +import javax.annotation.Nullable; + /** An interface that allows listening on the checkpoint lifecycle. */ public interface CheckpointStatsListener { @@ -26,8 +28,13 @@ default void onCompletedCheckpoint() { // No-op. } - /** Called when a checkpoint failed. */ - default void onFailedCheckpoint() { + /** + * Called when a checkpoint failed. + * + * @param reason the {@link CheckpointFailureReason} categorizing the failure, or {@code null} + * if the cause could not be classified (e.g. a non-{@link CheckpointException} cause). + */ + default void onFailedCheckpoint(@Nullable CheckpointFailureReason reason) { // No-op. } } diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointStatsTracker.java b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointStatsTracker.java index 71a970ddd19594..11208552442ab5 100644 --- a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointStatsTracker.java +++ b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointStatsTracker.java @@ -92,8 +92,11 @@ PendingCheckpointStats reportPendingCheckpoint( /** * Callback when a checkpoint failure without in progress checkpoint. For example, it should be * callback when triggering checkpoint failure before creating PendingCheckpoint. + * + * @param reason the {@link CheckpointFailureReason} categorizing the failure, or {@code null} + * if unknown. */ - void reportFailedCheckpointsWithoutInProgress(); + void reportFailedCheckpointsWithoutInProgress(@Nullable CheckpointFailureReason reason); /** * Creates a new snapshot of the available stats. diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/DefaultCheckpointStatsTracker.java b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/DefaultCheckpointStatsTracker.java index d9c63d9d52f43a..c46b5c5a5e65bb 100644 --- a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/DefaultCheckpointStatsTracker.java +++ b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/DefaultCheckpointStatsTracker.java @@ -334,7 +334,7 @@ public void reportFailedCheckpoint(FailedCheckpointStats failed) { logCheckpointStatistics(failed); if (checkpointStatsListener != null) { - checkpointStatsListener.onFailedCheckpoint(); + checkpointStatsListener.onFailedCheckpoint(failed.getFailureReason()); } } finally { statsReadWriteLock.unlock(); @@ -519,7 +519,7 @@ private boolean shouldSkipSumMetricNameInCheckpointSpanForCompatibility(String m } @Override - public void reportFailedCheckpointsWithoutInProgress() { + public void reportFailedCheckpointsWithoutInProgress(@Nullable CheckpointFailureReason reason) { statsReadWriteLock.lock(); try { counts.incrementFailedCheckpointsWithoutInProgress(); @@ -527,7 +527,7 @@ public void reportFailedCheckpointsWithoutInProgress() { dirty = true; if (checkpointStatsListener != null) { - checkpointStatsListener.onFailedCheckpoint(); + checkpointStatsListener.onFailedCheckpoint(reason); } } finally { statsReadWriteLock.unlock(); diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/FailedCheckpointStats.java b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/FailedCheckpointStats.java index 3f63ec22226bc5..fd3a69a89dec02 100644 --- a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/FailedCheckpointStats.java +++ b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/FailedCheckpointStats.java @@ -19,6 +19,7 @@ package org.apache.flink.runtime.checkpoint; import org.apache.flink.runtime.jobgraph.JobVertexID; +import org.apache.flink.util.ExceptionUtils; import javax.annotation.Nullable; @@ -41,6 +42,8 @@ public class FailedCheckpointStats extends PendingCheckpointStats { /** Optional failure message. */ @Nullable private final String failureMsg; + @Nullable private final CheckpointFailureReason failureReason; + /** * Creates a tracker for a failed checkpoint. * @@ -58,7 +61,9 @@ public class FailedCheckpointStats extends PendingCheckpointStats { * @param unalignedCheckpoint Whether the checkpoint is unaligned. * @param failureTimestamp Timestamp when this checkpoint failed. * @param latestAcknowledgedSubtask The latest acknowledged subtask stats or null. - * @param cause Cause of the checkpoint failure or null. + * @param cause Cause of the checkpoint failure or null. If a {@link + * CheckpointException} is found in its cause chain, that exception's {@link + * CheckpointFailureReason} is captured. */ FailedCheckpointStats( long checkpointId, @@ -92,6 +97,10 @@ public class FailedCheckpointStats extends PendingCheckpointStats { checkArgument(numAcknowledgedSubtasks >= 0, "Negative number of ACKs"); this.failureTimestamp = failureTimestamp; this.failureMsg = cause != null ? cause.getMessage() : null; + this.failureReason = + ExceptionUtils.findThrowable(cause, CheckpointException.class) + .map(CheckpointException::getCheckpointFailureReason) + .orElse(null); } @Override @@ -123,4 +132,14 @@ public long getFailureTimestamp() { public String getFailureMessage() { return failureMsg; } + + /** + * Returns the structured failure reason captured from a {@link CheckpointException} in the + * cause chain, or null if the cause was unknown or contained no {@code + * CheckpointException}. + */ + @Nullable + public CheckpointFailureReason getFailureReason() { + return failureReason; + } } diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/NoOpCheckpointStatsTracker.java b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/NoOpCheckpointStatsTracker.java index 8bd7de9daa5b4d..7e575139a40674 100644 --- a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/NoOpCheckpointStatsTracker.java +++ b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/NoOpCheckpointStatsTracker.java @@ -72,7 +72,8 @@ public PendingCheckpointStats reportPendingCheckpoint( public void reportFailedCheckpoint(FailedCheckpointStats failed) {} @Override - public void reportFailedCheckpointsWithoutInProgress() {} + public void reportFailedCheckpointsWithoutInProgress( + @Nullable CheckpointFailureReason reason) {} @Override public CheckpointStatsSnapshot createSnapshot() { diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/AdaptiveScheduler.java b/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/AdaptiveScheduler.java index b75f0d09f29e1e..c63cb1f110e79d 100644 --- a/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/AdaptiveScheduler.java +++ b/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/AdaptiveScheduler.java @@ -1827,8 +1827,9 @@ private CheckpointStatsListener createCheckpointStatsListener() { return new CheckpointStatsListener() { @Override - public void onFailedCheckpoint() { - runIfSupported(CheckpointStatsListener::onFailedCheckpoint, "onFailedCheckpoint"); + public void onFailedCheckpoint(@Nullable CheckpointFailureReason reason) { + runIfSupported( + listener -> listener.onFailedCheckpoint(reason), "onFailedCheckpoint"); } @Override diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/Executing.java b/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/Executing.java index d0ecf284b7914e..ca14e28b5d61c7 100644 --- a/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/Executing.java +++ b/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/Executing.java @@ -23,6 +23,7 @@ import org.apache.flink.core.execution.SavepointFormatType; import org.apache.flink.runtime.JobException; import org.apache.flink.runtime.checkpoint.CheckpointCoordinator; +import org.apache.flink.runtime.checkpoint.CheckpointFailureReason; import org.apache.flink.runtime.checkpoint.CheckpointScheduling; import org.apache.flink.runtime.checkpoint.CheckpointStatsListener; import org.apache.flink.runtime.checkpoint.CompletedCheckpoint; @@ -389,13 +390,35 @@ public void onCompletedCheckpoint() { } @Override - public void onFailedCheckpoint() { + public void onFailedCheckpoint(@Nullable CheckpointFailureReason reason) { + if (!isRescaleRelevantFailure(reason)) { + return; + } + // The countdown advances once per counted failure notification; unlike + // CheckpointFailureManager it does not dedup by checkpoint id. This is acceptable because + // the countdown is reset (to null) on the next completed checkpoint or rescale trigger, so + // at most one spurious decrement could ever accumulate within a single countdown window. if (this.failedCheckpointCountdown != null && this.failedCheckpointCountdown.decrementAndGet() <= 0) { triggerPotentialRescale(); } } + /** + * Whether a failed checkpoint with the given reason should advance the + * rescale-on-failed-checkpoints countdown. Delegates to {@link + * CheckpointFailureReason#isCountedAgainstFailureThreshold()} — the single source of truth + * shared with {@link + * org.apache.flink.runtime.checkpoint.CheckpointFailureManager#checkFailureCounter} — so the + * countdown stays consistent with the job-level failure counter. A {@code null} reason carries + * no such classification and is ignored; no production notification path yields a countable + * {@code null} reason, as a failure with an in-progress checkpoint always originates from a + * {@link org.apache.flink.runtime.checkpoint.CheckpointException}. + */ + private static boolean isRescaleRelevantFailure(@Nullable CheckpointFailureReason reason) { + return reason != null && reason.isCountedAgainstFailureThreshold(); + } + private void triggerPotentialRescale() { stateTransitionManager.onTrigger(); this.failedCheckpointCountdown = null; diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinatorTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinatorTest.java index 2b5e6653dc3da3..f0ff8b7efada8f 100644 --- a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinatorTest.java +++ b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinatorTest.java @@ -21,6 +21,7 @@ import org.apache.flink.api.common.JobID; import org.apache.flink.api.common.JobStatus; import org.apache.flink.api.java.tuple.Tuple2; +import org.apache.flink.configuration.TraceOptions; import org.apache.flink.core.execution.RecoveryClaimMode; import org.apache.flink.core.execution.SavepointFormatType; import org.apache.flink.core.fs.FSDataInputStream; @@ -457,6 +458,142 @@ void testCheckpointAbortsIfTriggerTasksAreNotExecuted() throws Exception { checkpointCoordinator.shutdown(); } + // ------------------------------------------------------------------------ + // Checkpoint failure reason delivery to the CheckpointStatsListener. + // + // Integration coverage for FLINK-36512: the adaptive scheduler's + // rescale-on-failed-checkpoints countdown consumes the reason handed to + // CheckpointStatsListener#onFailedCheckpoint and must only advance for + // reasons attributable to in-flight checkpoint execution. These tests + // drive a real CheckpointCoordinator so the reason under assertion is the + // one production code actually generates (rather than a hand-supplied one), + // covering both listener paths: reportFailedCheckpointsWithoutInProgress + // (pre-flight / IO trigger failures) and reportFailedCheckpoint (declines). + // ------------------------------------------------------------------------ + + @Test + void testPreFlightFailureReasonDeliveredToStatsListenerIsNotRescaleRelevant() throws Exception { + // A checkpoint triggered while not all tasks are running fails before any + // PendingCheckpoint exists and must surface NOT_ALL_REQUIRED_TASKS_RUNNING, which the + // adaptive scheduler ignores for rescaling. + ExecutionGraph graph = + new CheckpointCoordinatorTestingUtils.CheckpointExecutionGraphBuilder() + .addJobVertex(new JobVertexID()) + .setTransitToRunning(false) + .build(EXECUTOR_RESOURCE.getExecutor()); + + List observedFailureReasons = new ArrayList<>(); + CheckpointCoordinator coordinator = + createCoordinatorWithFailureReasonListener(graph, null, observedFailureReasons); + + CompletableFuture future = coordinator.triggerCheckpoint(false); + manuallyTriggeredScheduledExecutor.triggerAll(); + + assertThat(future).isCompletedExceptionally(); + assertThat(observedFailureReasons) + .containsExactly(CheckpointFailureReason.NOT_ALL_REQUIRED_TASKS_RUNNING); + assertThat( + CheckpointFailureReason.NOT_ALL_REQUIRED_TASKS_RUNNING + .isCountedAgainstFailureThreshold()) + .as("Pre-flight failures must not advance the adaptive rescale countdown.") + .isFalse(); + + coordinator.shutdown(); + } + + @Test + void testIOExceptionFailureReasonDeliveredToStatsListenerIsRescaleRelevant() throws Exception { + ExecutionGraph graph = + new CheckpointCoordinatorTestingUtils.CheckpointExecutionGraphBuilder() + .addJobVertex(new JobVertexID()) + .build(EXECUTOR_RESOURCE.getExecutor()); + + List observedFailureReasons = new ArrayList<>(); + CheckpointCoordinator coordinator = + createCoordinatorWithFailureReasonListener( + graph, new IOExceptionCheckpointStorage(), observedFailureReasons); + + CompletableFuture future = coordinator.triggerCheckpoint(false); + manuallyTriggeredScheduledExecutor.triggerAll(); + + assertThat(future).isCompletedExceptionally(); + assertThat(observedFailureReasons).containsExactly(IO_EXCEPTION); + assertThat(IO_EXCEPTION.isCountedAgainstFailureThreshold()) + .as("In-flight IO failures must advance the adaptive rescale countdown.") + .isTrue(); + + coordinator.shutdown(); + } + + @Test + void testDeclineFailureReasonDeliveredToStatsListenerIsRescaleRelevant() throws Exception { + JobVertexID jobVertexID = new JobVertexID(); + ExecutionGraph graph = + new CheckpointCoordinatorTestingUtils.CheckpointExecutionGraphBuilder() + .addJobVertex(jobVertexID) + .build(EXECUTOR_RESOURCE.getExecutor()); + ExecutionAttemptID attemptID = + graph.getJobVertex(jobVertexID) + .getTaskVertices()[0] + .getCurrentExecutionAttempt() + .getAttemptId(); + + List observedFailureReasons = new ArrayList<>(); + CheckpointCoordinator coordinator = + createCoordinatorWithFailureReasonListener(graph, null, observedFailureReasons); + + // A declined in-flight checkpoint reaches the listener via the reportFailedCheckpoint + // path (a PendingCheckpoint exists). + CompletableFuture future = coordinator.triggerCheckpoint(false); + manuallyTriggeredScheduledExecutor.triggerAll(); + FutureUtils.throwIfCompletedExceptionally(future); + + long checkpointId = + coordinator.getPendingCheckpoints().entrySet().iterator().next().getKey(); + coordinator.receiveDeclineMessage( + new DeclineCheckpoint( + graph.getJobID(), + attemptID, + checkpointId, + new CheckpointException(CHECKPOINT_DECLINED)), + TASK_MANAGER_LOCATION_INFO); + + assertThat(observedFailureReasons).containsExactly(CHECKPOINT_DECLINED); + assertThat(CHECKPOINT_DECLINED.isCountedAgainstFailureThreshold()) + .as("Declined checkpoints must advance the adaptive rescale countdown.") + .isTrue(); + + coordinator.shutdown(); + } + + private CheckpointCoordinator createCoordinatorWithFailureReasonListener( + ExecutionGraph graph, + @Nullable CheckpointStorage checkpointStorage, + List observedFailureReasons) + throws Exception { + CheckpointStatsListener listener = + new CheckpointStatsListener() { + @Override + public void onFailedCheckpoint(@Nullable CheckpointFailureReason reason) { + observedFailureReasons.add(reason); + } + }; + DefaultCheckpointStatsTracker statsTracker = + new DefaultCheckpointStatsTracker( + Integer.MAX_VALUE, + UnregisteredMetricGroups.createUnregisteredJobManagerJobMetricGroup(), + TraceOptions.CheckpointSpanDetailLevel.SPAN_PER_CHECKPOINT, + listener); + CheckpointCoordinatorBuilder builder = + new CheckpointCoordinatorBuilder() + .setTimer(manuallyTriggeredScheduledExecutor) + .setCheckpointStatsTracker(statsTracker); + if (checkpointStorage != null) { + builder.setCheckpointStorage(checkpointStorage); + } + return builder.build(graph); + } + @Test void testCheckpointAbortsIfTriggerTasksAreFinished() throws Exception { JobVertexID jobVertexID1 = new JobVertexID(); diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/DefaultCheckpointStatsTrackerTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/DefaultCheckpointStatsTrackerTest.java index b3bbe4b3d9b02e..0196d1b96a3401 100644 --- a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/DefaultCheckpointStatsTrackerTest.java +++ b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/DefaultCheckpointStatsTrackerTest.java @@ -40,6 +40,8 @@ import org.junit.jupiter.api.Test; import org.junit.jupiter.api.extension.RegisterExtension; +import javax.annotation.Nullable; + import java.util.ArrayList; import java.util.Arrays; import java.util.Collection; @@ -194,7 +196,7 @@ void testCheckpointTracking() throws Exception { assertThat(counts.getNumberOfCompletedCheckpoints()).isEqualTo(2); assertThat(counts.getNumberOfFailedCheckpoints()).isOne(); - tracker.reportFailedCheckpointsWithoutInProgress(); + tracker.reportFailedCheckpointsWithoutInProgress(null); CheckpointStatsSnapshot snapshot1 = tracker.createSnapshot(); counts = snapshot1.getCounts(); @@ -280,7 +282,7 @@ public void onCompletedCheckpoint() { } @Override - public void onFailedCheckpoint() { + public void onFailedCheckpoint(@Nullable CheckpointFailureReason reason) { onFailedCheckpointCount.incrementAndGet(); } }; diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/FailedCheckpointStatsTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/FailedCheckpointStatsTest.java index 38cc9c850ba305..b0d4263b8ae69b 100644 --- a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/FailedCheckpointStatsTest.java +++ b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/FailedCheckpointStatsTest.java @@ -23,6 +23,8 @@ import org.junit.jupiter.api.Test; +import javax.annotation.Nullable; + import java.io.NotSerializableException; import java.util.HashMap; import java.util.Map; @@ -111,4 +113,63 @@ void testIsJavaSerializable() throws Exception { assertThat(copy.getStatus()).isEqualTo(failed.getStatus()); assertThat(copy.getFailureMessage()).isEqualTo(failed.getFailureMessage()); } + + @Test + void testFailureReasonExtractedFromCheckpointException() { + FailedCheckpointStats failed = + newFailedCheckpointStats( + new CheckpointException(CheckpointFailureReason.IO_EXCEPTION)); + + assertThat(failed.getFailureReason()).isEqualTo(CheckpointFailureReason.IO_EXCEPTION); + } + + @Test + void testFailureReasonExtractedFromWrappedCheckpointException() { + FailedCheckpointStats failed = + newFailedCheckpointStats( + new RuntimeException( + "wrapper", + new CheckpointException( + CheckpointFailureReason.CHECKPOINT_EXPIRED))); + + assertThat(failed.getFailureReason()).isEqualTo(CheckpointFailureReason.CHECKPOINT_EXPIRED); + } + + @Test + void testFailureReasonNullForNonCheckpointExceptionCause() { + FailedCheckpointStats failed = + newFailedCheckpointStats(new RuntimeException("some other failure")); + + assertThat(failed.getFailureReason()).isNull(); + } + + @Test + void testFailureReasonNullForNullCause() { + FailedCheckpointStats failed = newFailedCheckpointStats(null); + + assertThat(failed.getFailureReason()).isNull(); + } + + private static FailedCheckpointStats newFailedCheckpointStats(@Nullable Throwable cause) { + Map taskStats = new HashMap<>(); + JobVertexID jobVertexId = new JobVertexID(); + taskStats.put(jobVertexId, new TaskStateStats(jobVertexId, 1)); + + return new FailedCheckpointStats( + 0, + 10123L, + CheckpointProperties.forCheckpoint( + CheckpointRetentionPolicy.NEVER_RETAIN_AFTER_TERMINATION), + 1, + taskStats, + 0, + 0, + 0, + 0, + 0, + false, + 10123L, + null, + cause); + } } diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/TestingCheckpointStatsTracker.java b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/TestingCheckpointStatsTracker.java index 9f42c5b159eb25..a660197f023330 100644 --- a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/TestingCheckpointStatsTracker.java +++ b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/TestingCheckpointStatsTracker.java @@ -21,6 +21,8 @@ import org.apache.flink.runtime.executiongraph.ExecutionAttemptID; import org.apache.flink.runtime.jobgraph.JobVertexID; +import javax.annotation.Nullable; + import java.util.Collections; import java.util.Map; import java.util.Set; @@ -39,7 +41,7 @@ public class TestingCheckpointStatsTracker implements CheckpointStatsTracker { Collections.singletonMap(new JobVertexID(), 1)); @Override - public void reportFailedCheckpointsWithoutInProgress() { + public void reportFailedCheckpointsWithoutInProgress(@Nullable CheckpointFailureReason reason) { numFailedCheckpoints.incrementAndGet(); } diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/dispatcher/DispatcherTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/dispatcher/DispatcherTest.java index ad4f10be633b63..528d4658d3b40d 100755 --- a/flink-runtime/src/test/java/org/apache/flink/runtime/dispatcher/DispatcherTest.java +++ b/flink-runtime/src/test/java/org/apache/flink/runtime/dispatcher/DispatcherTest.java @@ -802,8 +802,8 @@ private CheckpointStatsSnapshot getTestCheckpointStatsSnapshotWithTwoFailedCheck CheckpointStatsTracker checkpointStatsTracker = new DefaultCheckpointStatsTracker( 10, UnregisteredMetricGroups.createUnregisteredJobManagerJobMetricGroup()); - checkpointStatsTracker.reportFailedCheckpointsWithoutInProgress(); - checkpointStatsTracker.reportFailedCheckpointsWithoutInProgress(); + checkpointStatsTracker.reportFailedCheckpointsWithoutInProgress(null); + checkpointStatsTracker.reportFailedCheckpointsWithoutInProgress(null); return checkpointStatsTracker.createSnapshot(); } diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/AdaptiveSchedulerTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/AdaptiveSchedulerTest.java index 4feef4ca0f54fc..0ae80c351cbf3c 100644 --- a/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/AdaptiveSchedulerTest.java +++ b/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/AdaptiveSchedulerTest.java @@ -2306,7 +2306,9 @@ void testOnCompletedCheckpointIsHandledInMainThread() throws Exception { @Test void testOnFailedCheckpointIsHandledInMainThread() throws Exception { testCheckpointStatsEventBeingExecutedInTheMainThread( - CheckpointStatsListener::onFailedCheckpoint, 2, 2); + listener -> listener.onFailedCheckpoint(CheckpointFailureReason.IO_EXCEPTION), + 2, + 2); } private void testCheckpointStatsEventBeingExecutedInTheMainThread( diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/ExecutingTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/ExecutingTest.java index 0d3f76d3aed891..530ebeb137e2f0 100644 --- a/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/ExecutingTest.java +++ b/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/ExecutingTest.java @@ -26,6 +26,7 @@ import org.apache.flink.runtime.blob.PermanentBlobKey; import org.apache.flink.runtime.checkpoint.CheckpointCoordinator; import org.apache.flink.runtime.checkpoint.CheckpointCoordinatorTestingUtils; +import org.apache.flink.runtime.checkpoint.CheckpointFailureReason; import org.apache.flink.runtime.checkpoint.CheckpointScheduling; import org.apache.flink.runtime.client.JobExecutionException; import org.apache.flink.runtime.concurrent.ComponentMainThreadExecutor; @@ -83,6 +84,7 @@ import org.junit.jupiter.api.Test; import org.junit.jupiter.api.extension.RegisterExtension; import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.EnumSource; import org.junit.jupiter.params.provider.ValueSource; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -242,7 +244,7 @@ public void testTriggerRescaleOnFailedCheckpoint() throws Exception { // trigger an initial failed checkpoint event to show that the counting only starts // with the subsequent change event - testInstance.onFailedCheckpoint(); + testInstance.onFailedCheckpoint(CheckpointFailureReason.IO_EXCEPTION); // trigger change testInstance.onNewResourceRequirements(); @@ -253,7 +255,7 @@ public void testTriggerRescaleOnFailedCheckpoint() throws Exception { "No rescale operation should have been triggered for iteration #%d, yet.", rescaleIteration) .hasValue(rescaleIteration - 1); - testInstance.onFailedCheckpoint(); + testInstance.onFailedCheckpoint(CheckpointFailureReason.IO_EXCEPTION); } assertThat(rescaleTriggerCount) @@ -284,13 +286,16 @@ public void testOnCompletedCheckpointResetsFailedCheckpointCount() throws Except // trigger an initial failed checkpoint event to show that the counting only starts with // the subsequent change event - testInstance.onFailedCheckpoint(); + testInstance.onFailedCheckpoint(CheckpointFailureReason.IO_EXCEPTION); // trigger change testInstance.onNewResourcesAvailable(); IntStream.range(0, rescaleOnFailedCheckpointsCount - 1) - .forEach(ignored -> testInstance.onFailedCheckpoint()); + .forEach( + ignored -> + testInstance.onFailedCheckpoint( + CheckpointFailureReason.IO_EXCEPTION)); assertThat(rescaleTriggeredCount) .as("No rescaling should have been trigger, yet.") @@ -306,14 +311,17 @@ public void testOnCompletedCheckpointResetsFailedCheckpointCount() throws Except .hasValue(1); IntStream.range(0, rescaleOnFailedCheckpointsCount - 1) - .forEach(ignored -> testInstance.onFailedCheckpoint()); + .forEach( + ignored -> + testInstance.onFailedCheckpoint( + CheckpointFailureReason.IO_EXCEPTION)); assertThat(rescaleTriggeredCount) .as( "No additional rescaling should have been trigger by any subsequent failed checkpoint, yet.") .hasValue(1); - testInstance.onFailedCheckpoint(); + testInstance.onFailedCheckpoint(CheckpointFailureReason.IO_EXCEPTION); assertThat(rescaleTriggeredCount) .as("The previous failed checkpoint should have triggered the rescale.") @@ -321,6 +329,98 @@ public void testOnCompletedCheckpointResetsFailedCheckpointCount() throws Except } } + @Test + public void testIgnoredFailureReasonDoesNotTriggerRescale() throws Exception { + runRescaleTriggerCheckForReason(CheckpointFailureReason.NOT_ALL_REQUIRED_TASKS_RUNNING, 0); + } + + @Test + public void testNullFailureReasonDoesNotTriggerRescale() throws Exception { + runRescaleTriggerCheckForReason(null, 0); + } + + /** + * Guards that the rescale countdown honors exactly the set of failure reasons classified as + * counted by {@link CheckpointFailureReason#isCountedAgainstFailureThreshold()} — the same + * predicate used by {@code CheckpointFailureManager} — for every reason, so the two mechanisms + * cannot silently drift apart when a new reason is added. + */ + @ParameterizedTest + @EnumSource(CheckpointFailureReason.class) + public void testRescaleCountdownHonorsFailureThresholdClassification( + CheckpointFailureReason reason) throws Exception { + runRescaleTriggerCheckForReason(reason, reason.isCountedAgainstFailureThreshold() ? 1 : 0); + } + + @Test + public void testMixedFailureReasonsOnlyCountRelevant() throws Exception { + final AtomicInteger rescaleTriggerCount = new AtomicInteger(); + final Function + stateTransitionManagerFactory = + context -> + TestingStateTransitionManager.withOnTriggerEventOnly( + rescaleTriggerCount::incrementAndGet); + + final int rescaleOnFailedCheckpointsCount = 2; + try (MockExecutingContext ctx = new MockExecutingContext()) { + final Executing testInstance = + new ExecutingStateBuilder() + .setStateTransitionManagerFactory(stateTransitionManagerFactory) + .setRescaleOnFailedCheckpointCount(rescaleOnFailedCheckpointsCount) + .build(ctx); + + testInstance.onNewResourceRequirements(); + + // ignored failures interspersed with a single relevant failure must not advance the + // countdown enough to trigger a rescale + testInstance.onFailedCheckpoint(CheckpointFailureReason.NOT_ALL_REQUIRED_TASKS_RUNNING); + testInstance.onFailedCheckpoint(CheckpointFailureReason.IO_EXCEPTION); + testInstance.onFailedCheckpoint(CheckpointFailureReason.CHECKPOINT_SUBSUMED); + testInstance.onFailedCheckpoint(null); + + assertThat(rescaleTriggerCount) + .as("Only one relevant failure has been observed so far.") + .hasValue(0); + + testInstance.onFailedCheckpoint(CheckpointFailureReason.CHECKPOINT_EXPIRED); + + assertThat(rescaleTriggerCount) + .as("Two relevant failures should have triggered the rescale.") + .hasValue(1); + } + } + + private void runRescaleTriggerCheckForReason( + @Nullable CheckpointFailureReason reason, int expectedRescaleCount) throws Exception { + final AtomicInteger rescaleTriggerCount = new AtomicInteger(); + final Function + stateTransitionManagerFactory = + context -> + TestingStateTransitionManager.withOnTriggerEventOnly( + rescaleTriggerCount::incrementAndGet); + + final int rescaleOnFailedCheckpointsCount = 2; + try (MockExecutingContext ctx = new MockExecutingContext()) { + final Executing testInstance = + new ExecutingStateBuilder() + .setStateTransitionManagerFactory(stateTransitionManagerFactory) + .setRescaleOnFailedCheckpointCount(rescaleOnFailedCheckpointsCount) + .build(ctx); + + testInstance.onNewResourceRequirements(); + + for (int i = 0; i < rescaleOnFailedCheckpointsCount * 2; i++) { + testInstance.onFailedCheckpoint(reason); + } + + assertThat(rescaleTriggerCount) + .as( + "Failures with reason %s should have triggered %d rescale(s).", + reason, expectedRescaleCount) + .hasValue(expectedRescaleCount); + } + } + @Test void testDisposalOfOperatorCoordinatorsOnLeaveOfStateWithExecutionGraph() throws Exception { try (MockExecutingContext ctx = new MockExecutingContext()) {