review commit

Author: witgo

core/src/main/scala/org/apache/spark/deploy/SparkSubmitArguments.scala

@@ -104,7 +104,8 @@ private[spark] class SparkSubmitArguments(args: Seq[String]) {
     totalExecutorCores = Option(totalExecutorCores)
       .getOrElse(defaultProperties.get("spark.cores.max").orNull)
     name = Option(name).getOrElse(defaultProperties.get("spark.app.name").orNull)
-    jars = Option(jars).getOrElse(defaultProperties.get("spark.jars").orNull)
+    jars = Option(jars).getOrElse(defaultProperties.get("spark.jars").
+      map(p => Utils.resolveURIs(p)).orNull)
 
     // This supports env vars in older versions of Spark
     master = Option(master).getOrElse(System.getenv("MASTER"))
@@ -131,6 +132,13 @@ private[spark] class SparkSubmitArguments(args: Seq[String]) {
     if (name == null && primaryResource != null) {
       name = Utils.stripDirectory(primaryResource)
     }
+
+    if (master.startsWith("yarn")) {
+      archives = Option(archives).getOrElse(defaultProperties
+        .get("spark.yarn.dist.archives").map(p => Utils.resolveURIs(p)).orNull)
+      files = Option(files).getOrElse(defaultProperties
+        .get("spark.yarn.dist.files").map(p => Utils.resolveURIs(p)).orNull)
+    }
   }
 
   /** Ensure that required fields exists. Call this only once all defaults are loaded. */

yarn/common/src/main/scala/org/apache/spark/deploy/yarn/ClientBase.scala

@@ -219,21 +219,10 @@ trait ClientBase extends Logging {
       }
     }
 
-    def getArg(arg: String, envVar: String, sysProp: String): String = {
-      if (arg != null && !arg.isEmpty) {
-        arg
-      } else if (System.getenv(envVar) != null && !System.getenv(envVar).isEmpty) {
-        System.getenv(envVar)
-      } else {
-        sparkConf.getOption(sysProp).orNull
-      }
-    }
-    var cachedSecondaryJarLinks = ListBuffer.empty[String]
-    val fileLists = List((args.addJars, LocalResourceType.FILE, true),
-      (getArg(args.files, "SPARK_YARN_DIST_FILES", "spark.yarn.dist.files"),
-        LocalResourceType.FILE, false),
-      (getArg(args.archives, "SPARK_YARN_DIST_ARCHIVES", "spark.yarn.dist.archives"),
-        LocalResourceType.ARCHIVE, false))
+    val cachedSecondaryJarLinks = ListBuffer.empty[String]
+    val fileLists = List( (args.addJars, LocalResourceType.FILE, true),
+      (args.files, LocalResourceType.FILE, false),
+      (args.archives, LocalResourceType.ARCHIVE, false) )
     fileLists.foreach { case (flist, resType, addToClasspath) =>
       if (flist != null && !flist.isEmpty()) {
         flist.split(',').foreach { case file: String =>
@@ -251,6 +240,7 @@ trait ClientBase extends Logging {
         }
       }
     }
+    logInfo("Prepared Local resources " + localResources)
     sparkConf.set(ClientBase.CONF_SPARK_YARN_SECONDARY_JARS, cachedSecondaryJarLinks.mkString(","))
 
     UserGroupInformation.getCurrentUser().addCredentials(credentials)