[FLINK-35713][common] Support set parallelism in source and sink

docs/content.zh/docs/core-concept/data-source.md

@@ -37,6 +37,7 @@ under the License.
 | name                          | source 的名称，允许用户配置 (提供了一个默认值)。     | optional          |
 | configurations of Data Source | 用于构建 source 组件的配置，例如连接参数或者表属性的配置。 | optional          |
 
+
 # 示例
 我们可以使用yaml文件来定义一个mysql source：
 ```yaml

docs/content/docs/core-concept/data-sink.md

@@ -35,6 +35,7 @@ To describe a data sink, the follows are required:
 |-----------------------------|-------------------------------------------------------------------------------------------------|-------------------|
 | type                        | The type of the sink, such as doris or starrocks.                                               | required          |
 | name                        | The name of the sink, which is user-defined (a default value provided).                         | optional          |
+| parallelism                 | The parallelism of the sink (or will use `pipeline.parallelism`).                               | optional          |
 | configurations of Data Sink | Configurations to build the Data Sink e.g. connection configurations and sink table properties. | optional          |
 
 # Example

docs/content/docs/core-concept/data-source.md

@@ -35,6 +35,7 @@ To describe a data source, the follows are required:
 |-------------------------------|-----------------------------------------------------------------------------------------------------|-------------------|
 | type                          | The type of the source, such as mysql.                                                              | required          |
 | name                          | The name of the source, which is user-defined (a default value provided).                           | optional          |
+| parallelism                   | The parallelism of the source (or will use `pipeline.parallelism`).                                 | optional          |
 | configurations of Data Source | Configurations to build the Data Source e.g. connection configurations and source table properties. | optional          |
 
 # Example

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

@@ -64,6 +64,7 @@ public class YamlPipelineDefinitionParser implements PipelineDefinitionParser {
     // Source / sink keys
     private static final String TYPE_KEY = "type";
     private static final String NAME_KEY = "name";
+    private static final String PARALLELISM_KEY = "parallelism";
     private static final String INCLUDE_SCHEMA_EVOLUTION_TYPES = "include.schema.changes";
     private static final String EXCLUDE_SCHEMA_EVOLUTION_TYPES = "exclude.schema.changes";
 
@@ -188,7 +189,14 @@ private SourceDef toSourceDef(JsonNode sourceNode) {
         // "name" field is optional
         String name = sourceMap.remove(NAME_KEY);
 
-        return new SourceDef(type, name, Configuration.fromMap(sourceMap));
+        // "parallelism" field is optional
+        Integer parallelismValue = null;
+        String parallelism = sourceMap.remove(PARALLELISM_KEY);
+        if (parallelism != null) {
+            parallelismValue = Integer.parseInt(parallelism);
+        }
+
+        return new SourceDef(type, name, Configuration.fromMap(sourceMap), parallelismValue);
     }
 
     private SinkDef toSinkDef(JsonNode sinkNode, SchemaChangeBehavior schemaChangeBehavior) {
@@ -239,7 +247,15 @@ private SinkDef toSinkDef(JsonNode sinkNode, SchemaChangeBehavior schemaChangeBe
         // "name" field is optional
         String name = sinkMap.remove(NAME_KEY);
 
-        return new SinkDef(type, name, Configuration.fromMap(sinkMap), declaredSETypes);
+        // "parallelism" field is optional
+        Integer parallelismValue = null;
+        String parallelism = sinkMap.remove(PARALLELISM_KEY);
+        if (parallelism != null) {
+            parallelismValue = Integer.parseInt(parallelism);
+        }
+
+        return new SinkDef(
+                type, name, Configuration.fromMap(sinkMap), declaredSETypes, parallelismValue);
     }
 
     private RouteDef toRouteDef(JsonNode routeNode) {

flink-cdc-composer/src/main/java/org/apache/flink/cdc/composer/PipelineExecution.java

@@ -17,12 +17,22 @@
 
 package org.apache.flink.cdc.composer;
 
+import org.apache.flink.cdc.common.annotation.VisibleForTesting;
+
 /** A pipeline execution that can be executed by a computing engine. */
 public interface PipelineExecution {
 
     /** Execute the pipeline. */
     ExecutionInfo execute() throws Exception;
 
+    /**
+     * Get the plan of the pipeline.
+     *
+     * @return The plan of the pipeline.
+     */
+    @VisibleForTesting
+    String getPlan() throws Exception;
+
     /** Information of the execution. */
     class ExecutionInfo {
         private final String id;

flink-cdc-composer/src/main/java/org/apache/flink/cdc/composer/definition/SinkDef.java

@@ -44,26 +44,37 @@
 public class SinkDef {
     private final String type;
     @Nullable private final String name;
+    @Nullable private final Integer parallelism;
     private final Configuration config;
     private final Set<SchemaChangeEventType> includedSchemaEvolutionTypes;
 
     public SinkDef(String type, @Nullable String name, Configuration config) {
-        this.type = type;
-        this.name = name;
-        this.config = config;
-        this.includedSchemaEvolutionTypes =
-                Arrays.stream(SchemaChangeEventTypeFamily.ALL).collect(Collectors.toSet());
+        this(
+                type,
+                name,
+                config,
+                Arrays.stream(SchemaChangeEventTypeFamily.ALL).collect(Collectors.toSet()));
     }
 
     public SinkDef(
             String type,
             @Nullable String name,
             Configuration config,
             Set<SchemaChangeEventType> includedSchemaEvolutionTypes) {
+        this(type, name, config, includedSchemaEvolutionTypes, null);
+    }
+
+    public SinkDef(
+            String type,
+            @Nullable String name,
+            Configuration config,
+            Set<SchemaChangeEventType> includedSchemaEvolutionTypes,
+            @Nullable Integer parallelism) {
         this.type = type;
         this.name = name;
         this.config = config;
         this.includedSchemaEvolutionTypes = includedSchemaEvolutionTypes;
+        this.parallelism = parallelism;
     }
 
     public String getType() {
@@ -82,6 +93,10 @@ public Set<SchemaChangeEventType> getIncludedSchemaEvolutionTypes() {
         return includedSchemaEvolutionTypes;
     }
 
+    public Optional<Integer> getParallelism() {
+        return Optional.ofNullable(parallelism);
+    }
+
     @Override
     public String toString() {
         return "SinkDef{"
@@ -91,6 +106,8 @@ public String toString() {
                 + ", name='"
                 + name
                 + '\''
+                + ", parallelism="
+                + parallelism
                 + ", config="
                 + config
                 + ", includedSchemaEvolutionTypes="
@@ -109,13 +126,14 @@ public boolean equals(Object o) {
         SinkDef sinkDef = (SinkDef) o;
         return Objects.equals(type, sinkDef.type)
                 && Objects.equals(name, sinkDef.name)
+                && Objects.equals(parallelism, sinkDef.parallelism)
                 && Objects.equals(config, sinkDef.config)
                 && Objects.equals(
                         includedSchemaEvolutionTypes, sinkDef.includedSchemaEvolutionTypes);
     }
 
     @Override
     public int hashCode() {
-        return Objects.hash(type, name, config, includedSchemaEvolutionTypes);
+        return Objects.hash(type, name, config, includedSchemaEvolutionTypes, parallelism);
     }
 }

flink-cdc-composer/src/main/java/org/apache/flink/cdc/composer/definition/SourceDef.java

@@ -39,12 +39,22 @@
 public class SourceDef {
     private final String type;
     @Nullable private final String name;
+    @Nullable private final Integer parallelism;
     private final Configuration config;
 
     public SourceDef(String type, @Nullable String name, Configuration config) {
+        this(type, name, config, null);
+    }
+
+    public SourceDef(
+            String type,
+            @Nullable String name,
+            Configuration config,
+            @Nullable Integer parallelism) {
         this.type = type;
         this.name = name;
         this.config = config;
+        this.parallelism = parallelism;
     }
 
     public String getType() {
@@ -59,6 +69,10 @@ public Configuration getConfig() {
         return config;
     }
 
+    public Optional<Integer> getParallelism() {
+        return Optional.ofNullable(parallelism);
+    }
+
     @Override
     public String toString() {
         return "SourceDef{"
@@ -68,6 +82,8 @@ public String toString() {
                 + ", name='"
                 + name
                 + '\''
+                + ", parallelism="
+                + parallelism
                 + ", config="
                 + config
                 + '}';
@@ -84,11 +100,12 @@ public boolean equals(Object o) {
         SourceDef sourceDef = (SourceDef) o;
         return Objects.equals(type, sourceDef.type)
                 && Objects.equals(name, sourceDef.name)
-                && Objects.equals(config, sourceDef.config);
+                && Objects.equals(config, sourceDef.config)
+                && Objects.equals(parallelism, sourceDef.parallelism);
     }
 
     @Override
     public int hashCode() {
-        return Objects.hash(type, name, config);
+        return Objects.hash(type, name, config, parallelism);
     }
 }

flink-cdc-composer/src/main/java/org/apache/flink/cdc/composer/flink/FlinkPipelineComposer.java

@@ -135,7 +135,11 @@ private void translate(StreamExecutionEnvironment env, PipelineDef pipelineDef)
 
         // O ---> Source
         DataStream<Event> stream =
-                sourceTranslator.translate(pipelineDef.getSource(), dataSource, env, parallelism);
+                sourceTranslator.translate(
+                        pipelineDef.getSource(),
+                        dataSource,
+                        env,
+                        pipelineDef.getSource().getParallelism().orElse(parallelism));
 
         // Source ---> PreTransform
         stream =
@@ -205,7 +209,11 @@ private void translate(StreamExecutionEnvironment env, PipelineDef pipelineDef)
 
         // Schema Operator -> Sink -> X
         sinkTranslator.translate(
-                pipelineDef.getSink(), stream, dataSink, schemaOperatorIDGenerator.generate());
+                pipelineDef.getSink(),
+                stream,
+                dataSink,
+                schemaOperatorIDGenerator.generate(),
+                pipelineDef.getSink().getParallelism().orElse(parallelism));
     }
 
     private void addFrameworkJars() {

flink-cdc-composer/src/main/java/org/apache/flink/cdc/composer/flink/FlinkPipelineExecution.java

@@ -46,4 +46,9 @@ public ExecutionInfo execute() throws Exception {
         }
         return new ExecutionInfo(jobClient.getJobID().toString(), jobName);
     }
+
+    @Override
+    public String getPlan() throws Exception {
+        return env.getExecutionPlan();
+    }
 }

flink-cdc-composer/src/main/java/org/apache/flink/cdc/composer/flink/translator/DataSinkTranslator.java

@@ -81,21 +81,22 @@ public void translate(
             SinkDef sinkDef,
             DataStream<Event> input,
             DataSink dataSink,
-            OperatorID schemaOperatorID) {
+            OperatorID schemaOperatorID,
+            int parallelism) {
         // Get sink provider
         EventSinkProvider eventSinkProvider = dataSink.getEventSinkProvider();
         String sinkName = generateSinkName(sinkDef);
         if (eventSinkProvider instanceof FlinkSinkProvider) {
             // Sink V2
             FlinkSinkProvider sinkProvider = (FlinkSinkProvider) eventSinkProvider;
             Sink<Event> sink = sinkProvider.getSink();
-            sinkTo(input, sink, sinkName, schemaOperatorID);
+            sinkTo(input, sink, sinkName, schemaOperatorID, parallelism);
         } else if (eventSinkProvider instanceof FlinkSinkFunctionProvider) {
             // SinkFunction
             FlinkSinkFunctionProvider sinkFunctionProvider =
                     (FlinkSinkFunctionProvider) eventSinkProvider;
             SinkFunction<Event> sinkFunction = sinkFunctionProvider.getSinkFunction();
-            sinkTo(input, sinkFunction, sinkName, schemaOperatorID);
+            sinkTo(input, sinkFunction, sinkName, schemaOperatorID, parallelism);
         }
     }
 
@@ -104,8 +105,11 @@ void sinkTo(
             DataStream<Event> input,
             Sink<Event> sink,
             String sinkName,
-            OperatorID schemaOperatorID) {
+            OperatorID schemaOperatorID,
+            int parallelism) {
         DataStream<Event> stream = input;
+        stream.getExecutionEnvironment().setParallelism(parallelism);
+
         // Pre-write topology
         if (sink instanceof WithPreWriteTopology) {
             stream = ((WithPreWriteTopology<Event>) sink).addPreWriteTopology(stream);
@@ -125,7 +129,8 @@ private void sinkTo(
             DataStream<Event> input,
             SinkFunction<Event> sinkFunction,
             String sinkName,
-            OperatorID schemaOperatorID) {
+            OperatorID schemaOperatorID,
+            int parallelism) {
         DataSinkFunctionOperator sinkOperator =
                 new DataSinkFunctionOperator(sinkFunction, schemaOperatorID);
         final StreamExecutionEnvironment executionEnvironment = input.getExecutionEnvironment();
@@ -134,7 +139,7 @@ private void sinkTo(
                         input.getTransformation(),
                         SINK_WRITER_PREFIX + sinkName,
                         sinkOperator,
-                        executionEnvironment.getParallelism(),
+                        parallelism,
                         false);
         executionEnvironment.addOperator(transformation);
     }