[FLINK-39977][runtime] Recovery of merged channel state handles

[Zakelly]

What is the purpose of the change

This pull request fixes recovery from checkpoints containing channel state written by the unified file-merging mechanism.

During recovery, the file-merging snapshot manager previously restored keyed and operator state handles but ignored channel state handles. Consequently, merged files referenced by input channel state, upstream output buffer state, or result subpartition state were not restored into the file-merging snapshot manager's tracking information.

Brief change log

• Restore file-merged state handles referenced by input channel state, upstream output buffer state, and result subpartition state.
• Deduplicate shared channel state delegates before restoring their corresponding logical files.
• Extend file-merging snapshot manager tests to cover merged channel state handles and shared delegates.
• Add FileMergingChannelStateITCase to verify recovery after replacing all TaskManagers.

Verifying this change

This change added and extended tests and can be verified as follows:

• Added FileMergingChannelStateITCase, which:
  • Enables unaligned checkpoints and unified file merging.
  • Processes 16,000 records through a WordCount pipeline.
  • Verifies that checkpointed channel state uses SegmentFileStateHandle.
  • Replaces all TaskManagers and restores the job from the checkpoint.
  • Takes another checkpoint after recovery and verifies obsolete merged file cleanup.
  • Verifies the final count for every word to ensure data correctness.
• Extended FileMergingSnapshotManagerTestBase to verify that input channel, upstream output buffer, and result subpartition state handles are restored and tracked without duplicating shared delegates.
• Verified FileMergingChannelStateITCase successfully; it completed in approximately 22 seconds locally.

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; checkpoint recovery is affected
• 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

---

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

• Yes (please specify the tool below)

Generated-by: OpenAI Codex (GPT-5.6 Sol)

[flinkbot]

CI report:

• 38ebba0 Azure: SUCCESS

Bot commands

The @flinkbot bot supports the following commands:

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

[Zakelly]

@1996fanrui Would you please take a look?

[1996fanrui]

Hey @Zakelly , thanks for the pr! I will take a look next week

[1996fanrui]

Thanks @Zakelly for the fix, lgtm

The fix is simple than I expected

[Zakelly]

@1996fanrui Yeah, it's also beyond my imagine :D. Thanks for the review

[snuyanzin]

@Zakelly , @1996fanrui
I'm going to revert this: it is very flaky and it fails everywhere in master Azure CI
see

https://dev.azure.com/apache-flink/apache-flink/_build/results?buildId=78673&view=logs&j=5c8e7682-d68f-54d1-16a2-a09310218a49&s=ae4f8708-9994-57d3-c2d7-b892156e7812&t=9d734c8c-6253-55e6-3bce-47e7cdf68ac4&l=7778

https://dev.azure.com/apache-flink/apache-flink/_build/results?buildId=78686&view=logs&j=5c8e7682-d68f-54d1-16a2-a09310218a49&t=9d734c8c-6253-55e6-3bce-47e7cdf68ac4&l=7775

https://dev.azure.com/apache-flink/apache-flink/_build/results?buildId=78678&view=logs&j=5c8e7682-d68f-54d1-16a2-a09310218a49&t=9d734c8c-6253-55e6-3bce-47e7cdf68ac4&l=7240

P.S.: there are more failures
e.g. every test job in nightly (every jdk) fails one that test https://dev.azure.com/apache-flink/apache-flink/_build/results?buildId=78686&view=logs&j=2c3cbe13-dee0-5837-cf47-3053da9a8a78&t=d102aafb-3bbd-55e4-a35f-e8935afffc31

Sep 04 01:31:43 01:31:43.543 [ERROR] Errors: 
Sep 04 01:31:43 01:31:43.543 [ERROR]   FileMergingChannelStateITCase.testRestoreFileMergedChannelState:147 » Execution org.apache.flink.runtime.checkpoint.CheckpointException: task Map (1/1) of job c62d634c19662df1005dea1fe1998200 is in INITIALIZING state instead of RUNNING or FINISHED. Aborting checkpoint. Failure reason: Not all required tasks are currently running.
Sep 04 01:31:43 01:31:43.543 [INFO] 

[snuyanzin]

also looking at master CI
it succeded right before this PR merged https://dev.azure.com/apache-flink/apache-flink/_build/results?buildId=78655&view=results
and failed right after https://dev.azure.com/apache-flink/apache-flink/_build/results?buildId=78657&view=results

[snuyanzin]

reverted at #29100

cc @Zakelly