feat: Make EnvVarResolver extendable, add more config options (#352)

Author: ferenc-csaky

README.md

@@ -143,7 +143,9 @@ helm install sql-example -f <your-helm-values>.yaml <your-helm-chart>
 
 ### 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

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<String, String> getEnvWithDeploymentDefaults() {
-    var env = new HashMap<>(System.getenv());
+    return addDeploymentDefaults(System.getenv());
+  }
+
+  /**
+   * Returns a copy of the supplied environment variables with deployment-specific defaults added.
+   *
+   * <p>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<String, String> addDeploymentDefaults(Map<String, String> envVars) {
+    var env = new HashMap<>(envVars);
     getDeploymentDefaults().forEach(env::putIfAbsent);
 
     return Map.copyOf(env);

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<String, String> envVars;
-  private final ObjectMapper objectMapper;
-  private final boolean strict;
-
-  private EnvVarResolver(Map<String, String> 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<String, String> 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<String, String> 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<String, String> envVars) {
-    return withDeploymentDefaults(envVars, true);
-  }
-
-  /**
-   * Creates a resolver backed by the supplied environment variables plus deployment defaults.
-   *
-   * <p>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<String, String> 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<String, String> envVars = System.getenv();
+  @Builder.Default private final boolean strict = true;
+  @Builder.Default private final Set<String> 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);

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<String, String> 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<String, String> 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,22 +139,52 @@ void givenMissingEnvWithoutDefault_whenResolve_thenThrowException(String command
   })
   void givenFallbackWithColons_whenResolve_thenParseCorrectly(String command, String expected) {
     Map<String, String> 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("   ");
   }
 
   @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<String, String> withDeploymentDefaults(Map<String, String> 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);
+    }
+  }
 }

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