[FLINK-36794] [cdc-composer/cli] pipeline cdc connector support multiple data sources

docs/content.zh/docs/connectors/pipeline-connectors/mysql.md

@@ -50,9 +50,9 @@ MySQL CDC Pipeline 连接器允许从 MySQL 数据库读取快照数据和增量
 </table>
 </div>
 
-## 示例
+## 单数据源示例
 
-从 MySQL 读取数据同步到 Doris 的 Pipeline 可以定义如下：
+单数据源，从单个 MySQL 读取数据同步到 Doris 的 Pipeline 可以定义如下：
 
 ```yaml
 source:
@@ -77,6 +77,32 @@ pipeline:
    parallelism: 4
 ```
 
+## 多数据源示例
+
+单数据源，从多个 MySQL 读取数据同步到 Doris 的 Pipeline 可以定义如下：
+
+```yaml
+source:
+   type: mysql_mutiple
+   name: MySQL mutiple Source
+   host_list: 127.0.0.1:3306,127.0.0.2:3307
+   username: admin
+   password: pass
+   tables: adb.\.*, bdb.user_table_[0-9]+, [app|web].order_\.*
+   server-id: 5401-5404
+
+sink:
+  type: doris
+  name: Doris Sink
+  fenodes: 127.0.0.1:8030
+  username: root
+  password: pass
+
+pipeline:
+   name: MySQL to Doris Pipeline
+   parallelism: 4
+```
+
 ## 连接器配置项
 
 <div class="highlight">

docs/content/docs/connectors/pipeline-connectors/mysql.md

@@ -51,9 +51,9 @@ You may need to configure the following dependencies manually, and pass it with
 </table>
 </div>
 
-## Example
+## single data source Example
 
-An example of the pipeline for reading data from MySQL and sink to Doris can be defined as follows:
+An example of the pipeline for reading data from single MySQL and sink to Doris can be defined as follows:
 
 ```yaml
 source:
@@ -78,6 +78,32 @@ pipeline:
    parallelism: 4
 ```
 
+## mutiple data source Example
+
+An example of the pipeline for reading data from mutiple MySQL and sink to Doris can be defined as follows:
+
+```yaml
+source:
+   type: mysql_mutiple
+   name: MySQL mutiple Source
+   host_list: 127.0.0.1:3306,127.0.0.2:3307
+   username: admin
+   password: pass
+   tables: adb.\.*, bdb.user_table_[0-9]+, [app|web].order_\.*
+   server-id: 5401-5404
+
+sink:
+  type: doris
+  name: Doris Sink
+  fenodes: 127.0.0.1:8030
+  username: root
+  password: pass
+
+pipeline:
+   name: MySQL to Doris Pipeline
+   parallelism: 4
+```
+
 ## Connector Options
 
 <div class="highlight">

flink-cdc-cli/src/main/java/org/apache/flink/cdc/cli/parser/YamlPipelineDefinitionParser.java

@@ -97,6 +97,13 @@ public class YamlPipelineDefinitionParser implements PipelineDefinitionParser {
 
     public static final String TRANSFORM_TABLE_OPTION_KEY = "table-options";
 
+    private static final String HOST_LIST = "host_list";
+    private static final String COMMA = ",";
+    private static final String HOST_NAME = "hostname";
+    private static final String PORT = "port";
+    private static final String COLON = ":";
+    private static final String MUTIPLE = "_mutiple";
+
     private final ObjectMapper mapper = new ObjectMapper(new YAMLFactory());
 
     /** Parse the specified pipeline definition file. */
@@ -135,13 +142,35 @@ private PipelineDef parse(JsonNode pipelineDefJsonNode, Configuration globalPipe
         SchemaChangeBehavior schemaChangeBehavior =
                 userPipelineConfig.get(PIPELINE_SCHEMA_CHANGE_BEHAVIOR);
 
+        //        JsonNode root = mapper.readTree(pipelineDefPath.toFile());
+        JsonNode sourceNode = pipelineDefJsonNode.get(SOURCE_KEY);
+        JsonNode hostList = sourceNode.get(HOST_LIST);
+        String type = sourceNode.get(TYPE_KEY).asText();
+        List<SourceDef> sourceDefs = new ArrayList<>();
+        if (hostList != null && type.contains(MUTIPLE)) {
+            String hostString = hostList.asText();
+            String[] hosts = hostString.split(COMMA);
+            Arrays.stream(hosts)
+                    .forEach(
+                            e -> {
+                                ((ObjectNode) sourceNode)
+                                        .put(TYPE_KEY, type.substring(0, type.indexOf("_")));
+                                ((ObjectNode) sourceNode).put(HOST_NAME, e.split(COLON)[0]);
+                                ((ObjectNode) sourceNode).put(PORT, e.split(COLON)[1]);
+                                ((ObjectNode) sourceNode).remove(HOST_LIST);
+                                getSourceDefs(sourceNode, sourceDefs);
+                            });
+        } else {
+            // Source is required
+            getSourceDefs(sourceNode, sourceDefs);
+        }
         // Source is required
-        SourceDef sourceDef =
-                toSourceDef(
-                        checkNotNull(
-                                pipelineDefJsonNode.get(SOURCE_KEY),
-                                "Missing required field \"%s\" in pipeline definition",
-                                SOURCE_KEY));
+        //        SourceDef sourceDef =
+        //                toSourceDef(
+        //                        checkNotNull(
+        //                                pipelineDefJsonNode.get(SOURCE_KEY),
+        //                                "Missing required field \"%s\" in pipeline definition",
+        //                                SOURCE_KEY));
 
         // Sink is required
         SinkDef sinkDef =
@@ -171,7 +200,17 @@ private PipelineDef parse(JsonNode pipelineDefJsonNode, Configuration globalPipe
         pipelineConfig.addAll(userPipelineConfig);
 
         return new PipelineDef(
-                sourceDef, sinkDef, routeDefs, transformDefs, udfDefs, modelDefs, pipelineConfig);
+                sourceDefs, sinkDef, routeDefs, transformDefs, udfDefs, modelDefs, pipelineConfig);
+    }
+
+    private void getSourceDefs(JsonNode root, List<SourceDef> sourceDefs) {
+        SourceDef sourceDef =
+                toSourceDef(
+                        checkNotNull(
+                                root,
+                                "Missing required field \"%s\" in pipeline definition",
+                                SOURCE_KEY));
+        sourceDefs.add(sourceDef);
     }
 
     private SourceDef toSourceDef(JsonNode sourceNode) {