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

[prashantbh]

What is the purpose of the change

The Adaptive Scheduler can trigger a rescale after a configurable number of consecutive failed checkpoints (jobmanager.adaptive-scheduler.rescale-trigger.max-checkpoint-failures). The countdown was decremented on every failed-checkpoint notification, including failures that do not count against the tolerable checkpoint-failure threshold — e.g. checkpoints triggered before all tasks are running during startup/recovery (NOT_ALL_REQUIRED_TASKS_RUNNING), coordinator shutdown, or checkpoint subsumption.

As a result, transient pre-flight failures during startup/recovery could drive the countdown to zero and cause a spurious, premature rescale even though no checkpoint ever actually ran and failed.

This change makes the rescale trigger count only the failure reasons that count against the tolerable checkpoint-failure threshold — the exact same set that CheckpointFailureManager enforces at the job level — so the two failure-counting mechanisms stay consistent.

Brief change log

• Added CheckpointFailureReason#isCountedAgainstFailureThreshold() as the single source of truth for which reasons count toward a failure threshold; CheckpointFailureManager#checkFailureCounter and the Adaptive Scheduler rescale countdown now both consult it (previously the classification was duplicated).
• Threaded the CheckpointFailureReason through the checkpoint-stats failure notification path: CheckpointStatsListener#onFailedCheckpoint, CheckpointStatsTracker#reportFailedCheckpointsWithoutInProgress, and FailedCheckpointStats (extracted from the CheckpointException cause chain).
• Executing#onFailedCheckpoint now ignores non-counted reasons, so pre-flight failures no longer advance the rescale countdown.
• Tightened the description of SCHEDULER_RESCALE_TRIGGER_MAX_CHECKPOINT_FAILURES to reflect the above.

Verifying this change

This change added tests and can be verified as follows:

• ExecutingTest: parameterized over all CheckpointFailureReason values, asserting the rescale countdown advances iff the reason is counted; plus null-reason and mixed relevant/irrelevant interleaving cases.
• CheckpointFailureManagerTest: existing tests confirm the refactor to the shared predicate is behavior-preserving.
• FailedCheckpointStatsTest: reason extraction from a direct CheckpointException, a wrapped CheckpointException in the cause chain, a non-CheckpointException cause, and a null cause.
• CheckpointCoordinatorTest (integration): new cases drive a real CheckpointCoordinator → CheckpointFailureManager → DefaultCheckpointStatsTracker → CheckpointStatsListener and assert the exact reason delivered for a pre-flight failure (NOT_ALL_REQUIRED_TASKS_RUNNING), an IO failure, and a declined checkpoint — covering both listener paths (reportFailedCheckpointsWithoutInProgress and reportFailedCheckpoint).

Does this pull request potentially affect one of the following parts:

• Dependencies (does it add or upgrade a dependency): no
• The public API, i.e., is any changed class annotated with @Public(Evolving): no
• The serializers: no
• The runtime per-record code paths (performance sensitive): no
• Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn, ZooKeeper: yes — changes checkpoint-failure classification and the Adaptive Scheduler's rescale-on-failed-checkpoints behavior. No changes to checkpoint format, wire protocol, or public API.
• The S3 file system connector: no

Documentation

• Does this pull request introduce a new feature? no (bug fix to existing behavior)
• If yes, how is the feature documented? not applicable (the description of the existing rescale-trigger.max-checkpoint-failures config option was updated to reflect the corrected semantics)

---

Was generative AI tooling used to co-author this PR?

• Yes (Claude Code — Claude Opus 4.8)

[flinkbot]

CI report:

• 6a31226 Azure: SUCCESS

Bot commands

The @flinkbot bot supports the following commands:

• @flinkbot run azure re-run the last Azure build