[mysql] support binlog deserialize with parallel threads

Author: guoxuanlin

flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-mysql/src/main/java/org/apache/flink/cdc/connectors/mysql/factory/MySqlDataSourceFactory.java

@@ -77,6 +77,7 @@
 import static org.apache.flink.cdc.connectors.mysql.source.MySqlDataSourceOptions.PORT;
 import static org.apache.flink.cdc.connectors.mysql.source.MySqlDataSourceOptions.SCAN_BINLOG_NEWLY_ADDED_TABLE_ENABLED;
 import static org.apache.flink.cdc.connectors.mysql.source.MySqlDataSourceOptions.SCAN_INCREMENTAL_CLOSE_IDLE_READER_ENABLED;
+import static org.apache.flink.cdc.connectors.mysql.source.MySqlDataSourceOptions.SCAN_INCREMENTAL_DESERIALIZE_PARALLELISM;
 import static org.apache.flink.cdc.connectors.mysql.source.MySqlDataSourceOptions.SCAN_INCREMENTAL_SNAPSHOT_BACKFILL_SKIP;
 import static org.apache.flink.cdc.connectors.mysql.source.MySqlDataSourceOptions.SCAN_INCREMENTAL_SNAPSHOT_CHUNK_KEY_COLUMN;
 import static org.apache.flink.cdc.connectors.mysql.source.MySqlDataSourceOptions.SCAN_INCREMENTAL_SNAPSHOT_CHUNK_SIZE;
@@ -170,6 +171,7 @@ public DataSource createDataSource(Context context) {
                 config.get(SCAN_INCREMENTAL_SNAPSHOT_UNBOUNDED_CHUNK_FIRST_ENABLED);
         boolean onlyDeserializeCapturedTablesChangelog =
                 config.get(SCAN_ONLY_DESERIALIZE_CAPTURED_TABLES_CHANGELOG_ENABLED);
+        int deserializeParallelism = config.get(SCAN_INCREMENTAL_DESERIALIZE_PARALLELISM);
 
         validateIntegerOption(SCAN_INCREMENTAL_SNAPSHOT_CHUNK_SIZE, splitSize, 1);
         validateIntegerOption(CHUNK_META_GROUP_SIZE, splitMetaGroupSize, 1);
@@ -225,7 +227,8 @@ public DataSource createDataSource(Context context) {
                         .assignUnboundedChunkFirst(isAssignUnboundedChunkFirst)
                         .skipSnapshotBackfill(skipSnapshotBackfill)
                         .onlyDeserializeCapturedTablesChangelog(
-                                onlyDeserializeCapturedTablesChangelog);
+                                onlyDeserializeCapturedTablesChangelog)
+                        .deserializeParallelism(deserializeParallelism);
 
         List<TableId> tableIds = MySqlSchemaUtils.listTables(configFactory.createConfig(0), null);
 
@@ -364,6 +367,7 @@ public Set<ConfigOption<?>> optionalOptions() {
         options.add(SCAN_INCREMENTAL_SNAPSHOT_UNBOUNDED_CHUNK_FIRST_ENABLED);
         options.add(SCAN_INCREMENTAL_SNAPSHOT_BACKFILL_SKIP);
         options.add(SCAN_ONLY_DESERIALIZE_CAPTURED_TABLES_CHANGELOG_ENABLED);
+        options.add(SCAN_INCREMENTAL_DESERIALIZE_PARALLELISM);
         return options;
     }
 

flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-mysql/src/main/java/org/apache/flink/cdc/connectors/mysql/source/MySqlDataSourceOptions.java

@@ -341,4 +341,15 @@ public class MySqlDataSourceOptions {
                                     "Whether to only deserialize changelog events for captured tables during incremental phase. "
                                             + "When set to true, only changelog events for the target tables will be deserialized, "
                                             + "which can speed up binlog reading. Defaults to false.");
+
+    @Experimental
+    public static final ConfigOption<Integer> SCAN_INCREMENTAL_DESERIALIZE_PARALLELISM =
+            ConfigOptions.key("scan.incremental.deserialize.parallelism")
+                    .intType()
+                    .defaultValue(1)
+                    .withDescription(
+                            "The number of parallel threads used to deserialize binlog events "
+                                    + "during the incremental phase. Default is 1 (single-threaded, "
+                                    + "original behavior). Setting this to a value greater than 1 "
+                                    + "enables concurrent deserialization to improve throughput.");
 }

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

@@ -98,7 +98,8 @@ public MySqlPipelineRecordEmitter(
                 sourceReaderMetrics,
                 sourceConfig.isIncludeSchemaChanges(),
                 false, // Explicitly disable heartbeat events
-                false); // Explicitly disable transaction metadata events
+                false, // Explicitly disable transaction metadata events
+                sourceConfig.getDeserializeParallelism());
         this.debeziumDeserializationSchema = debeziumDeserializationSchema;
         this.sourceConfig = sourceConfig;
         this.alreadySendCreateTableTables = new HashSet<>();

flink-cdc-connect/flink-cdc-source-connectors/flink-connector-debezium/src/main/java/org/apache/flink/cdc/debezium/event/DebeziumEventDeserializationSchema.java

@@ -68,7 +68,6 @@
 import java.time.Instant;
 import java.util.Collections;
 import java.util.Date;
-import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
@@ -98,7 +97,7 @@ public DebeziumEventDeserializationSchema(
             SchemaDataTypeInference schemaDataTypeInference, DebeziumChangelogMode changelogMode) {
         this.schemaDataTypeInference = schemaDataTypeInference;
         this.changelogMode = changelogMode;
-        this.createTableEventCache = new HashMap<>();
+        this.createTableEventCache = new ConcurrentHashMap<>();
     }
 
     @Override

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

@@ -131,7 +131,8 @@ public static <T> MySqlSourceBuilder<T> builder() {
                                 sourceReaderMetrics,
                                 sourceConfig.isIncludeSchemaChanges(),
                                 sourceConfig.isIncludeHeartbeatEvents(),
-                                sourceConfig.isIncludeTransactionMetadataEvents()));
+                                sourceConfig.isIncludeTransactionMetadataEvents(),
+                                sourceConfig.getDeserializeParallelism()));
     }
 
     MySqlSource(

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

@@ -312,6 +312,15 @@ public MySqlSourceBuilder<T> assignUnboundedChunkFirst(boolean assignUnboundedCh
         return this;
     }
 
+    /**
+     * The number of parallel threads used to deserialize binlog events during the incremental
+     * phase. Default is 1 (single-threaded, original behavior).
+     */
+    public MySqlSourceBuilder<T> deserializeParallelism(int deserializeParallelism) {
+        this.configFactory.deserializeParallelism(deserializeParallelism);
+        return this;
+    }
+
     /**
      * Build the {@link MySqlSource}.
      *

flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/config/MySqlSourceConfig.java

@@ -73,6 +73,7 @@ public class MySqlSourceConfig implements Serializable {
     public static boolean useLegacyJsonFormat = true;
     private final boolean assignUnboundedChunkFirst;
     private final boolean onlyDeserializeCapturedTablesChangelog;
+    private final int deserializeParallelism;
 
     // --------------------------------------------------------------------------------------------
     // Debezium Configurations
@@ -114,7 +115,8 @@ public class MySqlSourceConfig implements Serializable {
             boolean treatTinyInt1AsBoolean,
             boolean useLegacyJsonFormat,
             boolean assignUnboundedChunkFirst,
-            boolean onlyDeserializeCapturedTablesChangelog) {
+            boolean onlyDeserializeCapturedTablesChangelog,
+            int deserializeParallelism) {
         this.hostname = checkNotNull(hostname);
         this.port = port;
         this.username = checkNotNull(username);
@@ -161,6 +163,7 @@ public class MySqlSourceConfig implements Serializable {
         this.useLegacyJsonFormat = useLegacyJsonFormat;
         this.assignUnboundedChunkFirst = assignUnboundedChunkFirst;
         this.onlyDeserializeCapturedTablesChangelog = onlyDeserializeCapturedTablesChangelog;
+        this.deserializeParallelism = deserializeParallelism;
     }
 
     public String getHostname() {
@@ -306,4 +309,8 @@ public boolean isTreatTinyInt1AsBoolean() {
     public boolean isOnlyDeserializeCapturedTablesChangelog() {
         return onlyDeserializeCapturedTablesChangelog;
     }
+
+    public int getDeserializeParallelism() {
+        return deserializeParallelism;
+    }
 }

flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/config/MySqlSourceConfigFactory.java

@@ -79,6 +79,7 @@ public class MySqlSourceConfigFactory implements Serializable {
     private boolean useLegacyJsonFormat = true;
     private boolean assignUnboundedChunkFirst = false;
     private boolean onlyDeserializeCapturedTablesChangelog = false;
+    private int deserializeParallelism = 1;
 
     public MySqlSourceConfigFactory hostname(String hostname) {
         this.hostname = hostname;
@@ -352,6 +353,28 @@ public MySqlSourceConfigFactory onlyDeserializeCapturedTablesChangelog(
         return this;
     }
 
+    /**
+     * The number of parallel threads used to deserialize binlog events during the incremental
+     * phase. Default is 1 (single-threaded, original behavior). Setting this to a value greater
+     * than 1 enables concurrent deserialization to improve throughput.
+     */
+    public MySqlSourceConfigFactory deserializeParallelism(int deserializeParallelism) {
+        if (deserializeParallelism <= 0) {
+            throw new IllegalArgumentException(
+                    String.format(
+                            "The value of 'scan.incremental.deserialize.parallelism' must be positive, but is %d",
+                            deserializeParallelism));
+        }
+        if (deserializeParallelism > 256) {
+            throw new IllegalArgumentException(
+                    String.format(
+                            "The value of 'scan.incremental.deserialize.parallelism' must not exceed 256, but is %d",
+                            deserializeParallelism));
+        }
+        this.deserializeParallelism = deserializeParallelism;
+        return this;
+    }
+
     /** Creates a new {@link MySqlSourceConfig} for the given subtask {@code subtaskId}. */
     public MySqlSourceConfig createConfig(int subtaskId) {
         // hard code server name, because we don't need to distinguish it, docs:
@@ -456,6 +479,7 @@ public MySqlSourceConfig createConfig(int subtaskId, String serverName) {
                 treatTinyInt1AsBoolean,
                 useLegacyJsonFormat,
                 assignUnboundedChunkFirst,
-                onlyDeserializeCapturedTablesChangelog);
+                onlyDeserializeCapturedTablesChangelog,
+                deserializeParallelism);
     }
 }

flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/config/MySqlSourceOptions.java

@@ -292,4 +292,15 @@ public class MySqlSourceOptions {
                     .defaultValue(true)
                     .withDescription(
                             "Whether to assign the unbounded chunks first during snapshot reading phase. This might help reduce the risk of the TaskManager experiencing an out-of-memory (OOM) error when taking a snapshot of the largest unbounded chunk.");
+
+    @Experimental
+    public static final ConfigOption<Integer> SCAN_INCREMENTAL_DESERIALIZE_PARALLELISM =
+            ConfigOptions.key("scan.incremental.deserialize.parallelism")
+                    .intType()
+                    .defaultValue(1)
+                    .withDescription(
+                            "The number of parallel threads used to deserialize binlog events "
+                                    + "during the incremental phase. Default is 1 (single-threaded, "
+                                    + "original behavior). Setting this to a value greater than 1 "
+                                    + "enables concurrent deserialization to improve throughput.");
 }