[SPARK-42060][K8S][WIP] add new config to override driver/executor k8s containers names

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

@@ -146,6 +146,20 @@ private[spark] object Config extends Logging {
       .checkValues(Set("Always", "Never", "IfNotPresent"))
       .createWithDefault("IfNotPresent")
 
+  val DRIVER_CONTAINER_NAME =
+    ConfigBuilder("spark.kubernetes.driver.container.name")
+      .doc("Name of the driver container.")
+      .version("3.4.0")
+      .stringConf
+      .createWithDefault(DEFAULT_DRIVER_CONTAINER_NAME)
+
+  val EXECUTOR_CONTAINER_NAME =
+    ConfigBuilder("spark.kubernetes.executor.container.name")
+      .doc("Name of the executors containers.")
+      .version("3.4.0")
+      .stringConf
+      .createWithDefault(DEFAULT_EXECUTOR_CONTAINER_NAME)
+
   val IMAGE_PULL_SECRETS =
     ConfigBuilder("spark.kubernetes.container.image.pullSecrets")
       .doc("Comma separated list of the Kubernetes secrets used " +

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

@@ -101,7 +101,7 @@ private[spark] class BasicDriverFeatureStep(conf: KubernetesDriverConf)
     )
     val driverUIPort = SparkUI.getUIPort(conf.sparkConf)
     val driverContainer = new ContainerBuilder(pod.container)
-      .withName(Option(pod.container.getName).getOrElse(DEFAULT_DRIVER_CONTAINER_NAME))
+      .withName(Option(pod.container.getName).getOrElse(DRIVER_CONTAINER_NAME))
       .withImage(driverContainerImage)
       .withImagePullPolicy(conf.imagePullPolicy)
       .addNewPort()

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

@@ -187,7 +187,7 @@ private[spark] class BasicExecutorFeatureStep(
     }
 
     val executorContainer = new ContainerBuilder(pod.container)
-      .withName(Option(pod.container.getName).getOrElse(DEFAULT_EXECUTOR_CONTAINER_NAME))
+      .withName(Option(pod.container.getName).getOrElse(EXECUTOR_CONTAINER_NAME))
       .withImage(executorContainerImage)
       .withImagePullPolicy(kubernetesConf.imagePullPolicy)
       .editOrNewResources()

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

@@ -23,6 +23,7 @@ import scala.collection.JavaConverters._
 import io.fabric8.kubernetes.api.model.ContainerStateTerminated
 import io.fabric8.kubernetes.api.model.Pod
 
+import org.apache.spark.deploy.k8s.Config._
 import org.apache.spark.deploy.k8s.Constants._
 import org.apache.spark.internal.Logging
 
@@ -43,7 +44,7 @@ private[spark] case class ExecutorPodsSnapshot(
 
 object ExecutorPodsSnapshot extends Logging {
   private var shouldCheckAllContainers: Boolean = _
-  private var sparkContainerName: String = DEFAULT_EXECUTOR_CONTAINER_NAME
+  private var sparkContainerName: String = EXECUTOR_CONTAINER_NAME
 
   def apply(executorPods: Seq[Pod], fullSnapshotTs: Long): ExecutorPodsSnapshot = {
     ExecutorPodsSnapshot(toStatesByExecutorId(executorPods), fullSnapshotTs)

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

@@ -24,7 +24,6 @@ import io.fabric8.kubernetes.client.KubernetesClient
 import org.apache.spark.{SparkConf, SparkContext}
 import org.apache.spark.deploy.k8s.{KubernetesConf, KubernetesUtils, SparkKubernetesClientFactory}
 import org.apache.spark.deploy.k8s.Config._
-import org.apache.spark.deploy.k8s.Constants.DEFAULT_EXECUTOR_CONTAINER_NAME
 import org.apache.spark.internal.Logging
 import org.apache.spark.scheduler.{ExternalClusterManager, SchedulerBackend, TaskScheduler, TaskSchedulerImpl}
 import org.apache.spark.util.{Clock, SystemClock, ThreadUtils, Utils}
@@ -98,7 +97,7 @@ private[spark] class KubernetesClusterManager extends ExternalClusterManager wit
     ExecutorPodsSnapshot.setShouldCheckAllContainers(
       sc.conf.get(KUBERNETES_EXECUTOR_CHECK_ALL_CONTAINERS))
     val sparkContainerName = sc.conf.get(KUBERNETES_EXECUTOR_PODTEMPLATE_CONTAINER_NAME)
-      .getOrElse(DEFAULT_EXECUTOR_CONTAINER_NAME)
+      .getOrElse(EXECUTOR_CONTAINER_NAME)
     ExecutorPodsSnapshot.setSparkContainerName(sparkContainerName)
     val subscribersExecutor = ThreadUtils
       .newDaemonThreadPoolScheduledExecutor(

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

@@ -19,13 +19,13 @@ package org.apache.spark.scheduler.cluster.k8s
 import io.fabric8.kubernetes.api.model.Pod
 import scala.collection.mutable
 
-import org.apache.spark.deploy.k8s.Constants.DEFAULT_EXECUTOR_CONTAINER_NAME
+import org.apache.spark.deploy.k8s.Config._
 import org.apache.spark.util.ManualClock
 
 class DeterministicExecutorPodsSnapshotsStore extends ExecutorPodsSnapshotsStore {
 
   ExecutorPodsSnapshot.setShouldCheckAllContainers(false)
-  ExecutorPodsSnapshot.setSparkContainerName(DEFAULT_EXECUTOR_CONTAINER_NAME)
+  ExecutorPodsSnapshot.setSparkContainerName(EXECUTOR_CONTAINER_NAME)
 
   val clock = new ManualClock()