Modify FlinkConfMountDecorator for seamless multiversion support

Author: gyfora

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

@@ -22,29 +22,28 @@
 import org.apache.flink.configuration.Configuration;
 import org.apache.flink.configuration.ConfigurationUtils;
 import org.apache.flink.configuration.DeploymentOptionsInternal;
-import org.apache.flink.configuration.GlobalConfiguration;
 import org.apache.flink.configuration.JobManagerOptions;
 import org.apache.flink.configuration.RestOptions;
 import org.apache.flink.configuration.TaskManagerOptions;
 import org.apache.flink.kubernetes.configuration.KubernetesConfigOptions;
 import org.apache.flink.kubernetes.kubeclient.FlinkPod;
 import org.apache.flink.kubernetes.kubeclient.parameters.AbstractKubernetesParameters;
+import org.apache.flink.kubernetes.operator.api.spec.FlinkVersion;
+import org.apache.flink.kubernetes.shaded.io.fabric8.kubernetes.api.model.ConfigMap;
+import org.apache.flink.kubernetes.shaded.io.fabric8.kubernetes.api.model.ConfigMapBuilder;
+import org.apache.flink.kubernetes.shaded.io.fabric8.kubernetes.api.model.Container;
+import org.apache.flink.kubernetes.shaded.io.fabric8.kubernetes.api.model.ContainerBuilder;
+import org.apache.flink.kubernetes.shaded.io.fabric8.kubernetes.api.model.HasMetadata;
+import org.apache.flink.kubernetes.shaded.io.fabric8.kubernetes.api.model.KeyToPath;
+import org.apache.flink.kubernetes.shaded.io.fabric8.kubernetes.api.model.KeyToPathBuilder;
+import org.apache.flink.kubernetes.shaded.io.fabric8.kubernetes.api.model.Pod;
+import org.apache.flink.kubernetes.shaded.io.fabric8.kubernetes.api.model.PodBuilder;
+import org.apache.flink.kubernetes.shaded.io.fabric8.kubernetes.api.model.Volume;
+import org.apache.flink.kubernetes.shaded.io.fabric8.kubernetes.api.model.VolumeBuilder;
 import org.apache.flink.kubernetes.utils.Constants;
 
 import org.apache.flink.shaded.guava31.com.google.common.io.Files;
 
-import io.fabric8.kubernetes.api.model.ConfigMap;
-import io.fabric8.kubernetes.api.model.ConfigMapBuilder;
-import io.fabric8.kubernetes.api.model.Container;
-import io.fabric8.kubernetes.api.model.ContainerBuilder;
-import io.fabric8.kubernetes.api.model.HasMetadata;
-import io.fabric8.kubernetes.api.model.KeyToPath;
-import io.fabric8.kubernetes.api.model.KeyToPathBuilder;
-import io.fabric8.kubernetes.api.model.Pod;
-import io.fabric8.kubernetes.api.model.PodBuilder;
-import io.fabric8.kubernetes.api.model.Volume;
-import io.fabric8.kubernetes.api.model.VolumeBuilder;
-
 import java.io.File;
 import java.io.IOException;
 import java.io.PrintWriter;
@@ -57,16 +56,14 @@
 import java.util.Map;
 import java.util.stream.Collectors;
 
+import static org.apache.flink.kubernetes.operator.config.FlinkConfigBuilder.FLINK_VERSION;
 import static org.apache.flink.kubernetes.utils.Constants.CONFIG_FILE_LOG4J_NAME;
 import static org.apache.flink.kubernetes.utils.Constants.CONFIG_FILE_LOGBACK_NAME;
 import static org.apache.flink.kubernetes.utils.Constants.CONFIG_MAP_PREFIX;
 import static org.apache.flink.kubernetes.utils.Constants.FLINK_CONF_VOLUME;
 import static org.apache.flink.util.Preconditions.checkNotNull;
 
-/**
- * Mounts the log4j.properties, logback.xml, and config.yaml configuration on the JobManager or
- * TaskManager pod.
- */
+/** Copied from Flink core and modified to handle all Flink versions. */
 public class FlinkConfMountDecorator extends AbstractKubernetesStepDecorator {
 
     private final AbstractKubernetesParameters kubernetesComponentConf;
@@ -105,8 +102,8 @@ private Pod decoratePod(Pod pod) {
                         .collect(Collectors.toList());
         keyToPaths.add(
                 new KeyToPathBuilder()
-                        .withKey(GlobalConfiguration.getFlinkConfFilename())
-                        .withPath(GlobalConfiguration.getFlinkConfFilename())
+                        .withKey(getFlinkConfFilename())
+                        .withPath(getFlinkConfFilename())
                         .build());
 
         final Volume flinkConfVolume =
@@ -139,7 +136,7 @@ public List<HasMetadata> buildAccompanyingKubernetesResources() throws IOExcepti
 
         final List<String> confData =
                 getClusterSideConfData(kubernetesComponentConf.getFlinkConfiguration());
-        data.put(GlobalConfiguration.getFlinkConfFilename(), getFlinkConfData(confData));
+        data.put(getFlinkConfFilename(), getFlinkConfData(confData));
 
         final ConfigMap flinkConfConfigMap =
                 new ConfigMapBuilder()
@@ -156,21 +153,26 @@ public List<HasMetadata> buildAccompanyingKubernetesResources() throws IOExcepti
 
     /** Get properties map for the cluster-side after removal of some keys. */
     private List<String> getClusterSideConfData(Configuration flinkConfig) {
-        final Configuration clusterSideConfig = flinkConfig.clone();
+        // For Flink versions that use the standard config we have to set the standardYaml flag in
+        // the Configuration object manually instead of simply cloning, otherwise it would simply
+        // inherit it from the base config (which would always be false currently).
+        Configuration clusterSideConfig = new Configuration(useStandardYamlConfig());
+        clusterSideConfig.addAll(flinkConfig);
         // Remove some configuration options that should not be taken to cluster side.
         clusterSideConfig.removeConfig(KubernetesConfigOptions.KUBE_CONFIG_FILE);
         clusterSideConfig.removeConfig(DeploymentOptionsInternal.CONF_DIR);
         clusterSideConfig.removeConfig(RestOptions.BIND_ADDRESS);
         clusterSideConfig.removeConfig(JobManagerOptions.BIND_HOST);
         clusterSideConfig.removeConfig(TaskManagerOptions.BIND_HOST);
         clusterSideConfig.removeConfig(TaskManagerOptions.HOST);
+
         return ConfigurationUtils.convertConfigToWritableLines(clusterSideConfig, false);
     }
 
     @VisibleForTesting
     String getFlinkConfData(List<String> confData) throws IOException {
         try (StringWriter sw = new StringWriter();
-             PrintWriter out = new PrintWriter(sw)) {
+                PrintWriter out = new PrintWriter(sw)) {
             confData.forEach(out::println);
 
             return sw.toString();
@@ -197,4 +199,28 @@ private List<File> getLocalLogConfFiles() {
     public static String getFlinkConfConfigMapName(String clusterId) {
         return CONFIG_MAP_PREFIX + clusterId;
     }
+
+    /**
+     * We have to override the GlobalConfiguration#getFlinkConfFilename() logic to make sure we can
+     * separate operator level (Global) and Flink deployment specific config handling.
+     *
+     * @return conf file name
+     */
+    public String getFlinkConfFilename() {
+        return useStandardYamlConfig() ? "config.yaml" : "flink-conf.yaml";
+    }
+
+    /**
+     * Determine based on the Flink Version if we should use the new standard config.yaml vs the old
+     * flink-conf.yaml. While technically 1.19+ could use this we don't want to change the behaviour
+     * for already released Flink versions, so only switch to new yaml from Flink 2.0 onwards.
+     *
+     * @return True for Flink version 2.0 and above
+     */
+    boolean useStandardYamlConfig() {
+        return kubernetesComponentConf
+                .getFlinkConfiguration()
+                .get(FLINK_VERSION)
+                .isEqualOrNewer(FlinkVersion.v2_0);
+    }
 }