[FLINK-38218] Require assigned snapshot splits to be ordered

Author: morozov

flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/assigners/MySqlSnapshotSplitAssigner.java

@@ -26,6 +26,7 @@
 import org.apache.flink.cdc.connectors.mysql.source.config.MySqlSourceOptions;
 import org.apache.flink.cdc.connectors.mysql.source.connection.JdbcConnectionPools;
 import org.apache.flink.cdc.connectors.mysql.source.offset.BinlogOffset;
+import org.apache.flink.cdc.connectors.mysql.source.reader.MySqlSourceReader;
 import org.apache.flink.cdc.connectors.mysql.source.split.FinishedSnapshotSplitInfo;
 import org.apache.flink.cdc.connectors.mysql.source.split.MySqlSchemalessSnapshotSplit;
 import org.apache.flink.cdc.connectors.mysql.source.split.MySqlSnapshotSplit;
@@ -73,7 +74,21 @@ public class MySqlSnapshotSplitAssigner implements MySqlSplitAssigner {
 
     private final List<TableId> alreadyProcessedTables;
     private final List<MySqlSchemalessSnapshotSplit> remainingSplits;
-    private final Map<String, MySqlSchemalessSnapshotSplit> assignedSplits;
+
+    /**
+     * The splits that have been assigned to a reader. Once a split is finished, it remains in this
+     * map. An entry added to {@link #splitFinishedOffsets} indicates that the split has been
+     * finished. If reading the split fails, it is removed from this map.
+     *
+     * <p>{@link MySqlSourceReader} relies on the order of elements within the map:
+     *
+     * <ol>
+     *   <li>It must correspond to the order of assignment of the splits to readers.
+     *   <li>The order must be retained across job restarts.
+     * </ol>
+     */
+    private final LinkedHashMap<String, MySqlSchemalessSnapshotSplit> assignedSplits;
+
     private final Map<TableId, TableChanges.TableChange> tableSchemas;
     private final Map<String, BinlogOffset> splitFinishedOffsets;
     private final MySqlSourceConfig sourceConfig;
@@ -141,7 +156,7 @@ private MySqlSnapshotSplitAssigner(
             int currentParallelism,
             List<TableId> alreadyProcessedTables,
             List<MySqlSchemalessSnapshotSplit> remainingSplits,
-            Map<String, MySqlSchemalessSnapshotSplit> assignedSplits,
+            LinkedHashMap<String, MySqlSchemalessSnapshotSplit> assignedSplits,
             Map<TableId, TableChanges.TableChange> tableSchemas,
             Map<String, BinlogOffset> splitFinishedOffsets,
             AssignerStatus assignerStatus,

flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/assigners/state/PendingSplitsStateSerializer.java

@@ -35,6 +35,7 @@
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.HashMap;
+import java.util.LinkedHashMap;
 import java.util.List;
 import java.util.Map;
 
@@ -196,12 +197,12 @@ private SnapshotPendingSplitsState deserializeLegacySnapshotPendingSplitsState(
             int splitVersion, DataInputDeserializer in) throws IOException {
         List<TableId> alreadyProcessedTables = readTableIds(in);
         List<MySqlSnapshotSplit> remainingSplits = readMySqlSnapshotSplits(splitVersion, in);
-        Map<String, MySqlSnapshotSplit> assignedSnapshotSplits =
+        LinkedHashMap<String, MySqlSnapshotSplit> assignedSnapshotSplits =
                 readAssignedSnapshotSplits(splitVersion, in);
 
         final List<MySqlSchemalessSnapshotSplit> remainingSchemalessSplits = new ArrayList<>();
-        final Map<String, MySqlSchemalessSnapshotSplit> assignedSchemalessSnapshotSplits =
-                new HashMap<>();
+        final LinkedHashMap<String, MySqlSchemalessSnapshotSplit> assignedSchemalessSnapshotSplits =
+                new LinkedHashMap<>();
         final Map<TableId, TableChanges.TableChange> tableSchemas = new HashMap<>();
         remainingSplits.forEach(
                 split -> {
@@ -267,8 +268,8 @@ private SnapshotPendingSplitsState deserializeSnapshotPendingSplitsState(
         List<TableId> remainingTableIds = readTableIds(in);
         boolean isTableIdCaseSensitive = in.readBoolean();
         final List<MySqlSchemalessSnapshotSplit> remainingSchemalessSplits = new ArrayList<>();
-        final Map<String, MySqlSchemalessSnapshotSplit> assignedSchemalessSnapshotSplits =
-                new HashMap<>();
+        final LinkedHashMap<String, MySqlSchemalessSnapshotSplit> assignedSchemalessSnapshotSplits =
+                new LinkedHashMap<>();
         final Map<TableId, TableChanges.TableChange> tableSchemas = new HashMap<>();
         remainingSplits.forEach(
                 split -> {
@@ -368,9 +369,9 @@ private void writeAssignedSnapshotSplits(
         }
     }
 
-    private Map<String, MySqlSnapshotSplit> readAssignedSnapshotSplits(
+    private LinkedHashMap<String, MySqlSnapshotSplit> readAssignedSnapshotSplits(
             int splitVersion, DataInputDeserializer in) throws IOException {
-        Map<String, MySqlSnapshotSplit> assignedSplits = new HashMap<>();
+        LinkedHashMap<String, MySqlSnapshotSplit> assignedSplits = new LinkedHashMap<>();
         final int size = in.readInt();
         for (int i = 0; i < size; i++) {
             String splitId = in.readUTF();

flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/assigners/state/SnapshotPendingSplitsState.java

@@ -27,6 +27,7 @@
 import io.debezium.relational.TableId;
 import io.debezium.relational.history.TableChanges.TableChange;
 
+import java.util.LinkedHashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
@@ -50,7 +51,7 @@ public class SnapshotPendingSplitsState extends PendingSplitsState {
      * The snapshot splits that the {@link MySqlSourceEnumerator} has assigned to {@link
      * MySqlSplitReader}s.
      */
-    private final Map<String, MySqlSchemalessSnapshotSplit> assignedSplits;
+    private final LinkedHashMap<String, MySqlSchemalessSnapshotSplit> assignedSplits;
 
     /**
      * The offsets of finished (snapshot) splits that the {@link MySqlSourceEnumerator} has received
@@ -75,7 +76,7 @@ public class SnapshotPendingSplitsState extends PendingSplitsState {
     public SnapshotPendingSplitsState(
             List<TableId> alreadyProcessedTables,
             List<MySqlSchemalessSnapshotSplit> remainingSplits,
-            Map<String, MySqlSchemalessSnapshotSplit> assignedSplits,
+            LinkedHashMap<String, MySqlSchemalessSnapshotSplit> assignedSplits,
             Map<TableId, TableChange> tableSchemas,
             Map<String, BinlogOffset> splitFinishedOffsets,
             AssignerStatus assignerStatus,
@@ -103,7 +104,7 @@ public List<MySqlSchemalessSnapshotSplit> getRemainingSplits() {
         return remainingSplits;
     }
 
-    public Map<String, MySqlSchemalessSnapshotSplit> getAssignedSplits() {
+    public LinkedHashMap<String, MySqlSchemalessSnapshotSplit> getAssignedSplits() {
         return assignedSplits;
     }
 

flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/test/java/org/apache/flink/cdc/connectors/mysql/source/assigners/MySqlHybridSplitAssignerTest.java

@@ -47,6 +47,7 @@
 import java.util.Arrays;
 import java.util.Comparator;
 import java.util.HashMap;
+import java.util.LinkedHashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
@@ -81,7 +82,7 @@ void testAssignMySqlBinlogSplitAfterAllSnapshotSplitsFinished() {
         List<TableId> alreadyProcessedTables = Lists.newArrayList(tableId);
         List<MySqlSchemalessSnapshotSplit> remainingSplits = new ArrayList<>();
 
-        Map<String, MySqlSchemalessSnapshotSplit> assignedSplits = new HashMap<>();
+        LinkedHashMap<String, MySqlSchemalessSnapshotSplit> assignedSplits = new LinkedHashMap<>();
         Map<String, BinlogOffset> splitFinishedOffsets = new HashMap<>();
 
         for (int i = 0; i < 5; i++) {

flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/test/java/org/apache/flink/cdc/connectors/mysql/source/assigners/MySqlSnapshotSplitAssignerTest.java

@@ -42,6 +42,7 @@
 import java.util.Arrays;
 import java.util.Collections;
 import java.util.HashMap;
+import java.util.LinkedHashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
@@ -647,7 +648,7 @@ private List<String> getTestAssignSnapshotSplitsFromCheckpoint(AssignerStatus as
                                 null,
                                 null));
 
-        Map<String, MySqlSchemalessSnapshotSplit> assignedSplits = new HashMap<>();
+        LinkedHashMap<String, MySqlSchemalessSnapshotSplit> assignedSplits = new LinkedHashMap<>();
         assignedSplits.put(
                 processedTable + ":0",
                 new MySqlSchemalessSnapshotSplit(

flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/test/java/org/apache/flink/cdc/connectors/mysql/source/assigners/state/PendingSplitsStateSerializerTest.java

@@ -39,6 +39,7 @@
 import java.util.Arrays;
 import java.util.Collections;
 import java.util.HashMap;
+import java.util.LinkedHashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.stream.Stream;
@@ -147,7 +148,8 @@ private static SnapshotPendingSplitsState getTestSnapshotPendingSplitsState(
         remainingSplits.add(getTestSchemalessSnapshotSplit(tableId1, 2));
         remainingSplits.add(getTestSchemalessSnapshotSplit(tableId1, 3));
 
-        final Map<String, MySqlSchemalessSnapshotSplit> assignedSnapshotSplits = new HashMap<>();
+        final LinkedHashMap<String, MySqlSchemalessSnapshotSplit> assignedSnapshotSplits =
+                new LinkedHashMap<>();
         Arrays.asList(
                         getTestSchemalessSnapshotSplit(tableId0, 0),
                         getTestSchemalessSnapshotSplit(tableId0, 1),