[FLINK-38218] Enforce no duplicate split infos in MySqlBinlogSplit

Author: morozov

flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/split/MySqlBinlogSplit.java

@@ -28,6 +28,7 @@
 import javax.annotation.Nullable;
 
 import java.util.ArrayList;
+import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
@@ -42,7 +43,10 @@ public class MySqlBinlogSplit extends MySqlSplit {
 
     private final BinlogOffset startingOffset;
     private final BinlogOffset endingOffset;
+
+    /** Split IDs of all elements must be unique. */
     private final List<FinishedSnapshotSplitInfo> finishedSnapshotSplitInfos;
+
     private final Map<TableId, TableChange> tableSchemas;
     private final int totalFinishedSplitSize;
     private final boolean isSuspended;
@@ -58,6 +62,19 @@ public MySqlBinlogSplit(
             int totalFinishedSplitSize,
             boolean isSuspended) {
         super(splitId);
+
+        Set<String> seenSplitIds = new HashSet<>();
+        for (FinishedSnapshotSplitInfo splitInfo : finishedSnapshotSplitInfos) {
+            if (seenSplitIds.contains(splitInfo.getSplitId())) {
+                throw new IllegalArgumentException(
+                        String.format(
+                                "Found duplicate split ID %s in finished snapshot split infos",
+                                splitInfo.getSplitId()));
+            }
+
+            seenSplitIds.add(splitInfo.getSplitId());
+        }
+
         this.startingOffset = startingOffset;
         this.endingOffset = endingOffset;
         this.finishedSnapshotSplitInfos = finishedSnapshotSplitInfos;

flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/test/java/org/apache/flink/cdc/connectors/mysql/source/split/MySqlBinlogSplitTest.java

@@ -148,6 +148,32 @@ void testTruncatedTablesForLog() {
         Assertions.assertThat(binlogSplit.getTables()).isEqualTo(expectedTables);
     }
 
+    @Test
+    public void duplicateSplitInfo() {
+        FinishedSnapshotSplitInfo info =
+                new FinishedSnapshotSplitInfo(
+                        new TableId("catalog", "schema", "table"),
+                        "split",
+                        null,
+                        null,
+                        BinlogOffset.ofLatest());
+        List<FinishedSnapshotSplitInfo> infos = new ArrayList<>();
+        infos.add(info);
+        infos.add(info);
+
+        Assertions.assertThatThrownBy(
+                        () ->
+                                new MySqlBinlogSplit(
+                                        "binlog-split",
+                                        BinlogOffset.ofLatest(),
+                                        null,
+                                        infos,
+                                        Collections.emptyMap(),
+                                        0,
+                                        false))
+                .isExactlyInstanceOf(IllegalArgumentException.class);
+    }
+
     /** A mock implementation for {@link Table} which is used for unit tests. */
     private static class MockTable implements Table {
         private final TableId tableId;