This is an automated email from the ASF dual-hosted git repository.
renqs 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 21032263c [FLINK-34969][cdc][cli] Add support for both new and old
Flink config files in Flink CDC (#3194)
21032263c is described below
commit 21032263c54b9b14999dc234fea551b7253c6e4f
Author: sky <[email protected]>
AuthorDate: Tue Apr 23 17:58:08 2024 +0800
[FLINK-34969][cdc][cli] Add support for both new and old Flink config files
in Flink CDC (#3194)
---
flink-cdc-cli/pom.xml | 8 +-
.../java/org/apache/flink/cdc/cli/CliFrontend.java | 4 +-
.../flink/cdc/cli/utils/ConfigurationUtils.java | 54 ++--
.../flink/cdc/cli/utils/FlinkEnvironmentUtils.java | 19 +-
.../flink/cdc/cli/utils/YamlParserUtils.java | 293 +++++++++++++++++++++
.../cdc/cli/utils/ConfigurationUtilsTest.java | 87 ++++++
.../src/test/resources/flink-home/conf/config.yaml | 41 +++
7 files changed, 478 insertions(+), 28 deletions(-)
diff --git a/flink-cdc-cli/pom.xml b/flink-cdc-cli/pom.xml
index 285462031..1aa57f336 100644
--- a/flink-cdc-cli/pom.xml
+++ b/flink-cdc-cli/pom.xml
@@ -29,6 +29,7 @@ limitations under the License.
<properties>
<commons-cli.version>1.6.0</commons-cli.version>
+ <snakeyaml.version>2.6</snakeyaml.version>
</properties>
<dependencies>
@@ -37,7 +38,12 @@ limitations under the License.
<artifactId>flink-cdc-common</artifactId>
<version>${project.version}</version>
</dependency>
-
+ <!-- YAML parser utilities -->
+ <dependency>
+ <groupId>org.snakeyaml</groupId>
+ <artifactId>snakeyaml-engine</artifactId>
+ <version>${snakeyaml.version}</version>
+ </dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-cdc-composer</artifactId>
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 663f03942..c67c3cf99 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
@@ -169,7 +169,7 @@ public class CliFrontend {
if (globalConfig != null) {
Path globalConfigPath = Paths.get(globalConfig);
LOG.info("Using global config in command line: {}",
globalConfigPath);
- return ConfigurationUtils.loadMapFormattedConfig(globalConfigPath);
+ return ConfigurationUtils.loadConfigFile(globalConfigPath);
}
// Fallback to Flink CDC home
@@ -178,7 +178,7 @@ public class CliFrontend {
Path globalConfigPath =
Paths.get(flinkCdcHome).resolve("conf").resolve("flink-cdc.yaml");
LOG.info("Using global config in FLINK_CDC_HOME: {}",
globalConfigPath);
- return ConfigurationUtils.loadMapFormattedConfig(globalConfigPath);
+ return ConfigurationUtils.loadConfigFile(globalConfigPath);
}
// Fallback to empty configuration
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 efe0a5143..38fa608f5 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
@@ -19,33 +19,41 @@ package org.apache.flink.cdc.cli.utils;
import org.apache.flink.cdc.common.configuration.Configuration;
-import
org.apache.flink.shaded.jackson2.com.fasterxml.jackson.core.type.TypeReference;
-import
org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.ObjectMapper;
-import
org.apache.flink.shaded.jackson2.com.fasterxml.jackson.dataformat.yaml.YAMLFactory;
-
-import java.io.FileNotFoundException;
-import java.nio.file.Files;
import java.nio.file.Path;
+import java.util.HashMap;
+import java.util.List;
import java.util.Map;
/** Utilities for handling {@link Configuration}. */
public class ConfigurationUtils {
- public static Configuration loadMapFormattedConfig(Path configPath) throws
Exception {
- if (!Files.exists(configPath)) {
- throw new FileNotFoundException(
- String.format("Cannot find configuration file at \"%s\"",
configPath));
- }
- ObjectMapper mapper = new ObjectMapper(new YAMLFactory());
- try {
- Map<String, String> configMap =
- mapper.readValue(
- configPath.toFile(), new TypeReference<Map<String,
String>>() {});
- return Configuration.fromMap(configMap);
- } catch (Exception e) {
- throw new IllegalStateException(
- String.format(
- "Failed to load config file \"%s\" to key-value
pairs", configPath),
- e);
- }
+
+ private static final String KEY_SEPARATOR = ".";
+
+ public static Configuration loadConfigFile(Path configPath) throws
Exception {
+ Map<String, Object> configMap =
YamlParserUtils.loadYamlFile(configPath.toFile());
+ return Configuration.fromMap(flattenConfigMap(configMap, ""));
+ }
+
+ @SuppressWarnings("unchecked")
+ private static Map<String, String> flattenConfigMap(
+ Map<String, Object> config, String keyPrefix) {
+ final Map<String, String> flattenedMap = new HashMap<>();
+
+ config.forEach(
+ (key, value) -> {
+ String flattenedKey = keyPrefix + key;
+ if (value instanceof Map) {
+ Map<String, Object> e = (Map<String, Object>) value;
+ flattenedMap.putAll(flattenConfigMap(e, flattenedKey +
KEY_SEPARATOR));
+ } else {
+ if (value instanceof List) {
+ flattenedMap.put(flattenedKey,
YamlParserUtils.toYAMLString(value));
+ } else {
+ flattenedMap.put(flattenedKey, value.toString());
+ }
+ }
+ });
+
+ return flattenedMap;
}
}
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 4880fa47b..d250ea835 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
@@ -21,18 +21,33 @@ import
org.apache.flink.cdc.common.configuration.Configuration;
import org.apache.flink.cdc.composer.flink.FlinkPipelineComposer;
import org.apache.flink.runtime.jobgraph.SavepointRestoreSettings;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.FileNotFoundException;
import java.nio.file.Path;
import java.util.List;
/** Utilities for handling Flink configuration and environment. */
public class FlinkEnvironmentUtils {
+ private static final Logger LOG =
LoggerFactory.getLogger(FlinkEnvironmentUtils.class);
private static final String FLINK_CONF_DIR = "conf";
- private static final String FLINK_CONF_FILENAME = "flink-conf.yaml";
+ 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 {
Path flinkConfPath =
flinkHome.resolve(FLINK_CONF_DIR).resolve(FLINK_CONF_FILENAME);
- return ConfigurationUtils.loadMapFormattedConfig(flinkConfPath);
+ try {
+ return ConfigurationUtils.loadConfigFile(flinkConfPath);
+ } catch (FileNotFoundException e) {
+ LOG.warn(
+ "Failed to load the configuration file from {}. Trying to
use legacy YAML parser to load flink configuration file from {}.",
+ FLINK_CONF_FILENAME,
+ LEGACY_FLINK_CONF_FILENAME);
+ return ConfigurationUtils.loadConfigFile(
+
flinkHome.resolve(FLINK_CONF_DIR).resolve(LEGACY_FLINK_CONF_FILENAME));
+ }
}
public static FlinkPipelineComposer createComposer(
diff --git
a/flink-cdc-cli/src/main/java/org/apache/flink/cdc/cli/utils/YamlParserUtils.java
b/flink-cdc-cli/src/main/java/org/apache/flink/cdc/cli/utils/YamlParserUtils.java
new file mode 100644
index 000000000..d47332b01
--- /dev/null
+++
b/flink-cdc-cli/src/main/java/org/apache/flink/cdc/cli/utils/YamlParserUtils.java
@@ -0,0 +1,293 @@
+/*
+ * 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.cdc.cli.utils;
+
+import org.apache.flink.cdc.common.utils.TimeUtils;
+import org.apache.flink.configuration.MemorySize;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.snakeyaml.engine.v2.api.Dump;
+import org.snakeyaml.engine.v2.api.DumpSettings;
+import org.snakeyaml.engine.v2.api.Load;
+import org.snakeyaml.engine.v2.api.LoadSettings;
+import org.snakeyaml.engine.v2.common.FlowStyle;
+import org.snakeyaml.engine.v2.exceptions.Mark;
+import org.snakeyaml.engine.v2.exceptions.MarkedYamlEngineException;
+import org.snakeyaml.engine.v2.exceptions.YamlEngineException;
+import org.snakeyaml.engine.v2.nodes.Node;
+import org.snakeyaml.engine.v2.nodes.ScalarNode;
+import org.snakeyaml.engine.v2.nodes.Tag;
+import org.snakeyaml.engine.v2.representer.StandardRepresenter;
+import org.snakeyaml.engine.v2.schema.CoreSchema;
+
+import javax.annotation.Nonnull;
+
+import java.io.File;
+import java.io.FileInputStream;
+import java.io.FileNotFoundException;
+import java.io.IOException;
+import java.time.Duration;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+
+/**
+ * This class contains utility methods to load standard yaml file and convert
object to standard
+ * yaml syntax.
+ */
+public class YamlParserUtils {
+
+ private static final Logger LOG =
LoggerFactory.getLogger(YamlParserUtils.class);
+
+ private static final DumpSettings blockerDumperSettings =
+ DumpSettings.builder()
+ .setDefaultFlowStyle(FlowStyle.BLOCK)
+ // Disable split long lines to avoid add unexpected line
breaks
+ .setSplitLines(false)
+ .setSchema(new CoreSchema())
+ .build();
+
+ private static final DumpSettings flowDumperSettings =
+ DumpSettings.builder()
+ .setDefaultFlowStyle(FlowStyle.FLOW)
+ // Disable split long lines to avoid add unexpected line
breaks
+ .setSplitLines(false)
+ .setSchema(new CoreSchema())
+ .build();
+
+ private static final Dump blockerDumper =
+ new Dump(blockerDumperSettings, new
FlinkConfigRepresenter(blockerDumperSettings));
+
+ private static final Dump flowDumper =
+ new Dump(flowDumperSettings, new
FlinkConfigRepresenter(flowDumperSettings));
+
+ private static final Load loader =
+ new Load(LoadSettings.builder().setSchema(new
CoreSchema()).build());
+
+ /**
+ * Loads the contents of the given YAML file into a map.
+ *
+ * @param file the YAML file to load.
+ * @return a non-null map representing the YAML content. If the file is
empty or only contains
+ * comments, an empty map is returned.
+ * @throws FileNotFoundException if the YAML file is not found.
+ * @throws YamlEngineException if the file cannot be parsed.
+ * @throws IOException if an I/O error occurs while reading from the file
stream.
+ */
+ @SuppressWarnings("unchecked")
+ public static synchronized @Nonnull Map<String, Object> loadYamlFile(File
file)
+ throws Exception {
+ try (FileInputStream inputStream = new FileInputStream((file))) {
+ Map<String, Object> yamlResult =
+ (Map<String, Object>)
loader.loadFromInputStream(inputStream);
+ return yamlResult == null ? new HashMap<>() : yamlResult;
+ } catch (FileNotFoundException e) {
+ LOG.error("Failed to find YAML file", e);
+ throw e;
+ } catch (IOException | YamlEngineException e) {
+ if (e instanceof MarkedYamlEngineException) {
+ YamlEngineException exception =
+
wrapExceptionToHiddenSensitiveData((MarkedYamlEngineException) e);
+ LOG.error("Failed to parse YAML configuration", exception);
+ throw exception;
+ } else {
+ throw e;
+ }
+ }
+ }
+
+ /**
+ * Converts the given value to a string representation in the YAML syntax.
This method uses a
+ * YAML parser to convert the object to YAML format.
+ *
+ * <p>The resulting YAML string may have line breaks at the end of each
line. This method
+ * removes the line break at the end of the string if it exists.
+ *
+ * <p>Note: This method may perform escaping on certain characters in the
value to ensure proper
+ * YAML syntax.
+ *
+ * @param value The value to be converted.
+ * @return The string representation of the value in YAML syntax.
+ */
+ public static synchronized String toYAMLString(Object value) {
+ try {
+ String output = flowDumper.dumpToString(value);
+ // remove the line break
+ String linebreak = flowDumperSettings.getBestLineBreak();
+ if (output.endsWith(linebreak)) {
+ output = output.substring(0, output.length() -
linebreak.length());
+ }
+ return output;
+ } catch (MarkedYamlEngineException exception) {
+ throw wrapExceptionToHiddenSensitiveData(exception);
+ }
+ }
+
+ /**
+ * Converts a flat map into a nested map structure and outputs the result
as a list of
+ * YAML-formatted strings. Each item in the list represents a single line
of the YAML data. The
+ * method is synchronized and thus thread-safe.
+ *
+ * @param flattenMap A map containing flattened keys (e.g.,
"parent.child.key") associated with
+ * their values.
+ * @return A list of strings that represents the YAML data, where each
item corresponds to a
+ * line of the data.
+ */
+ @SuppressWarnings("unchecked")
+ public static synchronized List<String> convertAndDumpYamlFromFlatMap(
+ Map<String, Object> flattenMap) {
+ try {
+ Map<String, Object> nestedMap = new LinkedHashMap<>();
+ for (Map.Entry<String, Object> entry : flattenMap.entrySet()) {
+ String[] keys = entry.getKey().split("\\.");
+ Map<String, Object> currentMap = nestedMap;
+ for (int i = 0; i < keys.length - 1; i++) {
+ currentMap =
+ (Map<String, Object>)
+ currentMap.computeIfAbsent(keys[i], k ->
new LinkedHashMap<>());
+ }
+ currentMap.put(keys[keys.length - 1], entry.getValue());
+ }
+ String data = blockerDumper.dumpToString(nestedMap);
+ String linebreak = blockerDumperSettings.getBestLineBreak();
+ return Arrays.asList(data.split(linebreak));
+ } catch (MarkedYamlEngineException exception) {
+ throw wrapExceptionToHiddenSensitiveData(exception);
+ }
+ }
+
+ public static synchronized <T> T convertToObject(String value, Class<T>
type) {
+ try {
+ return type.cast(loader.loadFromString(value));
+ } catch (MarkedYamlEngineException exception) {
+ throw wrapExceptionToHiddenSensitiveData(exception);
+ }
+ }
+
+ /**
+ * This method wraps a MarkedYAMLException to hide sensitive data in its
message. Before using
+ * this method, an exception message might include sensitive information
like:
+ *
+ * <pre>{@code
+ * while constructing a mapping
+ * in 'reader', line 1, column 1:
+ * key1: secret1
+ * ^
+ * found duplicate key key1
+ * in 'reader', line 2, column 1:
+ * key1: secret2
+ * ^
+ * }</pre>
+ *
+ * <p>After using this method, the message will be sanitized to hide the
sensitive details:
+ *
+ * <pre>{@code
+ * while constructing a mapping
+ * in 'reader', line 1, column 1
+ * found duplicate key key1
+ * in 'reader', line 2, column 1
+ * }</pre>
+ *
+ * @param exception The MarkedYamlEngineException containing potentially
sensitive data.
+ * @return A YamlEngineException with a message that has sensitive data
hidden.
+ */
+ private static YamlEngineException wrapExceptionToHiddenSensitiveData(
+ MarkedYamlEngineException exception) {
+ StringBuilder lines = new StringBuilder();
+ String context = exception.getContext();
+ Optional<Mark> contextMark = exception.getContextMark();
+ Optional<Mark> problemMark = exception.getProblemMark();
+ String problem = exception.getProblem();
+
+ if (context != null) {
+ lines.append(context);
+ lines.append("\n");
+ }
+
+ if (contextMark.isPresent()
+ && (problem == null
+ || !problemMark.isPresent()
+ ||
contextMark.get().getName().equals(problemMark.get().getName())
+ || contextMark.get().getLine() !=
problemMark.get().getLine()
+ || contextMark.get().getColumn() !=
problemMark.get().getColumn())) {
+ lines.append(hiddenSensitiveDataInMark(contextMark.get()));
+ lines.append("\n");
+ }
+
+ if (problem != null) {
+ lines.append(problem);
+ lines.append("\n");
+ }
+
+ if (problemMark.isPresent()) {
+ lines.append(hiddenSensitiveDataInMark(problemMark.get()));
+ lines.append("\n");
+ }
+
+ Throwable cause = exception.getCause();
+ if (cause instanceof MarkedYamlEngineException) {
+ cause =
wrapExceptionToHiddenSensitiveData((MarkedYamlEngineException) cause);
+ }
+
+ YamlEngineException yamlException = new
YamlEngineException(lines.toString(), cause);
+ yamlException.setStackTrace(exception.getStackTrace());
+ return yamlException;
+ }
+
+ /**
+ * This method is a mock implementation of the Mark#toString() method,
specifically designed to
+ * exclude the Mark#get_snippet(), to prevent leaking any sensitive data.
+ */
+ private static String hiddenSensitiveDataInMark(Mark mark) {
+ return " in "
+ + mark.getName()
+ + ", line "
+ + (mark.getLine() + 1)
+ + ", column "
+ + (mark.getColumn() + 1);
+ }
+
+ private static class FlinkConfigRepresenter extends StandardRepresenter {
+ public FlinkConfigRepresenter(DumpSettings dumpSettings) {
+ super(dumpSettings);
+ representers.put(Duration.class, this::representDuration);
+ representers.put(MemorySize.class, this::representMemorySize);
+ parentClassRepresenters.put(Enum.class, this::representEnum);
+ }
+
+ private Node representDuration(Object data) {
+ Duration duration = (Duration) data;
+ String durationString = TimeUtils.formatWithHighestUnit(duration);
+ return new ScalarNode(Tag.STR, durationString,
settings.getDefaultScalarStyle());
+ }
+
+ private Node representMemorySize(Object data) {
+ MemorySize memorySize = (MemorySize) data;
+ return new ScalarNode(Tag.STR, memorySize.toString(),
settings.getDefaultScalarStyle());
+ }
+
+ private Node representEnum(Object data) {
+ return new ScalarNode(Tag.STR, data.toString(),
settings.getDefaultScalarStyle());
+ }
+ }
+}
diff --git
a/flink-cdc-cli/src/test/java/org/apache/flink/cdc/cli/utils/ConfigurationUtilsTest.java
b/flink-cdc-cli/src/test/java/org/apache/flink/cdc/cli/utils/ConfigurationUtilsTest.java
new file mode 100644
index 000000000..7e012fce9
--- /dev/null
+++
b/flink-cdc-cli/src/test/java/org/apache/flink/cdc/cli/utils/ConfigurationUtilsTest.java
@@ -0,0 +1,87 @@
+/*
+ * 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.cdc.cli.utils;
+
+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.shaded.curator5.com.google.common.io.Resources;
+
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+
+import java.net.URL;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.util.HashMap;
+import java.util.Map;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/** Unit test for {@link org.apache.flink.cdc.cli.utils.ConfigurationUtils}. */
+class ConfigurationUtilsTest {
+
+ private static final Map<ConfigOption<?>, Object> CONFIG_OPTIONS = new
HashMap<>();
+
+ static {
+ CONFIG_OPTIONS.put(
+
ConfigOptions.key("jobmanager.rpc.address").stringType().noDefaultValue(),
+ "localhost");
+ CONFIG_OPTIONS.put(
+
ConfigOptions.key("jobmanager.rpc.port").stringType().noDefaultValue(), "6123");
+ CONFIG_OPTIONS.put(
+
ConfigOptions.key("jobmanager.bind-host").stringType().noDefaultValue(),
+ "localhost");
+ CONFIG_OPTIONS.put(
+
ConfigOptions.key("jobmanager.memory.process.size").stringType().noDefaultValue(),
+ "1600m");
+ CONFIG_OPTIONS.put(
+ ConfigOptions.key("jobmanager.execution.failover-strategy")
+ .stringType()
+ .noDefaultValue(),
+ "region");
+ CONFIG_OPTIONS.put(
+
ConfigOptions.key("taskmanager.bind-host").stringType().noDefaultValue(),
+ "localhost");
+ CONFIG_OPTIONS.put(
+
ConfigOptions.key("taskmanager.host").stringType().noDefaultValue(),
"localhost");
+ CONFIG_OPTIONS.put(
+
ConfigOptions.key("taskmanager.memory.process.size").stringType().noDefaultValue(),
+ "1728m");
+ CONFIG_OPTIONS.put(
+
ConfigOptions.key("taskmanager.numberOfTaskSlots").stringType().noDefaultValue(),
+ "1");
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = {"flink-home/conf/config.yaml",
"flink-home/conf/flink-conf.yaml"})
+ void loadConfigFile(String resourcePath) throws Exception {
+ URL resource = Resources.getResource(resourcePath);
+ Path path = Paths.get(resource.toURI());
+ Configuration configuration = ConfigurationUtils.loadConfigFile(path);
+ Map<String, String> configMap = configuration.toMap();
+ for (Map.Entry<ConfigOption<?>, Object> entry :
CONFIG_OPTIONS.entrySet()) {
+ String key = entry.getKey().key();
+ Object expectedValue = entry.getValue();
+ assertTrue(configMap.containsKey(key));
+ assertEquals(expectedValue, configMap.get(key));
+ }
+ }
+}
diff --git a/flink-cdc-cli/src/test/resources/flink-home/conf/config.yaml
b/flink-cdc-cli/src/test/resources/flink-home/conf/config.yaml
new file mode 100644
index 000000000..30c44f547
--- /dev/null
+++ b/flink-cdc-cli/src/test/resources/flink-home/conf/config.yaml
@@ -0,0 +1,41 @@
+# 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.
+################################################################################
+
+jobmanager:
+ rpc:
+ address: localhost
+ port: 6123
+ bind-host: localhost
+ memory:
+ process:
+ size: 1600m
+ execution:
+ failover-strategy: region
+
+taskmanager:
+ bind-host: localhost
+ host: localhost
+ memory:
+ process:
+ size: 1728m
+ numberOfTaskSlots: 1
+
+
+parallelism:
+ default: 1
+
+# The rest of your YAML file...
