[hotfix] Fix the issue related to mounting the Logback configuration file.

Author: tclxyuhang2-zhang

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

@@ -57,8 +57,7 @@
 import java.util.Map;
 import java.util.stream.Collectors;
 
-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_FILE_NAME_LIST;
 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;
@@ -179,15 +178,14 @@ String getFlinkConfData(List<String> confData) throws IOException {
 
     private List<File> getLocalLogConfFiles() {
         final String confDir = kubernetesComponentConf.getConfigDirectory();
-        final File logbackFile = new File(confDir, CONFIG_FILE_LOGBACK_NAME);
-        final File log4jFile = new File(confDir, CONFIG_FILE_LOG4J_NAME);
 
         List<File> localLogConfFiles = new ArrayList<>();
-        if (logbackFile.exists()) {
-            localLogConfFiles.add(logbackFile);
-        }
-        if (log4jFile.exists()) {
-            localLogConfFiles.add(log4jFile);
+
+        for (String fileName : CONFIG_FILE_NAME_LIST) {
+            final File file = new File(confDir, fileName);
+            if (file.exists()) {
+                localLogConfFiles.add(file);
+            }
         }
 
         return localLogConfFiles;

flink-kubernetes/src/main/java/org/apache/flink/kubernetes/utils/Constants.java

@@ -18,6 +18,10 @@
 
 package org.apache.flink.kubernetes.utils;
 
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+
 /** Constants for kubernetes. */
 public class Constants {
 
@@ -30,6 +34,15 @@ public class Constants {
 
     public static final String CONFIG_FILE_LOGBACK_NAME = "logback-console.xml";
     public static final String CONFIG_FILE_LOG4J_NAME = "log4j-console.properties";
+    public static final List<String> CONFIG_FILE_NAME_LIST = Arrays.asList(
+            "logback.xml",
+            "log4j.properties",
+            "logback-console.xml",
+            "log4j-console.properties",
+            "logback-session.xml",
+            "log4j-session.properties",
+            "log4j-cli.properties"
+    );
     public static final String ENV_FLINK_LOG_DIR = "FLINK_LOG_DIR";
 
     public static final String MAIN_CONTAINER_NAME = "flink-main-container";

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

@@ -38,6 +38,7 @@
 import org.junit.jupiter.api.Test;
 
 import java.io.IOException;
+import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collections;
 import java.util.List;
@@ -47,6 +48,7 @@
 import static org.apache.flink.kubernetes.kubeclient.decorators.FlinkConfMountDecorator.getFlinkConfConfigMapName;
 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_FILE_NAME_LIST;
 import static org.assertj.core.api.Assertions.assertThat;
 
 /** General tests for the {@link FlinkConfMountDecorator}. */
@@ -201,25 +203,27 @@ void testDecoratedFlinkPodWithLogback() throws IOException {
 
     @Test
     void testDecoratedFlinkPodWithLog4jAndLogback() throws IOException {
-        KubernetesTestUtils.createTemporyFile("some data", flinkConfDir, CONFIG_FILE_LOG4J_NAME);
-        KubernetesTestUtils.createTemporyFile("some data", flinkConfDir, CONFIG_FILE_LOGBACK_NAME);
+
+        for (String fileName : CONFIG_FILE_NAME_LIST) {
+            KubernetesTestUtils.createTemporyFile("some data", flinkConfDir, fileName);
+        }
 
         final FlinkPod resultFlinkPod = flinkConfMountDecorator.decorateFlinkPod(baseFlinkPod);
 
-        final List<KeyToPath> expectedKeyToPaths =
-                Arrays.asList(
-                        new KeyToPathBuilder()
-                                .withKey(CONFIG_FILE_LOGBACK_NAME)
-                                .withPath(CONFIG_FILE_LOGBACK_NAME)
-                                .build(),
-                        new KeyToPathBuilder()
-                                .withKey(CONFIG_FILE_LOG4J_NAME)
-                                .withPath(CONFIG_FILE_LOG4J_NAME)
-                                .build(),
-                        new KeyToPathBuilder()
-                                .withKey(FLINK_CONF_FILENAME)
-                                .withPath(FLINK_CONF_FILENAME)
-                                .build());
+        final List<KeyToPath> expectedKeyToPaths = new ArrayList<>();
+        for (String fileName : CONFIG_FILE_NAME_LIST) {
+            KeyToPath keyToPath = new KeyToPathBuilder()
+                    .withKey(fileName)
+                    .withPath(fileName)
+                    .build();
+            expectedKeyToPaths.add(keyToPath);
+        }
+        expectedKeyToPaths.add(
+                new KeyToPathBuilder()
+                        .withKey(FLINK_CONF_FILENAME)
+                        .withPath(FLINK_CONF_FILENAME)
+                        .build());
+
         final List<Volume> expectedVolumes =
                 Collections.singletonList(
                         new VolumeBuilder()