[FLINK-36529] Allow flink version configs to be set to greater than given version

Signed-off-by: Thomas Cooper <[email protected]>

Author: tomncooper

docs/content/docs/operations/configuration.md

@@ -52,21 +52,43 @@ To learn more about metrics and logging configuration please refer to the dedica
 
 ### Flink Version and Namespace specific defaults
 
-The operator also supports default configuration overrides for selected Flink versions and namespaces. This can be important if some behaviour changed across Flink versions or we want to treat certain namespaces differently (such as reconcile it more or less frequently etc).
+The operator also supports default configuration overrides for selected Flink versions and namespaces. This can be important if some behaviour changed across Flink versions, or we want to treat certain namespaces differently (such as reconcile it more or less frequently etc.):
 
 ```
-# Flink Version specific defaults 
-kubernetes.operator.default-configuration.flink-version.v1_17.k1: v1
-kubernetes.operator.default-configuration.flink-version.v1_17.k2: v2
-kubernetes.operator.default-configuration.flink-version.v1_17.k3: v3
-
 # Namespace specific defaults
 kubernetes.operator.default-configuration.namespace.ns1.k1: v1
 kubernetes.operator.default-configuration.namespace.ns1.k2: v2
 kubernetes.operator.default-configuration.namespace.ns2.k1: v1
 ```
 
-Flink version specific defaults will have a higher precedence so namespace defaults would be overridden by the same key.
+Flink version specific defaults have a higher precedence, so namespace defaults will be overridden by version defaults with the same key.
+
+Flink version defaults can also be suffixed by a `+` character after the version string. This indicates that the default applies to this Flink version and any higher version.
+
+For example, taking the configuration below:
+```
+# Flink Version specific defaults 
+kubernetes.operator.default-configuration.flink-version.v1_16+.k4: v4
+kubernetes.operator.default-configuration.flink-version.v1_16+.k5: v5
+kubernetes.operator.default-configuration.flink-version.v1_17.k1: v1
+kubernetes.operator.default-configuration.flink-version.v1_17.k2: v2
+kubernetes.operator.default-configuration.flink-version.v1_17.k3: v3
+kubernetes.operator.default-configuration.flink-version.v1_17.k5: v5.1
+```
+This would result in the defaults for Flink 1.17 being:
+```
+k1: v1
+k2: v2
+k3: v3
+k4: v4
+k5: v5.1
+```
+
+**Note**: The configuration above sets `k5: v5` for all versions >= 1.16. 
+However, this is overridden for Flink 1.17 to `v5.1`. 
+But if you ran a Flink 1.18 deployment with this configuration, then the value of `k5` would be `v5` not `v5.1`. The `k5` override only applies to Flink 1.17. 
+Adding a `+` to the Flink 1.17 `k5` default would apply the new value to all future versions.
+
 
 ## Dynamic Operator Configuration
 

flink-kubernetes-operator-api/src/main/java/org/apache/flink/kubernetes/operator/api/spec/FlinkVersion.java

@@ -25,21 +25,38 @@
 public enum FlinkVersion {
     /** No longer supported since 1.7 operator release. */
     @Deprecated
-    v1_13,
+    v1_13(1, 13),
     /** No longer supported since 1.7 operator release. */
     @Deprecated
-    v1_14,
+    v1_14(1, 14),
     /** Deprecated since 1.10 operator release. */
     @Deprecated
-    v1_15,
-    v1_16,
-    v1_17,
-    v1_18,
-    v1_19,
-    v1_20;
+    v1_15(1, 15),
+    v1_16(1, 16),
+    v1_17(1, 17),
+    v1_18(1, 18),
+    v1_19(1, 19),
+    v1_20(1, 20);
+
+    /** The major integer from the Flink semver. For example for Flink 1.18.1 this would be 1. */
+    private final int majorVersion;
+
+    /** The minor integer from the Flink semver. For example for Flink 1.18.1 this would be 18. */
+    private final int minorVersion;
+
+    FlinkVersion(int major, int minor) {
+        this.majorVersion = major;
+        this.minorVersion = minor;
+    }
 
     public boolean isEqualOrNewer(FlinkVersion otherVersion) {
-        return this.ordinal() >= otherVersion.ordinal();
+        if (this.majorVersion > otherVersion.majorVersion) {
+            return true;
+        }
+        if (this.majorVersion == otherVersion.majorVersion) {
+            return this.minorVersion >= otherVersion.minorVersion;
+        }
+        return false;
     }
 
     /**
@@ -54,4 +71,22 @@ public static FlinkVersion current() {
     public static boolean isSupported(FlinkVersion version) {
         return version != null && version.isEqualOrNewer(FlinkVersion.v1_15);
     }
+
+    /**
+     * Returns the FlinkVersion associated with the supplied major and minor version integers.
+     *
+     * @param major The major part of the Flink version (e.g. 1 for 1.18.1).
+     * @param minor The minor part of the Flink version (e.g. 18 for 1.18.1).
+     * @throws IllegalArgumentException If the supplied major and minor version do not correspond to
+     *     a supported FlinkVersion.
+     * @return The FlinkVersion associated with the supplied major and minor version integers.
+     */
+    public static FlinkVersion fromMajorMinor(int major, int minor) {
+        for (FlinkVersion version : values()) {
+            if (version.majorVersion == major && version.minorVersion == minor) {
+                return version;
+            }
+        }
+        throw new IllegalArgumentException("Unknown Flink version: " + major + "." + minor);
+    }
 }

flink-kubernetes-operator-api/src/test/java/org/apache/flink/kubernetes/operator/api/spec/FlinkVersionTest.java

@@ -0,0 +1,37 @@
+package org.apache.flink.kubernetes.operator.api.spec;
+
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+class FlinkVersionTest {
+
+    @Test
+    void isEqualOrNewer() {
+        assertFalse(FlinkVersion.v1_16.isEqualOrNewer(FlinkVersion.v1_17));
+        assertTrue(FlinkVersion.v1_17.isEqualOrNewer(FlinkVersion.v1_17));
+        assertTrue(FlinkVersion.v1_18.isEqualOrNewer(FlinkVersion.v1_17));
+    }
+
+    @Test
+    void current() {
+        assertEquals(FlinkVersion.v1_20, FlinkVersion.current());
+    }
+
+    @Test
+    void isSupported() {
+        assertTrue(FlinkVersion.isSupported(FlinkVersion.v1_20));
+    }
+
+    @Test
+    void fromMajorMinor() {
+        assertEquals(FlinkVersion.fromMajorMinor(1, 17), FlinkVersion.v1_17);
+        assertEquals(FlinkVersion.fromMajorMinor(1, 18), FlinkVersion.v1_18);
+        assertEquals(FlinkVersion.fromMajorMinor(1, 19), FlinkVersion.v1_19);
+        assertEquals(FlinkVersion.fromMajorMinor(1, 20), FlinkVersion.v1_20);
+        assertThrows(IllegalArgumentException.class, () -> FlinkVersion.fromMajorMinor(0, 1));
+    }
+}

flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/config/FlinkConfigManager.java

@@ -49,14 +49,20 @@
 import org.slf4j.LoggerFactory;
 
 import java.time.Duration;
+import java.util.ArrayList;
+import java.util.List;
 import java.util.Map;
 import java.util.Optional;
 import java.util.Set;
+import java.util.SortedMap;
+import java.util.TreeMap;
 import java.util.concurrent.Executors;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicLong;
 import java.util.function.Consumer;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
 
 import static org.apache.flink.configuration.CheckpointingOptions.SAVEPOINT_DIRECTORY;
 import static org.apache.flink.kubernetes.operator.config.FlinkConfigBuilder.applyJobConfig;
@@ -81,6 +87,13 @@ public class FlinkConfigManager {
     private final LoadingCache<Key, Configuration> cache;
     private final Consumer<Set<String>> namespaceListener;
 
+    protected static final Pattern FLINK_VERSION_PATTERN =
+            Pattern.compile(
+                    VERSION_CONF_PREFIX.replaceAll("\\.", "\\\\\\.")
+                            + "v(?<major>\\d+)_(?<minor>\\d+)(?<gt>\\"
+                            + KubernetesOperatorConfigOptions.FLINK_VERSION_GREATER_THAN_SUFFIX
+                            + ")?\\..*");
+
     @VisibleForTesting
     public FlinkConfigManager(Configuration defaultConfig) {
         this(defaultConfig, ns -> {}, true);
@@ -202,6 +215,65 @@ public FlinkOperatorConfiguration getOperatorConfiguration(
                 getDefaultConfig(namespace, flinkVersion));
     }
 
+    /**
+     * This method will search the keys of the supplied map and find any that contain a flink
+     * version string that is relevant to the supplied flink version.
+     *
+     * <p>Relevance is defined as any key with the {@link
+     * KubernetesOperatorConfigOptions#VERSION_CONF_PREFIX} followed by either the supplied flink
+     * version (with or without the {@link
+     * KubernetesOperatorConfigOptions#FLINK_VERSION_GREATER_THAN_SUFFIX}) or a lower flink version
+     * string followed by the {@link
+     * KubernetesOperatorConfigOptions#FLINK_VERSION_GREATER_THAN_SUFFIX}.
+     *
+     * <p>Prefixes are returned in ascending order of flink version.
+     *
+     * @param baseConfMap The configuration map that should be searched for relevant Flink version
+     *     prefixes.
+     * @return A list of relevant Flink version prefixes in order of ascending Flink version.
+     */
+    protected static List<String> getRelevantVersionPrefixes(
+            Map<String, String> baseConfMap, FlinkVersion flinkVersion) {
+        SortedMap<FlinkVersion, String> greaterThanVersionPrefixes = new TreeMap<>();
+
+        for (Map.Entry<String, String> entry : baseConfMap.entrySet()) {
+            Matcher versionMatcher = FLINK_VERSION_PATTERN.matcher(entry.getKey());
+            if (versionMatcher.matches() && versionMatcher.group("gt") != null) {
+                try {
+                    FlinkVersion keyFlinkVersion =
+                            FlinkVersion.fromMajorMinor(
+                                    Integer.parseInt(versionMatcher.group("major")),
+                                    Integer.parseInt(versionMatcher.group("minor")));
+                    if (flinkVersion.isEqualOrNewer(keyFlinkVersion)) {
+                        greaterThanVersionPrefixes.put(
+                                keyFlinkVersion,
+                                VERSION_CONF_PREFIX
+                                        + keyFlinkVersion
+                                        + KubernetesOperatorConfigOptions
+                                                .FLINK_VERSION_GREATER_THAN_SUFFIX
+                                        + ".");
+                    }
+                } catch (NumberFormatException numberFormatException) {
+                    LOG.warn("Unable to parse version number in config key: {}", entry.getKey());
+                } catch (IllegalArgumentException illegalArgumentException) {
+                    LOG.warn("Unknown Flink version in config key: {}", entry.getKey());
+                }
+            }
+        }
+
+        // Extract the prefixes from the sorted map, these will be ascending Flink version order
+        List<String> sortedRelevantVersionPrefixes =
+                new ArrayList<>(greaterThanVersionPrefixes.values());
+
+        // Add the current flink version prefix (without the greater than symbol) to the set.
+        // Any current flink version prefix with the greater than symbol would already have been
+        // added
+        // in the loop above.
+        sortedRelevantVersionPrefixes.add(VERSION_CONF_PREFIX + flinkVersion + ".");
+
+        return sortedRelevantVersionPrefixes;
+    }
+
     /**
      * Get the base configuration for the given namespace and flink version combination. This is
      * different from the platform level base config as it may contain namespaces or version
@@ -220,7 +292,17 @@ public Configuration getDefaultConfig(String namespace, FlinkVersion flinkVersio
         }
 
         if (flinkVersion != null) {
-            applyDefault(VERSION_CONF_PREFIX + flinkVersion + ".", baseConfMap, conf);
+            // Get a list of flink version configs that apply to this current flink version
+            // That will include all versions that are equal to or lower than the current one
+            // that are suffixed by a `+`
+            List<String> versionPrefixes = getRelevantVersionPrefixes(baseConfMap, flinkVersion);
+
+            // The version prefixes are returned in ascending order of Flink version, so configs
+            // attached to newer versions will override older ones. For example v1_16+.conf1 will
+            // be overridden if a key containing v1_18+.conf1 is present.
+            for (String versionPrefix : versionPrefixes) {
+                applyDefault(versionPrefix, baseConfMap, conf);
+            }
         }
 
         return conf;

flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/config/KubernetesOperatorConfigOptions.java

@@ -40,6 +40,7 @@ public class KubernetesOperatorConfigOptions {
 
     private static final String DEFAULT_CONF_PREFIX = K8S_OP_CONF_PREFIX + "default-configuration.";
     public static final String VERSION_CONF_PREFIX = DEFAULT_CONF_PREFIX + "flink-version.";
+    public static final String FLINK_VERSION_GREATER_THAN_SUFFIX = "+";
     public static final String NAMESPACE_CONF_PREFIX = DEFAULT_CONF_PREFIX + "namespace.";
     public static final String SECTION_SYSTEM = "system";
     public static final String SECTION_ADVANCED = "system_advanced";