[FLINK-36512][runtime] Ignore non-rescale-relevant checkpoint failure reasons in adaptive rescale trigger

docs/layouts/shortcodes/generated/all_jobmanager_section.html

@@ -60,7 +60,7 @@
             <td><h5>jobmanager.adaptive-scheduler.rescale-trigger.max-checkpoint-failures</h5></td>
             <td style="word-wrap: break-word;">2</td>
             <td>Integer</td>
-            <td>The number of consecutive failed checkpoints that will trigger rescaling even in the absence of a completed checkpoint.</td>
+            <td>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.</td>
         </tr>
         <tr>
             <td><h5>jobmanager.adaptive-scheduler.rescale-trigger.max-delay</h5></td>

docs/layouts/shortcodes/generated/expert_scheduling_section.html

@@ -114,7 +114,7 @@
             <td><h5>jobmanager.adaptive-scheduler.rescale-trigger.max-checkpoint-failures</h5></td>
             <td style="word-wrap: break-word;">2</td>
             <td>Integer</td>
-            <td>The number of consecutive failed checkpoints that will trigger rescaling even in the absence of a completed checkpoint.</td>
+            <td>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.</td>
         </tr>
         <tr>
             <td><h5>jobmanager.adaptive-scheduler.rescale-trigger.max-delay</h5></td>

docs/layouts/shortcodes/generated/job_manager_configuration.html

@@ -60,7 +60,7 @@
             <td><h5>jobmanager.adaptive-scheduler.rescale-trigger.max-checkpoint-failures</h5></td>
             <td style="word-wrap: break-word;">2</td>
             <td>Integer</td>
-            <td>The number of consecutive failed checkpoints that will trigger rescaling even in the absence of a completed checkpoint.</td>
+            <td>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.</td>
         </tr>
         <tr>
             <td><h5>jobmanager.adaptive-scheduler.rescale-trigger.max-delay</h5></td>

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

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();
+            }
         }
     }
 

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());
+        }
+    }
 }

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

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.

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,15 +519,15 @@ private boolean shouldSkipSumMetricNameInCheckpointSpanForCompatibility(String m
     }
 
     @Override
-    public void reportFailedCheckpointsWithoutInProgress() {
+    public void reportFailedCheckpointsWithoutInProgress(@Nullable CheckpointFailureReason reason) {
         statsReadWriteLock.lock();
         try {
             counts.incrementFailedCheckpointsWithoutInProgress();
 
             dirty = true;
 
             if (checkpointStatsListener != null) {
-                checkpointStatsListener.onFailedCheckpoint();
+                checkpointStatsListener.onFailedCheckpoint(reason);
             }
         } finally {
             statsReadWriteLock.unlock();

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 <code>null</code>.
-     * @param cause Cause of the checkpoint failure or <code>null</code>.
+     * @param cause Cause of the checkpoint failure or <code>null</code>. 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 <code>null</code> if the cause was unknown or contained no {@code
+     * CheckpointException}.
+     */
+    @Nullable
+    public CheckpointFailureReason getFailureReason() {
+        return failureReason;
+    }
 }

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() {

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

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;