[FLINK-38218] Extract duplicate split ID check

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

@@ -63,17 +63,7 @@ public MySqlBinlogSplit(
             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());
-        }
+        ensureNoDuplicates(finishedSnapshotSplitInfos);
 
         this.startingOffset = startingOffset;
         this.endingOffset = endingOffset;
@@ -101,6 +91,21 @@ public MySqlBinlogSplit(
                 false);
     }
 
+    private static void ensureNoDuplicates(
+            List<FinishedSnapshotSplitInfo> finishedSnapshotSplitInfos) {
+        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());
+        }
+    }
+
     public BinlogOffset getStartingOffset() {
         return startingOffset;
     }