yuxiqian commented on code in PR #3918:
URL: https://github.com/apache/flink-cdc/pull/3918#discussion_r4068331037
##########
docs/content.zh/docs/core-concept/data-pipeline.md:
##########
@@ -125,5 +127,6 @@ under the License.
| `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 |
+| `flink-conf` |
用于配置[Flink相关参数](https://nightlies.apache.org/flink/flink-docs-master/docs/deployment/config/)。
<br/>Flink参数优先级：config.yaml < job command-line < pipeline.yaml | optional
|
Review Comment:
```suggestion
| `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/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
Review Comment:
I suggest we not to modify each connectors' docs.
##########
flink-cdc-composer/src/main/java/org/apache/flink/cdc/composer/flink/FlinkPipelineComposer.java:
##########
@@ -70,32 +70,34 @@ public class FlinkPipelineComposer implements
PipelineComposer {
public static FlinkPipelineComposer ofRemoteCluster(
org.apache.flink.configuration.Configuration flinkConfig,
List<Path> additionalJars) {
StreamExecutionEnvironment env = new
StreamExecutionEnvironment(flinkConfig);
- additionalJars.forEach(
- jarPath -> {
- try {
- FlinkEnvironmentUtils.addJar(
- env,
-
jarPath.makeQualified(jarPath.getFileSystem()).toUri().toURL());
- } catch (Exception e) {
- throw new RuntimeException(
- String.format(
- "Unable to convert JAR path \"%s\" to
URL when adding JAR to Flink environment",
- jarPath),
- e);
- }
- });
+ addAdditionalJars(env, additionalJars);
return new FlinkPipelineComposer(env, false);
}
public static FlinkPipelineComposer
ofApplicationCluster(StreamExecutionEnvironment env) {
return new FlinkPipelineComposer(env, false);
}
+ public static FlinkPipelineComposer ofMiniCluster(
Review Comment:
Seems unnecessary?
##########
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
Review Comment:
Seems the `execution.target` is always overridden by the `--target` command
and could not be specified in this way.
##########
docs/content/docs/core-concept/data-pipeline.md:
##########
@@ -127,5 +129,6 @@ Note that whilst the parameters are each individually
optional, at least one of
| `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 |
| `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.
| optional |
| `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).
| 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 |
Review Comment:
```suggestion
| `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 |
```
##########
flink-cdc-cli/src/main/java/org/apache/flink/cdc/cli/parser/YamlPipelineDefinitionParser.java:
##########
@@ -642,4 +646,19 @@ private JsonNode validateArray(String contextName,
JsonNode jsonNode) {
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))
+ .map(
+ flinkNode ->
+ mapper.convertValue(
+ flinkNode, new
TypeReference<Map<String, String>>() {}))
+ .orElse(Collections.emptyMap());
+ }
+ }
Review Comment:
Nested config isn't supported here. If it's intended, please specify it in
docs.
##########
flink-cdc-cli/src/main/java/org/apache/flink/cdc/cli/utils/FlinkEnvironmentUtils.java:
##########
@@ -17,38 +17,211 @@
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.lang.reflect.InvocationTargetException;
+import java.util.Arrays;
+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 configuration from flink_home、command line and
flink pipeline.yaml.flink
+ * config priority: CDC CLI arguments > YAML Pipeline config blocks >
Flink `config.yaml`.
+ */
Review Comment:
```suggestion
/**
* Load and merge Flink configurations by priority rules.
*/
```
##########
flink-cdc-cli/src/test/java/org/apache/flink/cdc/cli/CliFrontendTest.java:
##########
@@ -112,6 +114,22 @@ void testSavePointConfiguration() throws Exception {
assertThat(executor.getFlinkConfig().get(SAVEPOINT_IGNORE_UNCLAIMED_STATE)).isTrue();
}
+ @Test
+ void testFlinkConfigurationWithPriority() throws Exception {
+ CliExecutor executor =
+ createExecutor(
+ pipelineDef(),
+ "--flink-home",
+ flinkHome(),
+ "-D",
+ "execution.checkpointing.timeout=11min");
+ assertThat(executor.getFlinkConfig().get(CHECKPOINTING_TIMEOUT))
+ .as(
+ "execution.checkpointing.timeout config priority test:
"
+ + "10min(default: config.yml) < pipeline.yml:
12min < command-line: 11min")
+ .isEqualTo(Duration.ofMinutes(11));
+ }
Review Comment:
Tests are insufficient here.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
