This is an automated email from the ASF dual-hosted git repository.
yuxiqian pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink-cdc.git
The following commit(s) were added to refs/heads/master by this push:
new f16709718 [FLINK-36581][cli] Allow specifying Flink configuration in
YAML job file (#3918)
f16709718 is described below
commit f16709718dc0dd4f22ad494113ba9d7cf8672415
Author: MOBIN <[email protected]>
AuthorDate: Tue Sep 29 09:51:47 2026 +0800
[FLINK-36581][cli] Allow specifying Flink configuration in YAML job file
(#3918)
---
docs/content.zh/docs/core-concept/data-pipeline.md | 37 +++---
docs/content.zh/docs/deployment/yarn.md | 8 +-
docs/content/docs/core-concept/data-pipeline.md | 37 +++---
docs/content/docs/deployment/yarn.md | 8 +-
.../java/org/apache/flink/cdc/cli/CliExecutor.java | 3 +-
.../java/org/apache/flink/cdc/cli/CliFrontend.java | 134 +------------------
.../cli/parser/YamlPipelineDefinitionParser.java | 27 +++-
.../flink/cdc/cli/utils/ConfigurationUtils.java | 20 +--
.../flink/cdc/cli/utils/FlinkEnvironmentUtils.java | 145 +++++++++++++++++++--
.../org/apache/flink/cdc/cli/CliFrontendTest.java | 136 +++++++++++++++++--
.../definitions/pipeline-definition-full.yaml | 2 +
...peline-definition-with-flink-conf-priority.yaml | 34 +++++
...pipeline-definition-with-nested-flink-conf.yaml | 31 +++++
.../cdc/composer/flink/FlinkPipelineComposer.java | 34 +++--
.../composer/flink/FlinkPipelineComposerTest.java | 29 ++++-
.../flink/core/execution/RestoreModeAdapter.java | 104 +++++++++++++++
.../flink/core/execution/RestoreModeAdapter.java | 106 +++++++++++++++
17 files changed, 676 insertions(+), 219 deletions(-)
diff --git a/docs/content.zh/docs/core-concept/data-pipeline.md
b/docs/content.zh/docs/core-concept/data-pipeline.md
index 2e228574c..5e2c02517 100644
--- a/docs/content.zh/docs/core-concept/data-pipeline.md
+++ b/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
```
## 包含可选部分
@@ -112,23 +114,24 @@ under the License.
下面是 Data Pipeline 级别支持的配置选项。
请注意，虽然这些参数都是可选的，但至少需要指定其中一个。也就是说，`pipeline` 部分是必需的，不能为空。
-| 参数 | 含义
[...]
-|-------------------------------|---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
[...]
-| `name` | 这个 pipeline 的名称，会用在 Flink 集群中作为作业的名称。
[...]
-| `parallelism` | pipeline的全局并发度，默认值是1。
[...]
-| `local-time-zone` | 作业级别的本地时区。
[...]
-| `execution.runtime-mode` | pipeline 的运行模式，包含 STREAMING 和 BATCH，默认值是
STREAMING。
[...]
-| `route-mode` | [路由]({{< ref "docs/core-concept/route"
>}}#路由模式)规则的匹配模式。可选值：`ALL_MATCH`（默认，应用所有匹配的规则）或 `FIRST_MATCH`（只应用第一个匹配的规则）。
| optional
|
-| `schema.change.behavior` | 如何处理 [schema 变更]({{< ref
"docs/core-concept/schema-evolution" >}})。可选值：[`exception`]({{< ref
"docs/core-concept/schema-evolution" >}}#exception-mode)、[`evolve`]({{< ref
"docs/core-concept/schema-evolution" >}}#evolve-mode)、[`try_evolve`]({{< ref
"docs/core-concept/schema-evolution" >}}#tryevolve-mode)、[`lenient`]({{< ref
"docs/core-concept/schema-evolution" >}}#lenient-mode)（默认值）或 [`ignore`]({{< ref
"docs/core-concept/schema-evolution" >}}#ignore-mode)。 [...]
-| `schema.operator.uid` | Schema 算子的唯一 ID。此 ID
用于算子间通信，必须在所有算子中保持唯一。**已废弃**：请使用 `operator.uid.prefix` 代替。
[...]
-| `schema-operator.rpc-timeout` | SchemaOperator 等待下游 SchemaChangeEvent
应用完成的超时时间，默认值是 3 分钟。
[...]
-| `operator.uid.prefix` | Pipeline 中算子 UID 的前缀。如果不设置，Flink 会为每个算子生成唯一的
UID。 建议设置这个参数以提供稳定和可识别的算子 ID，这有助于有状态升级、问题排查和在 Flink UI 上的诊断。
[...]
-| `sink.partitioning.strategy` | Sink
写入数据时使用的分区策略。数据类型：String。默认值：`SINK_DEFINED`。备注：可配置的值如下：`SINK_DEFINED`：使用 Sink
定义的分区策略；`PRIMARY_KEY`：按表 ID 和主键分区；`TABLE_ID`：仅按表 ID 分区。
| optional |
-| `transform.decimal.precision.mode` | transform 表达式求值中 DECIMAL
类型的最大精度模式。可选值：`UP_TO_19`（默认，使用 Calcite 默认类型系统）或 `UP_TO_38`（允许 DECIMAL 精度最高为 38
位）。
| optional |
-| `transform.async-execution.enabled` | 是否为 PostTransform 开启有序异步执行，默认值为
`false`。
| op [...]
-| `transform.async-execution.timeout` | 有序异步执行中每个 PostTransform 事件的超时时间，默认值为 5
分钟。
| option [...]
-| `transform.async-execution.capacity` | 异步执行中最多允许同时处理的 PostTransform 事件数，默认值为
100。
| option [...]
-| `transform.async-execution.worker-threads` | 每个异步 PostTransform
任务使用的工作线程数，默认值为 16。
| option [...]
+| 参数 | 含义
[...]
+|--------------------------------------------|--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
[...]
+| `name` | 这个 pipeline 的名称，会用在 Flink
集群中作为作业的名称。
[...]
+| `parallelism` | pipeline的全局并发度，默认值是1。
[...]
+| `local-time-zone` | 作业级别的本地时区。
[...]
+| `execution.runtime-mode` | pipeline 的运行模式，包含 STREAMING 和
BATCH，默认值是 STREAMING。
[...]
+| `route-mode` | [路由]({{< ref
"docs/core-concept/route" >}}#路由模式)规则的匹配模式。可选值：`ALL_MATCH`（默认，应用所有匹配的规则）或
`FIRST_MATCH`（只应用第一个匹配的规则）。
| optional [...]
+| `schema.change.behavior` | 如何处理 [schema 变更]({{< ref
"docs/core-concept/schema-evolution" >}})。可选值：[`exception`]({{< ref
"docs/core-concept/schema-evolution" >}}#exception-mode)、[`evolve`]({{< ref
"docs/core-concept/schema-evolution" >}}#evolve-mode)、[`try_evolve`]({{< ref
"docs/core-concept/schema-evolution" >}}#tryevolve-mode)、[`lenient`]({{< ref
"docs/core-concept/schema-evolution" >}}#lenient-mode)（默认值）或 [`ignore`]({{< ref
"docs/core-concept/schema-evolution" >}}#i [...]
+| `schema.operator.uid` | Schema 算子的唯一 ID。此 ID
用于算子间通信，必须在所有算子中保持唯一。**已废弃**：请使用 `operator.uid.prefix` 代替。
| op [...]
+| `schema-operator.rpc-timeout` | SchemaOperator 等待下游
SchemaChangeEvent 应用完成的超时时间，默认值是 3 分钟。
[...]
+| `operator.uid.prefix` | Pipeline 中算子 UID
的前缀。如果不设置，Flink 会为每个算子生成唯一的 UID。 建议设置这个参数以提供稳定和可识别的算子 ID，这有助于有状态升级、问题排查和在 Flink
UI 上的诊断。
|
optional |
+| `sink.partitioning.strategy` | Sink
写入数据时使用的分区策略。数据类型：String。默认值：`SINK_DEFINED`。备注：可配置的值如下：`SINK_DEFINED`：使用 Sink
定义的分区策略；`PRIMARY_KEY`：按表 ID 和主键分区；`TABLE_ID`：仅按表 ID 分区。
| optional |
+| `transform.decimal.precision.mode` | transform 表达式求值中 DECIMAL
类型的最大精度模式。可选值：`UP_TO_19`（默认，使用 Calcite 默认类型系统）或 `UP_TO_38`（允许 DECIMAL 精度最高为 38
位）。
| optional [...]
+| `transform.async-execution.enabled` | 是否为 PostTransform 开启有序异步执行，默认值为
`false`。
[...]
+| `transform.async-execution.timeout` | 有序异步执行中每个 PostTransform
事件的超时时间，默认值为 5 分钟。
[...]
+| `transform.async-execution.capacity` | 异步执行中最多允许同时处理的 PostTransform
事件数，默认值为 100。
[...]
+| `transform.async-execution.worker-threads` | 每个异步 PostTransform
任务使用的工作线程数，默认值为 16。
[...]
+| `flink-conf` | 用于配置 [Flink
相关参数](https://nightlies.apache.org/flink/flink-docs-master/docs/deployment/config/)。<br/>配置优先级：命令行参数（CLI
arguments）高于 pipeline.yaml 高于 Flink 默认配置 config.yaml。同时支持扁平键值对与嵌套 YAML 结构。
|
optional [...]
> **实验性且不稳定：** 异步 PostTransform 执行目前是实验性功能，其配置和行为可能会在未来版本中发生变化。
diff --git a/docs/content.zh/docs/deployment/yarn.md
b/docs/content.zh/docs/deployment/yarn.md
index 89d587d5b..1bfaaae27 100644
--- a/docs/content.zh/docs/deployment/yarn.md
+++ b/docs/content.zh/docs/deployment/yarn.md
@@ -126,7 +126,13 @@ sink:
pipeline:
name: Sync MySQL Database to Doris
parallelism: 2
-
+ flink-conf:
+ rest.bind-port: {{REST_PORT}}
+ rest.address: {{NODE_IP}}
+ yarn.application.id: {{YARN_APPLICATION_ID}}
+ execution.checkpointing.interval: 2min
+ # 如需从 savepoint 恢复，可配置以下参数：
+ # execution.savepoint.path: hdfs:///flink/savepoint-1537
```
你可以按需修改配置文件。
diff --git a/docs/content/docs/core-concept/data-pipeline.md
b/docs/content/docs/core-concept/data-pipeline.md
index 664c5c5db..64e709192 100644
--- a/docs/content/docs/core-concept/data-pipeline.md
+++ b/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
@@ -114,23 +116,24 @@ The following config options of Data Pipeline level are
supported.
Note that whilst the parameters are each individually optional, at least one
of them must be specified. That is to say, The `pipeline` section is mandatory
and cannot be empty.
-| parameter | meaning
[...]
-|-------------------------------|---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
[...]
-| `name` | The name of the pipeline, which will be
submitted to the Flink cluster as the job name.
[...]
-| `parallelism` | The global parallelism of the pipeline.
Defaults to 1.
[...]
-| `local-time-zone` | The local time zone defines current session
time zone id.
[...]
-| `execution.runtime-mode` | The runtime mode of the pipeline includes
STREAMING and BATCH, with the default value being STREAMING.
[...]
-| `route-mode` | The matching mode for [route]({{< ref
"docs/core-concept/route" >}}#route-mode) rules. One of: `ALL_MATCH` (default,
apply all matching rules) or `FIRST_MATCH` (apply only the first matching
rule).
[...]
-| `schema.change.behavior` | How to handle [changes in schema]({{< ref
"docs/core-concept/schema-evolution" >}}). One of: [`exception`]({{< ref
"docs/core-concept/schema-evolution" >}}#exception-mode), [`evolve`]({{< ref
"docs/core-concept/schema-evolution" >}}#evolve-mode), [`try_evolve`]({{< ref
"docs/core-concept/schema-evolution" >}}#tryevolve-mode), [`lenient`]({{< ref
"docs/core-concept/schema-evolution" >}}#lenient-mode) (default) or
[`ignore`]({{< ref "docs/core-concept/sche [...]
-| `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.
[...]
-| `schema-operator.rpc-timeout` | The timeout time for SchemaOperator to wait
downstream SchemaChangeEvent applying finished, the default value is 3 minutes.
[...]
-| `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.
[...]
-| `sink.partitioning.strategy` | The partitioning strategy used when writing
data to the sink. Data type: String. Default value: `SINK_DEFINED`. Available
values: `SINK_DEFINED`: uses the partitioning strategy defined by the sink;
`PRIMARY_KEY`: partitions by table ID and primary key; `TABLE_ID`: partitions
only by table ID.
[...]
-| `transform.decimal.precision.mode` | Maximum precision mode for DECIMAL type
in transform expression evaluation. One of: `UP_TO_19` (default, match
Calcite's default type system) or `UP_TO_38` (allow DECIMAL precision up to 38
digits).
[...]
-| `transform.async-execution.enabled` | Whether to enable ordered asynchronous
execution for post-transform. Defaults to `false`.
[...]
-| `transform.async-execution.timeout` | The timeout for each post-transform
event in ordered asynchronous execution. Defaults to 5 minutes.
[...]
-| `transform.async-execution.capacity` | The maximum number of in-flight
post-transform events. Defaults to 100.
[...]
-| `transform.async-execution.worker-threads` | The number of worker threads
used by each asynchronous post-transform task. Defaults to 16.
[...]
+| parameter | meaning
[...]
+|--------------------------------------------|--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
[...]
+| `name` | The name of the pipeline, which
will be submitted to the Flink cluster as the job name.
[...]
+| `parallelism` | The global parallelism of the
pipeline. Defaults to 1.
[...]
+| `local-time-zone` | The local time zone defines
current session time zone id.
[...]
+| `execution.runtime-mode` | The runtime mode of the
pipeline includes STREAMING and BATCH, with the default value being STREAMING.
[...]
+| `route-mode` | The matching mode for
[route]({{< ref "docs/core-concept/route" >}}#route-mode) rules. One of:
`ALL_MATCH` (default, apply all matching rules) or `FIRST_MATCH` (apply only
the first matching rule).
[...]
+| `schema.change.behavior` | How to handle [changes in
schema]({{< ref "docs/core-concept/schema-evolution" >}}). One of:
[`exception`]({{< ref "docs/core-concept/schema-evolution" >}}#exception-mode),
[`evolve`]({{< ref "docs/core-concept/schema-evolution" >}}#evolve-mode),
[`try_evolve`]({{< ref "docs/core-concept/schema-evolution"
>}}#tryevolve-mode), [`lenient`]({{< ref "docs/core-concept/schema-evolution"
>}}#lenient-mode) (default) or [`ignore`]({{< ref "docs/core [...]
+| `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.
[...]
+| `schema-operator.rpc-timeout` | The timeout time for
SchemaOperator to wait downstream SchemaChangeEvent applying finished, the
default value is 3 minutes.
[...]
+| `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.
[...]
+| `sink.partitioning.strategy` | The partitioning strategy used
when writing data to the sink. Data type: String. Default value:
`SINK_DEFINED`. Available values: `SINK_DEFINED`: uses the partitioning
strategy defined by the sink; `PRIMARY_KEY`: partitions by table ID and primary
key; `TABLE_ID`: partitions only by table ID.
[...]
+| `transform.decimal.precision.mode` | Maximum precision mode for
DECIMAL type in transform expression evaluation. One of: `UP_TO_19` (default,
match Calcite's default type system) or `UP_TO_38` (allow DECIMAL precision up
to 38 digits).
[...]
+| `transform.async-execution.enabled` | Whether to enable ordered
asynchronous execution for post-transform. Defaults to `false`.
[...]
+| `transform.async-execution.timeout` | The timeout for each
post-transform event in ordered asynchronous execution. Defaults to 5 minutes.
[...]
+| `transform.async-execution.capacity` | The maximum number of in-flight
post-transform events. Defaults to 100.
[...]
+| `transform.async-execution.worker-threads` | The number of worker threads
used by each asynchronous post-transform task. Defaults to 16.
[...]
+| `flink-conf` | Used to configure
[Flink-related
parameters](https://nightlies.apache.org/flink/flink-docs-master/docs/deployment/config/).
<br/>Configuration priority: CLI arguments > pipeline.yaml > Flink
config.yaml. Both flat key-value pairs and nested YAML structure are supported.
[...]
> **Experimental and unstable:** Asynchronous post-transform execution is an
> experimental feature. Its configuration and behavior may change in future
> releases.
diff --git a/docs/content/docs/deployment/yarn.md
b/docs/content/docs/deployment/yarn.md
index 4969ccf22..6fb021c65 100644
--- a/docs/content/docs/deployment/yarn.md
+++ b/docs/content/docs/deployment/yarn.md
@@ -128,7 +128,13 @@ sink:
pipeline:
name: Sync MySQL Database to Doris
parallelism: 2
-
+ flink-conf:
+ rest.bind-port: {{REST_PORT}}
+ rest.address: {{NODE_IP}}
+ yarn.application.id: {{YARN_APPLICATION_ID}}
+ execution.checkpointing.interval: 2min
+ # If you need to restore from a savepoint, uncomment the next line:
+ # execution.savepoint.path: hdfs:///flink/savepoint-1537
```
You need to modify the configuration file according to your needs.
diff --git
a/flink-cdc-cli/src/main/java/org/apache/flink/cdc/cli/CliExecutor.java
b/flink-cdc-cli/src/main/java/org/apache/flink/cdc/cli/CliExecutor.java
index 31468e472..2b2a92655 100644
--- a/flink-cdc-cli/src/main/java/org/apache/flink/cdc/cli/CliExecutor.java
+++ b/flink-cdc-cli/src/main/java/org/apache/flink/cdc/cli/CliExecutor.java
@@ -74,7 +74,8 @@ public class CliExecutor {
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(
diff --git
a/flink-cdc-cli/src/main/java/org/apache/flink/cdc/cli/CliFrontend.java
b/flink-cdc-cli/src/main/java/org/apache/flink/cdc/cli/CliFrontend.java
index 0f170367a..2d130a0d8 100644
--- a/flink-cdc-cli/src/main/java/org/apache/flink/cdc/cli/CliFrontend.java
+++ b/flink-cdc-cli/src/main/java/org/apache/flink/cdc/cli/CliFrontend.java
@@ -18,16 +18,10 @@
package org.apache.flink.cdc.cli;
import org.apache.flink.cdc.cli.utils.ConfigurationUtils;
-import org.apache.flink.cdc.cli.utils.FlinkEnvironmentUtils;
import org.apache.flink.cdc.common.annotation.VisibleForTesting;
-import org.apache.flink.cdc.common.configuration.ConfigOption;
-import org.apache.flink.cdc.common.configuration.ConfigOptions;
import org.apache.flink.cdc.common.configuration.Configuration;
-import org.apache.flink.cdc.common.utils.StringUtils;
import org.apache.flink.cdc.composer.PipelineExecution;
-import org.apache.flink.configuration.DeploymentOptions;
import org.apache.flink.core.fs.Path;
-import org.apache.flink.runtime.jobgraph.SavepointRestoreSettings;
import org.apache.flink.shaded.guava31.com.google.common.base.Joiner;
@@ -40,26 +34,17 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.File;
-import java.lang.reflect.InvocationTargetException;
import java.util.Arrays;
import java.util.List;
import java.util.Optional;
-import java.util.Properties;
import java.util.stream.Collectors;
-import static org.apache.flink.cdc.cli.CliFrontendOptions.FLINK_CONFIG;
-import static
org.apache.flink.cdc.cli.CliFrontendOptions.SAVEPOINT_ALLOW_NON_RESTORED_OPTION;
-import static org.apache.flink.cdc.cli.CliFrontendOptions.SAVEPOINT_CLAIM_MODE;
-import static
org.apache.flink.cdc.cli.CliFrontendOptions.SAVEPOINT_PATH_OPTION;
-import static org.apache.flink.cdc.cli.CliFrontendOptions.TARGET;
-import static org.apache.flink.cdc.cli.CliFrontendOptions.USE_MINI_CLUSTER;
-import static
org.apache.flink.cdc.composer.flink.deployment.ComposeDeployment.LOCAL;
-import static
org.apache.flink.cdc.composer.flink.deployment.ComposeDeployment.REMOTE;
+import static
org.apache.flink.cdc.cli.utils.FlinkEnvironmentUtils.getFlinkHome;
+import static
org.apache.flink.cdc.cli.utils.FlinkEnvironmentUtils.mergeFlinkConfigurationsWithPriority;
/** The frontend entrypoint for the command-line interface of Flink CDC. */
public class CliFrontend {
private static final Logger LOG =
LoggerFactory.getLogger(CliFrontend.class);
- private static final String FLINK_HOME_ENV_VAR = "FLINK_HOME";
private static final String FLINK_CDC_HOME_ENV_VAR = "FLINK_CDC_HOME";
public static void main(String[] args) throws Exception {
@@ -98,19 +83,9 @@ public class CliFrontend {
// Global pipeline configuration
Configuration globalPipelineConfig = getGlobalConfig(commandLine);
- // Load Flink environment
Path flinkHome = getFlinkHome(commandLine);
- Configuration configuration =
FlinkEnvironmentUtils.loadFlinkConfiguration(flinkHome);
-
- // To override the Flink configuration
- overrideFlinkConfiguration(configuration, commandLine);
-
org.apache.flink.configuration.Configuration flinkConfig =
-
org.apache.flink.configuration.Configuration.fromMap(configuration.toMap());
-
- // Savepoint
- SavepointRestoreSettings savepointSettings =
createSavepointRestoreSettings(commandLine);
- SavepointRestoreSettings.toConfiguration(savepointSettings,
flinkConfig);
+ mergeFlinkConfigurationsWithPriority(pipelineDefPath,
commandLine);
// Additional JARs
List<Path> additionalJars =
@@ -131,109 +106,6 @@ public class CliFrontend {
flinkHome);
}
- private static void overrideFlinkConfiguration(
- Configuration flinkConfig, CommandLine commandLine) {
-
- String target =
- commandLine.hasOption(USE_MINI_CLUSTER)
- ? LOCAL.getName()
- : commandLine.getOptionValue(TARGET, REMOTE.getName());
- flinkConfig.set(
-
ConfigOptions.key(DeploymentOptions.TARGET.key()).stringType().defaultValue(target),
- target);
-
- Properties properties =
commandLine.getOptionProperties(FLINK_CONFIG.getOpt());
- LOG.info("Dynamic flink config items found: {}", properties);
- for (String key : properties.stringPropertyNames()) {
- String value = properties.getProperty(key);
- if (StringUtils.isNullOrWhitespaceOnly(key)
- || StringUtils.isNullOrWhitespaceOnly(value)) {
- throw new IllegalArgumentException(
- String.format(
- "null or white space argument for key or
value: %s=%s",
- key, value));
- }
- ConfigOption<String> configOption =
-
ConfigOptions.key(key.trim()).stringType().defaultValue(value.trim());
- flinkConfig.set(configOption, value.trim());
- }
- }
-
- private static SavepointRestoreSettings createSavepointRestoreSettings(
- CommandLine commandLine) {
- if (commandLine.hasOption(SAVEPOINT_PATH_OPTION.getOpt())) {
- String savepointPath =
commandLine.getOptionValue(SAVEPOINT_PATH_OPTION.getOpt());
- boolean allowNonRestoredState =
-
commandLine.hasOption(SAVEPOINT_ALLOW_NON_RESTORED_OPTION.getOpt());
- final Object restoreMode;
- if (commandLine.hasOption(SAVEPOINT_CLAIM_MODE)) {
- restoreMode =
-
org.apache.flink.configuration.ConfigurationUtils.convertValue(
-
commandLine.getOptionValue(SAVEPOINT_CLAIM_MODE),
- ConfigurationUtils.getClaimModeClass());
- } else {
- try {
- restoreMode =
-
ConfigurationUtils.getClaimModeClass().getField("DEFAULT").get(null);
- } catch (NoSuchFieldException | IllegalAccessException e) {
- throw new RuntimeException(
- "Failed to get DEFAULT value from RestoreMode
class.", e);
- }
- }
- // allowNonRestoredState is always false because all operators are
predefined.
-
- return (SavepointRestoreSettings)
- Arrays.stream(SavepointRestoreSettings.class.getMethods())
- .filter(
- method ->
- method.getName().equals("forPath")
- &&
method.getParameterCount() == 3)
- .findFirst()
- .map(
- method -> {
- try {
- return method.invoke(
- null,
- savepointPath,
- allowNonRestoredState,
- restoreMode);
- } catch (IllegalAccessException
- | InvocationTargetException e)
{
- throw new RuntimeException(
- "Failed to invoke
SavepointRestoreSettings#forPath nethod.",
- e);
- }
- })
- .orElseThrow(
- () ->
- new RuntimeException(
- "Failed to resolve
SavepointRestoreSettings#forPath method."));
- } else {
- return SavepointRestoreSettings.none();
- }
- }
-
- private static Path getFlinkHome(CommandLine commandLine) {
- // Check command line arguments first
- String flinkHomeFromArgs =
commandLine.getOptionValue(CliFrontendOptions.FLINK_HOME);
- if (flinkHomeFromArgs != null) {
- LOG.debug("Flink home is loaded by command-line argument: {}",
flinkHomeFromArgs);
- return new Path(flinkHomeFromArgs);
- }
-
- // Fallback to environment variable
- String flinkHomeFromEnvVar = System.getenv(FLINK_HOME_ENV_VAR);
- if (flinkHomeFromEnvVar != null) {
- LOG.debug("Flink home is loaded by environment variable: {}",
flinkHomeFromEnvVar);
- return new Path(flinkHomeFromEnvVar);
- }
-
- throw new IllegalArgumentException(
- "Cannot find Flink home from either command line arguments
\"--flink-home\" "
- + "or the environment variable \"FLINK_HOME\". "
- + "Please make sure Flink home is properly set. ");
- }
-
private static Configuration getGlobalConfig(CommandLine commandLine)
throws Exception {
// Try to get global config path from command line
String globalConfig =
commandLine.getOptionValue(CliFrontendOptions.GLOBAL_CONFIG);
diff --git
a/flink-cdc-cli/src/main/java/org/apache/flink/cdc/cli/parser/YamlPipelineDefinitionParser.java
b/flink-cdc-cli/src/main/java/org/apache/flink/cdc/cli/parser/YamlPipelineDefinitionParser.java
index 70d3a4615..717e9222f 100644
---
a/flink-cdc-cli/src/main/java/org/apache/flink/cdc/cli/parser/YamlPipelineDefinitionParser.java
+++
b/flink-cdc-cli/src/main/java/org/apache/flink/cdc/cli/parser/YamlPipelineDefinitionParser.java
@@ -17,6 +17,7 @@
package org.apache.flink.cdc.cli.parser;
+import org.apache.flink.cdc.cli.utils.ConfigurationUtils;
import org.apache.flink.cdc.common.configuration.Configuration;
import org.apache.flink.cdc.common.event.SchemaChangeEventType;
import org.apache.flink.cdc.common.event.SchemaChangeEventTypeFamily;
@@ -42,6 +43,7 @@ import
org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.ObjectMap
import
org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.node.ObjectNode;
import
org.apache.flink.shaded.jackson2.com.fasterxml.jackson.dataformat.yaml.YAMLFactory;
+import java.io.IOException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
@@ -71,6 +73,7 @@ public class YamlPipelineDefinitionParser implements
PipelineDefinitionParser {
private static final String TRANSFORM_KEY = "transform";
private static final String PIPELINE_KEY = "pipeline";
private static final String MODEL_KEY = "model";
+ private static final String FLINK_KEY = "flink-conf";
// Source / sink keys
private static final String TYPE_KEY = "type";
@@ -119,7 +122,7 @@ public class YamlPipelineDefinitionParser implements
PipelineDefinitionParser {
public static final String TRANSFORM_TABLE_OPTION_DELIMITER_KEY =
"table-options.delimiter";
- private final ObjectMapper mapper = new ObjectMapper(new YAMLFactory());
+ private static final ObjectMapper mapper = new ObjectMapper(new
YAMLFactory());
/** Parse the specified pipeline definition file. */
@Override
@@ -157,6 +160,8 @@ public class YamlPipelineDefinitionParser implements
PipelineDefinitionParser {
Optional.ofNullable(
((ObjectNode)
pipelineDefJsonNode.get(PIPELINE_KEY)).remove(MODEL_KEY))
.ifPresent(node -> modelDefs.addAll(parseModels(node)));
+
+ ((ObjectNode)
pipelineDefJsonNode.get(PIPELINE_KEY)).remove(FLINK_KEY);
}
// Pipeline configs are optional
@@ -675,4 +680,24 @@ public class YamlPipelineDefinitionParser implements
PipelineDefinitionParser {
contextName, jsonNode.getNodeType(), jsonNode));
}
}
+
+ public static Map<String, String> getFlinkConfigFromPipelineDef(Path
pipelineDefPath)
+ throws IOException {
+ FileSystem fileSystem = FileSystem.get(pipelineDefPath.toUri());
+ try (FSDataInputStream pipelineInStream =
fileSystem.open(pipelineDefPath)) {
+ JsonNode pipelineDefJsonNode = mapper.readTree(pipelineInStream);
+ return Optional.ofNullable(pipelineDefJsonNode.get(PIPELINE_KEY))
+ .map(node -> node.get(FLINK_KEY))
+ .filter(JsonNode::isObject)
+ .map(
+ flinkNode -> {
+ Map<String, Object> rawMap =
+ mapper.convertValue(
+ flinkNode,
+ new TypeReference<Map<String,
Object>>() {});
+ return
ConfigurationUtils.flattenConfigMap(rawMap, "");
+ })
+ .orElse(Collections.emptyMap());
+ }
+ }
}
diff --git
a/flink-cdc-cli/src/main/java/org/apache/flink/cdc/cli/utils/ConfigurationUtils.java
b/flink-cdc-cli/src/main/java/org/apache/flink/cdc/cli/utils/ConfigurationUtils.java
index eafd4da09..9e2de35ab 100644
---
a/flink-cdc-cli/src/main/java/org/apache/flink/cdc/cli/utils/ConfigurationUtils.java
+++
b/flink-cdc-cli/src/main/java/org/apache/flink/cdc/cli/utils/ConfigurationUtils.java
@@ -43,7 +43,7 @@ public class ConfigurationUtils {
}
@SuppressWarnings("unchecked")
- private static Map<String, String> flattenConfigMap(
+ public static Map<String, String> flattenConfigMap(
Map<String, Object> config, String keyPrefix) {
final Map<String, String> flattenedMap = new HashMap<>();
@@ -53,7 +53,7 @@ public class ConfigurationUtils {
if (value instanceof Map) {
Map<String, Object> e = (Map<String, Object>) value;
flattenedMap.putAll(flattenConfigMap(e, flattenedKey +
KEY_SEPARATOR));
- } else {
+ } else if (value != null) {
if (value instanceof List) {
flattenedMap.put(flattenedKey,
YamlParserUtils.toYAMLString(value));
} else {
@@ -64,20 +64,4 @@ public class ConfigurationUtils {
return flattenedMap;
}
-
- public static Class<?> getClaimModeClass() {
- try {
- return
Class.forName("org.apache.flink.core.execution.RecoveryClaimMode");
- } catch (ClassNotFoundException classNotFoundException) {
- try {
- return
Class.forName("org.apache.flink.core.execution.RestoreMode");
- } catch (ClassNotFoundException ignored) {
- try {
- return
Class.forName("org.apache.flink.runtime.jobgraph.RestoreMode");
- } catch (ClassNotFoundException e) {
- throw new RuntimeException(e);
- }
- }
- }
- }
}
diff --git
a/flink-cdc-cli/src/main/java/org/apache/flink/cdc/cli/utils/FlinkEnvironmentUtils.java
b/flink-cdc-cli/src/main/java/org/apache/flink/cdc/cli/utils/FlinkEnvironmentUtils.java
index f10b2fa06..2ea8f9aa2 100644
---
a/flink-cdc-cli/src/main/java/org/apache/flink/cdc/cli/utils/FlinkEnvironmentUtils.java
+++
b/flink-cdc-cli/src/main/java/org/apache/flink/cdc/cli/utils/FlinkEnvironmentUtils.java
@@ -17,38 +17,165 @@
package org.apache.flink.cdc.cli.utils;
-import org.apache.flink.cdc.common.configuration.Configuration;
+import org.apache.flink.cdc.cli.CliFrontendOptions;
+import org.apache.flink.cdc.cli.parser.YamlPipelineDefinitionParser;
+import org.apache.flink.cdc.common.utils.StringUtils;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.configuration.DeploymentOptions;
+import org.apache.flink.core.execution.RestoreModeAdapter;
import org.apache.flink.core.fs.Path;
+import org.apache.flink.runtime.jobgraph.SavepointRestoreSettings;
import org.apache.flink.shaded.guava31.com.google.common.base.Joiner;
+import org.apache.commons.cli.CommandLine;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.File;
+import java.io.IOException;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Properties;
+
+import static org.apache.flink.cdc.cli.CliFrontendOptions.FLINK_CONFIG;
+import static
org.apache.flink.cdc.cli.CliFrontendOptions.SAVEPOINT_ALLOW_NON_RESTORED_OPTION;
+import static org.apache.flink.cdc.cli.CliFrontendOptions.SAVEPOINT_CLAIM_MODE;
+import static
org.apache.flink.cdc.cli.CliFrontendOptions.SAVEPOINT_PATH_OPTION;
+import static org.apache.flink.cdc.cli.CliFrontendOptions.TARGET;
+import static org.apache.flink.cdc.cli.CliFrontendOptions.USE_MINI_CLUSTER;
+import static
org.apache.flink.cdc.composer.flink.deployment.ComposeDeployment.LOCAL;
+import static
org.apache.flink.cdc.composer.flink.deployment.ComposeDeployment.REMOTE;
/** Utilities for handling Flink configuration and environment. */
public class FlinkEnvironmentUtils {
private static final Logger LOG =
LoggerFactory.getLogger(FlinkEnvironmentUtils.class);
+ private static final String FLINK_HOME_ENV_VAR = "FLINK_HOME";
private static final String FLINK_CONF_DIR = "conf";
private static final String LEGACY_FLINK_CONF_FILENAME = "flink-conf.yaml";
private static final String FLINK_CONF_FILENAME = "config.yaml";
- public static Configuration loadFlinkConfiguration(Path flinkHome) throws
Exception {
+ /** Load and merge Flink configurations by priority rules. */
+ public static Configuration mergeFlinkConfigurationsWithPriority(
+ Path pipelineDefPath, CommandLine commandLine) throws Exception {
+ Configuration flinkConfig = loadBaseFlinkConfig(commandLine);
+
+ mergePipelineFlinkConfig(pipelineDefPath, flinkConfig);
+
+ mergeCommandLineFlinkConfig(commandLine, flinkConfig);
+
+ applySavepointSettings(flinkConfig);
+ return flinkConfig;
+ }
+
+ private static Configuration loadBaseFlinkConfig(CommandLine commandLine)
throws Exception {
+ Path flinkHome = getFlinkHome(commandLine);
+ Map<String, String> flinkConfigurationMap;
Path flinkConfPath =
new Path(
flinkHome,
Joiner.on(File.separator).join(FLINK_CONF_DIR,
FLINK_CONF_FILENAME));
if (flinkConfPath.getFileSystem().exists(flinkConfPath)) {
- return ConfigurationUtils.loadConfigFile(flinkConfPath);
+ flinkConfigurationMap =
ConfigurationUtils.loadConfigFile(flinkConfPath).toMap();
} else {
- return ConfigurationUtils.loadConfigFile(
- new Path(
- flinkHome,
- Joiner.on(File.separator)
- .join(FLINK_CONF_DIR,
LEGACY_FLINK_CONF_FILENAME)),
- true);
+ flinkConfigurationMap =
+ ConfigurationUtils.loadConfigFile(
+ new Path(
+ flinkHome,
+ Joiner.on(File.separator)
+ .join(
+ FLINK_CONF_DIR,
+
LEGACY_FLINK_CONF_FILENAME)),
+ true)
+ .toMap();
+ }
+ return Configuration.fromMap(flinkConfigurationMap);
+ }
+
+ private static void mergeCommandLineFlinkConfig(
+ CommandLine commandLine, Configuration flinkConfig) {
+ Properties commandLineProperties =
commandLine.getOptionProperties(FLINK_CONFIG.getOpt());
+ // Use "remote" as the default target
+ String target =
+ commandLine.hasOption(USE_MINI_CLUSTER)
+ ? LOCAL.getName()
+ : commandLine.getOptionValue(TARGET, REMOTE.getName());
+ flinkConfig.set(DeploymentOptions.TARGET, target);
+
+ Optional.ofNullable(commandLine.getOptionValue(SAVEPOINT_PATH_OPTION))
+ .ifPresent(value ->
RestoreModeAdapter.setSavepointPath(flinkConfig, value));
+
+ Optional.ofNullable(commandLine.getOptionValue(SAVEPOINT_CLAIM_MODE))
+ .ifPresent(value ->
RestoreModeAdapter.setRestoreMode(flinkConfig, value));
+
+ if (commandLine.hasOption(SAVEPOINT_ALLOW_NON_RESTORED_OPTION)) {
+ RestoreModeAdapter.setSavepointIgnoreUnclaimedState(flinkConfig,
true);
+ }
+
+ LOG.info(
+ "Dynamic Flink configuration items found from command line:
{}",
+ commandLineProperties);
+ commandLineProperties.forEach(
+ (key, value) -> validateAndApplyCommandLineEntry(flinkConfig,
key, value));
+ }
+
+ private static void validateAndApplyCommandLineEntry(
+ Configuration flinkConfig, Object key, Object value) {
+ String keyStr = key.toString();
+ String valueStr = value.toString();
+ if (StringUtils.isNullOrWhitespaceOnly(keyStr)
+ || StringUtils.isNullOrWhitespaceOnly(valueStr)) {
+ throw new IllegalArgumentException(
+ String.format(
+ "null or white space argument for key or value:
%s=%s", key, value));
+ }
+ flinkConfig.setString(keyStr.trim(), valueStr.trim());
+ }
+
+ private static void mergePipelineFlinkConfig(Path pipelineDefPath,
Configuration flinkConfig)
+ throws IOException {
+ Configuration flinkConfigFromPipelineDef =
+ Configuration.fromMap(
+
YamlPipelineDefinitionParser.getFlinkConfigFromPipelineDef(
+ pipelineDefPath));
+ LOG.info(
+ "Flink configuration items found from pipeline definition: {}",
+ flinkConfigFromPipelineDef);
+ flinkConfig.addAll(flinkConfigFromPipelineDef);
+ }
+
+ private static void applySavepointSettings(Configuration flinkConfig) {
+ SavepointRestoreSettings savepointSettings =
createSavepointRestoreSettings(flinkConfig);
+ SavepointRestoreSettings.toConfiguration(savepointSettings,
flinkConfig);
+ }
+
+ public static Path getFlinkHome(CommandLine commandLine) {
+ // Check command line arguments first
+ String flinkHomeFromArgs =
commandLine.getOptionValue(CliFrontendOptions.FLINK_HOME);
+ if (flinkHomeFromArgs != null) {
+ LOG.debug("Flink home is loaded by command-line argument: {}",
flinkHomeFromArgs);
+ return new Path(flinkHomeFromArgs);
+ }
+
+ // Fallback to environment variable
+ String flinkHomeFromEnvVar = System.getenv(FLINK_HOME_ENV_VAR);
+ if (flinkHomeFromEnvVar != null) {
+ LOG.debug("Flink home is loaded by environment variable: {}",
flinkHomeFromEnvVar);
+ return new Path(flinkHomeFromEnvVar);
}
+
+ throw new IllegalArgumentException(
+ "Cannot find Flink home from either command line arguments
\"--flink-home\" "
+ + "or the environment variable \"FLINK_HOME\". "
+ + "Please make sure Flink home is properly set. ");
+ }
+
+ public static SavepointRestoreSettings createSavepointRestoreSettings(
+ Configuration flinkConfig) {
+ RestoreModeAdapter.getSavepointPath(flinkConfig)
+ .ifPresent(
+ savepointPath -> LOG.info("Load savepoint from path:
{}", savepointPath));
+ return RestoreModeAdapter.createSavepointRestoreSettings(flinkConfig);
}
}
diff --git
a/flink-cdc-cli/src/test/java/org/apache/flink/cdc/cli/CliFrontendTest.java
b/flink-cdc-cli/src/test/java/org/apache/flink/cdc/cli/CliFrontendTest.java
index a9b8035b5..f42fa9f09 100644
--- a/flink-cdc-cli/src/test/java/org/apache/flink/cdc/cli/CliFrontendTest.java
+++ b/flink-cdc-cli/src/test/java/org/apache/flink/cdc/cli/CliFrontendTest.java
@@ -20,7 +20,7 @@ package org.apache.flink.cdc.cli;
import org.apache.flink.cdc.composer.PipelineComposer;
import org.apache.flink.cdc.composer.PipelineExecution;
import org.apache.flink.cdc.composer.definition.PipelineDef;
-import org.apache.flink.core.execution.RestoreMode;
+import org.apache.flink.core.execution.RestoreModeAdapter;
import org.apache.flink.core.fs.Path;
import org.apache.flink.shaded.guava31.com.google.common.io.Resources;
@@ -28,7 +28,6 @@ import
org.apache.flink.shaded.guava31.com.google.common.io.Resources;
import org.apache.commons.cli.CommandLineParser;
import org.apache.commons.cli.DefaultParser;
import org.apache.commons.cli.Options;
-import org.assertj.core.api.Assertions;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
@@ -36,11 +35,18 @@ import java.io.ByteArrayOutputStream;
import java.io.PrintStream;
import java.net.URL;
import java.nio.file.Paths;
+import java.time.Duration;
import java.util.Map;
-import static org.apache.flink.configuration.StateRecoveryOptions.RESTORE_MODE;
+import static
org.apache.flink.configuration.CheckpointingOptions.CHECKPOINTING_INTERVAL;
+import static
org.apache.flink.configuration.CheckpointingOptions.CHECKPOINTING_TIMEOUT;
+import static
org.apache.flink.configuration.CheckpointingOptions.MAX_CONCURRENT_CHECKPOINTS;
+import static org.apache.flink.configuration.CoreOptions.DEFAULT_PARALLELISM;
+import static org.apache.flink.configuration.JobManagerOptions.ADDRESS;
+import static org.apache.flink.configuration.JobManagerOptions.PORT;
import static
org.apache.flink.configuration.StateRecoveryOptions.SAVEPOINT_IGNORE_UNCLAIMED_STATE;
import static
org.apache.flink.configuration.StateRecoveryOptions.SAVEPOINT_PATH;
+import static org.apache.flink.configuration.TaskManagerOptions.NUM_TASK_SLOTS;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
@@ -107,11 +113,109 @@ class CliFrontendTest {
"-n");
assertThat(executor.getFlinkConfig().get(SAVEPOINT_PATH))
.isEqualTo(flinkHome() + "/savepoints/savepoint-1");
- assertThat(executor.getFlinkConfig().get(RESTORE_MODE).toString())
- .isEqualTo(RestoreMode.NO_CLAIM.toString());
+
assertThat(RestoreModeAdapter.getRestoreMode(executor.getFlinkConfig()).toString())
+ .isEqualTo("NO_CLAIM");
assertThat(executor.getFlinkConfig().get(SAVEPOINT_IGNORE_UNCLAIMED_STATE)).isTrue();
}
+ @Test
+ void testFlinkConfigurationWithPriority() throws Exception {
+ // 1. Command-line options have higher priority than pipeline
definition options
+ CliExecutor executorWithCliOverride =
+ createExecutor(
+ pipelineDef(),
+ "--flink-home",
+ flinkHome(),
+ "-D",
+ "execution.checkpointing.timeout=11min");
+
assertThat(executorWithCliOverride.getFlinkConfig().get(CHECKPOINTING_TIMEOUT))
+ .isEqualTo(Duration.ofMinutes(11));
+
assertThat(executorWithCliOverride.getFlinkConfig().get(DEFAULT_PARALLELISM)).isEqualTo(1);
+
+ // 2. Pipeline definition options have higher priority than cluster
config.yaml options
+ CliExecutor executorWithoutCliOverride =
+ createExecutor(pipelineDef(), "--flink-home", flinkHome());
+
assertThat(executorWithoutCliOverride.getFlinkConfig().get(CHECKPOINTING_TIMEOUT))
+ .isEqualTo(Duration.ofMinutes(12));
+
assertThat(executorWithoutCliOverride.getFlinkConfig().get(DEFAULT_PARALLELISM))
+ .isEqualTo(1);
+ }
+
+ @Test
+ void testFlinkConfigurationMutualOverridePriority() throws Exception {
+ // Verify all 3 configuration provision methods (CLI args,
pipeline.yaml, config.yaml)
+ // interacting simultaneously with different override relationships:
+ // 1. CLI overrides both pipeline.yaml and config.yaml
(parallelism.default: 3 vs 2 vs 1)
+ // 2. pipeline.yaml overrides config.yaml when CLI does not specify
+ // (taskmanager.numberOfTaskSlots: 4 vs 1)
+ // 3. CLI overrides config.yaml directly when pipeline.yaml does not
specify
+ // (jobmanager.rpc.port: 9999 vs 6123)
+ // 4. CLI overrides pipeline.yaml directly
(execution.checkpointing.timeout: 11min vs 12min)
+ // 5. pipeline.yaml value used when not overridden
(execution.checkpointing.interval: 3min)
+ // 6. CLI value used for option only set via CLI
+ // (execution.checkpointing.max-concurrent-checkpoints: 5)
+ // 7. config.yaml value preserved for option only set in cluster config
+ // (jobmanager.rpc.address: localhost)
+ CliExecutor executorWithCli =
+ createExecutor(
+ pipelineDefWithPriority(),
+ "--flink-home",
+ flinkHome(),
+ "-D",
+ "parallelism.default=3",
+ "-D",
+ "jobmanager.rpc.port=9999",
+ "-D",
+ "execution.checkpointing.timeout=11min",
+ "-D",
+
"execution.checkpointing.max-concurrent-checkpoints=5");
+ org.apache.flink.configuration.Configuration flinkConfigWithCli =
+ executorWithCli.getFlinkConfig();
+
+ assertThat(flinkConfigWithCli.get(DEFAULT_PARALLELISM)).isEqualTo(3);
+ assertThat(flinkConfigWithCli.get(NUM_TASK_SLOTS)).isEqualTo(4);
+ assertThat(flinkConfigWithCli.get(PORT)).isEqualTo(9999);
+ assertThat(flinkConfigWithCli.get(ADDRESS)).isEqualTo("localhost");
+
assertThat(flinkConfigWithCli.get(CHECKPOINTING_TIMEOUT)).isEqualTo(Duration.ofMinutes(11));
+
assertThat(flinkConfigWithCli.get(CHECKPOINTING_INTERVAL)).isEqualTo(Duration.ofMinutes(3));
+
assertThat(flinkConfigWithCli.get(MAX_CONCURRENT_CHECKPOINTS)).isEqualTo(5);
+
+ // When no CLI arguments are provided, verify pipeline.yaml overrides
config.yaml
+ // while non-overridden cluster configs are safely preserved
+ CliExecutor executorWithoutCli =
+ createExecutor(pipelineDefWithPriority(), "--flink-home",
flinkHome());
+ org.apache.flink.configuration.Configuration flinkConfigWithoutCli =
+ executorWithoutCli.getFlinkConfig();
+
+
assertThat(flinkConfigWithoutCli.get(DEFAULT_PARALLELISM)).isEqualTo(2);
+ assertThat(flinkConfigWithoutCli.get(NUM_TASK_SLOTS)).isEqualTo(4);
+ assertThat(flinkConfigWithoutCli.get(PORT)).isEqualTo(6123);
+ assertThat(flinkConfigWithoutCli.get(ADDRESS)).isEqualTo("localhost");
+ assertThat(flinkConfigWithoutCli.get(CHECKPOINTING_TIMEOUT))
+ .isEqualTo(Duration.ofMinutes(12));
+ assertThat(flinkConfigWithoutCli.get(CHECKPOINTING_INTERVAL))
+ .isEqualTo(Duration.ofMinutes(3));
+ }
+
+ @Test
+ void testFlinkConfigurationWithNestedYaml() throws Exception {
+ CliExecutor executor =
+ createExecutor(
+ pipelineDefWithNestedFlinkConf(),
+ "--flink-home",
+ flinkHome(),
+ "-D",
+ "execution.checkpointing.interval=5min");
+ org.apache.flink.configuration.Configuration flinkConfig =
executor.getFlinkConfig();
+
+ // Command-line override takes precedence over nested YAML
+
assertThat(flinkConfig.get(CHECKPOINTING_INTERVAL)).isEqualTo(Duration.ofMinutes(5));
+ // Nested YAML takes effect
+
assertThat(flinkConfig.get(CHECKPOINTING_TIMEOUT)).isEqualTo(Duration.ofMinutes(15));
+ // Nested YAML overrides config.yaml default parallelism
+ assertThat(flinkConfig.get(DEFAULT_PARALLELISM)).isEqualTo(4);
+ }
+
@Test
void testDeploymentTargetConfiguration() throws Exception {
CliExecutor executor =
@@ -183,8 +287,8 @@ class CliFrontendTest {
}
@Test
- void testPipelineExecutingWithUnValidFlinkConfig() throws Exception {
- Assertions.assertThatThrownBy(
+ void testPipelineExecutingWithInvalidFlinkConfig() throws Exception {
+ assertThatThrownBy(
() ->
createExecutor(
pipelineDef(),
@@ -198,7 +302,7 @@ class CliFrontendTest {
"null or white space argument for key or
value: %s=%s",
"", "execution.target"));
- Assertions.assertThatThrownBy(
+ assertThatThrownBy(
() ->
createExecutor(
pipelineDef(),
@@ -212,7 +316,7 @@ class CliFrontendTest {
"null or white space argument for key or
value: %s=%s",
"execution.target", ""));
- Assertions.assertThatThrownBy(
+ assertThatThrownBy(
() -> createExecutor(pipelineDef(), "--flink-home",
flinkHome(), "-D", "="))
.isInstanceOf(IllegalArgumentException.class)
.hasMessage(
@@ -231,6 +335,20 @@ class CliFrontendTest {
return Paths.get(resource.toURI()).toString();
}
+ private String pipelineDefWithPriority() throws Exception {
+ URL resource =
+ Resources.getResource(
+
"definitions/pipeline-definition-with-flink-conf-priority.yaml");
+ return Paths.get(resource.toURI()).toString();
+ }
+
+ private String pipelineDefWithNestedFlinkConf() throws Exception {
+ URL resource =
+ Resources.getResource(
+
"definitions/pipeline-definition-with-nested-flink-conf.yaml");
+ return Paths.get(resource.toURI()).toString();
+ }
+
private String flinkHome() throws Exception {
URL resource = Resources.getResource("flink-home");
return Paths.get(resource.toURI()).toString();
diff --git
a/flink-cdc-cli/src/test/resources/definitions/pipeline-definition-full.yaml
b/flink-cdc-cli/src/test/resources/definitions/pipeline-definition-full.yaml
index d1a694d70..9b877440b 100644
--- a/flink-cdc-cli/src/test/resources/definitions/pipeline-definition-full.yaml
+++ b/flink-cdc-cli/src/test/resources/definitions/pipeline-definition-full.yaml
@@ -66,3 +66,5 @@ pipeline:
openai.model: text-embedding-3-small
openai.host: https://xxxx
openai.apikey: abcd1234
+ flink-conf:
+ execution.checkpointing.timeout: 12min
diff --git
a/flink-cdc-cli/src/test/resources/definitions/pipeline-definition-with-flink-conf-priority.yaml
b/flink-cdc-cli/src/test/resources/definitions/pipeline-definition-with-flink-conf-priority.yaml
new file mode 100644
index 000000000..9697ba784
--- /dev/null
+++
b/flink-cdc-cli/src/test/resources/definitions/pipeline-definition-with-flink-conf-priority.yaml
@@ -0,0 +1,34 @@
+################################################################################
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+################################################################################
+source:
+ type: values
+
+sink:
+ type: values
+
+pipeline:
+ name: priority-test-pipeline
+ parallelism: 2
+ flink-conf:
+ # 1. Option present in config.yaml (1), pipeline.yaml (2), and overridden
by CLI (3)
+ parallelism.default: 2
+ # 2. Option present in config.yaml (1) and pipeline.yaml (4), NOT in CLI
-> pipeline.yaml wins
+ taskmanager.numberOfTaskSlots: 4
+ # 3. Option present in pipeline.yaml (12min) and overridden by CLI (11min)
+ execution.checkpointing.timeout: 12min
+ # 4. Option present only in pipeline.yaml
+ execution.checkpointing.interval: 3min
diff --git
a/flink-cdc-cli/src/test/resources/definitions/pipeline-definition-with-nested-flink-conf.yaml
b/flink-cdc-cli/src/test/resources/definitions/pipeline-definition-with-nested-flink-conf.yaml
new file mode 100644
index 000000000..65cc33f3f
--- /dev/null
+++
b/flink-cdc-cli/src/test/resources/definitions/pipeline-definition-with-nested-flink-conf.yaml
@@ -0,0 +1,31 @@
+################################################################################
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+################################################################################
+source:
+ type: values
+
+sink:
+ type: values
+
+pipeline:
+ name: nested-flink-conf-pipe
+ parallelism: 2
+ flink-conf:
+ execution:
+ checkpointing:
+ interval: 3min
+ timeout: 15min
+ parallelism.default: 4
diff --git
a/flink-cdc-composer/src/main/java/org/apache/flink/cdc/composer/flink/FlinkPipelineComposer.java
b/flink-cdc-composer/src/main/java/org/apache/flink/cdc/composer/flink/FlinkPipelineComposer.java
index b21ef2ce3..ad297d448 100644
---
a/flink-cdc-composer/src/main/java/org/apache/flink/cdc/composer/flink/FlinkPipelineComposer.java
+++
b/flink-cdc-composer/src/main/java/org/apache/flink/cdc/composer/flink/FlinkPipelineComposer.java
@@ -70,6 +70,30 @@ public class FlinkPipelineComposer implements
PipelineComposer {
public static FlinkPipelineComposer ofRemoteCluster(
org.apache.flink.configuration.Configuration flinkConfig,
List<Path> additionalJars) {
StreamExecutionEnvironment env = new
StreamExecutionEnvironment(flinkConfig);
+ addAdditionalJars(env, additionalJars);
+ return new FlinkPipelineComposer(env, false);
+ }
+
+ public static FlinkPipelineComposer
ofApplicationCluster(StreamExecutionEnvironment env) {
+ return new FlinkPipelineComposer(env, false);
+ }
+
+ @VisibleForTesting
+ public static FlinkPipelineComposer ofMiniCluster() {
+ return new FlinkPipelineComposer(
+ StreamExecutionEnvironment.getExecutionEnvironment(), true);
+ }
+
+ public static FlinkPipelineComposer ofMiniCluster(
+ org.apache.flink.configuration.Configuration flinkConfig,
List<Path> additionalJars) {
+ StreamExecutionEnvironment env =
+
StreamExecutionEnvironment.getExecutionEnvironment(flinkConfig);
+ addAdditionalJars(env, additionalJars);
+ return new FlinkPipelineComposer(env, true);
+ }
+
+ private static void addAdditionalJars(
+ StreamExecutionEnvironment env, List<Path> additionalJars) {
additionalJars.forEach(
jarPath -> {
try {
@@ -84,16 +108,6 @@ public class FlinkPipelineComposer implements
PipelineComposer {
e);
}
});
- return new FlinkPipelineComposer(env, false);
- }
-
- public static FlinkPipelineComposer
ofApplicationCluster(StreamExecutionEnvironment env) {
- return new FlinkPipelineComposer(env, false);
- }
-
- public static FlinkPipelineComposer ofMiniCluster() {
- return new FlinkPipelineComposer(
- StreamExecutionEnvironment.getExecutionEnvironment(), true);
}
private FlinkPipelineComposer(StreamExecutionEnvironment env, boolean
isBlocking) {
diff --git
a/flink-cdc-composer/src/test/java/org/apache/flink/cdc/composer/flink/FlinkPipelineComposerTest.java
b/flink-cdc-composer/src/test/java/org/apache/flink/cdc/composer/flink/FlinkPipelineComposerTest.java
index a9da9c398..98277b1ee 100644
---
a/flink-cdc-composer/src/test/java/org/apache/flink/cdc/composer/flink/FlinkPipelineComposerTest.java
+++
b/flink-cdc-composer/src/test/java/org/apache/flink/cdc/composer/flink/FlinkPipelineComposerTest.java
@@ -42,20 +42,26 @@ import
org.apache.flink.cdc.connectors.values.factory.ValuesDataFactory;
import org.apache.flink.cdc.connectors.values.source.ValuesDataSource;
import org.apache.flink.cdc.connectors.values.source.ValuesDataSourceHelper;
import org.apache.flink.configuration.DeploymentOptions;
+import org.apache.flink.configuration.ReadableConfig;
+import org.apache.flink.core.fs.Path;
import org.apache.flink.shaded.guava31.com.google.common.collect.ImmutableMap;
-import org.assertj.core.api.Assertions;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.Arguments;
import org.junit.jupiter.params.provider.MethodSource;
+import java.time.Duration;
+import java.util.ArrayList;
import java.util.Collections;
import java.util.HashSet;
+import java.util.List;
import java.util.Set;
import java.util.stream.Stream;
+import static
org.apache.flink.configuration.CheckpointingOptions.CHECKPOINTING_INTERVAL;
+import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatCode;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
@@ -84,9 +90,8 @@ class FlinkPipelineComposerTest {
new Configuration(),
Thread.currentThread().getContextClassLoader()));
-
Assertions.assertThat(dataSink).isExactlyInstanceOf(DataSinkFactory1.TestDataSink.class);
- Assertions.assertThat(((DataSinkFactory1.TestDataSink)
dataSink).getHost())
- .isEqualTo("0.0.0.0");
+
assertThat(dataSink).isExactlyInstanceOf(DataSinkFactory1.TestDataSink.class);
+ assertThat(((DataSinkFactory1.TestDataSink)
dataSink).getHost()).isEqualTo("0.0.0.0");
}
@Test
@@ -321,4 +326,20 @@ class FlinkPipelineComposerTest {
Collections.emptyList(),
pipelineConfig);
}
+
+ @Test
+ void testOfMiniCluster() {
+ org.apache.flink.configuration.Configuration flinkConfig =
+ new org.apache.flink.configuration.Configuration();
+ flinkConfig.set(CHECKPOINTING_INTERVAL, Duration.ofSeconds(30));
+ List<Path> additionalJars = new ArrayList<>();
+ additionalJars.add(new Path("/path/to/additionalJars.jar"));
+ FlinkPipelineComposer flinkPipelineComposer =
+ FlinkPipelineComposer.ofMiniCluster(flinkConfig,
additionalJars);
+ ReadableConfig configuration =
flinkPipelineComposer.getEnv().getConfiguration();
+
+
assertThat(configuration.get(CHECKPOINTING_INTERVAL)).isEqualTo(Duration.ofSeconds(30));
+
assertThat(configuration.get(org.apache.flink.configuration.PipelineOptions.JARS))
+ .contains("file:/path/to/additionalJars.jar");
+ }
}
diff --git
a/flink-cdc-flink1-compat/src/main/java/org/apache/flink/core/execution/RestoreModeAdapter.java
b/flink-cdc-flink1-compat/src/main/java/org/apache/flink/core/execution/RestoreModeAdapter.java
new file mode 100644
index 000000000..8a16f79da
--- /dev/null
+++
b/flink-cdc-flink1-compat/src/main/java/org/apache/flink/core/execution/RestoreModeAdapter.java
@@ -0,0 +1,104 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.core.execution;
+
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.configuration.ConfigurationUtils;
+import org.apache.flink.configuration.StateRecoveryOptions;
+import org.apache.flink.runtime.jobgraph.SavepointRestoreSettings;
+
+import java.util.Optional;
+
+/**
+ * Compatibility adapter for Flink restore mode configuration.
+ *
+ * <p>This class is part of the multi-version compatibility layer that allows
Flink CDC to work
+ * across different Flink versions. It isolates version-specific Flink APIs
for restore mode
+ * configuration from the common Flink CDC code.
+ *
+ * <p>The adapter converts the restore mode object to the corresponding
Flink-specific restore mode
+ * and configures it in the Flink {@code Configuration}.
+ *
+ * <p>Different Flink versions may use different restore mode types. For
example, Flink 2.x uses
+ * {@code RecoveryClaimMode}, while Flink 1.x uses {@code RestoreMode}.
+ */
+public final class RestoreModeAdapter {
+
+ private RestoreModeAdapter() {}
+
+ public static void setRestoreMode(Configuration configuration, String
restoreModeStr) {
+ RestoreMode restoreMode =
+ ConfigurationUtils.convertValue(restoreModeStr,
RestoreMode.class);
+ setRestoreMode(configuration, restoreMode);
+ }
+
+ public static void setRestoreMode(Configuration configuration, Object
restoreMode) {
+ configuration.set(StateRecoveryOptions.RESTORE_MODE,
convert((RestoreMode) restoreMode));
+ }
+
+ public static void setSavepointIgnoreUnclaimedState(
+ Configuration configuration, boolean ignoreUnclaimedState) {
+ configuration.set(
+ StateRecoveryOptions.SAVEPOINT_IGNORE_UNCLAIMED_STATE,
ignoreUnclaimedState);
+ }
+
+ public static void setSavepointPath(Configuration configuration, String
savepointPath) {
+ configuration.set(StateRecoveryOptions.SAVEPOINT_PATH, savepointPath);
+ }
+
+ public static Object getRestoreMode(Configuration configuration) {
+ return configuration
+ .getOptional(StateRecoveryOptions.RESTORE_MODE)
+ .orElse(StateRecoveryOptions.RESTORE_MODE.defaultValue());
+ }
+
+ public static Optional<String> getSavepointPath(Configuration
configuration) {
+ return configuration.getOptional(StateRecoveryOptions.SAVEPOINT_PATH);
+ }
+
+ public static boolean getSavepointIgnoreUnclaimedState(Configuration
configuration) {
+ return configuration
+
.getOptional(StateRecoveryOptions.SAVEPOINT_IGNORE_UNCLAIMED_STATE)
+
.orElse(StateRecoveryOptions.SAVEPOINT_IGNORE_UNCLAIMED_STATE.defaultValue());
+ }
+
+ public static SavepointRestoreSettings createSavepointRestoreSettings(
+ Configuration configuration) {
+ Optional<String> savepointPath = getSavepointPath(configuration);
+ if (savepointPath.isEmpty()) {
+ return SavepointRestoreSettings.none();
+ }
+ return SavepointRestoreSettings.forPath(
+ savepointPath.get(),
+ getSavepointIgnoreUnclaimedState(configuration),
+ (RestoreMode) getRestoreMode(configuration));
+ }
+
+ private static RestoreMode convert(RestoreMode restoreMode) {
+ switch (restoreMode) {
+ case CLAIM:
+ return RestoreMode.CLAIM;
+ case NO_CLAIM:
+ return RestoreMode.NO_CLAIM;
+ case LEGACY:
+ return RestoreMode.LEGACY;
+ default:
+ throw new IllegalArgumentException("Unsupported restore mode:
" + restoreMode);
+ }
+ }
+}
diff --git
a/flink-cdc-flink2-compat/src/main/java/org/apache/flink/core/execution/RestoreModeAdapter.java
b/flink-cdc-flink2-compat/src/main/java/org/apache/flink/core/execution/RestoreModeAdapter.java
new file mode 100644
index 000000000..b0fcac2cd
--- /dev/null
+++
b/flink-cdc-flink2-compat/src/main/java/org/apache/flink/core/execution/RestoreModeAdapter.java
@@ -0,0 +1,106 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.core.execution;
+
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.configuration.ConfigurationUtils;
+import org.apache.flink.configuration.StateRecoveryOptions;
+import org.apache.flink.runtime.jobgraph.SavepointRestoreSettings;
+
+import java.util.Optional;
+
+/**
+ * Compatibility adapter for Flink restore mode configuration.
+ *
+ * <p>This class is part of the multi-version compatibility layer that allows
Flink CDC to work
+ * across different Flink versions. It isolates version-specific Flink APIs
for restore mode
+ * configuration from the common Flink CDC code.
+ *
+ * <p>The adapter converts the restore mode object to the corresponding
Flink-specific restore mode
+ * and configures it in the Flink {@code Configuration}.
+ *
+ * <p>Different Flink versions may use different restore mode types. For
example, Flink 2.x uses
+ * {@code RecoveryClaimMode}, while Flink 1.x uses {@code RestoreMode}.
+ */
+public final class RestoreModeAdapter {
+
+ private RestoreModeAdapter() {}
+
+ public static void setRestoreMode(Configuration configuration, String
restoreModeStr) {
+ RecoveryClaimMode restoreMode =
+ ConfigurationUtils.convertValue(restoreModeStr,
RecoveryClaimMode.class);
+ setRestoreMode(configuration, restoreMode);
+ }
+
+ public static void setRestoreMode(Configuration configuration, Object
restoreMode) {
+ configuration.set(
+ StateRecoveryOptions.RESTORE_MODE, convert((RecoveryClaimMode)
restoreMode));
+ }
+
+ public static void setSavepointIgnoreUnclaimedState(
+ Configuration configuration, boolean ignoreUnclaimedState) {
+ configuration.set(
+ StateRecoveryOptions.SAVEPOINT_IGNORE_UNCLAIMED_STATE,
ignoreUnclaimedState);
+ }
+
+ public static void setSavepointPath(Configuration configuration, String
savepointPath) {
+ configuration.set(StateRecoveryOptions.SAVEPOINT_PATH, savepointPath);
+ }
+
+ public static Object getRestoreMode(Configuration configuration) {
+ return configuration
+ .getOptional(StateRecoveryOptions.RESTORE_MODE)
+ .orElse(StateRecoveryOptions.RESTORE_MODE.defaultValue());
+ }
+
+ public static Optional<String> getSavepointPath(Configuration
configuration) {
+ return configuration.getOptional(StateRecoveryOptions.SAVEPOINT_PATH);
+ }
+
+ public static boolean getSavepointIgnoreUnclaimedState(Configuration
configuration) {
+ return configuration
+
.getOptional(StateRecoveryOptions.SAVEPOINT_IGNORE_UNCLAIMED_STATE)
+
.orElse(StateRecoveryOptions.SAVEPOINT_IGNORE_UNCLAIMED_STATE.defaultValue());
+ }
+
+ public static SavepointRestoreSettings createSavepointRestoreSettings(
+ Configuration configuration) {
+ Optional<String> savepointPath = getSavepointPath(configuration);
+ if (savepointPath.isEmpty()) {
+ return SavepointRestoreSettings.none();
+ }
+ return SavepointRestoreSettings.forPath(
+ savepointPath.get(),
+ getSavepointIgnoreUnclaimedState(configuration),
+ (RecoveryClaimMode) getRestoreMode(configuration));
+ }
+
+ private static RecoveryClaimMode convert(RecoveryClaimMode
recoveryClaimMode) {
+ switch (recoveryClaimMode) {
+ case CLAIM:
+ return RecoveryClaimMode.CLAIM;
+ case NO_CLAIM:
+ return RecoveryClaimMode.NO_CLAIM;
+ case LEGACY:
+ return RecoveryClaimMode.LEGACY;
+ default:
+ throw new IllegalArgumentException(
+ "Unsupported restore mode: " + recoveryClaimMode);
+ }
+ }
+}
