[SPARK-37713][K8S] Assign namespace to executor configmap

### What changes were proposed in this pull request?

Since Spark 3.X, Executor pod needs separate executor configmap. But, no namespace is assigned in configmap while building it. K8s views configmap without namespace as global resource. Once pod access is restricted (global resources cannot be read), and executor cannot obtain configmap itself.

### Why are the changes needed?

Fix executor pod authorization error in K8S. When executor pod cannot read global resources, executor configmap is not available for executor.

### Does this PR introduce _any_ user-facing change?

No

### How was this patch tested?

Pass Integration Test and CIs.

Closes #34983 from dcoliversun/SPARK-37713.

Authored-by: Qian.Sun <[email protected]>
Signed-off-by: Dongjoon Hyun <[email protected]>

Author: dcoliversun

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

@@ -29,7 +29,7 @@ import io.fabric8.kubernetes.api.model.{ConfigMap, ConfigMapBuilder, KeyToPath}
 
 import org.apache.spark.SparkConf
 import org.apache.spark.deploy.k8s.{Config, Constants, KubernetesUtils}
-import org.apache.spark.deploy.k8s.Config.KUBERNETES_DNSNAME_MAX_LENGTH
+import org.apache.spark.deploy.k8s.Config.{KUBERNETES_DNSNAME_MAX_LENGTH, KUBERNETES_NAMESPACE}
 import org.apache.spark.deploy.k8s.Constants.ENV_SPARK_CONF_DIR
 import org.apache.spark.internal.Logging
 
@@ -89,9 +89,12 @@ private[spark] object KubernetesClientUtils extends Logging {
    */
   def buildConfigMap(configMapName: String, confFileMap: Map[String, String],
       withLabels: Map[String, String] = Map()): ConfigMap = {
+    val configMapNameSpace =
+      confFileMap.getOrElse(KUBERNETES_NAMESPACE.key, KUBERNETES_NAMESPACE.defaultValueString)
     new ConfigMapBuilder()
       .withNewMetadata()
         .withName(configMapName)
+        .withNamespace(configMapNameSpace)
         .withLabels(withLabels.asJava)
         .endMetadata()
       .withImmutable(true)

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

@@ -76,8 +76,10 @@ private[spark] class KubernetesClusterSchedulerBackend(
 
   private def setUpExecutorConfigMap(driverPod: Option[Pod]): Unit = {
     val configMapName = KubernetesClientUtils.configMapNameExecutor
+    val resolvedExecutorProperties =
+      Map(KUBERNETES_NAMESPACE.key -> conf.get(KUBERNETES_NAMESPACE))
     val confFilesMap = KubernetesClientUtils
-      .buildSparkConfDirFilesMap(configMapName, conf, Map.empty)
+      .buildSparkConfDirFilesMap(configMapName, conf, resolvedExecutorProperties)
     val labels =
       Map(SPARK_APP_ID_LABEL -> applicationId(), SPARK_ROLE_LABEL -> SPARK_POD_EXECUTOR_ROLE)
     val configMap = KubernetesClientUtils.buildConfigMap(configMapName, confFilesMap, labels)

resource-managers/kubernetes/core/src/test/scala/org/apache/spark/deploy/k8s/submit/KubernetesClientUtilsSuite.scala

@@ -20,7 +20,11 @@ package org.apache.spark.deploy.k8s.submit
 import java.io.File
 import java.nio.charset.StandardCharsets
 import java.nio.file.Files
+import java.util.UUID
 
+import scala.collection.JavaConverters._
+
+import io.fabric8.kubernetes.api.model.ConfigMapBuilder
 import org.scalatest.BeforeAndAfter
 
 import org.apache.spark.{SparkConf, SparkFunSuite}
@@ -76,4 +80,27 @@ class KubernetesClientUtilsSuite extends SparkFunSuite with BeforeAndAfter {
       "testConf.3" -> "test123456")
     assert(output === expectedOutput)
   }
+
+  test("verify that configmap built as expected") {
+    val configMapName = s"configmap-name-${UUID.randomUUID.toString}"
+    val configMapNameSpace = s"configmap-namespace-${UUID.randomUUID.toString}"
+    val properties = Map(Config.KUBERNETES_NAMESPACE.key -> configMapNameSpace)
+    val sparkConf =
+      testSetup(properties.map(f => f._1 -> f._2.getBytes(StandardCharsets.UTF_8)))
+    val confFileMap =
+      KubernetesClientUtils.buildSparkConfDirFilesMap(configMapName, sparkConf, properties)
+    val outputConfigMap =
+      KubernetesClientUtils.buildConfigMap(configMapName, confFileMap, properties)
+    val expectedConfigMap =
+      new ConfigMapBuilder()
+        .withNewMetadata()
+          .withName(configMapName)
+          .withNamespace(configMapNameSpace)
+          .withLabels(properties.asJava)
+        .endMetadata()
+        .withImmutable(true)
+        .addToData(confFileMap.asJava)
+        .build()
+    assert(outputConfigMap === expectedConfigMap)
+  }
 }