Fix duplicated secret volume in pod spec

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

@@ -31,6 +31,14 @@ private[spark] trait MountSecretsBootstrap {
    * @return the updated pod and container with the secrets mounted.
    */
   def mountSecrets(pod: Pod, container: Container): (Pod, Container)
+  /**
+   * Mounts Kubernetes secrets as secret volumes into the given container. Make sure that
+   * the secret volumes are being added to its pod spec.
+   *
+   * @param container the container into which the secret volumes are being mounted.
+   * @return the updated pod and container with the secrets mounted.
+   */
+  def mountSecrets(container: Container): (Container)
 }
 
 private[spark] class MountSecretsBootstrapImpl(
@@ -49,16 +57,20 @@ private[spark] class MountSecretsBootstrapImpl(
             .endVolume()
           .endSpec())
 
+    (podBuilder.build(), mountSecrets(container))
+  }
+
+  override def mountSecrets(container: Container): (Container) = {
     var containerBuilder = new ContainerBuilder(container)
     secretNamesToMountPaths.foreach(namePath =>
       containerBuilder = containerBuilder
         .addNewVolumeMount()
-          .withName(secretVolumeName(namePath._1))
-          .withMountPath(namePath._2)
-          .endVolumeMount()
+        .withName(secretVolumeName(namePath._1))
+        .withMountPath(namePath._2)
+        .endVolumeMount()
     )
 
-    (podBuilder.build(), containerBuilder.build())
+    containerBuilder.build()
   }
 
   private def secretVolumeName(secretName: String): String = {

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

@@ -28,12 +28,9 @@ private[spark] class InitContainerMountSecretsStep(
     mountSecretsBootstrap: MountSecretsBootstrap) extends InitContainerConfigurationStep {
 
   override def configureInitContainer(initContainerSpec: InitContainerSpec) : InitContainerSpec = {
-    val (podWithSecretsMounted, initContainerWithSecretsMounted) =
-      mountSecretsBootstrap.mountSecrets(
-        initContainerSpec.podToInitialize,
-        initContainerSpec.initContainer)
+    val initContainerWithSecretsMounted =
+      mountSecretsBootstrap.mountSecrets(initContainerSpec.initContainer)
     initContainerSpec.copy(
-      podToInitialize = podWithSecretsMounted,
       initContainer = initContainerWithSecretsMounted
     )
   }

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

@@ -256,13 +256,13 @@ private[spark] class ExecutorPodFactoryImpl(
               podWithDetachedInitContainer.initContainer)
         }.getOrElse(podWithDetachedInitContainer.initContainer)
 
-        val (mayBePodWithSecretsMountedToInitContainer, mayBeInitContainerWithSecretsMounted) =
+        val mayBeInitContainerWithSecretsMounted =
           executorInitContainerMountSecretsBootstrap.map { bootstrap =>
-            bootstrap.mountSecrets(podWithDetachedInitContainer.pod, resolvedInitContainer)
-        }.getOrElse(podWithDetachedInitContainer.pod, resolvedInitContainer)
+            bootstrap.mountSecrets(resolvedInitContainer)
+          }.getOrElse(resolvedInitContainer)
 
         val podWithAttachedInitContainer = InitContainerUtil.appendInitContainer(
-          mayBePodWithSecretsMountedToInitContainer, mayBeInitContainerWithSecretsMounted)
+          podWithDetachedInitContainer.pod, mayBeInitContainerWithSecretsMounted)
 
         val resolvedPodWithMountedSecret = executorMountInitContainerSecretPlugin.map { plugin =>
           plugin.addResourceStagingServerSecretVolumeToPod(podWithAttachedInitContainer)

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

@@ -221,7 +221,7 @@ class ExecutorPodFactorySuite extends SparkFunSuite with BeforeAndAfter with Bef
     val factory = new ExecutorPodFactoryImpl(
       conf,
       nodeAffinityExecutorPodModifier,
-      None,
+      Some(secretsBootstrap),
       None,
       Some(initContainerBootstrap),
       Some(secretsBootstrap),
@@ -242,6 +242,10 @@ class ExecutorPodFactorySuite extends SparkFunSuite with BeforeAndAfter with Bef
     assert(executor.getSpec.getInitContainers.get(0).getVolumeMounts.get(0)
       .getMountPath === "/var/secret1")
 
+    // check volume mounted.
+    assert(executor.getSpec.getVolumes.size() === 1)
+    assert(executor.getSpec.getVolumes.get(0).getSecret.getSecretName === "secret1")
+
     checkOwnerReferences(executor, driverPodUid)
   }