[FLINK-40551][tests] Select the checkpoint to restore by content in FileMergingChannelStateITCase

[pedromazala]

The test asserted that the checkpoint it restores from carries in-flight input
channel state for the slow mapper, but selected that checkpoint by waiting for
the second completed checkpoint that persisted any in-flight data anywhere in
the job. Which subtask's buffers land in a given checkpoint depends on where the
barriers are when it is triggered, so the count was a calibration rather than a
condition: instrumenting the scan shows checkpoint 1 persists ~295 KB of
in-flight data while carrying no input channel state for the mapper at all, and
only checkpoint 2 has it on this hardware.

Walk the completed checkpoint history instead and restore from the first
checkpoint whose metadata really contains file-merged input channel state for
the slow mapper. The lookup reports which checkpoints it inspected if the job
terminates first, and skips checkpoints cleaned up while being read. The
SegmentFileStateHandle and state size checks stay hard assertions on the
selected checkpoint, so a file merging regression still fails loudly instead of
timing out.

What is the purpose of the change

Reduce flakiness around FileMergingChannelStateITCase

Brief change log

• Replace the checkpoint selection in testRestoreFileMergedChannelState: instead of CommonTestUtils.waitForCheckpointWithInflightBuffers(..., INITIAL_CHECKPOINTS_TO_WAIT), which accepts the latest completed checkpoint that persisted any in-flight data anywhere in the job, the test now restores from a checkpoint selected by content
• Add waitForCheckpointWithSlowMapperChannelState, which walks the completed checkpoint history oldest-first and returns the first checkpoint whose metadata actually carries file-merged input channel state for the slow-word-mapper operator
• Bound that lookup with a deadline (CHECKPOINT_LOOKUP_TIMEOUT, polled every CHECKPOINT_LOOKUP_INTERVAL_MS) and report the inspected checkpoint ids on timeout, rather than looping until the surefire timeout kills the fork
• Fail fast with the job's own failure cause if the job reaches a globally terminal state before such a checkpoint completes
• Skip checkpoints that are subsumed and cleaned up while their metadata is being read, instead of failing the test on the resulting IOException
• Extract collectChannelStateDelegates (and the ChannelStateDelegates holder) so the new predicate and assertFileMergedChannelState share one traversal of the checkpoint metadata
• Keep the assertions themselves unchanged — all channel state delegates are SegmentFileStateHandle, total state size is positive, and the slow mapper's input channel state is non-empty — so a file merging regression still fails loudly on the selected checkpoint instead of timing out
• Drop the now-unused INITIAL_CHECKPOINTS_TO_WAIT constant. Test-only change confined to FileMergingChannelStateITCase; the 3-arg waitForCheckpointWithInflightBuffers overload stays in use by UnalignedCheckpointRescaleSameUpstreamITCase

Verifying this change

The test was executed 20 times to ensure the flakiness was over.

This change is a trivial rework / code cleanup without any test coverage.

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: no
• The S3 file system connector: no

Documentation

• Does this pull request introduce a new feature? no
• If yes, how is the feature documented? (not applicable / docs / JavaDocs / not documented)

---

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

• Yes (please specify the tool below)

Generated-by: Claude Opus 5 (1M context).

[flinkbot]

CI report:

• 26cc91b Azure: SUCCESS

Bot commands

The @flinkbot bot supports the following commands:

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

[snuyanzin]

Thank you for the contribution

Can you explain please what exactly you are fixing and why?

[snuyanzin]

I'm asking since the FLINK-40551 was fixed by reverting original commit at
#29100

after that it was not fail

so why do we need to change anything else here?

[Jackeyzhe]

This catches more than the subsumption race described by the comment.loadCheckpointMetadata also reports corrupt headers, unsupported versions, truncated input, and deserialization failures as IOException. Since this checkpoint was already added to inspectedCheckpoints, those failures are silently discarded and a later checkpoint can make the test pass. Could we suppress only the missing/cleaned-up checkpoint case and rethrow other metadata read failures?

[Jackeyzhe]

Could we give this lookup an explicit deadline and include inspectedCheckpoints in the timeout failure? The one-argument waitUntilCondition loops without a timeout, while failIfJobStoppedCheckpointing only fires after the job becomes terminal. If the job remains running and completed checkpoints never contain the slow mapper's channel state, this test can hang until the outer build timeout, which keeps the original diagnosability problem.