Cleanups.

Mostly minor, except for ClientBase.scala where I cleaned up the
generated classpath to only include the necessary entries.

Author: Marcelo Vanzin

core/src/main/scala/org/apache/spark/SparkConf.scala

@@ -360,8 +360,8 @@ private[spark] object SparkConf extends Logging {
     val configs = Seq(
       DeprecatedConfig("spark.files.userClassPathFirst", "spark.executor.userClassPathFirst",
         "1.3"),
-      DeprecatedConfig("spark.yarn.user.classpath.first", null, "1.3.",
-        "Use spark.{driver,executor}.userClassPathFirst; this option will be removed the future."))
+      DeprecatedConfig("spark.yarn.user.classpath.first", null, "1.3",
+        "Use spark.{driver,executor}.userClassPathFirst instead."))
     configs.map { x => (x.oldName, x) }.toMap
   }
 
@@ -435,12 +435,12 @@ private[spark] object SparkConf extends Logging {
           val message = Option(deprecationMessage).getOrElse(
             s"Please use the alternative '$newName' instead.")
           logWarning(
-            s"The configuration option '$oldName' has been replaced as of Spark $version. " +
-            message)
+            s"The configuration option '$oldName' has been replaced as of Spark $version and " +
+            s"may be removed in the future. $message")
         } else {
           logWarning(
-            s"The configuration option '$oldName' has been deprecated as of Spark $version. " +
-            deprecationMessage)
+            s"The configuration option '$oldName' has been deprecated as of Spark $version and " +
+            s"may be removed in the future. $deprecationMessage")
         }
       }
     }

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

@@ -321,7 +321,7 @@ object SparkSubmit {
     }
 
     val loader =
-      if (sysProps.get("spark.driver.userClassPathFirst").getOrElse("false").toBoolean) {
+      if (sysProps.getOrElse("spark.driver.userClassPathFirst", "false").toBoolean) {
         new ChildExecutorURLClassLoader(new Array[URL](0),
           Thread.currentThread.getContextClassLoader)
       } else {

core/src/main/scala/org/apache/spark/deploy/worker/DriverRunner.scala

@@ -74,7 +74,6 @@ private[spark] class DriverRunner(
           val driverDir = createWorkingDirectory()
           val localJarFilename = downloadUserJar(driverDir)
 
-          /** Replace variables in a command argument passed to us */
           def substituteVariables(argument: String): String = argument match {
             case "{{WORKER_URL}}" => workerUrl
             case "{{USER_JAR}}" => localJarFilename

core/src/main/scala/org/apache/spark/deploy/worker/DriverWrapper.scala

@@ -18,7 +18,6 @@
 package org.apache.spark.deploy.worker
 
 import java.io.File
-import java.net.URL
 
 import akka.actor._
 

core/src/main/scala/org/apache/spark/executor/CoarseGrainedExecutorBackend.scala

@@ -211,7 +211,7 @@ private[spark] object CoarseGrainedExecutorBackend extends Logging {
       |   --hostname <hostname>
       |   --cores <cores>
       |   --app-id <appid>
-      |   [--worker-id <workerUrl>]
+      |   [--worker-url <workerUrl>]
       |   [--user-class-path <url>]
       |""".stripMargin)
     System.exit(1)

core/src/test/scala/org/apache/spark/SparkConfSuite.scala

@@ -185,12 +185,16 @@ class SparkConfSuite extends FunSuite with LocalSparkContext {
     serializer.newInstance().serialize(new StringBuffer())
   }
 
-  test("deprecated key renaming") {
-    val conf = new SparkConf().set("spark.files.userClassPathFirst", "true")
+  test("deprecated config keys") {
+    val conf = new SparkConf()
+      .set("spark.files.userClassPathFirst", "true")
+      .set("spark.yarn.user.classpath.first", "true")
     assert(conf.contains("spark.files.userClassPathFirst"))
     assert(conf.contains("spark.executor.userClassPathFirst"))
+    assert(conf.contains("spark.yarn.user.classpath.first"))
     assert(conf.getBoolean("spark.files.userClassPathFirst", false))
     assert(conf.getBoolean("spark.executor.userClassPathFirst", false))
+    assert(conf.getBoolean("spark.yarn.user.classpath.first", false))
   }
 
 }

core/src/test/scala/org/apache/spark/deploy/SparkSubmitSuite.scala

@@ -417,7 +417,7 @@ class SparkSubmitSuite extends FunSuite with Matchers {
     val systemJar = TestUtils.createJarWithFiles(Map("test.resource" -> "SYSTEM"))
     val userJar = TestUtils.createJarWithFiles(Map("test.resource" -> "USER"))
     val args = Seq(
-      "--class", ClassPathIsolationTest.getClass.getName.stripSuffix("$"),
+      "--class", UserClasspathFirstTest.getClass.getName.stripSuffix("$"),
       "--name", "testApp",
       "--master", "local",
       "--conf", "spark.driver.extraClassPath=" + systemJar,
@@ -514,7 +514,7 @@ object SimpleApplicationTest {
   }
 }
 
-object ClassPathIsolationTest {
+object UserClasspathFirstTest {
   def main(args: Array[String]) {
     val ccl = Thread.currentThread().getContextClassLoader()
     val resource = ccl.getResourceAsStream("test.resource")

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

@@ -21,7 +21,7 @@ import scala.util.control.NonFatal
 
 import java.io.{File, IOException}
 import java.lang.reflect.InvocationTargetException
-import java.net.{Socket, URI, URL}
+import java.net.{Socket, URL}
 import java.util.concurrent.atomic.AtomicReference
 
 import akka.actor._
@@ -452,13 +452,7 @@ private[spark] class ApplicationMaster(args: ApplicationMasterArguments,
 
     val classpath = ClientBase.getUserClasspath(null, sparkConf)
     val urls = classpath.map { entry =>
-      val absPath =
-        if (new File(entry.getPath()).isAbsolute()) {
-          entry.getPath()
-        } else {
-          new File(entry.getPath()).getAbsolutePath()
-        }
-      new URL("file:" + absPath)
+      new URL("file:" + new File(entry.getPath()).getAbsolutePath())
     }
     val userClassLoader =
       if (ClientBase.isUserClassPathFirst(sparkConf, true)) {

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

@@ -17,7 +17,6 @@
 
 package org.apache.spark.deploy.yarn
 
-import java.io.File
 import java.net.{InetAddress, UnknownHostException, URI}
 
 import scala.collection.JavaConversions._
@@ -161,17 +160,16 @@ private[spark] trait ClientBase extends Logging {
     /**
      * Copy the given main resource to the distributed cache if the scheme is not "local".
      * Otherwise, set the corresponding key in our SparkConf to handle it downstream.
-     * Each resource is represented by a 4-tuple of:
+     * Each resource is represented by a 3-tuple of:
      *   (1) destination resource name,
      *   (2) local path to the resource,
      *   (3) Spark property key to set if the scheme is not local, and
-     *   (4) whether to set permissions for this resource
      */
     List(
-      (SPARK_JAR, sparkJar(sparkConf), CONF_SPARK_JAR, false),
-      (APP_JAR, args.userJar, CONF_SPARK_USER_JAR, true),
-      ("log4j.properties", oldLog4jConf.orNull, null, true)
-    ).foreach { case (destName, _localPath, confKey, isUserArtifact) =>
+      (SPARK_JAR, sparkJar(sparkConf), CONF_SPARK_JAR),
+      (APP_JAR, args.userJar, CONF_SPARK_USER_JAR),
+      ("log4j.properties", oldLog4jConf.orNull, null)
+    ).foreach { case (destName, _localPath, confKey) =>
       val localPath: String = if (_localPath != null) _localPath.trim() else ""
       if (!localPath.isEmpty()) {
         val localURI = new URI(localPath)
@@ -661,9 +659,10 @@ private[spark] object ClientBase extends Logging {
   /**
    * Populate the classpath entry in the given environment map.
    *
-   * Class path isolation, when enabled, makes the user-added jars be loaded from a different
-   * class loader than other class path entries. The extra class path and other uploaded files
-   * are still made available through the system class path.
+   * User jars are generally not added to the JVM's classpath; those are handled by the AM and
+   * executor backend. When the deprecated `spark.yarn.user.classpath.first` is used, user jars are
+   * included in the system classpath, though. The extra class path and other uploaded files are
+   * always made available through the system class path.
    *
    * @param args Client arguments (when starting the AM) or null (when starting executors).
    */
@@ -675,22 +674,13 @@ private[spark] object ClientBase extends Logging {
       extraClassPath: Option[String] = None): Unit = {
     extraClassPath.foreach(addClasspathEntry(_, env))
     addClasspathEntry(Environment.PWD.$(), env)
-    if (isUserClassPathFirst(sparkConf, args == null)) {
-      addFileToClasspath(new URI(sparkJar(sparkConf)), SPARK_JAR, env)
-      populateHadoopClasspath(conf, env)
-    } else {
-      // Normally the users app.jar is last in case it conflicts with spark jars.
-      if (sparkConf.getBoolean("spark.yarn.user.classpath.first", false)) {
-        getUserClasspath(args, sparkConf).foreach { x =>
-          addFileToClasspath(x, null, env)
-        }
+    if (sparkConf.getBoolean("spark.yarn.user.classpath.first", false)) {
+      getUserClasspath(args, sparkConf).foreach { x =>
+        addFileToClasspath(x, null, env)
       }
-      addFileToClasspath(new URI(sparkJar(sparkConf)), SPARK_JAR, env)
-      populateHadoopClasspath(conf, env)
-
-      // Append all jar files under the directory holding cached files to the classpath.
-      addClasspathEntry(buildPath(Environment.PWD.$(), "*"), env)
     }
+    addFileToClasspath(new URI(sparkJar(sparkConf)), SPARK_JAR, env)
+    populateHadoopClasspath(conf, env)
   }
 
   /**

yarn/common/src/test/scala/org/apache/spark/deploy/yarn/ClientBaseSuite.scala

@@ -100,7 +100,6 @@ class ClientBaseSuite extends FunSuite with Matchers {
       }
     })
     cp should contain (Environment.PWD.$())
-    cp should contain (s"${Environment.PWD.$()}${File.separator}*")
     cp should not contain (ClientBase.SPARK_JAR)
     cp should not contain (ClientBase.APP_JAR)
   }