From e103225d9ff54ca17692279cc6a7999f9b8c3265 Mon Sep 17 00:00:00 2001
From: mcheah <mcheah@palantir.com>
Date: Thu, 29 Jun 2017 17:13:53 -0700
Subject: [PATCH] Don't add the init-container step if all URIs are local.

---
 ...ubernetesSubmissionStepsOrchestrator.scala | 50 +++++++++++--------
 1 file changed, 29 insertions(+), 21 deletions(-)

diff --git a/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/kubernetes/submit/KubernetesSubmissionStepsOrchestrator.scala b/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/kubernetes/submit/KubernetesSubmissionStepsOrchestrator.scala
index f42afafbcb41a..b0c98b2575607 100644
--- a/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/kubernetes/submit/KubernetesSubmissionStepsOrchestrator.scala
+++ b/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/kubernetes/submit/KubernetesSubmissionStepsOrchestrator.scala
@@ -23,6 +23,7 @@ import org.apache.spark.deploy.kubernetes.constants._
 import org.apache.spark.deploy.kubernetes.submit.submitsteps.{BaseSubmissionStep, DependencyResolutionStep, DriverKubernetesCredentialsStep, InitContainerBootstrapStep, KubernetesSubmissionStep, PythonStep}
 import org.apache.spark.deploy.kubernetes.submit.submitsteps.initcontainer.InitContainerStepsOrchestrator
 import org.apache.spark.launcher.SparkLauncher
+import org.apache.spark.util.Utils
 
 /**
  * Constructs the complete list of submission steps to run to deploy the Spark application.
@@ -93,26 +94,32 @@ private[spark] class KubernetesSubmissionStepsOrchestrator(
     val pythonStep = mainAppResource match {
       case PythonMainAppResource(mainPyResource) =>
         Option(new PythonStep(mainPyResource, additionalPythonFiles, filesDownloadPath))
-      case _ => Option.empty[PythonStep]
+      case _ => Option.empty[KubernetesSubmissionStep]
+    }
+    val initContainerBootstrapStep = if ((sparkJars ++ sparkFiles).exists { uri =>
+      Option(Utils.resolveURI(uri).getScheme).getOrElse("file") != "local"
+    }) {
+      val initContainerStepsOrchestrator = new InitContainerStepsOrchestrator(
+        namespace,
+        kubernetesResourceNamePrefix,
+        sparkJars,
+        sparkFiles,
+        jarsDownloadPath,
+        filesDownloadPath,
+        dockerImagePullPolicy,
+        allDriverLabels,
+        initContainerConfigMapName,
+        INIT_CONTAINER_CONFIG_MAP_KEY,
+        submissionSparkConf)
+      val initContainerSteps = initContainerStepsOrchestrator.getInitContainerSteps()
+      Some(new InitContainerBootstrapStep(
+        submissionSparkConf,
+        initContainerSteps,
+        initContainerConfigMapName,
+        INIT_CONTAINER_CONFIG_MAP_KEY))
+    } else {
+      Option.empty[KubernetesSubmissionStep]
     }
-    val initContainerStepsOrchestrator = new InitContainerStepsOrchestrator(
-      namespace,
-      kubernetesResourceNamePrefix,
-      sparkJars,
-      sparkFiles,
-      jarsDownloadPath,
-      filesDownloadPath,
-      dockerImagePullPolicy,
-      allDriverLabels,
-      initContainerConfigMapName,
-      INIT_CONTAINER_CONFIG_MAP_KEY,
-      submissionSparkConf)
-    val initContainerSteps = initContainerStepsOrchestrator.getInitContainerSteps()
-    val initContainerBootstrapStep = new InitContainerBootstrapStep(
-      submissionSparkConf,
-      initContainerSteps,
-      initContainerConfigMapName,
-      INIT_CONTAINER_CONFIG_MAP_KEY)
     val dependencyResolutionStep = new DependencyResolutionStep(
       sparkJars,
       sparkFiles,
@@ -121,7 +128,8 @@ private[spark] class KubernetesSubmissionStepsOrchestrator(
     Seq(
       initialSubmissionStep,
       kubernetesCredentialsStep,
-      initContainerBootstrapStep,
-      dependencyResolutionStep) ++ pythonStep.toSeq
+      dependencyResolutionStep) ++
+      initContainerBootstrapStep.toSeq ++
+      pythonStep.toSeq
   }
 }