[FLINK-38930][checkpoint] Filtering record before processing without spilling strategy

[1996fanrui]

This PR depends on #27782

What is the purpose of the change

[FLINK-38930][checkpoint] Filtering record before processing without spilling strategy

Brief change log

Core filtering mechanism for recovered channel state buffers:

• ChannelStateFilteringHandler with per-gate GateFilterHandler
• RecordFilterContext with VirtualChannelRecordFilterFactory
• Partial data check in SequentialChannelStateReaderImpl
• Fix RecordFilterContext for Union downscale scenario

Verifying this change

• Tons of unit tests

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

[flinkbot]

CI report:

• 997e3a3 Azure: SUCCESS

Bot commands

The @flinkbot bot supports the following commands:

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