[FLINK-36581][cli] Allow passing Flink configuration by yaml job file

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

@@ -46,7 +46,8 @@ sink:
 
 pipeline:
    parallelism: 1
-
+   flink-conf:
+     execution.checkpointing.interval: 2min
 ```
 
 ## 连接器配置项

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

@@ -58,6 +58,8 @@ route:
 pipeline:
   name: MySQL to Elasticsearch Pipeline
   parallelism: 2
+  flink-conf:
+    execution.checkpointing.interval: 2min
 ```
 
 Pipeline Connector Options

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

@@ -57,6 +57,8 @@ sink:
 pipeline:
   name: MySQL to Kafka Pipeline
   parallelism: 2
+  flink-conf:
+    execution.checkpointing.interval: 2min
 ```
 
 Pipeline 连接器配置项

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

@@ -63,6 +63,8 @@ sink:
 pipeline:
   name: MySQL to MaxCompute Pipeline
   parallelism: 2
+  flink-conf:
+    execution.checkpointing.interval: 2min
 ```
 
 ## 连接器配置项

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

@@ -75,6 +75,8 @@ sink:
 pipeline:
    name: MySQL to Doris Pipeline
    parallelism: 4
+   flink-conf:
+     execution.checkpointing.interval: 2min
 ```
 
 ## 连接器配置项

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

@@ -57,6 +57,8 @@ sink:
 pipeline:
   name: MySQL to OceanBase Pipeline
   parallelism: 1
+  flink-conf:
+    execution.checkpointing.interval: 2min
 ```
 
 ## 连接器配置项

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

@@ -58,6 +58,8 @@ sink:
 pipeline:
   name: MySQL to Paimon Pipeline
   parallelism: 2
+  flink-conf:
+    execution.checkpointing.interval: 2min
 ```
 
 Pipeline 连接器配置项

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

@@ -59,6 +59,8 @@ sink:
 pipeline:
    name: MySQL to StarRocks Pipeline
    parallelism: 2
+   flink-conf:
+     execution.checkpointing.interval: 2min
 ```
 
 ## 连接器配置项

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

@@ -60,6 +60,8 @@ under the License.
    pipeline:
      name: Sync MySQL Database to Doris
      parallelism: 2
+     flink-conf:
+       execution.checkpointing.interval: 2min
 ```
 
 ## 包含可选部分
@@ -106,6 +108,8 @@ under the License.
          classpath: com.example.functions.AddOneFunctionClass
        - name: format
          classpath: com.example.functions.FormatFunctionClass
+     flink-conf:
+       execution.checkpointing.interval: 2min
 ```
 
 # Pipeline 配置
@@ -118,3 +122,4 @@ under the License.
 | local-time-zone        | 作业级别的本地时区。                                                                                                | optional          |
 | execution.runtime-mode | pipeline 的运行模式，包含 STREAMING 和 BATCH，默认值是 STREAMING。                                                       | optional          |
 | operator.uid.prefix    | Pipeline 中算子 UID 的前缀。如果不设置，Flink 会为每个算子生成唯一的 UID。 建议设置这个参数以提供稳定和可识别的算子 ID，这有助于有状态升级、问题排查和在 Flink UI 上的诊断。 | optional          |
+| flink-conf             | 用于配置[Flink相关参数](https://nightlies.apache.org/flink/flink-docs-master/docs/deployment/config/)。 <br/>Flink参数优先级：config.yaml < job command-line < pipeline.yaml | optional          |

docs/content.zh/docs/deployment/yarn.md

@@ -126,7 +126,14 @@ sink:
 pipeline:
  name: Sync MySQL Database to Doris
  parallelism: 2
-
+ flink-conf:
+   rest.bind-port: {{REST_PORT}}
+   rest.address: {{NODE_IP}}
+   execution.target: yarn-session
+   yarn.application.id: {{YARN_APPLICATION_ID}}
+   execution.checkpointing.interval: 2min
+   #如果需要从savepoint恢复，则配置以下参数
+   #execution.savepoint.path: hdfs:///flink/savepoint-1537
 ```
 
 你可以按需修改配置文件。

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

@@ -46,7 +46,8 @@ sink:
 
 pipeline:
    parallelism: 1
-
+   flink-conf:
+     execution.checkpointing.interval: 2min
 ```
 
 ## Connector Options

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

@@ -58,6 +58,8 @@ route:
 pipeline:
   name: MySQL to Elasticsearch Pipeline
   parallelism: 2
+  flink-conf:
+    execution.checkpointing.interval: 2min
 ```
 
 Pipeline Connector Options

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

@@ -55,6 +55,8 @@ sink:
 pipeline:
   name: MySQL to Kafka Pipeline
   parallelism: 2
+  flink-conf:
+    execution.checkpointing.interval: 2min
 ```
 
 Pipeline Connector Options

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

@@ -63,6 +63,8 @@ sink:
 pipeline:
   name: MySQL to MaxCompute Pipeline
   parallelism: 2
+  flink-conf:
+    execution.checkpointing.interval: 2min
 ```
 
 ## Connector Options

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

@@ -76,6 +76,8 @@ sink:
 pipeline:
    name: MySQL to Doris Pipeline
    parallelism: 4
+   flink-conf:
+     execution.checkpointing.interval: 2min
 ```
 
 ## Connector Options

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

@@ -57,6 +57,8 @@ sink:
 pipeline:
   name: MySQL to OceanBase Pipeline
   parallelism: 1
+  flink-conf:
+    execution.checkpointing.interval: 2min
 ```
 
 ## Connector Options

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

@@ -58,6 +58,8 @@ sink:
 pipeline:
   name: MySQL to Paimon Pipeline
   parallelism: 2
+  flink-conf:
+    execution.checkpointing.interval: 2min
 ```
 
 Pipeline Connector Options

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

@@ -59,6 +59,8 @@ sink:
 pipeline:
    name: MySQL to StarRocks Pipeline
    parallelism: 2
+   flink-conf:
+     execution.checkpointing.interval: 2min
 ```
 
 ## Connector Options

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

@@ -106,6 +106,8 @@ We could use following yaml file to define a complicated Data Pipeline describin
          classpath: com.example.functions.AddOneFunctionClass
        - name: format
          classpath: com.example.functions.FormatFunctionClass
+     flink-conf:
+       execution.checkpointing.interval: 2min
 ```
 
 # Pipeline Configurations
@@ -124,6 +126,8 @@ Note that whilst the parameters are each individually optional, at least one of
 | `schema.operator.uid`         | The unique ID for schema operator. This ID will be used for inter-operator communications and must be unique across operators. **Deprecated**: use `operator.uid.prefix` instead.                                                                                                                                                                                                                                                                                                                         | optional          |
 | `schema-operator.rpc-timeout` | The timeout time for SchemaOperator to wait downstream SchemaChangeEvent applying finished, the default value is 3 minutes.                                                                                                                                                                                                                                                                                                                                                                               | optional          |
 | `operator.uid.prefix`         | The prefix to use for all pipeline operator UIDs. If not set, all pipeline operator UIDs will be generated by Flink. It is recommended to set this parameter to ensure stable and recognizable operator UIDs, which can help with stateful upgrades, troubleshooting, and Flink UI diagnostics.                                                                                                                                                                                                           | optional          |
-
+| flink-conf                    | Used to configure [Flink related parameters](https://nightlies.apache.org/flink/flink-docs-master/docs/deployment/config/). <br/>Flink parameter priority: config.yaml < job command-line < pipeline.yaml                                                                                                                                                                                                                                                                                                 | optional          |
 NOTE: Whilst the above parameters are each individually optional, at least one of them must be specified. The `pipeline` section is mandatory and cannot be empty.
 
+
+

docs/content/docs/deployment/yarn.md

@@ -128,7 +128,14 @@ sink:
 pipeline:
  name: Sync MySQL Database to Doris
  parallelism: 2
-
+ flink-conf:
+   rest.bind-port: {{REST_PORT}}
+   rest.address: {{NODE_IP}}
+   execution.target: yarn-session
+   yarn.application.id: {{YARN_APPLICATION_ID}}
+   execution.checkpointing.interval: 2min
+   #If you need to restore from a savepoint, configure the following parameters:
+   #execution.savepoint.path: hdfs:///flink/savepoint-1537
 ```
 
 You need to modify the configuration file according to your needs.

flink-cdc-cli/src/main/java/org/apache/flink/cdc/cli/CliExecutor.java

@@ -74,7 +74,8 @@ public PipelineExecution.ExecutionInfo run() throws Exception {
             case YARN_APPLICATION:
                 return deployWithApplicationComposer(new YarnApplicationDeploymentExecutor());
             case LOCAL:
-                return deployWithComposer(FlinkPipelineComposer.ofMiniCluster());
+                return deployWithComposer(
+                        FlinkPipelineComposer.ofMiniCluster(flinkConfig, additionalJars));
             case REMOTE:
             case YARN_SESSION:
                 return deployWithComposer(