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

Author: tclxyuhang2-zhang

Diff for: 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;

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

@@ -18,6 +18,9 @@
 
 package org.apache.flink.kubernetes.utils;
 
+import java.util.Arrays;
+import java.util.List;
+
 /** Constants for kubernetes. */
 public class Constants {
 
@@ -30,6 +33,17 @@ 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";

Diff for: 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}. */
@@ -233,6 +235,38 @@ void testDecoratedFlinkPodWithLog4jAndLogback() throws IOException {
                 .isEqualTo(expectedVolumes);
     }
 
+    @Test
+    void testDecoratedFlinkPodWithAllLog4jAndLogback() throws IOException {
+        for (String fileName : CONFIG_FILE_NAME_LIST) {
+            KubernetesTestUtils.createTemporyFile("some data", flinkConfDir, fileName);
+        }
+
+        final FlinkPod resultFlinkPod = flinkConfMountDecorator.decorateFlinkPod(baseFlinkPod);
+
+        final List<KeyToPath> expectedKeyToPaths = new ArrayList<>();
+        for (String fileName : CONFIG_FILE_NAME_LIST) {
+            expectedKeyToPaths.add(
+                    new KeyToPathBuilder().withKey(fileName).withPath(fileName).build());
+        }
+        expectedKeyToPaths.add(
+                new KeyToPathBuilder()
+                        .withKey(FLINK_CONF_FILENAME)
+                        .withPath(FLINK_CONF_FILENAME)
+                        .build());
+
+        final List<Volume> expectedVolumes =
+                Collections.singletonList(
+                        new VolumeBuilder()
+                                .withName(Constants.FLINK_CONF_VOLUME)
+                                .withNewConfigMap()
+                                .withName(getFlinkConfConfigMapName(CLUSTER_ID))
+                                .withItems(expectedKeyToPaths)
+                                .endConfigMap()
+                                .build());
+        assertThat(resultFlinkPod.getPodWithoutMainContainer().getSpec().getVolumes())
+                .isEqualTo(expectedVolumes);
+    }
+
     @Test
     void testDecoratedFlinkContainer() {
         final Container resultMainContainer =