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

Author: tomncooper

docs/content/docs/custom-resource/reference.md

@@ -117,6 +117,8 @@ This serves as a full reference for FlinkDeployment and FlinkSessionJob custom r
 | v1_18 |  |
 | v1_19 |  |
 | v1_20 |  |
+| majorVersion | int | The major integer from the Flink semver. For example for Flink 1.18.1 this would be 1. |
+| minorVersion | int | The minor integer from the Flink semver. For example for Flink 1.18.1 this would be 18. |
 
 ### IngressSpec
 **Class**: org.apache.flink.kubernetes.operator.api.spec.IngressSpec

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,54 @@
+/*
+ * 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.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,21 @@
 import org.slf4j.LoggerFactory;
 
 import java.time.Duration;
+import java.util.ArrayList;
+import java.util.HashMap;
+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;
@@ -80,6 +87,14 @@ public class FlinkConfigManager {
     private final AtomicLong defaultConfigVersion = new AtomicLong(0);
     private final LoadingCache<Key, Configuration> cache;
     private final Consumer<Set<String>> namespaceListener;
+    private volatile Map<FlinkVersion, List<String>> relevantFlinkVersionPrefixes;
+
+    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) {
@@ -99,6 +114,7 @@ public FlinkConfigManager(
         this.namespaceListener = namespaceListener;
         Duration cacheTimeout =
                 defaultConfig.get(KubernetesOperatorConfigOptions.OPERATOR_CONFIG_CACHE_TIMEOUT);
+        this.relevantFlinkVersionPrefixes = new HashMap<>();
         this.cache =
                 CacheBuilder.newBuilder()
                         .maximumSize(
@@ -169,6 +185,11 @@ public void updateDefaultConfig(Configuration newConf) {
         // We do not invalidate the cache to avoid deleting currently used temp files,
         // simply bump the version
         this.defaultConfigVersion.incrementAndGet();
+
+        // We clear the cached relevant Flink version prefixes as the base config may include new
+        // version overrides.
+        // This will trigger a regeneration of the prefixes in the next call to getDefaultConfig.
+        relevantFlinkVersionPrefixes = new HashMap<>();
     }
 
     /**
@@ -202,6 +223,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 +300,20 @@ public Configuration getDefaultConfig(String namespace, FlinkVersion flinkVersio
         }
 
         if (flinkVersion != null) {
-            applyDefault(VERSION_CONF_PREFIX + flinkVersion + ".", baseConfMap, conf);
+            // Fetch or create a list of Flink version configs that apply to this current
+            // FlinkVersion. That will include all versions that are equal to or lower than
+            // the current one that are suffixed by a `+`
+            List<String> versionPrefixes =
+                    relevantFlinkVersionPrefixes.computeIfAbsent(
+                            flinkVersion,
+                            fv -> 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";