From 9cbe474aa2804408442ef69fa91ffbd9e8cf0e2e Mon Sep 17 00:00:00 2001 From: Marvin Froeder Date: Tue, 21 Jul 2026 13:20:53 -0300 Subject: [PATCH 1/2] Resolve env-var placeholders in Flink config.yaml Signed-off-by: Marvin Froeder --- .../com/datasqrl/flinkrunner/BaseRunner.java | 19 +++++++++++++++ .../datasqrl/flinkrunner/CliRunnerTest.java | 24 +++++++++++++++++++ 2 files changed, 43 insertions(+) diff --git a/flink-sql-runner/src/main/java/com/datasqrl/flinkrunner/BaseRunner.java b/flink-sql-runner/src/main/java/com/datasqrl/flinkrunner/BaseRunner.java index eeb9ccbd..bda242f2 100644 --- a/flink-sql-runner/src/main/java/com/datasqrl/flinkrunner/BaseRunner.java +++ b/flink-sql-runner/src/main/java/com/datasqrl/flinkrunner/BaseRunner.java @@ -92,6 +92,7 @@ Configuration initConfiguration() { if (StringUtils.isNotBlank(configDir)) { log.info("Loading Flink configuration from '{}'", configDir); conf = GlobalConfiguration.loadConfiguration(configDir); + resolveEnvVars(conf); } // Do not overwrite runtime given in YAML @@ -102,6 +103,24 @@ Configuration initConfiguration() { return conf; } + /** + * Resolves {@code ${VAR}} placeholders in Flink configuration values from the environment, the + * same mechanism already applied to the SQL script and compiled plan. Lets secrets (e.g. the + * Iceberg maintenance lock JDBC credentials) be injected via env vars instead of landing in the + * plaintext config.yaml ConfigMap. {@link Configuration#toMap()} returns a copy, so mutating + * {@code conf} while iterating it is safe. + */ + private void resolveEnvVars(Configuration conf) { + conf.toMap() + .forEach( + (key, value) -> { + var resolved = resolver.resolve(value); + if (!resolved.equals(value)) { + conf.setString(key, resolved); + } + }); + } + static String readTextFile(String path) throws IOException { var f = new File(path); if (!f.exists()) { diff --git a/flink-sql-runner/src/test/java/com/datasqrl/flinkrunner/CliRunnerTest.java b/flink-sql-runner/src/test/java/com/datasqrl/flinkrunner/CliRunnerTest.java index d059acbc..9a65d4b6 100644 --- a/flink-sql-runner/src/test/java/com/datasqrl/flinkrunner/CliRunnerTest.java +++ b/flink-sql-runner/src/test/java/com/datasqrl/flinkrunner/CliRunnerTest.java @@ -26,6 +26,7 @@ import java.io.IOException; import java.nio.file.Files; import java.nio.file.Path; +import java.util.Map; import java.util.function.Supplier; import org.apache.flink.api.common.RuntimeExecutionMode; import org.apache.flink.configuration.ExecutionOptions; @@ -171,4 +172,27 @@ void initConfiguration_shouldCreateNewConfigIfNoPathGiven(RuntimeExecutionMode m assertThat(finalConf.keySet()).hasSize(1); assertThat(finalConf.get(ExecutionOptions.RUNTIME_MODE)).isEqualTo(mode); } + + @Test + void initConfiguration_shouldResolveEnvVarPlaceholdersInConfig() throws IOException { + // Arrange — a secret injected via env var, e.g. the Iceberg maintenance lock password + Path configFile = tempDir.resolve("config.yaml"); + Files.writeString( + configFile, + "flink-maintenance.lock.jdbc.password: ${ICEBERG_LOCK_PASSWORD}\ndummy.key: literal"); + + var resolver = + EnvVarResolver.builder().envVars(Map.of("ICEBERG_LOCK_PASSWORD", "s3cr3t")).build(); + var cliRunner = + new CliRunner( + RuntimeExecutionMode.STREAMING, resolver, null, null, tempDir.toString(), null); + + // Act + var finalConf = cliRunner.initConfiguration(); + + // Assert + assertThat(finalConf.toMap()) + .containsEntry("flink-maintenance.lock.jdbc.password", "s3cr3t") + .containsEntry("dummy.key", "literal"); + } } From 54a8bea7bdb8b1c769eca1dd6732a36bf5995722 Mon Sep 17 00:00:00 2001 From: Marvin Froeder Date: Tue, 21 Jul 2026 17:02:56 -0300 Subject: [PATCH 2/2] Return new Configuration and drop noise in resolveEnvVars Signed-off-by: Marvin Froeder --- .../com/datasqrl/flinkrunner/BaseRunner.java | 17 ++++++----------- 1 file changed, 6 insertions(+), 11 deletions(-) diff --git a/flink-sql-runner/src/main/java/com/datasqrl/flinkrunner/BaseRunner.java b/flink-sql-runner/src/main/java/com/datasqrl/flinkrunner/BaseRunner.java index bda242f2..f08553d4 100644 --- a/flink-sql-runner/src/main/java/com/datasqrl/flinkrunner/BaseRunner.java +++ b/flink-sql-runner/src/main/java/com/datasqrl/flinkrunner/BaseRunner.java @@ -91,8 +91,7 @@ Configuration initConfiguration() { var conf = new Configuration(); if (StringUtils.isNotBlank(configDir)) { log.info("Loading Flink configuration from '{}'", configDir); - conf = GlobalConfiguration.loadConfiguration(configDir); - resolveEnvVars(conf); + conf = resolveEnvVars(GlobalConfiguration.loadConfiguration(configDir)); } // Do not overwrite runtime given in YAML @@ -103,15 +102,10 @@ Configuration initConfiguration() { return conf; } - /** - * Resolves {@code ${VAR}} placeholders in Flink configuration values from the environment, the - * same mechanism already applied to the SQL script and compiled plan. Lets secrets (e.g. the - * Iceberg maintenance lock JDBC credentials) be injected via env vars instead of landing in the - * plaintext config.yaml ConfigMap. {@link Configuration#toMap()} returns a copy, so mutating - * {@code conf} while iterating it is safe. - */ - private void resolveEnvVars(Configuration conf) { - conf.toMap() + private Configuration resolveEnvVars(Configuration source) { + var conf = new Configuration(source); + source + .toMap() .forEach( (key, value) -> { var resolved = resolver.resolve(value); @@ -119,6 +113,7 @@ private void resolveEnvVars(Configuration conf) { conf.setString(key, resolved); } }); + return conf; } static String readTextFile(String path) throws IOException {