[FLINK-39775][mysql] Release snapshot split metadata after entering the stream phase

[spoorthibasu]

TL;DR: Fix JobManager OOM with a large number of snapshot splits by releasing the snapshot split metadata after the source enters the binlog phase, once the binlog split is assigned and covered by a completed checkpoint.

Root Cause

After the snapshot phase finishes and the binlog split is assigned, the source coordinator keeps the full snapshot split metadata indefinitely.

• MySqlSnapshotSplitAssigner holds assignedSplits, splitFinishedOffsets, and tableSchemas in memory.
• snapshotState() rebuilds the complete SnapshotPendingSplitsState on every checkpoint, MySqlHybridSplitAssigner wraps it, and notifyCheckpointComplete() never drops it.
• MySqlSourceEnumerator keeps a second copy of the same finished-split metadata in its binlogSplitMeta cache.

With many finished splits (for example ~300K for a 2.5B-row table at the default chunk size of 8096), every post-snapshot checkpoint re-serializes the entire snapshot state. This keeps JobManager heap high, and because the large coordinator state is copied during checkpoint serialization and transfer, it also drives up direct/off-heap memory and can OOM-kill the container even when the heap itself does not exceed its limit.

Fix

Release the snapshot split metadata once it is no longer needed for correctness: after the binlog split has been created, its finished-split metadata fully transferred to and assembled by the reader, and that is covered by a completed checkpoint.

Reader-side (completion signal)

• When a complete binlog split enters reading (whether its metadata arrived inline or was assembled over the divided meta-group requests), MySqlSourceReader sends a new BinlogSplitMetaAssembledEvent to the coordinator.
• The event is idempotent: it is re-sent whenever a complete binlog split (re-)enters reading, including after a restore.

Coordinator-side (gated release)

• On receiving the event, MySqlHybridSplitAssigner records that the reader has assembled the metadata.
• snapshotState() schedules the release at the current checkpoint id; the actual clearing happens in notifyCheckpointComplete() once that checkpoint completes.
• releaseSnapshotMetadata() clears assignedSplits, splitFinishedOffsets, and tableSchemas. alreadyProcessedTables, the assigner status, and the chunk splitter state are kept so a restore does not re-discover tables.
• MySqlSourceEnumerator clears its cached binlogSplitMeta at the same point.
• Release is skipped entirely when scan.newly-added-table.enabled is set, since that flow may need the metadata again to extend the binlog split.

Why the release is safe

• The clearing happens only in notifyCheckpointComplete(), never in snapshotState(). Flink only re-delivers (addSplitsBack) split assignments that are not covered by the last completed checkpoint. Because the binlog split was assigned before the release-scheduling checkpoint, that assignment is checkpoint-covered by the time release runs, so the binlog split can never be added back into an emptied assigner. createBinlogSplit() also asserts the metadata has not been released, turning any future regression into a fast failure instead of silent data loss.
• If the binlog split is added back before release (reader failover between scheduling and the checkpoint completing), the scheduled release is reset and the metadata is still intact, so the split is recreated normally.
• On restore from a released ("light") state, the maps are empty, isBinlogSplitAssigned is true, and alreadyProcessedTables is non-empty, so the assigner does not re-discover tables and does not recreate the binlog split. The reader restores its own complete binlog split, requests no metadata, and re-sends the assembled event, which makes the release idempotent.
• The state and serializer format are unchanged. A released state is a normal SnapshotPendingSplitsState with empty maps, so it serializes under the existing version and stays compatible on restore.

Tests Added

• MySqlSnapshotMetadataReleaseTest (new)
  Release after the assembled event plus a completed checkpoint; no release without the event; no release before the checkpoint completes; no release when newly-added-table is enabled; add-back resets the schedule and the binlog split is recreated with full metadata.

• PendingSplitsStateSerializerTest.testSerializeAndDeserializeReleasedHybridState
  Round-trips the released ("light") hybrid state, confirming empty maps and isBinlogSplitAssigned survive serialization under the current version.

• MySqlSourceReaderTest.testReaderSendsBinlogSplitMetaAssembledEventForCompleteSplit
  The reader emits the assembled event when a complete binlog split enters reading.

• MySqlSourceITCase
  TaskManager and JobManager failover in the binlog phase, plus reads across multiple tables at higher parallelism, all pass with no data loss.

Notes

• The common path is unchanged until the source is well into the binlog phase; release only happens after the binlog split is assigned and one covering checkpoint completes.
• No new configuration is introduced, and no state or serializer version bump is required.