[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