Josh's comments

Author: sryza

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

@@ -1018,7 +1018,6 @@ class SparkContext(config: SparkConf) extends Logging with ExecutorAllocationCli
    * supported for Hadoop-supported filesystems.
    */
   def addFile(path: String, recursive: Boolean): Unit = {
-    val isLocalMode = conf.get("spark.master").startsWith("local")
     val uri = new URI(path)
     val schemeCorrectedPath = uri.getScheme match {
       case null | "local" => "file:" + uri.getPath
@@ -1030,10 +1029,10 @@ class SparkContext(config: SparkConf) extends Logging with ExecutorAllocationCli
     if (!Array("http", "https", "ftp").contains(scheme)) {
       val fs = hadoopPath.getFileSystem(hadoopConfiguration)
       if (!fs.exists(hadoopPath)) {
-        throw new SparkException(s"Added file $hadoopPath does not exist.")
+        throw new FileNotFoundException(s"Added file $hadoopPath does not exist.")
       }
       val isDir = fs.isDirectory(hadoopPath)
-      if (!isLocalMode && scheme == "file" && isDir) {
+      if (!isLocal && scheme == "file" && isDir) {
         throw new SparkException(s"addFile does not support local directories when not running " +
           "local mode.")
       }
@@ -1043,7 +1042,7 @@ class SparkContext(config: SparkConf) extends Logging with ExecutorAllocationCli
       }
     }
 
-    val key = if (!isLocalMode && scheme == "file") {
+    val key = if (!isLocal && scheme == "file") {
       env.httpFileServer.addFile(new File(uri.getPath))
     } else {
       schemeCorrectedPath

core/src/main/scala/org/apache/spark/util/Utils.scala

@@ -442,7 +442,7 @@ private[spark] object Utils extends Logging {
       fileOverwrite: Boolean): Unit = {
     val tempFile = File.createTempFile("fetchFileTemp", null,
       new File(destFile.getParentFile.getAbsolutePath))
-    logInfo("Fetching " + url + " to " + tempFile)
+    logInfo(s"Fetching $url to $tempFile")
 
     try {
       val out = new FileOutputStream(tempFile)
@@ -527,9 +527,9 @@ private[spark] object Utils extends Logging {
       if (subfiles1.size != subfiles2.size) {
         return false
       }
-      subfiles1.sortBy(_.getName).zip(subfiles2.sortBy(_.getName)).dropWhile {
+      subfiles1.sortBy(_.getName).zip(subfiles2.sortBy(_.getName)).forall {
         case (f1, f2) => filesEqualRecursive(f1, f2)
-      }.isEmpty
+      }
     } else if (file1.isFile && file2.isFile) {
       Files.equal(file1, file2)
     } else {
@@ -611,7 +611,9 @@ private[spark] object Utils extends Logging {
       conf: SparkConf,
       hadoopConf: Configuration,
       fileOverwrite: Boolean): Unit = {
-    targetDir.mkdir()
+    if (!targetDir.mkdir()) {
+      throw new IOException(s"Failed to create directory ${targetDir.getPath}")
+    }
     fs.listStatus(path).foreach { fileStatus =>
       val innerPath = fileStatus.getPath
       if (fileStatus.isDir) {

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

@@ -26,6 +26,8 @@ import org.scalatest.FunSuite
 
 import org.apache.hadoop.io.BytesWritable
 
+import org.apache.spark.util.Utils
+
 class SparkContextSuite extends FunSuite with LocalSparkContext {
 
   test("Only one SparkContext may be active at a time") {
@@ -79,7 +81,7 @@ class SparkContextSuite extends FunSuite with LocalSparkContext {
   }
 
   test("addFile works") {
-    val file = new File("somefile")
+    val file = File.createTempFile("someprefix", "somesuffix")
     val absolutePath = file.getAbsolutePath
     try {
       Files.write("somewords", file, UTF_8)
@@ -102,60 +104,49 @@ class SparkContextSuite extends FunSuite with LocalSparkContext {
       }).count()
     } finally {
       sc.stop()
-      file.delete()
     }
   }
 
   test("addFile recursive works") {
-    val pluto = new File("pluto")
-    val neptune = new File(pluto, "neptune")
-    val saturn = new File(neptune, "saturn")
-    val alien1 = new File(neptune, "alien1")
-    val alien2 = new File(saturn, "alien2")
+    val pluto = Utils.createTempDir()
+    val neptune = Utils.createTempDir(pluto.getAbsolutePath)
+    val saturn = Utils.createTempDir(neptune.getAbsolutePath)
+    val alien1 = File.createTempFile("alien", "1", neptune)
+    val alien2 = File.createTempFile("alien", "2", saturn)
 
     try {
-      assert(neptune.mkdirs())
-      assert(saturn.mkdir())
-      assert(alien1.createNewFile())
-      assert(alien2.createNewFile())
-
       sc = new SparkContext(new SparkConf().setAppName("test").setMaster("local"))
       sc.addFile(neptune.getAbsolutePath, true)
       sc.parallelize(Array(1), 1).map(x => {
         val sep = File.separator
-        if (!new File(SparkFiles.get("neptune" + sep + "alien1")).exists()) {
+        if (!new File(SparkFiles.get(neptune.getName + sep + alien1.getName)).exists()) {
           throw new SparkException("can't access file under root added directory")
         }
-        if (!new File(SparkFiles.get("neptune" + sep + "saturn" + sep + "alien2")).exists()) {
+        if (!new File(SparkFiles.get(neptune.getName + sep + saturn.getName + sep + alien2.getName))
+            .exists()) {
           throw new SparkException("can't access file in nested directory")
         }
-        if (new File(SparkFiles.get("pluto" + sep + "neptune" + sep + "alien1")).exists()) {
+        if (new File(SparkFiles.get(pluto.getName + sep + neptune.getName + sep + alien1.getName))
+            .exists()) {
           throw new SparkException("file exists that shouldn't")
         }
         x
       }).count()
     } finally {
       sc.stop()
-      alien2.delete()
-      saturn.delete()
-      alien1.delete()
-      neptune.delete()
-      pluto.delete()
     }
   }
 
   test("addFile recursive can't add directories by default") {
-    val dir = new File("dir")
+    val dir = Utils.createTempDir()
 
     try {
       sc = new SparkContext(new SparkConf().setAppName("test").setMaster("local"))
-      sc.addFile(dir.getAbsolutePath)
-      assert(false, "should have thrown exception")
-    } catch {
-      case _: SparkException =>
+      intercept[SparkException] {
+        sc.addFile(dir.getAbsolutePath)
+      }
     } finally {
       sc.stop()
-      dir.delete()
     }
   }
 }