From f444457f664aba2abd954512e28601667f4b02e4 Mon Sep 17 00:00:00 2001 From: Ferenc Csaky Date: Mon, 8 Jun 2026 13:04:15 +0200 Subject: [PATCH] feat: Make `EnvVarResolver` extendable, and its pattern detection more strict --- README.md | 4 +- .../datasqrl/flinkrunner/utils/EnvUtils.java | 15 ++- .../flinkrunner/utils/EnvVarResolver.java | 113 +++--------------- .../flinkrunner/utils/EnvVarResolverTest.java | 99 ++++++++++++--- .../com/datasqrl/flinkrunner/CliRunner.java | 9 +- 5 files changed, 125 insertions(+), 115 deletions(-) diff --git a/README.md b/README.md index 38f7525c..7ae49e01 100644 --- a/README.md +++ b/README.md @@ -143,7 +143,9 @@ helm install sql-example -f .yaml ### Environment Variable Substitution -Flink SQL Runner automatically substitutes environment variables in your configuration files, SQL scripts, and compiled plans for secrets and environment specific configuration. Environment variables must be of the form `${ENV_VARIABLE}` and inside of strings. +Flink SQL Runner automatically substitutes environment variables in your configuration files, SQL scripts, and compiled plans for secrets and environment specific configuration. +Environment variables must be of the form `${ENV_VARIABLE}` and inside of strings. +They also support bash-style defaults using `${ENV_VARIABLE:-default}` or `${ENV_VARIABLE:=default}`. For example, `${DATA_PATH}` is an environment variable inside the connector configuration of a table that is substituted at runtime: ```sql diff --git a/env-utils/src/main/java/com/datasqrl/flinkrunner/utils/EnvUtils.java b/env-utils/src/main/java/com/datasqrl/flinkrunner/utils/EnvUtils.java index e0c3f089..a2149540 100644 --- a/env-utils/src/main/java/com/datasqrl/flinkrunner/utils/EnvUtils.java +++ b/env-utils/src/main/java/com/datasqrl/flinkrunner/utils/EnvUtils.java @@ -38,7 +38,20 @@ public final class EnvUtils { * @return an immutable map containing all environment variables with defaults applied */ public static Map getEnvWithDeploymentDefaults() { - var env = new HashMap<>(System.getenv()); + return addDeploymentDefaults(System.getenv()); + } + + /** + * Returns a copy of the supplied environment variables with deployment-specific defaults added. + * + *

Deployment defaults are only added when the supplied map does not already contain those + * keys. Existing values are preserved. + * + * @param envVars environment variables to augment with deployment defaults + * @return an immutable map containing the supplied variables plus any missing deployment defaults + */ + public static Map addDeploymentDefaults(Map envVars) { + var env = new HashMap<>(envVars); getDeploymentDefaults().forEach(env::putIfAbsent); return Map.copyOf(env); diff --git a/env-utils/src/main/java/com/datasqrl/flinkrunner/utils/EnvVarResolver.java b/env-utils/src/main/java/com/datasqrl/flinkrunner/utils/EnvVarResolver.java index c62db983..1f713832 100644 --- a/env-utils/src/main/java/com/datasqrl/flinkrunner/utils/EnvVarResolver.java +++ b/env-utils/src/main/java/com/datasqrl/flinkrunner/utils/EnvVarResolver.java @@ -21,11 +21,13 @@ import com.fasterxml.jackson.databind.ObjectMapper; import com.fasterxml.jackson.databind.module.SimpleModule; import java.io.IOException; -import java.util.HashMap; import java.util.HashSet; import java.util.Map; +import java.util.Set; import java.util.regex.Matcher; import java.util.regex.Pattern; +import lombok.Builder; +import lombok.experimental.SuperBuilder; import lombok.extern.slf4j.Slf4j; /** @@ -36,106 +38,15 @@ * non-strict mode. */ @Slf4j +@SuperBuilder public class EnvVarResolver { - private static final Pattern ENVIRONMENT_VARIABLE_PATTERN = Pattern.compile("\\$\\{(.*?)\\}"); + private static final Pattern ENVIRONMENT_VARIABLE_PATTERN = + Pattern.compile("\\$\\{(?!\\{)(.*?)\\}"); - private final Map envVars; - private final ObjectMapper objectMapper; - private final boolean strict; - - private EnvVarResolver(Map envVars, boolean strict) { - this.envVars = envVars; - this.strict = strict; - objectMapper = initObjectMapper(); - } - - /** - * Creates a strict resolver backed by the current process environment. - * - * @return a resolver backed by {@link System#getenv()} - */ - public static EnvVarResolver of() { - return of(true); - } - - /** - * Creates a resolver backed by the current process environment. - * - * @param strict whether missing variables should fail resolution - * @return a resolver backed by {@link System#getenv()} - */ - public static EnvVarResolver of(boolean strict) { - return of(System.getenv(), strict); - } - - /** - * Creates a strict resolver backed by the supplied environment variables. - * - * @param envVars environment variables used for placeholder resolution - * @return a resolver backed by the supplied environment variables - */ - public static EnvVarResolver of(Map envVars) { - return of(envVars, true); - } - - /** - * Creates a resolver backed by the supplied environment variables. - * - * @param envVars environment variables used for placeholder resolution - * @param strict whether missing variables should fail resolution - * @return a resolver backed by the supplied environment variables - */ - public static EnvVarResolver of(Map envVars, boolean strict) { - return new EnvVarResolver(envVars, strict); - } - - /** - * Creates a strict resolver backed by the current process environment plus deployment defaults. - * - * @return a resolver with deployment defaults applied - */ - public static EnvVarResolver withDeploymentDefaults() { - return withDeploymentDefaults(true); - } - - /** - * Creates a resolver backed by the current process environment plus deployment defaults. - * - * @param strict whether missing variables should fail resolution - * @return a resolver with deployment defaults applied - */ - public static EnvVarResolver withDeploymentDefaults(boolean strict) { - return new EnvVarResolver(EnvUtils.getEnvWithDeploymentDefaults(), strict); - } - - /** - * Creates a strict resolver backed by the supplied environment variables plus deployment - * defaults. - * - * @param envVars environment variables used for placeholder resolution - * @return a resolver with deployment defaults applied - */ - public static EnvVarResolver withDeploymentDefaults(Map envVars) { - return withDeploymentDefaults(envVars, true); - } - - /** - * Creates a resolver backed by the supplied environment variables plus deployment defaults. - * - *

Deployment defaults are only added when the supplied map does not already contain those - * keys. - * - * @param envVars environment variables used for placeholder resolution - * @param strict whether missing variables should fail resolution - * @return a resolver with deployment defaults applied - */ - public static EnvVarResolver withDeploymentDefaults(Map envVars, boolean strict) { - var modifiedEnvVars = new HashMap<>(envVars); - EnvUtils.getDeploymentDefaults().forEach(modifiedEnvVars::putIfAbsent); - - return new EnvVarResolver(Map.copyOf(modifiedEnvVars), strict); - } + @Builder.Default private final Map envVars = System.getenv(); + @Builder.Default private final boolean strict = true; + @Builder.Default private final Set exclusions = Set.of(); /** * Resolves environment variables referenced in a given source string. Searches for environment @@ -174,6 +85,11 @@ public String resolve(String src) { key = rawKey; } + // If excluded, don't do anything + if (exclusions.contains(key)) { + continue; + } + if (envVars.containsKey(key)) { var envValue = envVars.get(key); matcher.appendReplacement(res, Matcher.quoteReplacement(envValue)); @@ -205,6 +121,7 @@ public String resolve(String src) { * @throws IOException if the JSON processing fails in any way */ public String resolveInJson(String jsonSrc) throws IOException { + var objectMapper = initObjectMapper(); var res = objectMapper.readValue(jsonSrc, Map.class); return objectMapper.writeValueAsString(res); diff --git a/env-utils/src/test/java/com/datasqrl/flinkrunner/utils/EnvVarResolverTest.java b/env-utils/src/test/java/com/datasqrl/flinkrunner/utils/EnvVarResolverTest.java index e155d724..e09fe44c 100644 --- a/env-utils/src/test/java/com/datasqrl/flinkrunner/utils/EnvVarResolverTest.java +++ b/env-utils/src/test/java/com/datasqrl/flinkrunner/utils/EnvVarResolverTest.java @@ -20,7 +20,10 @@ import com.fasterxml.jackson.databind.ObjectMapper; import java.io.IOException; +import java.util.HashMap; import java.util.Map; +import java.util.Set; +import lombok.experimental.SuperBuilder; import org.junit.jupiter.api.Test; import org.junit.jupiter.params.ParameterizedTest; import org.junit.jupiter.params.provider.CsvSource; @@ -43,7 +46,7 @@ void givenEnvVariables_whenReplaceWithEnv_thenReplaceCorrectlyVars( Map.of( "USER", "John", "PATH", "/usr/bin"); - resolver = EnvVarResolver.of(envVariables); + resolver = EnvVarResolver.builder().envVars(envVariables).build(); var result = resolver.resolve(command); assertThat(result).isEqualTo(expected); } @@ -57,7 +60,7 @@ void givenEnvVariables_whenReplaceWithEnv_thenReplaceCorrectlyVars( }) void givenMissingEnvVariables_whenResolveEnv_Vars_thenThrowException(String command) { Map envVariables = Map.of(); - resolver = EnvVarResolver.of(envVariables); // Empty map to simulate missing variables + resolver = EnvVarResolver.builder().envVars(envVariables).build(); assertThatThrownBy(() -> resolver.resolve(command)) .isInstanceOf(IllegalStateException.class) .hasMessageStartingWith( @@ -76,7 +79,7 @@ void givenSimilarEnvVariableNames_whenResolveEnv_Vars_thenPartialMatchDoesNotOcc Map.of( "USER", "John", "NAME", "exists"); - resolver = EnvVarResolver.of(envVariables); + resolver = EnvVarResolver.builder().envVars(envVariables).build(); assertThatThrownBy(() -> resolver.resolve(command)) .isInstanceOf(IllegalStateException.class) .hasMessage( @@ -105,7 +108,7 @@ void givenDefaultEnvValues_whenResolve_thenFallbackOrUseEnvValue( Map.of( "USER", "John", "PATH", "/usr/bin"); - resolver = EnvVarResolver.of(envVariables); + resolver = EnvVarResolver.builder().envVars(envVariables).build(); var result = resolver.resolve(command); assertThat(result).isEqualTo(expected); } @@ -120,7 +123,7 @@ void givenDefaultEnvValues_whenResolve_thenFallbackOrUseEnvValue( }) void givenMissingEnvWithoutDefault_whenResolve_thenThrowException(String command) { Map envVariables = Map.of("A", "something"); - resolver = EnvVarResolver.of(envVariables); + resolver = EnvVarResolver.builder().envVars(envVariables).build(); assertThatThrownBy(() -> resolver.resolve(command)) .isInstanceOf(IllegalStateException.class) .hasMessageContaining("referenced, but not found"); @@ -136,14 +139,44 @@ void givenMissingEnvWithoutDefault_whenResolve_thenThrowException(String command }) void givenFallbackWithColons_whenResolve_thenParseCorrectly(String command, String expected) { Map envVariables = Map.of(); // No TOKEN set - resolver = EnvVarResolver.of(envVariables); + resolver = EnvVarResolver.builder().envVars(envVariables).build(); var result = resolver.resolve(command); assertThat(result).isEqualTo(expected); } + @ParameterizedTest + @CsvSource({ + "'${{SECRET}}', '${{SECRET}}'", + "'Token=${{SECRET}}', 'Token=${{SECRET}}'", + "'Host=${HOST}, password=${{DB_PASSWORD}}', 'Host=db.example.com, password=${{DB_PASSWORD}}'" + }) + void givenSecretEnvTemplate_whenResolve_thenLeaveUnchanged(String command, String expected) { + resolver = EnvVarResolver.builder().envVars(Map.of("HOST", "db.example.com")).build(); + + var result = resolver.resolve(command); + + assertThat(result).isEqualTo(expected); + } + + @Test + void givenExcludedEnvVariables_whenResolve_thenLeavePlaceholdersUnchanged() { + resolver = + EnvVarResolver.builder() + .envVars( + Map.of( + "USER", "John", + "EXCLUDED", "ignored")) + .exclusions(Set.of("EXCLUDED", "MISSING_EXCLUDED")) + .build(); + + var result = resolver.resolve("${USER}|${EXCLUDED}|${EXCLUDED:-fallback}|${MISSING_EXCLUDED}"); + + assertThat(result).isEqualTo("John|${EXCLUDED}|${EXCLUDED:-fallback}|${MISSING_EXCLUDED}"); + } + @Test void givenNullOrBlankSource_whenResolve_thenReturnSource() { - resolver = EnvVarResolver.of(Map.of()); + resolver = EnvVarResolver.builder().envVars(Map.of()).build(); assertThat(resolver.resolve(null)).isNull(); assertThat(resolver.resolve(" ")).isEqualTo(" "); @@ -151,7 +184,7 @@ void givenNullOrBlankSource_whenResolve_thenReturnSource() { @Test void givenNonStrictResolver_whenMissingEnvVariables_thenLeavesPlaceholdersUnresolved() { - resolver = EnvVarResolver.of(Map.of("USER", "John"), false); + resolver = EnvVarResolver.builder().envVars(Map.of("USER", "John")).strict(false).build(); var result = resolver.resolve("Hello ${USER}, ${MISSING}!"); @@ -161,10 +194,13 @@ void givenNonStrictResolver_whenMissingEnvVariables_thenLeavesPlaceholdersUnreso @Test void givenDeploymentDefaults_whenResolve_thenUseDefaultsAndSuppliedValues() { resolver = - EnvVarResolver.withDeploymentDefaults( - Map.of( - "DEPLOYMENT_ID", "deployment-1", - "USER", "John")); + EnvVarResolver.builder() + .envVars( + withDeploymentDefaults( + Map.of( + "DEPLOYMENT_ID", "deployment-1", + "USER", "John"))) + .build(); var result = resolver.resolve("${DEPLOYMENT_ID}|${DEPLOYMENT_TIMESTAMP}|${USER}"); var parts = result.split("\\|"); @@ -178,7 +214,10 @@ void givenDeploymentDefaults_whenResolve_thenUseDefaultsAndSuppliedValues() { @Test void givenNonStrictDeploymentDefaults_whenMissingNonDefaultEnvVariable_thenLeavesPlaceholder() { resolver = - EnvVarResolver.withDeploymentDefaults(Map.of("DEPLOYMENT_ID", "deployment-1"), false); + EnvVarResolver.builder() + .envVars(withDeploymentDefaults(Map.of("DEPLOYMENT_ID", "deployment-1"))) + .strict(false) + .build(); var result = resolver.resolve("${DEPLOYMENT_ID}|${MISSING}"); @@ -187,7 +226,7 @@ void givenNonStrictDeploymentDefaults_whenMissingNonDefaultEnvVariable_thenLeave @Test void givenJsonSource_whenResolveInJson_thenResolveStringLeafNodes() throws IOException { - resolver = EnvVarResolver.of(Map.of("USER", "John")); + resolver = EnvVarResolver.builder().envVars(Map.of("USER", "John")).build(); var result = resolver.resolveInJson( @@ -198,4 +237,36 @@ void givenJsonSource_whenResolveInJson_thenResolveStringLeafNodes() throws IOExc assertThat(json.get("count").asInt()).isEqualTo(1); assertThat(json.get("nested").get("path").asText()).isEqualTo("/tmp"); } + + @Test + void givenSubclass_whenBuild_thenInheritedBuilderFieldsAreApplied() { + resolver = + TestEnvVarResolver.builder() + .envVars(Map.of("USER", "John")) + .strict(false) + .prefix("resolved=") + .build(); + + var result = resolver.resolve("${USER}:${MISSING}"); + + assertThat(result).isEqualTo("resolved=John:${MISSING}"); + } + + private Map withDeploymentDefaults(Map envVars) { + var modifiedEnvVars = new HashMap<>(envVars); + EnvUtils.getDeploymentDefaults().forEach(modifiedEnvVars::putIfAbsent); + + return Map.copyOf(modifiedEnvVars); + } + + @SuperBuilder + private static class TestEnvVarResolver extends EnvVarResolver { + + private final String prefix; + + @Override + public String resolve(String src) { + return prefix + super.resolve(src); + } + } } diff --git a/flink-sql-runner/src/main/java/com/datasqrl/flinkrunner/CliRunner.java b/flink-sql-runner/src/main/java/com/datasqrl/flinkrunner/CliRunner.java index 3ff76ac1..4a6977fd 100644 --- a/flink-sql-runner/src/main/java/com/datasqrl/flinkrunner/CliRunner.java +++ b/flink-sql-runner/src/main/java/com/datasqrl/flinkrunner/CliRunner.java @@ -15,6 +15,7 @@ */ package com.datasqrl.flinkrunner; +import com.datasqrl.flinkrunner.utils.EnvUtils; import com.datasqrl.flinkrunner.utils.EnvVarResolver; import java.util.Arrays; import java.util.concurrent.Callable; @@ -76,7 +77,13 @@ public CliRunner( @Nullable String planFile, @Nullable String configDir, @Nullable String udfPath) { - this(mode, EnvVarResolver.withDeploymentDefaults(), sqlFile, planFile, configDir, udfPath); + this( + mode, + EnvVarResolver.builder().envVars(EnvUtils.getEnvWithDeploymentDefaults()).build(), + sqlFile, + planFile, + configDir, + udfPath); } @VisibleForTesting