Merge branch 'spark-2.0-preview'

Author: noorul

job-server/src/main/scala/spark/jobserver/BinaryManager.scala

@@ -33,6 +33,7 @@ case object InvalidBinary
 case object BinaryStored
 case object BinaryDeleted
 case class BinaryStorageFailure(ex: Throwable)
+case class BinaryDeletionFailure(ex: Throwable)
 
 /**
  * An Actor that manages the jars stored by the job server.   It's important that threads do not try to
@@ -106,6 +107,9 @@ class BinaryManager(jobDao: ActorRef) extends InstrumentedActor {
 
     case DeleteBinary(appName) =>
       logger.info(s"Deleting binary $appName")
-      deleteBinary(appName).pipeTo(sender)
+      deleteBinary(appName).map {
+        case Success(_) => BinaryDeleted
+        case Failure(ex) => BinaryDeletionFailure(ex)
+      }.pipeTo(sender)
   }
 }

job-server/src/main/scala/spark/jobserver/io/FileCacher.scala

@@ -52,8 +52,9 @@ trait FileCacher {
         if (name.startsWith(prefix)) {
           val suffix = name.substring(prefix.length)
           (Pattern findFirstIn suffix).isDefined
+        } else {
+          false
         }
-        false
       }
     })
     if (binaries != null) {

job-server/src/test/scala/spark/jobserver/BinaryManagerSpec.scala

@@ -68,7 +68,7 @@ class BinaryManagerSpec extends TestKit(BinaryManagerSpec.system) with ImplicitS
 
     it("should respond when deleted successfully") {
       binaryManager ! DeleteBinary("valid")
-      expectMsg(Success({}))
+      expectMsg(BinaryDeleted)
     }
   }
 }

job-server/src/test/scala/spark/jobserver/io/FileCacherSpec.scala

@@ -16,6 +16,8 @@ class FileCacherSpec extends FileCacher with FunSpecLike with Matchers {
   }
 
   it("clean cache binaries") {
-    cleanCacheBinaries("job")
+    val f = File.createTempFile("jobTest-20161010_010000_000.jar", ".jar", new File(rootDir))
+    cleanCacheBinaries("jobTest")
+    f.exists() should be(false)
   }
 }