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

[Zakelly]

Rework of #29043 as it was reverted by #29100.

The IT case has been rewritten. 5-run validate is ongoing.

What is the purpose of the change

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

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.

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 160,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.

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:

• 740deaa Azure: SUCCESS

Bot commands

The @flinkbot bot supports the following commands:

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