Addressed latest round of comments

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

@@ -159,6 +159,13 @@ private[spark] object Config extends Logging {
       .timeConf(TimeUnit.MINUTES)
       .createWithDefault(5)
 
+  val INIT_CONTAINER_MAX_THREAD_POOL_SIZE =
+    ConfigBuilder("spark.kubernetes.initContainer.maxThreadPoolSize")
+      .doc("Maximum size of the thread pool in the init-container for downloading remote " +
+        "dependencies.")
+      .intConf
+      .createWithDefault(5)
+
   val INIT_CONTAINER_REMOTE_JARS =
     ConfigBuilder("spark.kubernetes.initContainer.remoteJars")
       .doc("Comma-separated list of jar URIs to download in the init-container. This is " +

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

@@ -59,11 +59,12 @@ private[spark] class InitContainerBootstrap(
       case SPARK_POD_EXECUTOR_ROLE => "spark.executorEnv."
       case _ => throw new SparkException(s"$sparkRole is not a valid Spark pod role")
     }
-    val customEnvVars = sparkConf.getAllWithPrefix(customEnvVarKeyPrefix).toSeq.map { env =>
-      new EnvVarBuilder()
-        .withName(env._1)
-        .withValue(env._2)
-        .build()
+    val customEnvVars = sparkConf.getAllWithPrefix(customEnvVarKeyPrefix).toSeq.map {
+      case (key, value) =>
+        new EnvVarBuilder()
+          .withName(key)
+          .withValue(value)
+          .build()
     }
 
     val initContainer = new ContainerBuilder(original.initContainer)
@@ -102,14 +103,13 @@ private[spark] class InitContainerBootstrap(
       .endSpec()
       .build()
 
-    val mainContainer = new ContainerBuilder(
-      original.mainContainer)
-        .addToVolumeMounts(sharedVolumeMounts: _*)
-        .addNewEnv()
-          .withName(ENV_MOUNTED_FILES_DIR)
-          .withValue(filesDownloadPath)
-          .endEnv()
-        .build()
+    val mainContainer = new ContainerBuilder(original.mainContainer)
+      .addToVolumeMounts(sharedVolumeMounts: _*)
+      .addNewEnv()
+        .withName(ENV_MOUNTED_FILES_DIR)
+        .withValue(filesDownloadPath)
+        .endEnv()
+      .build()
 
     PodWithDetachedInitContainer(
       podWithBasicVolumes,

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

@@ -14,13 +14,49 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package org.apache.spark.deploy.k8s.submit
+package org.apache.spark.deploy.k8s
 
 import java.io.File
 
+import io.fabric8.kubernetes.api.model.{Container, Pod, PodBuilder}
+
+import org.apache.spark.SparkConf
 import org.apache.spark.util.Utils
 
-private[spark] object KubernetesFileUtils {
+private[spark] object KubernetesUtils {
+
+  /**
+   * Extract and parse Spark configuration properties with a given name prefix and
+   * return the result as a Map. Keys must not have more than one value.
+   *
+   * @param sparkConf Spark configuration
+   * @param prefix the given property name prefix
+   * @return a Map storing the configuration property keys and values
+   */
+  def parsePrefixedKeyValuePairs(
+      sparkConf: SparkConf,
+      prefix: String): Map[String, String] = {
+    sparkConf.getAllWithPrefix(prefix).toMap
+  }
+
+  def requireNandDefined(opt1: Option[_], opt2: Option[_], errMessage: String): Unit = {
+    opt1.foreach { _ => require(opt2.isEmpty, errMessage) }
+  }
+
+  /**
+   * Append the given init-container to a pod's list of init-containers..
+   *
+   * @param originalPodSpec original specification of the pod
+   * @param initContainer the init-container to add to the pod
+   * @return the pod with the init-container added to the list of InitContainers
+   */
+  def appendInitContainer(originalPodSpec: Pod, initContainer: Container): Pod = {
+    new PodBuilder(originalPodSpec)
+      .editOrNewSpec()
+        .addToInitContainers(initContainer)
+        .endSpec()
+      .build()
+  }
 
   /**
    * For the given collection of file URIs, resolves them as follows:

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

@@ -45,12 +45,13 @@ private[spark] class MountSecretsBootstrap(secretNamesToMountPaths: Map[String,
     }
 
     var containerBuilder = new ContainerBuilder(container)
-    secretNamesToMountPaths.foreach { namePath =>
-      containerBuilder = containerBuilder
-        .addNewVolumeMount()
-          .withName(secretVolumeName(namePath._1))
-          .withMountPath(namePath._2)
-          .endVolumeMount()
+    secretNamesToMountPaths.foreach {
+      case (name, path) =>
+        containerBuilder = containerBuilder
+          .addNewVolumeMount()
+            .withName(secretVolumeName(name))
+            .withMountPath(path)
+            .endVolumeMount()
     }
 
     (podBuilder.build(), containerBuilder.build())

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

@@ -48,7 +48,7 @@ private[spark] object SparkKubernetesClientFactory {
       .map(new File(_))
       .orElse(defaultServiceAccountToken)
     val oauthTokenValue = sparkConf.getOption(oauthTokenConf)
-    ConfigurationUtils.requireNandDefined(
+    KubernetesUtils.requireNandDefined(
       oauthTokenFile,
       oauthTokenValue,
       s"Cannot specify OAuth token through both a file $oauthTokenFileConf and a " +

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

@@ -21,7 +21,7 @@ import java.util.UUID
 import com.google.common.primitives.Longs
 
 import org.apache.spark.SparkConf
-import org.apache.spark.deploy.k8s.{ConfigurationUtils, MountSecretsBootstrap}
+import org.apache.spark.deploy.k8s.{KubernetesUtils, MountSecretsBootstrap}
 import org.apache.spark.deploy.k8s.Config._
 import org.apache.spark.deploy.k8s.Constants._
 import org.apache.spark.deploy.k8s.submit.steps._
@@ -61,7 +61,7 @@ private[spark] class DriverConfigOrchestrator(
   private val filesDownloadPath = sparkConf.get(FILES_DOWNLOAD_LOCATION)
 
   def getAllConfigurationSteps: Seq[DriverConfigurationStep] = {
-    val driverCustomLabels = ConfigurationUtils.parsePrefixedKeyValuePairs(
+    val driverCustomLabels = KubernetesUtils.parsePrefixedKeyValuePairs(
       sparkConf,
       KUBERNETES_DRIVER_LABEL_PREFIX)
     require(!driverCustomLabels.contains(SPARK_APP_ID_LABEL), "Label with key " +
@@ -71,15 +71,15 @@ private[spark] class DriverConfigOrchestrator(
       s"$SPARK_ROLE_LABEL is not allowed as it is reserved for Spark bookkeeping " +
       "operations.")
 
-    val secretNamesToMountPaths = ConfigurationUtils.parsePrefixedKeyValuePairs(
+    val secretNamesToMountPaths = KubernetesUtils.parsePrefixedKeyValuePairs(
       sparkConf,
       KUBERNETES_DRIVER_SECRETS_PREFIX)
 
     val allDriverLabels = driverCustomLabels ++ Map(
       SPARK_APP_ID_LABEL -> kubernetesAppId,
       SPARK_ROLE_LABEL -> SPARK_POD_DRIVER_ROLE)
 
-    val initialSubmissionStep = new BaseDriverConfigurationStep(
+    val initialSubmissionStep = new BasicDriverConfigurationStep(
       kubernetesAppId,
       kubernetesResourceNamePrefix,
       allDriverLabels,
@@ -117,50 +117,49 @@ private[spark] class DriverConfigOrchestrator(
       .map(_.split(","))
       .getOrElse(Array.empty[String])
 
-    val maybeDependencyResolutionStep = if (sparkJars.nonEmpty || sparkFiles.nonEmpty) {
-      Some(new DependencyResolutionStep(
+    val dependencyResolutionStep = if (sparkJars.nonEmpty || sparkFiles.nonEmpty) {
+      Seq(new DependencyResolutionStep(
         sparkJars,
         sparkFiles,
         jarsDownloadPath,
         filesDownloadPath))
     } else {
-      None
+      Nil
     }
 
-    val mayBeInitContainerBootstrapStep =
-      if (areAnyFilesNonContainerLocal(sparkJars ++ sparkFiles)) {
-        val orchestrator = new InitContainerConfigOrchestrator(
-          sparkJars,
-          sparkFiles,
-          jarsDownloadPath,
-          filesDownloadPath,
-          imagePullPolicy,
-          initContainerConfigMapName,
-          INIT_CONTAINER_PROPERTIES_FILE_NAME,
-          sparkConf)
-        val bootstrapStep = new DriverInitContainerBootstrapStep(
-          orchestrator.getAllConfigurationSteps,
-          initContainerConfigMapName,
-          INIT_CONTAINER_PROPERTIES_FILE_NAME)
-
-        Some(bootstrapStep)
-      } else {
-        None
-      }
+    val initContainerBootstrapStep = if (areAnyFilesNonContainerLocal(sparkJars ++ sparkFiles)) {
+      val orchestrator = new InitContainerConfigOrchestrator(
+        sparkJars,
+        sparkFiles,
+        jarsDownloadPath,
+        filesDownloadPath,
+        imagePullPolicy,
+        initContainerConfigMapName,
+        INIT_CONTAINER_PROPERTIES_FILE_NAME,
+        sparkConf)
+      val bootstrapStep = new DriverInitContainerBootstrapStep(
+        orchestrator.getAllConfigurationSteps,
+        initContainerConfigMapName,
+        INIT_CONTAINER_PROPERTIES_FILE_NAME)
+
+      Seq(bootstrapStep)
+    } else {
+      Nil
+    }
 
-    val mayBeMountSecretsStep = if (secretNamesToMountPaths.nonEmpty) {
-      Some(new DriverMountSecretsStep(new MountSecretsBootstrap(secretNamesToMountPaths)))
+    val mountSecretsStep = if (secretNamesToMountPaths.nonEmpty) {
+      Seq(new DriverMountSecretsStep(new MountSecretsBootstrap(secretNamesToMountPaths)))
     } else {
-      None
+      Nil
     }
 
     Seq(
       initialSubmissionStep,
       serviceBootstrapStep,
       kubernetesCredentialsStep) ++
-      maybeDependencyResolutionStep.toSeq ++
-      mayBeInitContainerBootstrapStep.toSeq ++
-      mayBeMountSecretsStep.toSeq
+      dependencyResolutionStep ++
+      initContainerBootstrapStep ++
+      mountSecretsStep
   }
 
   private def areAnyFilesNonContainerLocal(files: Seq[String]): Boolean = {

resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/submit/steps/BaseDriverConfigurationStep.scala → resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/submit/steps/BasicDriverConfigurationStep.scala

@@ -22,15 +22,15 @@ import io.fabric8.kubernetes.api.model.{ContainerBuilder, EnvVarBuilder, EnvVarS
 
 import org.apache.spark.{SparkConf, SparkException}
 import org.apache.spark.deploy.k8s.Config._
-import org.apache.spark.deploy.k8s.ConfigurationUtils
 import org.apache.spark.deploy.k8s.Constants._
+import org.apache.spark.deploy.k8s.KubernetesUtils
 import org.apache.spark.deploy.k8s.submit.KubernetesDriverSpec
 import org.apache.spark.internal.config.{DRIVER_CLASS_PATH, DRIVER_MEMORY, DRIVER_MEMORY_OVERHEAD}
 
 /**
  * Performs basic configuration for the driver pod.
  */
-private[spark] class BaseDriverConfigurationStep(
+private[spark] class BasicDriverConfigurationStep(
     kubernetesAppId: String,
     resourceNamePrefix: String,
     driverLabels: Map[String, String],
@@ -71,7 +71,7 @@ private[spark] class BaseDriverConfigurationStep(
         .build()
     }
 
-    val driverCustomAnnotations = ConfigurationUtils.parsePrefixedKeyValuePairs(
+    val driverCustomAnnotations = KubernetesUtils.parsePrefixedKeyValuePairs(
       sparkConf, KUBERNETES_DRIVER_ANNOTATION_PREFIX)
     require(!driverCustomAnnotations.contains(SPARK_APP_NAME_ANNOTATION),
       s"Annotation with key $SPARK_APP_NAME_ANNOTATION is not allowed as it is reserved for" +
@@ -87,7 +87,7 @@ private[spark] class BaseDriverConfigurationStep(
 
     val driverAnnotations = driverCustomAnnotations ++ Map(SPARK_APP_NAME_ANNOTATION -> appName)
 
-    val nodeSelector = ConfigurationUtils.parsePrefixedKeyValuePairs(
+    val nodeSelector = KubernetesUtils.parsePrefixedKeyValuePairs(
       sparkConf, KUBERNETES_NODE_SELECTOR_PREFIX)
 
     val driverCpuQuantity = new QuantityBuilder(false)

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

@@ -21,7 +21,8 @@ import java.io.File
 import io.fabric8.kubernetes.api.model.ContainerBuilder
 
 import org.apache.spark.deploy.k8s.Constants._
-import org.apache.spark.deploy.k8s.submit.{KubernetesDriverSpec, KubernetesFileUtils}
+import org.apache.spark.deploy.k8s.KubernetesUtils
+import org.apache.spark.deploy.k8s.submit.KubernetesDriverSpec
 
 /**
  * Step that configures the classpath, spark.jars, and spark.files for the driver given that the
@@ -34,8 +35,8 @@ private[spark] class DependencyResolutionStep(
     filesDownloadPath: String) extends DriverConfigurationStep {
 
   override def configureDriver(driverSpec: KubernetesDriverSpec): KubernetesDriverSpec = {
-    val resolvedSparkJars = KubernetesFileUtils.resolveFileUris(sparkJars, jarsDownloadPath)
-    val resolvedSparkFiles = KubernetesFileUtils.resolveFileUris(
+    val resolvedSparkJars = KubernetesUtils.resolveFileUris(sparkJars, jarsDownloadPath)
+    val resolvedSparkFiles = KubernetesUtils.resolveFileUris(
       sparkFiles, filesDownloadPath)
 
     val sparkConf = driverSpec.driverSparkConf.clone()
@@ -46,7 +47,7 @@ private[spark] class DependencyResolutionStep(
       sparkConf.set("spark.files", resolvedSparkFiles.mkString(","))
     }
 
-    val resolvedClasspath = KubernetesFileUtils.resolveFilePaths(sparkJars, jarsDownloadPath)
+    val resolvedClasspath = KubernetesUtils.resolveFilePaths(sparkJars, jarsDownloadPath)
     val resolvedDriverContainer = if (resolvedClasspath.nonEmpty) {
       new ContainerBuilder(driverSpec.driverContainer)
         .addNewEnv()