This is an automated email from the ASF dual-hosted git repository.
wenjin272 pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/flink-agents.git
The following commit(s) were added to refs/heads/main by this push:
new 10ba7294e [runtime][java] Fall back to legacy flink-conf.yaml when
loading agent configuration (#1337)
10ba7294e is described below
commit 10ba7294e07b307c5e75db8354ff64db2b1bb593
Author: Samuele Langhi <[email protected]>
AuthorDate: Wed Nov 4 03:17:50 2026 +0100
[runtime][java] Fall back to legacy flink-conf.yaml when loading agent
configuration (#1337)
Generated-by: Claude Code 2.1.312 (Claude Opus 5.5)
---
docs/content/docs/operations/configuration.md | 21 ++-
.../agents/runtime/env/LegacyFlinkConfParser.java | 85 ++++++++++++
.../runtime/env/RemoteExecutionEnvironment.java | 84 ++++++++++-
.../runtime/env/LegacyFlinkConfParserTest.java | 90 ++++++++++++
.../env/RemoteExecutionEnvironmentTest.java | 153 +++++++++++++++++++++
5 files changed, 427 insertions(+), 6 deletions(-)
diff --git a/docs/content/docs/operations/configuration.md
b/docs/content/docs/operations/configuration.md
index a5efff5b9..5118a95b3 100644
--- a/docs/content/docs/operations/configuration.md
+++ b/docs/content/docs/operations/configuration.md
@@ -94,6 +94,19 @@ agent:
async: true
```
+In Java, settings can also be written as flattened keys prefixed with `agent.`:
+
+```yaml
+agent.max-retries: 3
+agent.chat.async: true
+```
+
+This is the only form the legacy `flink-conf.yaml` supports. Java reads that
file with Flink's
+legacy rules rather than as YAML, as Flink 1.20 does: one `key: value` pair
per line, split on
+the first `": "`, with `#` starting a comment, invalid lines skipped and a
repeated key overriding
+earlier ones. Values are read as strings and converted when an option is read,
so list-typed
+options such as `event-listeners` cannot be set in `flink-conf.yaml`.
+
#### Loading Behavior
When creating an `AgentsExecutionEnvironment` for a Flink job, Flink Agents
loads
@@ -104,7 +117,13 @@ Set this environment variable before creating the
environment:
export FLINK_CONF_DIR="/path/to/flink/conf"
```
-`FLINK_CONF_DIR` points to a directory. Java reads `config.yaml` from that
directory.
+`FLINK_CONF_DIR` points to a directory. Java selects the file the way Flink
itself does:
+
+| Flink version | Files read |
+| --- | --- |
+| 1.20 | `flink-conf.yaml` if it exists (legacy rules), otherwise
`config.yaml` (standard YAML) |
+| 2.0 and later | `config.yaml` only (standard YAML) |
+
Python reads `flink-conf.yaml` if it exists, otherwise `config.yaml`.
If `FLINK_CONF_DIR` is unset, no YAML configuration file is loaded.
Configuration
diff --git
a/runtime/src/main/java/org/apache/flink/agents/runtime/env/LegacyFlinkConfParser.java
b/runtime/src/main/java/org/apache/flink/agents/runtime/env/LegacyFlinkConfParser.java
new file mode 100644
index 000000000..29db22741
--- /dev/null
+++
b/runtime/src/main/java/org/apache/flink/agents/runtime/env/LegacyFlinkConfParser.java
@@ -0,0 +1,85 @@
+/*
+ * 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.agents.runtime.env;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.BufferedReader;
+import java.io.File;
+import java.io.IOException;
+import java.nio.file.Files;
+import java.util.LinkedHashMap;
+import java.util.Map;
+
+/**
+ * Parser for the legacy Flink configuration file {@code flink-conf.yaml}.
+ *
+ * <p>Mirrors {@code GlobalConfiguration#loadLegacyYAMLResource} of Flink
1.20, which reads the file
+ * line by line rather than as YAML: everything after the first {@code #} is a
comment, key and
+ * value are split on the first {@code ": "}, lines without a valid pair are
skipped with a warning,
+ * values are kept as strings, and a duplicate key overrides the earlier one.
+ */
+final class LegacyFlinkConfParser {
+
+ private static final Logger LOG =
LoggerFactory.getLogger(LegacyFlinkConfParser.class);
+
+ private LegacyFlinkConfParser() {}
+
+ static Map<String, String> parse(File file) throws IOException {
+ final Map<String, String> config = new LinkedHashMap<>();
+
+ try (BufferedReader reader = Files.newBufferedReader(file.toPath())) {
+ String line;
+ int lineNo = 0;
+ while ((line = reader.readLine()) != null) {
+ lineNo++;
+ String conf = line.split("#", 2)[0].trim();
+ if (conf.isEmpty()) {
+ continue;
+ }
+
+ String[] kv = conf.split(": ", 2);
+ if (kv.length == 1) {
+ LOG.warn(
+ "Error while trying to split key and value in
configuration file {}:{}: "
+ + "Line is not a key-value pair (missing
space after ':'?)",
+ file,
+ lineNo);
+ continue;
+ }
+
+ String key = kv[0].trim();
+ String value = kv[1].trim();
+ if (key.isEmpty() || value.isEmpty()) {
+ LOG.warn(
+ "Error after splitting key and value in
configuration file {}:{}: "
+ + "Key or value was empty",
+ file,
+ lineNo);
+ continue;
+ }
+
+ config.put(key, value);
+ }
+ }
+
+ return config;
+ }
+}
diff --git
a/runtime/src/main/java/org/apache/flink/agents/runtime/env/RemoteExecutionEnvironment.java
b/runtime/src/main/java/org/apache/flink/agents/runtime/env/RemoteExecutionEnvironment.java
index e2218f665..ecea46faf 100644
---
a/runtime/src/main/java/org/apache/flink/agents/runtime/env/RemoteExecutionEnvironment.java
+++
b/runtime/src/main/java/org/apache/flink/agents/runtime/env/RemoteExecutionEnvironment.java
@@ -26,12 +26,14 @@ import org.apache.flink.agents.plan.AgentConfiguration;
import org.apache.flink.agents.plan.AgentPlan;
import org.apache.flink.agents.runtime.CompileUtils;
import org.apache.flink.agents.runtime.OutputTypeUtils;
+import org.apache.flink.annotation.VisibleForTesting;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.typeutils.RowTypeInfo;
import org.apache.flink.configuration.ConfigConstants;
import org.apache.flink.configuration.YamlParserUtils;
+import org.apache.flink.runtime.util.EnvironmentInformation;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
@@ -39,6 +41,8 @@ import org.apache.flink.table.api.Schema;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.types.Row;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
import javax.annotation.Nullable;
@@ -60,8 +64,14 @@ public class RemoteExecutionEnvironment extends
AgentsExecutionEnvironment {
private final AgentConfiguration config;
+ private static final Logger LOG =
LoggerFactory.getLogger(RemoteExecutionEnvironment.class);
+
public static final String FLINK_CONF_FILENAME = "config.yaml";
+ public static final String LEGACY_FLINK_CONF_FILENAME = "flink-conf.yaml";
+
+ private static final String AGENT_CONFIG_PREFIX = "agent";
+
public RemoteExecutionEnvironment(
StreamExecutionEnvironment env, @Nullable StreamTableEnvironment
tEnv) {
this.env = env;
@@ -113,16 +123,50 @@ public class RemoteExecutionEnvironment extends
AgentsExecutionEnvironment {
env.execute(jobName);
}
- @SuppressWarnings("unchecked")
public static AgentConfiguration loadAgentConfiguration(String configDir) {
+ return loadAgentConfiguration(configDir,
EnvironmentInformation.getVersion());
+ }
+
+ /**
+ * Selects and parses the configuration file the way Flink's {@code
GlobalConfiguration} does:
+ * on Flink 1.x the legacy {@code flink-conf.yaml} wins and is parsed with
the legacy line-based
+ * rules, otherwise {@code config.yaml} is parsed as standard YAML; Flink
2.x reads only {@code
+ * config.yaml}.
+ */
+ @VisibleForTesting
+ @SuppressWarnings("unchecked")
+ static AgentConfiguration loadAgentConfiguration(
+ String configDir, @Nullable String flinkVersion) {
try {
if (configDir == null) {
return new AgentConfiguration();
}
- final Map<String, Object> configData =
- (Map<String, Object>)
- YamlParserUtils.loadYamlFile(new File(configDir,
FLINK_CONF_FILENAME))
- .getOrDefault("agent", new HashMap<>());
+
+ File legacyConfigFile = new File(configDir,
LEGACY_FLINK_CONF_FILENAME);
+ if (isFlink2OrLater(flinkVersion)) {
+ if (legacyConfigFile.exists()) {
+ LOG.warn(
+ "Ignoring {}: Flink {} reads only {}",
+ legacyConfigFile,
+ flinkVersion,
+ FLINK_CONF_FILENAME);
+ }
+ } else if (legacyConfigFile.exists()) {
+ LOG.info(
+ "Using legacy YAML parser to load Flink Agents
configuration from {}",
+ legacyConfigFile);
+ return new AgentConfiguration(
+
flatAgentOptions(LegacyFlinkConfParser.parse(legacyConfigFile)));
+ }
+
+ final Map<String, Object> yaml =
+ YamlParserUtils.loadYamlFile(new File(configDir,
FLINK_CONF_FILENAME));
+ final Map<String, Object> configData = new HashMap<>();
+ Object nested = yaml.get(AGENT_CONFIG_PREFIX);
+ if (nested instanceof Map) {
+ configData.putAll((Map<String, Object>) nested);
+ }
+ configData.putAll(flatAgentOptions(yaml));
return new AgentConfiguration(configData);
} catch (Exception e) {
@@ -131,6 +175,36 @@ public class RemoteExecutionEnvironment extends
AgentsExecutionEnvironment {
}
}
+ /** Collects top-level {@code agent.<key>} entries under {@code <key>}. */
+ private static Map<String, Object> flatAgentOptions(Map<String, ?> config)
{
+ String flatPrefix = AGENT_CONFIG_PREFIX + ".";
+ Map<String, Object> options = new HashMap<>();
+ config.forEach(
+ (key, value) -> {
+ if (key.startsWith(flatPrefix)) {
+ options.put(key.substring(flatPrefix.length()), value);
+ }
+ });
+ return options;
+ }
+
+ /**
+ * Whether {@code version} names Flink 2.0 or later. Vendor suffixes after
the first {@code -}
+ * are ignored; an unknown or unparseable version counts as 1.x, which
tries both files.
+ */
+ @VisibleForTesting
+ static boolean isFlink2OrLater(@Nullable String version) {
+ if (version == null) {
+ return false;
+ }
+ String major = version.split("-", 2)[0].split("\\.", 2)[0];
+ try {
+ return Integer.parseInt(major.trim()) >= 2;
+ } catch (NumberFormatException e) {
+ return false;
+ }
+ }
+
/** Implementation of AgentBuilder for remote execution environment. */
private static class RemoteAgentBuilder<T, K> implements AgentBuilder {
diff --git
a/runtime/src/test/java/org/apache/flink/agents/runtime/env/LegacyFlinkConfParserTest.java
b/runtime/src/test/java/org/apache/flink/agents/runtime/env/LegacyFlinkConfParserTest.java
new file mode 100644
index 000000000..19791f3c7
--- /dev/null
+++
b/runtime/src/test/java/org/apache/flink/agents/runtime/env/LegacyFlinkConfParserTest.java
@@ -0,0 +1,90 @@
+/*
+ * 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.agents.runtime.env;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import java.io.File;
+import java.io.IOException;
+import java.nio.file.Files;
+import java.util.Arrays;
+import java.util.Map;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.assertj.core.api.Assertions.entry;
+
+class LegacyFlinkConfParserTest {
+ @TempDir private File tmpDir;
+
+ @Test
+ void testDuplicateKeyOverridesEarlierValue() throws IOException {
+ assertThat(parse("key: first", "key:
second")).containsOnly(entry("key", "second"));
+ }
+
+ @Test
+ void testValueContainingColonSpaceIsKeptWhole() throws IOException {
+ assertThat(parse("key: a: b: c")).containsOnly(entry("key", "a: b:
c"));
+ }
+
+ @Test
+ void testCommentsAreDropped() throws IOException {
+ assertThat(parse("# full-line comment", "key: value # trailing
comment", "other: a#b"))
+ .containsOnly(entry("key", "value"), entry("other", "a"));
+ }
+
+ @Test
+ void testLinesWithoutKeyValuePairAreSkipped() throws IOException {
+ assertThat(parse("key:value", "nested:", "key-only", "valid: yes"))
+ .containsOnly(entry("valid", "yes"));
+ }
+
+ @Test
+ void testEmptyKeyOrValueIsSkipped() throws IOException {
+ assertThat(parse(": value", "key: ", "key2: # only a comment",
"valid: yes"))
+ .containsOnly(entry("valid", "yes"));
+ }
+
+ @Test
+ void testKeysAndValuesAreTrimmedAndKeptAsStrings() throws IOException {
+ assertThat(parse(" key : 1 ", "flag: true", "quoted: 'v'",
"list: [a, b]"))
+ .containsOnly(
+ entry("key", "1"),
+ entry("flag", "true"),
+ entry("quoted", "'v'"),
+ entry("list", "[a, b]"));
+ }
+
+ @Test
+ void testIndentedLinesAreReadAsTopLevelKeys() throws IOException {
+ assertThat(parse("agent:", " key1: v1")).containsOnly(entry("key1",
"v1"));
+ }
+
+ @Test
+ void testMissingFileFails() {
+ assertThatThrownBy(() -> LegacyFlinkConfParser.parse(new File(tmpDir,
"missing.yaml")))
+ .isInstanceOf(IOException.class);
+ }
+
+ private Map<String, String> parse(String... lines) throws IOException {
+ File file = new File(tmpDir, "flink-conf.yaml");
+ Files.write(file.toPath(), Arrays.asList(lines));
+ return LegacyFlinkConfParser.parse(file);
+ }
+}
diff --git
a/runtime/src/test/java/org/apache/flink/agents/runtime/env/RemoteExecutionEnvironmentTest.java
b/runtime/src/test/java/org/apache/flink/agents/runtime/env/RemoteExecutionEnvironmentTest.java
index d49e291f8..ec44221f9 100644
---
a/runtime/src/test/java/org/apache/flink/agents/runtime/env/RemoteExecutionEnvironmentTest.java
+++
b/runtime/src/test/java/org/apache/flink/agents/runtime/env/RemoteExecutionEnvironmentTest.java
@@ -24,6 +24,7 @@ import org.apache.flink.agents.api.InputEvent;
import org.apache.flink.agents.api.OutputEvent;
import org.apache.flink.agents.api.agents.Agent;
import org.apache.flink.agents.api.annotation.Action;
+import org.apache.flink.agents.api.configuration.AgentConfigOptions;
import org.apache.flink.agents.api.context.MemoryObject;
import org.apache.flink.agents.api.context.RunnerContext;
import org.apache.flink.agents.plan.AgentConfiguration;
@@ -41,20 +42,28 @@ import org.junit.jupiter.api.io.TempDir;
import java.io.File;
import java.io.FileNotFoundException;
+import java.io.IOException;
import java.io.PrintWriter;
+import java.nio.file.Files;
import java.util.ArrayList;
+import java.util.Arrays;
import java.util.List;
import java.util.UUID;
import static
org.apache.flink.agents.runtime.env.RemoteExecutionEnvironment.FLINK_CONF_FILENAME;
+import static
org.apache.flink.agents.runtime.env.RemoteExecutionEnvironment.LEGACY_FLINK_CONF_FILENAME;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.assertj.core.api.Assertions.entry;
import static org.mockito.ArgumentMatchers.anyString;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
public class RemoteExecutionEnvironmentTest {
+ private static final String FLINK_1_20 = "1.20.3";
+ private static final String FLINK_2_3 = "2.3.0";
+
@TempDir private File tmpDir;
@Test
@@ -84,6 +93,150 @@ public class RemoteExecutionEnvironmentTest {
assert conf.getConfData().get("key7").equals(true);
}
+ @Test
+ void testIsFlink2OrLater() {
+
assertThat(RemoteExecutionEnvironment.isFlink2OrLater("1.20.3")).isFalse();
+
assertThat(RemoteExecutionEnvironment.isFlink2OrLater("1.20-vvr-11.5.0")).isFalse();
+
assertThat(RemoteExecutionEnvironment.isFlink2OrLater("2.3.0")).isTrue();
+
assertThat(RemoteExecutionEnvironment.isFlink2OrLater("2.0-SNAPSHOT")).isTrue();
+
assertThat(RemoteExecutionEnvironment.isFlink2OrLater("<unknown>")).isFalse();
+ assertThat(RemoteExecutionEnvironment.isFlink2OrLater("")).isFalse();
+ assertThat(RemoteExecutionEnvironment.isFlink2OrLater(null)).isFalse();
+ }
+
+ @Test
+ void testFlink1LoadsLegacyFileWhenOnlyLegacyFileExists() throws
IOException {
+ writeLegacyConf("agent.key1: fromLegacy");
+
+ assertThat(load(FLINK_1_20).getConfData()).containsOnly(entry("key1",
"fromLegacy"));
+ }
+
+ @Test
+ void testFlink1LoadsConfigYamlWhenOnlyConfigYamlExists() throws
IOException {
+ writeConfigYaml("agent:", " key1: fromConfigYaml");
+
+ assertThat(load(FLINK_1_20).getConfData()).containsOnly(entry("key1",
"fromConfigYaml"));
+ }
+
+ @Test
+ void testFlink1PrefersLegacyFileOverConfigYaml() throws IOException {
+ writeConfigYaml("agent:", " key1: fromConfigYaml");
+ writeLegacyConf("agent.key1: fromLegacy");
+
+ assertThat(load(FLINK_1_20).getConfData()).containsOnly(entry("key1",
"fromLegacy"));
+ }
+
+ @Test
+ void testFlink1FailsWhenNoConfigFileExists() {
+ assertThatThrownBy(() ->
load(FLINK_1_20)).isInstanceOf(RuntimeException.class);
+ }
+
+ @Test
+ void testFlink2IgnoresLegacyFile() throws IOException {
+ writeLegacyConf("agent.key1: fromLegacy");
+
+ assertThatThrownBy(() -> load(FLINK_2_3))
+ .isInstanceOf(RuntimeException.class)
+ .hasMessageContaining(tmpDir.getAbsolutePath());
+ }
+
+ @Test
+ void testFlink2LoadsConfigYamlWhenOnlyConfigYamlExists() throws
IOException {
+ writeConfigYaml("agent:", " key1: fromConfigYaml");
+
+ assertThat(load(FLINK_2_3).getConfData()).containsOnly(entry("key1",
"fromConfigYaml"));
+ }
+
+ @Test
+ void testFlink2LoadsConfigYamlWhenBothFilesExist() throws IOException {
+ writeConfigYaml("agent:", " key1: fromConfigYaml");
+ writeLegacyConf("agent.key1: fromLegacy");
+
+ assertThat(load(FLINK_2_3).getConfData()).containsOnly(entry("key1",
"fromConfigYaml"));
+ }
+
+ @Test
+ void testFlink2FailsWhenNoConfigFileExists() {
+ assertThatThrownBy(() ->
load(FLINK_2_3)).isInstanceOf(RuntimeException.class);
+ }
+
+ @Test
+ void testLegacyFileAcceptsDuplicateKeysAndValuesContainingColonSpace()
throws IOException {
+ // Both lines are rejected by a standard YAML parser but accepted by
Flink 1.20.
+ writeLegacyConf(
+ "env.java.opts.all: -Dfoo=bar",
+ "env.java.opts.all: -Dfoo=baz",
+ "kubernetes.jobmanager.annotations: owner: team-a",
+ "agent.key1: first",
+ "agent.key1: second",
+ "agent.key2: a: b");
+
+ assertThat(load(FLINK_1_20).getConfData())
+ .containsOnly(entry("key1", "second"), entry("key2", "a: b"));
+ }
+
+ @Test
+ void testLegacyFileWithoutAgentOptions() throws IOException {
+ writeLegacyConf(
+ "# Flink configuration as shipped by a distribution",
+ "jobmanager.rpc.address: localhost",
+ "taskmanager.memory.process.size: 1728m",
+ "env.java.opts.all: -XX:+UseG1GC -Dfoo=bar");
+
+ assertThat(load(FLINK_1_20).getConfData()).isEmpty();
+ }
+
+ @Test
+ void testLegacyFileIgnoresNestedAgentBlock() throws IOException {
+ writeLegacyConf("agent:", " key1: nested", "agent.key2: flat");
+
+ assertThat(load(FLINK_1_20).getConfData()).containsOnly(entry("key2",
"flat"));
+ }
+
+ @Test
+ void testLegacyFileValuesAreStringsAndConvertOnRead() throws IOException {
+ writeLegacyConf(
+ "agent.trace-log.base-dir: /tmp/agent-traces",
+ "agent.trace-log.pretty-print: true");
+
+ AgentConfiguration conf = load(FLINK_1_20);
+
+ assertThat(conf.getConfData()).containsEntry("trace-log.pretty-print",
"true");
+ assertThat(conf.get(AgentConfigOptions.TRACE_LOG_OUTPUT_BASE_DIR))
+ .isEqualTo("/tmp/agent-traces");
+
assertThat(conf.get(AgentConfigOptions.TRACE_LOG_OUTPUT_PRETTY_PRINT)).isTrue();
+ }
+
+ @Test
+ void testConfigYamlAcceptsNestedAndFlattenedKeys() throws IOException {
+ writeConfigYaml("agent:", " key1: v1", "agent.key2: v2");
+
+ assertThat(load(FLINK_2_3).getConfData())
+ .containsOnly(entry("key1", "v1"), entry("key2", "v2"));
+ }
+
+ @Test
+ void testFailIfConfigYamlUnparseable() throws IOException {
+ writeConfigYaml("agent.key1: [unclosed");
+
+ assertThatThrownBy(() -> load(FLINK_2_3))
+ .isInstanceOf(RuntimeException.class)
+ .hasMessageContaining(tmpDir.getAbsolutePath());
+ }
+
+ private AgentConfiguration load(String flinkVersion) {
+ return RemoteExecutionEnvironment.loadAgentConfiguration(
+ tmpDir.getAbsolutePath(), flinkVersion);
+ }
+
+ private void writeLegacyConf(String... lines) throws IOException {
+ Files.write(new File(tmpDir, LEGACY_FLINK_CONF_FILENAME).toPath(),
Arrays.asList(lines));
+ }
+
+ private void writeConfigYaml(String... lines) throws IOException {
+ Files.write(new File(tmpDir, FLINK_CONF_FILENAME).toPath(),
Arrays.asList(lines));
+ }
+
@Test
void testLoadAgentConfigurationFailIfNotLoaded() {
assertThatThrownBy(
