[FLINK-37405] Validate config prefixes for Flink 2.0

Author: gyfora

flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/kubeclient/decorators/FlinkConfMountDecorator.java

@@ -22,6 +22,7 @@
 import org.apache.flink.configuration.Configuration;
 import org.apache.flink.configuration.ConfigurationUtils;
 import org.apache.flink.configuration.DeploymentOptionsInternal;
+import org.apache.flink.configuration.IllegalConfigurationException;
 import org.apache.flink.configuration.JobManagerOptions;
 import org.apache.flink.configuration.RestOptions;
 import org.apache.flink.configuration.TaskManagerOptions;
@@ -166,9 +167,32 @@ private List<String> getClusterSideConfData(Configuration flinkConfig) {
         clusterSideConfig.removeConfig(TaskManagerOptions.BIND_HOST);
         clusterSideConfig.removeConfig(TaskManagerOptions.HOST);
 
+        validateConfigKeysForV2(clusterSideConfig);
+
         return ConfigurationUtils.convertConfigToWritableLines(clusterSideConfig, false);
     }
 
+    private void validateConfigKeysForV2(Configuration clusterSideConfig) {
+
+        // Only validate Flink 2.0 yaml configs
+        if (!useStandardYamlConfig()) {
+            return;
+        }
+
+        var keys = clusterSideConfig.keySet();
+
+        for (var key1 : keys) {
+            for (var key2 : keys) {
+                if (key2.startsWith(key1 + ".")) {
+                    throw new IllegalConfigurationException(
+                            String.format(
+                                    "Overlapping key prefixes detected (%s -> %s), please replace with Flink v2 compatible, non-deprecated keys.",
+                                    key1, key2));
+                }
+            }
+        }
+    }
+
     @VisibleForTesting
     String getFlinkConfData(List<String> confData) throws IOException {
         try (StringWriter sw = new StringWriter();

flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/kubeclient/decorators/FlinkConfMountDecoratorTest.java

@@ -20,6 +20,7 @@
 
 import org.apache.flink.client.deployment.ClusterSpecification;
 import org.apache.flink.configuration.Configuration;
+import org.apache.flink.configuration.IllegalConfigurationException;
 import org.apache.flink.configuration.YamlParserUtils;
 import org.apache.flink.kubernetes.configuration.KubernetesConfigOptions;
 import org.apache.flink.kubernetes.kubeclient.parameters.KubernetesJobManagerParameters;
@@ -32,6 +33,7 @@
 import org.junit.jupiter.params.ParameterizedTest;
 import org.junit.jupiter.params.provider.Arguments;
 import org.junit.jupiter.params.provider.MethodSource;
+import org.junit.jupiter.params.provider.ValueSource;
 
 import java.io.IOException;
 import java.util.HashMap;
@@ -95,6 +97,38 @@ void testConfigMap(FlinkVersion version, String expectedConfName, boolean standa
         }
     }
 
+    @ParameterizedTest
+    @ValueSource(booleans = {true, false})
+    void testOverlappingKeyDetection(boolean standardYaml) throws IOException {
+
+        flinkConfig.set(
+                FlinkConfigBuilder.FLINK_VERSION,
+                standardYaml ? FlinkVersion.v2_0 : FlinkVersion.v1_20);
+        flinkConfig.setString("k", "v");
+        flinkConfig.setString("kv", "v2");
+
+        // Non overlapping keys
+        flinkConfMountDecorator.buildAccompanyingKubernetesResources();
+        flinkConfig.setString("k.v", "v3");
+
+        IllegalConfigurationException err = null;
+        try {
+            var additionalResources =
+                    flinkConfMountDecorator.buildAccompanyingKubernetesResources();
+            assertThat(additionalResources).hasSize(1);
+        } catch (IllegalConfigurationException e) {
+            err = e;
+        }
+
+        if (standardYaml) {
+            assertThat(err)
+                    .isNotNull()
+                    .hasMessageContaining("Overlapping key prefixes detected (k -> k.v)");
+        } else {
+            assertThat(err).isNull();
+        }
+    }
+
     private static Stream<Arguments> testArgs() {
         return Stream.of(
                 Arguments.arguments(FlinkVersion.v1_19, "flink-conf.yaml", false),