[SPARK-50758][K8S] Mounts the krb5 config map on the executor pod

resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/Constants.scala

@@ -121,6 +121,8 @@ object Constants {
     "spark.kubernetes.executor.hadoopConfigMapName"
 
   // Kerberos Configuration
+  val KRB_CONFIG_MAP_NAME =
+    "spark.kubernetes.executor.krbConfigMapName"
   val KERBEROS_DT_SECRET_NAME =
     "spark.kubernetes.kerberos.dt-secret-name"
   val KERBEROS_DT_SECRET_KEY =

resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/features/KerberosConfDriverFeatureStep.scala

@@ -215,14 +215,17 @@ private[spark] class KerberosConfDriverFeatureStep(kubernetesConf: KubernetesDri
   }
 
   override def getAdditionalPodSystemProperties(): Map[String, String] = {
+    val props = scala.collection.mutable.Map.empty[String, String]
     // If a submission-local keytab is provided, update the Spark config so that it knows the
     // path of the keytab in the driver container.
     if (needKeytabUpload) {
       val ktName = new File(keytab.get).getName()
-      Map(KEYTAB.key -> s"$KERBEROS_KEYTAB_MOUNT_POINT/$ktName")
-    } else {
-      Map.empty
+      props += (KEYTAB.key -> s"$KERBEROS_KEYTAB_MOUNT_POINT/$ktName")
+    }
+    if (hasKerberosConf) {
+      props += (KRB_CONFIG_MAP_NAME -> krb5CMap.getOrElse(newConfigMapName))
     }
+    props.toMap
   }
 
   override def getAdditionalKubernetesResources(): Seq[HasMetadata] = {

resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/features/KerberosConfExecutorFeatureStep.scala

@@ -0,0 +1,59 @@
+/*
+ * 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.spark.deploy.k8s.features
+
+import io.fabric8.kubernetes.api.model.{ContainerBuilder, PodBuilder, VolumeBuilder}
+
+import org.apache.spark.deploy.k8s.{KubernetesConf, SparkPod}
+import org.apache.spark.deploy.k8s.Constants._
+
+/**
+ * Mounts the krb5.conf ConfigMap on the executor pod.
+ */
+private[spark] class KerberosConfExecutorFeatureStep(conf: KubernetesConf)
+  extends KubernetesFeatureConfigStep {
+
+  private val krbConfigMapName = conf.getOption(KRB_CONFIG_MAP_NAME)
+
+  override def configurePod(original: SparkPod): SparkPod = {
+    original.transform { case pod if krbConfigMapName.isDefined =>
+      val configMapVolume = new VolumeBuilder()
+        .withName(KRB_FILE_VOLUME)
+        .withNewConfigMap()
+          .withName(krbConfigMapName.get)
+          .endConfigMap()
+        .build()
+
+      val podWithVolume = new PodBuilder(pod.pod)
+        .editSpec()
+          .addNewVolumeLike(configMapVolume)
+            .endVolume()
+          .endSpec()
+        .build()
+
+      val containerWithMount = new ContainerBuilder(pod.container)
+        .addNewVolumeMount()
+          .withName(KRB_FILE_VOLUME)
+          .withMountPath(KRB_FILE_DIR_PATH + "/krb5.conf")
+          .withSubPath("krb5.conf")
+          .endVolumeMount()
+        .build()
+
+      SparkPod(podWithVolume, containerWithMount)
+    }
+  }
+}

resource-managers/kubernetes/core/src/main/scala/org/apache/spark/scheduler/cluster/k8s/KubernetesClusterSchedulerBackend.scala

@@ -16,6 +16,8 @@
  */
 package org.apache.spark.scheduler.cluster.k8s
 
+import java.io.File
+import java.nio.file.Files
 import java.util.concurrent.{ScheduledExecutorService, TimeUnit}
 import java.util.concurrent.atomic.AtomicInteger
 
@@ -98,6 +100,35 @@ private[spark] class KubernetesClusterSchedulerBackend(
     kubernetesClient.configMaps().inNamespace(namespace).resource(configMap).create()
   }
 
+  /**
+   * Publishes the krb5 ConfigMap name into [[KRB_CONFIG_MAP_NAME]] so executor pods can mount it.
+   * Cluster mode already has it set by the driver step; client mode creates or reuses it here.
+   */
+  private[k8s] def setUpExecutorKrb5ConfigMap(driverPod: Option[Pod]): Unit = {
+    // Already set by the driver feature step (cluster mode), nothing to do.
+    if (conf.getOption(KRB_CONFIG_MAP_NAME).isEmpty) {
+      conf.get(KUBERNETES_KERBEROS_KRB5_CONFIG_MAP) match {
+        case Some(existingMap) =>
+          // User supplied an existing ConfigMap, just publish its name to executors.
+          conf.set(KRB_CONFIG_MAP_NAME, existingMap)
+        case None =>
+          conf.get(KUBERNETES_KERBEROS_KRB5_FILE).foreach { localPath =>
+            val file = new File(localPath)
+            val configMapName = KubernetesClientUtils
+              .configMapName(s"spark-krb5-${KubernetesUtils.uniqueID()}")
+            val labels =
+              Map(SPARK_APP_ID_LABEL -> applicationId(),
+                SPARK_ROLE_LABEL -> SPARK_POD_EXECUTOR_ROLE)
+            val configMap = KubernetesClientUtils.buildConfigMap(
+              configMapName, Map(file.getName -> Files.readString(file.toPath)), labels)
+            KubernetesUtils.addOwnerReference(driverPod.orNull, Seq(configMap))
+            kubernetesClient.configMaps().inNamespace(namespace).resource(configMap).create()
+            conf.set(KRB_CONFIG_MAP_NAME, configMapName)
+          }
+      }
+    }
+  }
+
   /**
    * Get an application ID associated with the job.
    * This returns the string value of spark.app.id if set, otherwise
@@ -115,6 +146,7 @@ private[spark] class KubernetesClusterSchedulerBackend(
     // allocation is asynchronous (background thread pool), so requesting executors first
     // can race with ConfigMap creation, causing transient "configmap ... not found" mounts.
     if (!conf.get(KUBERNETES_EXECUTOR_DISABLE_CONFIGMAP)) {
+      setUpExecutorKrb5ConfigMap(podAllocator.driverPod)
       setUpExecutorConfigMap(podAllocator.driverPod)
     }
     val defaultProfile = scheduler.sc.resourceProfileManager.defaultResourceProfile

resource-managers/kubernetes/core/src/main/scala/org/apache/spark/scheduler/cluster/k8s/KubernetesExecutorBuilder.scala

@@ -72,6 +72,7 @@ private[spark] class KubernetesExecutorBuilder {
       new EnvSecretsFeatureStep(conf),
       new MountVolumesFeatureStep(conf),
       new HadoopConfExecutorFeatureStep(conf),
+      new KerberosConfExecutorFeatureStep(conf),
       new LocalDirsFeatureStep(conf)) ++ userFeatures
 
     val features = allFeatures.filterNot(f =>

resource-managers/kubernetes/core/src/test/scala/org/apache/spark/deploy/k8s/features/KerberosConfDriverFeatureStepSuite.scala

@@ -48,7 +48,7 @@ class KerberosConfDriverFeatureStepSuite extends SparkFunSuite {
       new SparkConf(false).set(KUBERNETES_KERBEROS_KRB5_CONFIG_MAP, configMap))
 
     checkPodForKrbConf(step.configurePod(SparkPod.initialPod()), configMap)
-    assert(step.getAdditionalPodSystemProperties().isEmpty)
+    assert(step.getAdditionalPodSystemProperties() === Map(KRB_CONFIG_MAP_NAME -> configMap))
     assert(filter[ConfigMap](step.getAdditionalKubernetesResources()).isEmpty)
   }
 
@@ -64,7 +64,8 @@ class KerberosConfDriverFeatureStepSuite extends SparkFunSuite {
     assert(confMap.getData().keySet().asScala === Set(krbConf.getName()))
 
     checkPodForKrbConf(step.configurePod(SparkPod.initialPod()), confMap.getMetadata().getName())
-    assert(step.getAdditionalPodSystemProperties().isEmpty)
+    assert(step.getAdditionalPodSystemProperties() ===
+      Map(KRB_CONFIG_MAP_NAME -> confMap.getMetadata().getName()))
   }
 
   test("create keytab secret if client keytab file used") {

resource-managers/kubernetes/core/src/test/scala/org/apache/spark/deploy/k8s/features/KerberosConfExecutorFeatureStepSuite.scala

@@ -0,0 +1,89 @@
+/*
+ * 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.spark.deploy.k8s.features
+
+import java.io.File
+import java.nio.file.Files
+
+import org.apache.spark.{SparkConf, SparkFunSuite}
+import org.apache.spark.deploy.k8s._
+import org.apache.spark.deploy.k8s.Config._
+import org.apache.spark.deploy.k8s.Constants._
+import org.apache.spark.util.Utils
+
+class KerberosConfExecutorFeatureStepSuite extends SparkFunSuite {
+  import SecretVolumeUtils._
+
+  test("SPARK-50758: mount krb5 ConfigMap when KRB_CONFIG_MAP_NAME is set") {
+    val executorSparkConf = new SparkConf(false).set(KRB_CONFIG_MAP_NAME, "testCM")
+    val executorConf = KubernetesTestConf.createExecutorConf(sparkConf = executorSparkConf)
+    val initial = SparkPod.initialPod()
+    val executorPod = new KerberosConfExecutorFeatureStep(executorConf).configurePod(initial)
+    checkPod(executorPod, hasKrb5 = true)
+  }
+
+  test("SPARK-50758: no-op when KRB_CONFIG_MAP_NAME is not set") {
+    val executorConf = KubernetesTestConf.createExecutorConf(sparkConf = new SparkConf(false))
+    val initial = SparkPod.initialPod()
+    val executorPod = new KerberosConfExecutorFeatureStep(executorConf).configurePod(initial)
+    checkPod(executorPod, hasKrb5 = false)
+  }
+
+  test("SPARK-50758: mount krb5 ConfigMap when driver step publishes its name") {
+    val tmpDir = Utils.createTempDir()
+    val krbConf = File.createTempFile("krb5", ".conf", tmpDir)
+    Files.writeString(krbConf.toPath, "some data")
+
+    Seq(
+      // (sparkConf, expectMount)
+      (new SparkConf(false).set(KUBERNETES_KERBEROS_KRB5_CONFIG_MAP, "userCM"), true),
+      (new SparkConf(false).set(KUBERNETES_KERBEROS_KRB5_FILE, krbConf.getAbsolutePath), true),
+      (new SparkConf(false), false)
+    ).foreach { case (driverSparkConf, expectMount) =>
+
+      val driverConf = KubernetesTestConf.createDriverConf(sparkConf = driverSparkConf)
+      val driverStep = new KerberosConfDriverFeatureStep(driverConf)
+
+      val executorSparkConf = new SparkConf(false)
+      val additionalProps = driverStep.getAdditionalPodSystemProperties()
+      if (expectMount) {
+        assert(additionalProps.contains(KRB_CONFIG_MAP_NAME),
+          s"Driver step must publish $KRB_CONFIG_MAP_NAME when krb5 conf is provided")
+        additionalProps.foreach { case (k, v) => executorSparkConf.set(k, v) }
+      } else {
+        assert(!additionalProps.contains(KRB_CONFIG_MAP_NAME))
+      }
+
+      val executorConf = KubernetesTestConf.createExecutorConf(sparkConf = executorSparkConf)
+      val initial = SparkPod.initialPod()
+      val executorPod = new KerberosConfExecutorFeatureStep(executorConf).configurePod(initial)
+      checkPod(executorPod, expectMount)
+    }
+  }
+
+  private def checkPod(pod: SparkPod, hasKrb5: Boolean): Unit = {
+    val mountPath = KRB_FILE_DIR_PATH + "/krb5.conf"
+    if (hasKrb5) {
+      assert(podHasVolume(pod.pod, KRB_FILE_VOLUME))
+      assert(containerHasVolume(pod.container, KRB_FILE_VOLUME, mountPath))
+    } else {
+      assert(!podHasVolume(pod.pod, KRB_FILE_VOLUME))
+      assert(!containerHasVolume(pod.container, KRB_FILE_VOLUME, mountPath))
+    }
+  }
+}

resource-managers/kubernetes/core/src/test/scala/org/apache/spark/scheduler/cluster/k8s/KubernetesClusterSchedulerBackendSuite.scala

@@ -16,6 +16,8 @@
  */
 package org.apache.spark.scheduler.cluster.k8s
 
+import java.io.File
+import java.nio.file.Files
 import java.util.Arrays
 import java.util.concurrent.TimeUnit
 
@@ -43,6 +45,7 @@ import org.apache.spark.scheduler.cluster.CoarseGrainedClusterMessages.{Decommis
 import org.apache.spark.scheduler.cluster.CoarseGrainedSchedulerBackend
 import org.apache.spark.scheduler.cluster.k8s.ExecutorLifecycleTestUtils.TEST_SPARK_APP_ID
 import org.apache.spark.storage.{BlockManager, BlockManagerMaster}
+import org.apache.spark.util.Utils
 
 class KubernetesClusterSchedulerBackendSuite extends SparkFunSuite with BeforeAndAfter {
 
@@ -510,4 +513,55 @@ class KubernetesClusterSchedulerBackendSuite extends SparkFunSuite with BeforeAn
     }
     assert(schedulerBackendUnderTest.supportsExecutorHold)
   }
+
+  test("SPARK-50758: setUpExecutorKrb5ConfigMap publishes or creates the krb5 ConfigMap") {
+    // 1. No krb5 config at all: no-op.
+    schedulerBackendUnderTest.setUpExecutorKrb5ConfigMap(None)
+    assert(sparkConf.getOption(KRB_CONFIG_MAP_NAME).isEmpty)
+    verify(configMapsWithNamespace, never()).resource(any[ConfigMap]())
+
+    // 2. Cluster mode: the driver step already set KRB_CONFIG_MAP_NAME.
+    sparkConf.set(KRB_CONFIG_MAP_NAME, "alreadySetByDriverStep")
+    sparkConf.set(KUBERNETES_KERBEROS_KRB5_FILE, "/does/not/matter")
+    try {
+      schedulerBackendUnderTest.setUpExecutorKrb5ConfigMap(None)
+      assert(sparkConf.get(KRB_CONFIG_MAP_NAME) === "alreadySetByDriverStep")
+      verify(configMapsWithNamespace, never()).resource(any[ConfigMap]())
+    } finally {
+      sparkConf.remove(KRB_CONFIG_MAP_NAME)
+      sparkConf.remove(KUBERNETES_KERBEROS_KRB5_FILE.key)
+    }
+
+    // 3. User-provided ConfigMap name is just published, nothing is created.
+    sparkConf.set(KUBERNETES_KERBEROS_KRB5_CONFIG_MAP, "userKrbCM")
+    try {
+      schedulerBackendUnderTest.setUpExecutorKrb5ConfigMap(None)
+      assert(sparkConf.get(KRB_CONFIG_MAP_NAME) === "userKrbCM")
+      verify(configMapsWithNamespace, never()).resource(any[ConfigMap]())
+    } finally {
+      sparkConf.remove(KUBERNETES_KERBEROS_KRB5_CONFIG_MAP.key)
+      sparkConf.remove(KRB_CONFIG_MAP_NAME)
+    }
+
+    // 4. Client mode: a local krb5.conf file triggers ConfigMap creation.
+    val tmpDir = Utils.createTempDir()
+    val krb5 = File.createTempFile("krb5", ".conf", tmpDir)
+    Files.writeString(krb5.toPath, "some krb5 data")
+    sparkConf.set(KUBERNETES_KERBEROS_KRB5_FILE, krb5.getAbsolutePath)
+    try {
+      schedulerBackendUnderTest.setUpExecutorKrb5ConfigMap(None)
+      val captor = ArgumentCaptor.forClass(classOf[ConfigMap])
+      verify(configMapsWithNamespace).resource(captor.capture())
+      verify(configMapResource).create()
+      val created = captor.getValue
+      val labels = created.getMetadata.getLabels.asScala
+      assert(labels(SPARK_APP_ID_LABEL) === TEST_SPARK_APP_ID)
+      assert(labels(SPARK_ROLE_LABEL) === SPARK_POD_EXECUTOR_ROLE)
+      assert(created.getData.keySet().asScala === Set(krb5.getName))
+      assert(sparkConf.get(KRB_CONFIG_MAP_NAME) === created.getMetadata.getName)
+    } finally {
+      sparkConf.remove(KUBERNETES_KERBEROS_KRB5_FILE.key)
+      sparkConf.remove(KRB_CONFIG_MAP_NAME)
+    }
+  }
 }