[FLINK-38218] Fix MySQL CDC binlog split metadata split transmission

[morozov]

The root cause is that the binlog split metadata transfer protocol relies on the order of finished snapshot split infos to be stable and corresponding to the order of split assignment (the infos of newly added/snapshotted tables are appended to the end of the list). However, when MySqlSnapshotSplitAssigner is restored from state, assignedSplits are reordered, which breaks this assumption.

Change summary

1. Require assigned snapshot splits to be ordered. This isn't strictly necessary to fix the bug but follows directly from the JavaDoc I added to MySqlSnapshotSplitAssigner#assignedSplits. If the order is important, the type should guarantee that it's preserved. Note the changes in the deserialization code. Not using an ordered map there while the order is important may cause other hard to diagnose issues.
2. Rely on stable order of assigned splits. Instead of identifying duplicate received split infos by split ID, ignore the first N elements that we know we already have.
3. Eliminate code duplication in MySqlBinlogSplit constructors. There are currently two constructors where one doesn't call the other. The subsequent commit adds a check that needs to be enforced regardless of which of the constructors was used, so I'm combining them.
4. Enforce no duplicate finished snapshot split infos in MySqlBinlogSplit. By design, a binlog split cannot contain duplicate finished snapshot split infos. If it does, it indicates the fact that it was constructed incorrectly. If it happens, it's a bug, and we want to fail as early as possible.

[morozov]

I'm not sure how to test this. The issue is reproducible if a source is restarted mid-snapshot of a newly added table and requires consuming the changes in the new table from the binlog. Could maintainers recommend an existing test on top of which I could build this?

[morozov]

@leonardBang could you restart the tests? The logs are no longer available.

[leonardBang]

Hey, @morozov Looks like Azure cannot re-trigger expired CI, could you rebase your PR to latest master branch to trigger new CI ?

[morozov]

@leonardBang, after the rebase, the build is green.

[morozov]

@leonardBang, @lvyanquan do you still want to get this issue fixed? I closed the PR due to the lack of interest but you reopened it.

[lvyanquan]

Hi @morozov, apologies for the delayed response. As this change impacts a critical code path, we are proceeding with utmost caution. I will review and verify both the issue and the proposed fix this week.

[Copilot]

Pull request overview

This PR fixes a bug in MySQL CDC where binlog split metadata transmission could fail due to inconsistent ordering of finished snapshot split information. The root cause was that assignedSplits were being reordered during restoration from state, breaking the protocol's assumption that split infos maintain their assignment order.

Changes:

• Enforces stable ordering of assigned snapshot splits using LinkedHashMap throughout the codebase
• Replaces split ID-based duplicate detection with simpler position-based logic in metadata transmission
• Adds validation to detect and fail fast on duplicate split IDs in binlog splits

Reviewed changes

Copilot reviewed 10 out of 10 changed files in this pull request and generated no comments.

Show a summary per file

File	Description
MySqlSnapshotSplitAssigner.java	Removes sorting of assignedSplits and adds documentation explaining order requirements; changes assignedSplits type to LinkedHashMap
SnapshotPendingSplitsState.java	Changes assignedSplits field type to LinkedHashMap to preserve insertion order
PendingSplitsStateSerializer.java	Updates deserialization to use LinkedHashMap instead of HashMap for assignedSplits
MySqlSourceReader.java	Replaces split ID-based duplicate filtering with position-based element skipping using modulo arithmetic
MySqlBinlogSplit.java	Adds duplicate split ID validation in constructor and consolidates two constructors to eliminate code duplication
MySqlHybridSplitAssigner.java	Removes sorting when creating binlog splits to preserve natural order from assignedSplits
MySqlSnapshotSplitAssignerTest.java	Adds parameterized test to verify finished split infos maintain assignment order across checkpoints; updates to use LinkedHashMap
MySqlBinlogSplitTest.java	Adds test to verify duplicate split IDs are rejected with IllegalArgumentException
MySqlHybridSplitAssignerTest.java	Updates test setup to use LinkedHashMap for assignedSplits
PendingSplitsStateSerializerTest.java	Updates test setup to use LinkedHashMap for assignedSplits

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

[lvyanquan]

Hi @morozov.

I understand the problem you described now. I'd like to confirm that similar issues can also occur in scenarios other than adding new tables. Another possible scenario is that due to missing some table splits, when we can't find the corresponding matchedSplit in BinlogSplitReader#shouldEmit and proceed to emit the data anyway, it leads to data duplication.

[morozov]

flink-cdc/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/debezium/reader/BinlogSplitReader.java

Line 291 in c4b698b

return matchedSplit != null && position.isAfter(matchedSplit.getHighWatermark());

If I'm reading the code correctly, if we cannot find the corresponding matchedSplit, it's null, so BinlogSplitReader#shouldEmit() will return false, which leads to data loss, not duplication. But this is a different issue (FLINK-38270).

The principal differences are:

1. Here, the connection gets stuck not producing any data; there, it transitions to reading the binlog but drops data from tables (much harder to spot).
2. This one is reproducible by addition of a table; the other one requires addition and removal during snapshot.

[lvyanquan]

+1.