Addressed the second round of comments

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

@@ -145,9 +145,9 @@ private[spark] object Config extends Logging {
       .stringConf
       .createWithDefault("/var/spark-data/spark-files")
 
-  val INIT_CONTAINER_DOCKER_IMAGE =
-    ConfigBuilder("spark.kubernetes.initContainer.docker.image")
-      .doc("Image for the driver and executor's init-container that downloads dependencies.")
+  val INIT_CONTAINER_IMAGE =
+    ConfigBuilder("spark.kubernetes.initContainer.image")
+      .doc("Image for the driver and executor's init-container for downloading dependencies.")
       .stringConf
       .createOptional
 

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

@@ -68,11 +68,12 @@ private[spark] class InitContainerBootstrapImpl(
     }
     val initContainerCustomEnvVars = sparkConf.getAllWithPrefix(initContainerCustomEnvVarKeyPrefix)
       .toSeq
-      .map(env =>
+      .map { env =>
         new EnvVarBuilder()
           .withName(env._1)
           .withValue(env._2)
-          .build())
+          .build()
+      }
 
     val initContainer = new ContainerBuilder(podWithDetachedInitContainer.initContainer)
       .withName("spark-init")

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

@@ -38,7 +38,7 @@ private[spark] class MountSecretsBootstrapImpl(
 
   override def mountSecrets(pod: Pod, container: Container): (Pod, Container) = {
     var podBuilder = new PodBuilder(pod)
-    secretNamesToMountPaths.keys.foreach(name =>
+    secretNamesToMountPaths.keys.foreach { name =>
       podBuilder = podBuilder
         .editOrNewSpec()
           .addNewVolume()
@@ -47,16 +47,17 @@ private[spark] class MountSecretsBootstrapImpl(
             .withSecretName(name)
             .endSecret()
           .endVolume()
-          .endSpec())
+          .endSpec()
+    }
 
     var containerBuilder = new ContainerBuilder(container)
-    secretNamesToMountPaths.foreach(namePath =>
+    secretNamesToMountPaths.foreach { namePath =>
       containerBuilder = containerBuilder
         .addNewVolumeMount()
           .withName(secretVolumeName(namePath._1))
           .withMountPath(namePath._2)
           .endVolumeMount()
-    )
+    }
 
     (podBuilder.build(), containerBuilder.build())
   }

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

@@ -25,15 +25,15 @@ import org.apache.spark.deploy.k8s.{ConfigurationUtils, MountSecretsBootstrapImp
 import org.apache.spark.deploy.k8s.Config._
 import org.apache.spark.deploy.k8s.Constants._
 import org.apache.spark.deploy.k8s.submit.steps._
-import org.apache.spark.deploy.k8s.submit.steps.initcontainer.InitContainerConfigurationStepsOrchestrator
+import org.apache.spark.deploy.k8s.submit.steps.initcontainer.InitContainerConfigOrchestrator
 import org.apache.spark.launcher.SparkLauncher
 import org.apache.spark.util.SystemClock
 import org.apache.spark.util.Utils
 
 /**
  * Constructs the complete list of driver configuration steps to run to deploy the Spark driver.
  */
-private[spark] class DriverConfigurationStepsOrchestrator(
+private[spark] class DriverConfigOrchestrator(
     namespace: String,
     kubernetesAppId: String,
     launchTime: Long,
@@ -125,6 +125,7 @@ private[spark] class DriverConfigurationStepsOrchestrator(
 
     val mayBeInitContainerBootstrapStep =
       if (areAnyFilesNonContainerLocal(sparkJars ++ sparkFiles)) {
+<<<<<<< HEAD:resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/submit/DriverConfigurationStepsOrchestrator.scala
         val initContainerConfigurationStepsOrchestrator =
           new InitContainerConfigurationStepsOrchestrator(
             namespace,
@@ -147,6 +148,26 @@ private[spark] class DriverConfigurationStepsOrchestrator(
             INIT_CONTAINER_PROPERTIES_FILE_NAME)
 
         Some(initContainerBootstrapStep)
+=======
+        val orchestrator = new InitContainerConfigOrchestrator(
+          namespace,
+          kubernetesResourceNamePrefix,
+          sparkJars,
+          sparkFiles,
+          jarsDownloadPath,
+          filesDownloadPath,
+          dockerImagePullPolicy,
+          allDriverLabels,
+          initContainerConfigMapName,
+          INIT_CONTAINER_PROPERTIES_FILE_NAME,
+          submissionSparkConf)
+        val bootstrapStep = new DriverInitContainerBootstrapStep(
+          orchestrator.getAllConfigurationSteps(),
+          initContainerConfigMapName,
+          INIT_CONTAINER_PROPERTIES_FILE_NAME)
+
+        Some(bootstrapStep)
+>>>>>>> Addressed the second round of comments:resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/submit/DriverConfigOrchestrator.scala
       } else {
         None
       }
@@ -169,7 +190,7 @@ private[spark] class DriverConfigurationStepsOrchestrator(
 
   private def areAnyFilesNonContainerLocal(files: Seq[String]): Boolean = {
     files.exists { uri =>
-      Option(Utils.resolveURI(uri).getScheme).getOrElse("file") != "local"
+      Utils.resolveURI(uri).getScheme != "local"
     }
   }
 }

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

@@ -210,7 +210,7 @@ private[spark] class KubernetesClientApplication extends SparkApplication {
     val loggingPodStatusWatcher = new LoggingPodStatusWatcherImpl(
       kubernetesAppId, loggingInterval)
 
-    val configurationStepsOrchestrator = new DriverConfigurationStepsOrchestrator(
+    val configurationStepsOrchestrator = new DriverConfigOrchestrator(
       namespace,
       kubernetesAppId,
       launchTime,

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

@@ -51,12 +51,10 @@ private[spark] object KubernetesFileUtils {
    * Get from a given collection of file URIs the ones that represent remote files.
    */
   def getOnlyRemoteFiles(uris: Iterable[String]): Iterable[String] = {
-    filterUriStringsByScheme(uris, scheme => scheme != "file" && scheme != "local")
-  }
-
-  private def filterUriStringsByScheme(
-      uris: Iterable[String], schemeFilter: (String => Boolean)): Iterable[String] = {
-    uris.filter(uri => schemeFilter(Option(Utils.resolveURI(uri).getScheme).getOrElse("file")))
+    uris.filter { uri =>
+      val scheme = Utils.resolveURI(uri).getScheme
+      scheme != "file" && scheme != "local"
+    }
   }
 
   private def resolveFileUri(

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

@@ -76,7 +76,8 @@ private[spark] class DriverInitContainerBootstrapStep(
       config: Map[String, String]): ConfigMap = {
     val properties = new Properties()
     config.foreach { entry =>
-      properties.setProperty(entry._1, entry._2) }
+      properties.setProperty(entry._1, entry._2)
+    }
     val propertiesWriter = new StringWriter()
     properties.store(propertiesWriter,
       s"Java properties built from Kubernetes config map with name: $configMapName " +

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

@@ -34,15 +34,15 @@ private[spark] class BaseInitContainerConfigurationStep(
     val remoteJarsConf = if (remoteJarsToDownload.nonEmpty) {
       Map(INIT_CONTAINER_REMOTE_JARS.key -> remoteJarsToDownload.mkString(","))
     } else {
-      Map.empty[String, String]
+      Map()
     }
     val remoteFilesConf = if (remoteFilesToDownload.nonEmpty) {
       Map(INIT_CONTAINER_REMOTE_FILES.key -> remoteFilesToDownload.mkString(","))
     } else {
-      Map.empty[String, String]
+      Map()
     }
 
-    val baseInitContainerConfig = Map[String, String](
+    val baseInitContainerConfig = Map(
       JARS_DOWNLOAD_LOCATION.key -> jarsDownloadPath,
       FILES_DOWNLOAD_LOCATION.key -> filesDownloadPath) ++
       remoteJarsConf ++

resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/submit/steps/initcontainer/InitContainerConfigurationStepsOrchestrator.scala → resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/submit/steps/initcontainer/InitContainerConfigOrchestrator.scala

@@ -25,7 +25,7 @@ import org.apache.spark.deploy.k8s.Constants._
  * Returns the complete ordered list of steps required to configure the init-container. This is
  * only used when there are remote application dependencies to localize.
  */
-private[spark] class InitContainerConfigurationStepsOrchestrator(
+private[spark] class InitContainerConfigOrchestrator(
     namespace: String,
     kubernetesResourceNamePrefix: String,
     sparkJars: Seq[String],
@@ -39,9 +39,9 @@ private[spark] class InitContainerConfigurationStepsOrchestrator(
     submissionSparkConf: SparkConf) {
 
   private val initContainerImage = submissionSparkConf
-    .get(INIT_CONTAINER_DOCKER_IMAGE)
+    .get(INIT_CONTAINER_IMAGE)
     .getOrElse(throw new SparkException(
-      "Must specify the init-container Docker image when there are remote dependencies"))
+      "Must specify the init-container image when there are remote dependencies"))
   private val downloadTimeoutMinutes = submissionSparkConf.get(INIT_CONTAINER_MOUNT_TIMEOUT)
 
   def getAllConfigurationSteps(): Seq[InitContainerConfigurationStep] = {

resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/rest/k8s/KubernetesSparkDependencyDownloadInitContainer.scala → resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/rest/k8s/SparkPodInitContainer.scala

@@ -36,7 +36,7 @@ import org.apache.spark.util.{ThreadUtils, Utils}
  * with different configurations for different download sources, or using the same container to
  * download everything at once.
  */
-private[spark] class KubernetesSparkDependencyDownloadInitContainer(
+private[spark] class SparkPodInitContainer(
     sparkConf: SparkConf,
     fileFetcher: FileFetcher) extends Logging {
 
@@ -70,9 +70,10 @@ private[spark] class KubernetesSparkDependencyDownloadInitContainer(
         s"Remote files download directory specified at $filesDownloadDir does not exist " +
           "or is not a directory.")
     }
-    waitForFutures(
-      remoteJarsDownload,
-      remoteFilesDownload)
+
+    Seq(remoteJarsDownload, remoteFilesDownload).foreach {
+      ThreadUtils.awaitResult(_, Duration.create(downloadTimeoutMinutes, TimeUnit.MINUTES))
+    }
   }
 
   private def downloadFiles(
@@ -86,16 +87,9 @@ private[spark] class KubernetesSparkDependencyDownloadInitContainer(
       fileFetcher.fetchFile(file, downloadDir)
     }
   }
-
-  private def waitForFutures(futures: Future[_]*) {
-    futures.foreach {
-      ThreadUtils.awaitResult(_, Duration.create(downloadTimeoutMinutes, TimeUnit.MINUTES))
-    }
-  }
 }
 
-private class FileFetcherImpl(sparkConf: SparkConf, securityManager: SparkSecurityManager)
-  extends FileFetcher {
+private class FileFetcher(sparkConf: SparkConf, securityManager: SparkSecurityManager) {
 
   def fetchFile(uri: String, targetDir: File): Unit = {
     Utils.fetchFile(
@@ -109,7 +103,7 @@ private class FileFetcherImpl(sparkConf: SparkConf, securityManager: SparkSecuri
   }
 }
 
-object KubernetesSparkDependencyDownloadInitContainer extends Logging {
+object SparkPodInitContainer extends Logging {
 
   def main(args: Array[String]): Unit = {
     logInfo("Starting init-container to download Spark application dependencies.")
@@ -120,10 +114,8 @@ object KubernetesSparkDependencyDownloadInitContainer extends Logging {
     }
 
     val securityManager = new SparkSecurityManager(sparkConf)
-    val fileFetcher = new FileFetcherImpl(sparkConf, securityManager)
-    new KubernetesSparkDependencyDownloadInitContainer(
-      sparkConf,
-      fileFetcher).run()
+    val fileFetcher = new FileFetcher(sparkConf, securityManager)
+    new SparkPodInitContainer(sparkConf, fileFetcher).run()
     logInfo("Finished downloading application dependencies.")
   }
 }

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

@@ -66,9 +66,9 @@ private[spark] class KubernetesClusterManager extends ExternalClusterManager wit
       configMapKey <- maybeInitContainerConfigMapKey
     } yield {
       val initContainerImage = sparkConf
-        .get(INIT_CONTAINER_DOCKER_IMAGE)
+        .get(INIT_CONTAINER_IMAGE)
         .getOrElse(throw new SparkException(
-          "Must specify the init-container Docker image when there are remote dependencies"))
+          "Must specify the init-container image when there are remote dependencies"))
       new InitContainerBootstrapImpl(
         initContainerImage,
         sparkConf.get(CONTAINER_IMAGE_PULL_POLICY),

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

@@ -20,7 +20,7 @@ import org.apache.spark.{SparkConf, SparkFunSuite}
 import org.apache.spark.deploy.k8s.Config._
 import org.apache.spark.deploy.k8s.submit.steps._
 
-class DriverConfigurationStepsOrchestratorSuite extends SparkFunSuite {
+class DriverConfigOrchestratorSuite extends SparkFunSuite {
 
   private val NAMESPACE = "default"
   private val DRIVER_IMAGE = "driver-image"
@@ -38,7 +38,7 @@ class DriverConfigurationStepsOrchestratorSuite extends SparkFunSuite {
     val sparkConf = new SparkConf(false)
       .set(DRIVER_CONTAINER_IMAGE, DRIVER_IMAGE)
     val mainAppResource = JavaMainAppResource("local:///var/apps/jars/main.jar")
-    val orchestrator = new DriverConfigurationStepsOrchestrator(
+    val orchestrator = new DriverConfigOrchestrator(
       NAMESPACE,
       APP_ID,
       LAUNCH_TIME,
@@ -58,8 +58,13 @@ class DriverConfigurationStepsOrchestratorSuite extends SparkFunSuite {
 
   test("Base submission steps without a main app resource.") {
     val sparkConf = new SparkConf(false)
+<<<<<<< HEAD:resource-managers/kubernetes/core/src/test/scala/org/apache/spark/deploy/k8s/submit/DriverConfigurationStepsOrchestratorSuite.scala
       .set(DRIVER_CONTAINER_IMAGE, DRIVER_IMAGE)
     val orchestrator = new DriverConfigurationStepsOrchestrator(
+=======
+      .set(DRIVER_DOCKER_IMAGE, DRIVER_IMAGE)
+    val orchestrator = new DriverConfigOrchestrator(
+>>>>>>> Addressed the second round of comments:resource-managers/kubernetes/core/src/test/scala/org/apache/spark/deploy/k8s/submit/DriverConfigOrchestratorSuite.scala
       NAMESPACE,
       APP_ID,
       LAUNCH_TIME,
@@ -78,11 +83,16 @@ class DriverConfigurationStepsOrchestratorSuite extends SparkFunSuite {
 
   test("Submission steps with an init-container.") {
     val sparkConf = new SparkConf(false)
+<<<<<<< HEAD:resource-managers/kubernetes/core/src/test/scala/org/apache/spark/deploy/k8s/submit/DriverConfigurationStepsOrchestratorSuite.scala
       .set(DRIVER_CONTAINER_IMAGE, DRIVER_IMAGE)
       .set(INIT_CONTAINER_DOCKER_IMAGE, INIT_CONTAINER_IMAGE)
+=======
+      .set(DRIVER_DOCKER_IMAGE, DRIVER_IMAGE)
+      .set(INIT_CONTAINER_IMAGE, INIT_CONTAINER_IMAGE)
+>>>>>>> Addressed the second round of comments:resource-managers/kubernetes/core/src/test/scala/org/apache/spark/deploy/k8s/submit/DriverConfigOrchestratorSuite.scala
       .set("spark.jars", "hdfs://localhost:9000/var/apps/jars/jar1.jar")
     val mainAppResource = JavaMainAppResource("local:///var/apps/jars/main.jar")
-    val orchestrator = new DriverConfigurationStepsOrchestrator(
+    val orchestrator = new DriverConfigOrchestrator(
       NAMESPACE,
       APP_ID,
       LAUNCH_TIME,
@@ -106,7 +116,7 @@ class DriverConfigurationStepsOrchestratorSuite extends SparkFunSuite {
       .set(s"$KUBERNETES_DRIVER_SECRETS_PREFIX$SECRET_FOO", SECRET_MOUNT_PATH)
       .set(s"$KUBERNETES_DRIVER_SECRETS_PREFIX$SECRET_BAR", SECRET_MOUNT_PATH)
     val mainAppResource = JavaMainAppResource("local:///var/apps/jars/main.jar")
-    val orchestrator = new DriverConfigurationStepsOrchestrator(
+    val orchestrator = new DriverConfigOrchestrator(
       NAMESPACE,
       APP_ID,
       LAUNCH_TIME,
@@ -125,8 +135,8 @@ class DriverConfigurationStepsOrchestratorSuite extends SparkFunSuite {
   }
 
   private def validateStepTypes(
-      orchestrator: DriverConfigurationStepsOrchestrator,
-      types: Class[_ <: DriverConfigurationStep]*): Unit = {
+                                 orchestrator: DriverConfigOrchestrator,
+                                 types: Class[_ <: DriverConfigurationStep]*): Unit = {
     val steps = orchestrator.getAllConfigurationSteps()
     assert(steps.size === types.size)
     assert(steps.map(_.getClass) === types)