Create one task per jar/file to download in the init-container

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

@@ -125,38 +125,14 @@ private[spark] class DriverConfigOrchestrator(
 
     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,
-            kubernetesResourceNamePrefix,
-            sparkJars,
-            sparkFiles,
-            jarsDownloadPath,
-            filesDownloadPath,
-            imagePullPolicy,
-            allDriverLabels,
-            initContainerConfigMapName,
-            INIT_CONTAINER_PROPERTIES_FILE_NAME,
-            submissionSparkConf)
-        val initContainerConfigurationSteps =
-          initContainerConfigurationStepsOrchestrator.getAllConfigurationSteps()
-        val initContainerBootstrapStep =
-          new DriverInitContainerBootstrapStep(
-            initContainerConfigurationSteps,
-            initContainerConfigMapName,
-            INIT_CONTAINER_PROPERTIES_FILE_NAME)
-
-        Some(initContainerBootstrapStep)
-=======
         val orchestrator = new InitContainerConfigOrchestrator(
           namespace,
           kubernetesResourceNamePrefix,
           sparkJars,
           sparkFiles,
           jarsDownloadPath,
           filesDownloadPath,
-          dockerImagePullPolicy,
+          imagePullPolicy,
           allDriverLabels,
           initContainerConfigMapName,
           INIT_CONTAINER_PROPERTIES_FILE_NAME,
@@ -167,7 +143,6 @@ private[spark] class DriverConfigOrchestrator(
           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
       }

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

@@ -20,7 +20,6 @@ import java.io.File
 import java.util.concurrent.TimeUnit
 
 import scala.concurrent.{ExecutionContext, Future}
-import scala.concurrent.duration.Duration
 
 import org.apache.spark.{SecurityManager => SparkSecurityManager, SparkConf}
 import org.apache.spark.deploy.SparkHadoopUtil
@@ -54,26 +53,22 @@ private[spark] class SparkPodInitContainer(
   private val downloadTimeoutMinutes = sparkConf.get(INIT_CONTAINER_MOUNT_TIMEOUT)
 
   def run(): Unit = {
-    val remoteJarsDownload = Future[Unit] {
-      logInfo(s"Downloading remote jars: $remoteJars")
-      downloadFiles(
-        remoteJars,
-        jarsDownloadDir,
-        s"Remote jars download directory specified at $jarsDownloadDir does not exist " +
-          "or is not a directory.")
-    }
-    val remoteFilesDownload = Future[Unit] {
-      logInfo(s"Downloading remote files: $remoteFiles")
-      downloadFiles(
-        remoteFiles,
-        filesDownloadDir,
-        s"Remote files download directory specified at $filesDownloadDir does not exist " +
-          "or is not a directory.")
-    }
-
-    Seq(remoteJarsDownload, remoteFilesDownload).foreach {
-      ThreadUtils.awaitResult(_, Duration.create(downloadTimeoutMinutes, TimeUnit.MINUTES))
-    }
+    logInfo(s"Downloading remote jars: $remoteJars")
+    downloadFiles(
+      remoteJars,
+      jarsDownloadDir,
+      s"Remote jars download directory specified at $jarsDownloadDir does not exist " +
+        "or is not a directory.")
+
+    logInfo(s"Downloading remote files: $remoteFiles")
+    downloadFiles(
+      remoteFiles,
+      filesDownloadDir,
+      s"Remote files download directory specified at $filesDownloadDir does not exist " +
+        "or is not a directory.")
+
+    downloadExecutor.shutdown()
+    downloadExecutor.awaitTermination(downloadTimeoutMinutes, TimeUnit.MINUTES)
   }
 
   private def downloadFiles(
@@ -84,7 +79,9 @@ private[spark] class SparkPodInitContainer(
       require(downloadDir.isDirectory, errMessageOnDestinationNotADirectory)
     }
     filesCommaSeparated.map(_.split(",")).toSeq.flatten.foreach { file =>
-      fileFetcher.fetchFile(file, downloadDir)
+      Future[Unit] {
+        fileFetcher.fetchFile(file, downloadDir)
+      }
     }
   }
 }

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

@@ -24,7 +24,7 @@ class DriverConfigOrchestratorSuite extends SparkFunSuite {
 
   private val NAMESPACE = "default"
   private val DRIVER_IMAGE = "driver-image"
-  private val INIT_CONTAINER_IMAGE = "init-container-image"
+  private val IC_IMAGE = "init-container-image"
   private val APP_ID = "spark-app-id"
   private val LAUNCH_TIME = 975256L
   private val APP_NAME = "spark"
@@ -58,13 +58,8 @@ class DriverConfigOrchestratorSuite 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,
@@ -83,13 +78,8 @@ class DriverConfigOrchestratorSuite 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(INIT_CONTAINER_IMAGE, IC_IMAGE)
       .set("spark.jars", "hdfs://localhost:9000/var/apps/jars/jar1.jar")
     val mainAppResource = JavaMainAppResource("local:///var/apps/jars/main.jar")
     val orchestrator = new DriverConfigOrchestrator(

resource-managers/kubernetes/core/src/test/scala/org/apache/spark/deploy/rest/k8s/SparkPodInitContainerSuite.scala

@@ -68,7 +68,6 @@ class SparkPodInitContainerSuite
     Mockito.verify(fileFetcher).fetchFile("http://localhost:9000/jar1.jar", downloadJarsDir)
     Mockito.verify(fileFetcher).fetchFile("hdfs://localhost:9000/jar2.jar", downloadJarsDir)
     Mockito.verify(fileFetcher).fetchFile("http://localhost:9000/file.txt", downloadFilesDir)
-
   }
 
   private def getSparkConfForRemoteFileDownloads: SparkConf = {