[FLINK-38139][cdc/mysql][follow] Cache Online Schema change DDL Alter AUTO_INCREMENT

Author: beryllw

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

@@ -90,7 +90,7 @@ public class BinlogSplitReader implements DebeziumReader<SourceRecords, MySqlSpl
             new StoppableChangeEventSourceContext();
     private final boolean isParsingOnLineSchemaChanges;
     private final boolean isBackfillSkipped;
-    private final Map<String, SourceRecord> pendingSchemaChangeEvents;
+    private final Map<String, List<SourceRecord>> pendingSchemaChangeEvents;
 
     private static final long READER_CLOSE_TIMEOUT = 30L;
 
@@ -178,7 +178,9 @@ public Iterator<SourceRecords> pollSplitRecords() throws InterruptedException {
                             LOG.info(
                                     "Received the start event of online schema change: {}. Save it for later.",
                                     oscRecord.get());
-                            pendingSchemaChangeEvents.put(tableId.toString(), oscRecord.get());
+                            pendingSchemaChangeEvents
+                                    .computeIfAbsent(tableId.toString(), k -> new ArrayList<>())
+                                    .add(oscRecord.get());
                             continue;
                         }
                     }
@@ -193,7 +195,7 @@ public Iterator<SourceRecords> pollSplitRecords() throws InterruptedException {
                                 finishedTableId);
 
                         if (pendingSchemaChangeEvents.containsKey(finishedTableId)) {
-                            sourceRecords.add(pendingSchemaChangeEvents.remove(finishedTableId));
+                            sourceRecords.addAll(pendingSchemaChangeEvents.remove(finishedTableId));
                         } else {
                             LOG.error(
                                     "Error: met an unexpected osc finish event. Current pending events: {}, Record: {}",