[FLINK-38218] Rely on stable order of assigned splits

Author: morozov

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

@@ -33,12 +33,10 @@
 
 import java.util.ArrayList;
 import java.util.Collection;
-import java.util.Comparator;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
-import java.util.stream.Collectors;
 
 /**
  * A {@link MySqlSplitAssigner} that splits tables into small chunk splits based on primary key
@@ -208,9 +206,7 @@ public void close() {
 
     private MySqlBinlogSplit createBinlogSplit() {
         final List<MySqlSchemalessSnapshotSplit> assignedSnapshotSplit =
-                snapshotSplitAssigner.getAssignedSplits().values().stream()
-                        .sorted(Comparator.comparing(MySqlSplit::splitId))
-                        .collect(Collectors.toList());
+                new ArrayList<>(snapshotSplitAssigner.getAssignedSplits().values());
 
         Map<String, BinlogOffset> splitFinishedOffsets =
                 snapshotSplitAssigner.getSplitFinishedOffsets();

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

@@ -169,17 +169,7 @@ private MySqlSnapshotSplitAssigner(
         this.currentParallelism = currentParallelism;
         this.alreadyProcessedTables = alreadyProcessedTables;
         this.remainingSplits = new CopyOnWriteArrayList<>(remainingSplits);
-        // When job restore from savepoint, sort the existing tables and newly added tables
-        // to let enumerator only send newly added tables' BinlogSplitMetaEvent
-        this.assignedSplits =
-                assignedSplits.entrySet().stream()
-                        .sorted(Entry.comparingByKey())
-                        .collect(
-                                Collectors.toMap(
-                                        Entry::getKey,
-                                        Entry::getValue,
-                                        (o, o2) -> o,
-                                        LinkedHashMap::new));
+        this.assignedSplits = assignedSplits;
         this.tableSchemas = tableSchemas;
         this.splitFinishedOffsets = splitFinishedOffsets;
         this.assignerStatus = assignerStatus;

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

@@ -62,10 +62,8 @@
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.HashMap;
-import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
-import java.util.Set;
 import java.util.function.Supplier;
 import java.util.stream.Collectors;
 
@@ -421,15 +419,24 @@ private void fillMetadataForBinlogSplit(BinlogSplitMetaEvent metadataEvent) {
                                 binlogSplit, receivedTotalFinishedSplitSize);
                 uncompletedBinlogSplits.put(binlogSplit.splitId(), binlogSplit);
             } else if (receivedMetaGroupId == expectedMetaGroupId) {
-                List<FinishedSnapshotSplitInfo> newAddedMetadataGroup;
-                Set<String> existedSplitsOfLastGroup =
-                        getExistedSplitsOfLastGroup(
-                                binlogSplit.getFinishedSnapshotSplitInfos(),
-                                sourceConfig.getSplitMetaGroupSize());
-                newAddedMetadataGroup =
-                        metadataEvent.getMetaGroup().stream()
+                int expectedNumberOfAlreadyRetrievedElements =
+                        binlogSplit.getFinishedSnapshotSplitInfos().size()
+                                % sourceConfig.getSplitMetaGroupSize();
+                List<byte[]> metaGroup = metadataEvent.getMetaGroup();
+                if (expectedNumberOfAlreadyRetrievedElements > 0) {
+                    LOG.info(
+                            "Source reader {} is discarding the first {} out of {} elements of meta group {}.",
+                            subtaskId,
+                            expectedNumberOfAlreadyRetrievedElements,
+                            metaGroup.size(),
+                            receivedMetaGroupId);
+                    metaGroup =
+                            metaGroup.subList(
+                                    expectedNumberOfAlreadyRetrievedElements, metaGroup.size());
+                }
+                List<FinishedSnapshotSplitInfo> newAddedMetadataGroup =
+                        metaGroup.stream()
                                 .map(FinishedSnapshotSplitInfo::deserialize)
-                                .filter(r -> !existedSplitsOfLastGroup.contains(r.getSplitId()))
                                 .collect(Collectors.toList());
 
                 uncompletedBinlogSplits.put(
@@ -499,28 +506,6 @@ private MySqlBinlogSplit discoverTableSchemasForBinlogSplit(
         }
     }
 
-    private Set<String> getExistedSplitsOfLastGroup(
-            List<FinishedSnapshotSplitInfo> finishedSnapshotSplits, int metaGroupSize) {
-        int splitsNumOfLastGroup =
-                finishedSnapshotSplits.size() % sourceConfig.getSplitMetaGroupSize();
-        if (splitsNumOfLastGroup != 0) {
-            int lastGroupStart =
-                    ((int) (finishedSnapshotSplits.size() / sourceConfig.getSplitMetaGroupSize()))
-                            * metaGroupSize;
-            // Keep same order with MySqlHybridSplitAssigner.createBinlogSplit() to avoid
-            // 'invalid request meta group id' error
-            List<String> sortedFinishedSnapshotSplits =
-                    finishedSnapshotSplits.stream()
-                            .map(FinishedSnapshotSplitInfo::getSplitId)
-                            .sorted()
-                            .collect(Collectors.toList());
-            return new HashSet<>(
-                    sortedFinishedSnapshotSplits.subList(
-                            lastGroupStart, lastGroupStart + splitsNumOfLastGroup));
-        }
-        return new HashSet<>();
-    }
-
     private void logCurrentBinlogOffsets(List<MySqlSplit> splits, long checkpointId) {
         if (!LOG.isInfoEnabled()) {
             return;