Merge pull request spark-jobserver#425 from hntd187/dependencies

velvia · velvia · commit 58a47aec2bf9 · 2016-04-20T00:16:10.000-07:00
Upgraded to sbt 0.13.11, upgraded dependencies to latest minor verson…
diff --git a/.travis.yml b/.travis.yml
@@ -3,5 +3,8 @@ env:
   global:
     _JAVA_OPTIONS="-Xmx1500m -XX:MaxPermSize=512m -Dakka.test.timefactor=3"
 scala:
-   - 2.10.4
-   - 2.11.6
+   - 2.10.6
+   - 2.11.8
+jdk:
+   - oraclejdk8
+   - oraclejdk7
diff --git a/job-server/src/spark.jobserver/AkkaClusterSupervisorActor.scala b/job-server/src/spark.jobserver/AkkaClusterSupervisorActor.scala
@@ -184,6 +184,9 @@ class AkkaClusterSupervisorActor(daoActor: ActorRef) extends InstrumentedActor {
         contexts(ctxName) = (ref, resActor)
         context.watch(ref)
         successFunc(ref)
+      case _ => logger.info("Failed for unknown reason.")
+        ref ! PoisonPill
+        failureFunc(new RuntimeException("Failed for unknown reason."))
     }
   }
 
diff --git a/job-server/test/spark.jobserver/DataManagerActorSpec.scala b/job-server/test/spark.jobserver/DataManagerActorSpec.scala
@@ -15,12 +15,12 @@ class DataManagerActorSpec extends TestKit(DataManagerActorSpec.system) with Imp
     with FunSpecLike with Matchers with BeforeAndAfter with BeforeAndAfterAll {
 
   import com.typesafe.config._
-  import CommonMessages.NoSuchJobId
   import DataManagerActor._
 
   private val bytes = Array[Byte](0, 1, 2)
   private val tmpDir = Files.createTempDirectory("ut")
-  private val config = ConfigFactory.empty().withValue("spark.jobserver.datadao.rootdir", ConfigValueFactory.fromAnyRef(tmpDir.toString))
+  private val config = ConfigFactory.empty().withValue("spark.jobserver.datadao.rootdir",
+    ConfigValueFactory.fromAnyRef(tmpDir.toString))
 
   override def afterAll() {
     dao.shutdown()
diff --git a/job-server/test/spark.jobserver/InMemoryDAO.scala b/job-server/test/spark.jobserver/InMemoryDAO.scala
@@ -16,12 +16,12 @@ class InMemoryDAO extends JobDAO {
     jars((appName, uploadTime)) = jarBytes
   }
 
-  def getApps(): Map[String, DateTime] = {
+  def getApps: Map[String, DateTime] = {
     jars.keys
       .groupBy(_._1)
       .map { case (appName, appUploadTimeTuples) =>
         appName -> appUploadTimeTuples.map(_._2).toSeq.head
-      }.toMap
+      }
   }
 
   def retrieveJarFile(appName: String, uploadTime: DateTime): String = {
@@ -34,7 +34,7 @@ class InMemoryDAO extends JobDAO {
     } finally {
       bos.close()
     }
-    outFile.getAbsolutePath()
+    outFile.getAbsolutePath
   }
 
   val jobInfos = mutable.HashMap.empty[String, JobInfo]
@@ -49,5 +49,5 @@ class InMemoryDAO extends JobDAO {
 
   def saveJobConfig(jobId: String, jobConfig: Config) { jobConfigs(jobId) = jobConfig }
 
-  def getJobConfigs(): Map[String, Config] = jobConfigs.toMap
+  def getJobConfigs: Map[String, Config] = jobConfigs.toMap
 }
diff --git a/job-server/test/spark.jobserver/JobManagerActorSpec.scala b/job-server/test/spark.jobserver/JobManagerActorSpec.scala
@@ -1,7 +1,5 @@
 package spark.jobserver
 
-import akka.actor.Props
-import akka.testkit.TestProbe
 import spark.jobserver.CommonMessages.{JobErroredOut, JobResult}
 import spark.jobserver.io.JobDAOActor
 
diff --git a/job-server/test/spark.jobserver/JobManagerSpec.scala b/job-server/test/spark.jobserver/JobManagerSpec.scala
@@ -1,16 +1,15 @@
 package spark.jobserver
 
-import java.nio.file.Paths
+import scala.collection.mutable
 
 import com.typesafe.config.ConfigFactory
 import spark.jobserver.JobManagerActor.KillJob
-import scala.collection.mutable
-import spark.jobserver.io.JobDAO
 
 object JobManagerSpec extends JobSpecConfig
 
 abstract class JobManagerSpec extends JobSpecBase(JobManagerSpec.getNewSystem) {
   import scala.concurrent.duration._
+
   import CommonMessages._
   import JobManagerSpec.MaxJobsPerContext
   import akka.testkit._
@@ -95,7 +94,7 @@ abstract class JobManagerSpec extends JobSpecBase(JobManagerSpec.getNewSystem) {
       uploadTestJar()
       manager ! JobManagerActor.StartJob("demo", wordCountClass, stringConfig, syncEvents ++ errorEvents)
       expectMsgPF(startJobWait, "Did not get JobResult") {
-        case JobResult(_, result: Map[String, Int]) => println("I got results! " + result)
+        case JobResult(_, result) => println("I got results! " + result)
       }
       expectNoMsg()
     }
@@ -129,7 +128,7 @@ abstract class JobManagerSpec extends JobSpecBase(JobManagerSpec.getNewSystem) {
       manager ! JobManagerActor.StartJob("demo", classPrefix + "ConfigCheckerJob", jobConfig,
         syncEvents ++ errorEvents)
       expectMsgPF(startJobWait, "Did not get JobResult") {
-        case JobResult(_, keys: Seq[String]) =>
+        case JobResult(_, keys: Seq[_]) =>
           keys should contain ("foo")
       }
     }
@@ -203,10 +202,10 @@ abstract class JobManagerSpec extends JobSpecBase(JobManagerSpec.getNewSystem) {
       uploadTestJar()
       manager ! JobManagerActor.StartJob("demo", classPrefix + "LongPiJob", stringConfig, allEvents)
       expectMsgPF(5.seconds.dilated, "Did not get JobResult") {
-        case JobStarted(id, _, _) => {
+        case JobStarted(id, _, _) =>
           manager ! KillJob(id)
           expectMsgClass(classOf[JobKilled])
-        }
+
       }
     }
 
@@ -219,12 +218,13 @@ abstract class JobManagerSpec extends JobSpecBase(JobManagerSpec.getNewSystem) {
       val jobJarDepsConfigs = ConfigFactory.parseString(
         s"""
           |dependent-jar-uris = ["file://$getExtrasJarPath"]
-        """.stripMargin)
+        """.stripMargin
+      )
 
       manager ! JobManagerActor.StartJob("demo", classPrefix + "jobJarDependenciesJob", jobJarDepsConfigs,
         syncEvents ++ errorEvents)
       expectMsgPF(startJobWait, "Did not get JobResult") {
-        case JobResult(_, result: Map[String, Long]) => println("I got results! " + result)
+        case JobResult(_, result) => println("I got results! " + result)
       }
       expectNoMsg()
     }
diff --git a/job-server/test/spark.jobserver/JobSpecBase.scala b/job-server/test/spark.jobserver/JobSpecBase.scala
@@ -17,10 +17,12 @@ trait JobSpecConfig {
   import collection.JavaConverters._
 
   val JobResultCacheSize = Integer.valueOf(30)
-  val NumCpuCores = Integer.valueOf(Runtime.getRuntime.availableProcessors())  // number of cores to allocate. Required.
-  val MemoryPerNode = "512m"  // Executor memory per node, -Xmx style eg 512m, 1G, etc.
+  // number of cores to allocate. Required.
+  val NumCpuCores = Integer.valueOf(Runtime.getRuntime.availableProcessors())
+  // Executor memory per node, -Xmx style eg 512m, 1G, etc.
+  val MemoryPerNode = "512m"
   val MaxJobsPerContext = Integer.valueOf(2)
-  def contextFactory = classOf[DefaultSparkContextFactory].getName
+  def contextFactory: String = classOf[DefaultSparkContextFactory].getName
   lazy val config = {
     val ConfigMap = Map(
       "spark.jobserver.job-result-cache-size" -> JobResultCacheSize,
@@ -29,7 +31,7 @@ trait JobSpecConfig {
       "spark.jobserver.max-jobs-per-context" -> MaxJobsPerContext,
       "spark.jobserver.named-object-creation-timeout" -> "60 s",
       "akka.log-dead-letters" -> Integer.valueOf(0),
-      "spark.master" -> "local[4]",
+      "spark.master" -> "local[*]",
       "context-factory" -> contextFactory,
       "spark.context-settings.test" -> ""
     )
@@ -44,14 +46,14 @@ trait JobSpecConfig {
   lazy val contextConfig = {
     val ConfigMap = Map(
       "context-factory" -> contextFactory,
-      "streaming.batch_interval" -> new Integer(40),
-      "streaming.stopGracefully" -> false,
-      "streaming.stopSparkContext" -> true
+      "streaming.batch_interval" -> Integer.valueOf(40),
+      "streaming.stopGracefully" -> Boolean.box(false),
+      "streaming.stopSparkContext" -> Boolean.box(true)
     )
     ConfigFactory.parseMap(ConfigMap.asJava).withFallback(ConfigFactory.defaultOverrides())
   }
 
-  def getNewSystem = ActorSystem("test", config)
+  def getNewSystem: ActorSystem = ActorSystem("test", config)
 }
 
 abstract class JobSpecBaseBase(system: ActorSystem) extends TestKit(system) with ImplicitSender
diff --git a/job-server/test/spark.jobserver/JobStatusActorSpec.scala b/job-server/test/spark.jobserver/JobStatusActorSpec.scala
@@ -1,20 +1,19 @@
 package spark.jobserver
 
-import akka.actor.{Props, PoisonPill, ActorRef, ActorSystem}
+import akka.actor.{Props, ActorRef, ActorSystem}
 import akka.testkit.{TestKit, ImplicitSender}
 import spark.jobserver.io.{JobDAOActor, JarInfo, JobInfo, JobDAO}
 import org.joda.time.DateTime
-import org.scalatest.matchers.ShouldMatchers
-import org.scalatest.{FunSpecLike, FunSpec, BeforeAndAfter, BeforeAndAfterAll}
+import org.scalatest.Matchers
+import org.scalatest.{FunSpecLike, BeforeAndAfter, BeforeAndAfterAll}
 
 object JobStatusActorSpec {
   val system = ActorSystem("test")
 }
 
 class JobStatusActorSpec extends TestKit(JobStatusActorSpec.system) with ImplicitSender
-with FunSpecLike with ShouldMatchers with BeforeAndAfter with BeforeAndAfterAll {
+with FunSpecLike with Matchers with BeforeAndAfter with BeforeAndAfterAll {
 
-  import com.typesafe.config._
   import CommonMessages._
   import JobStatusActor._
 
@@ -24,7 +23,6 @@ with FunSpecLike with ShouldMatchers with BeforeAndAfter with BeforeAndAfterAll
   private val jarInfo = JarInfo(appName, DateTime.now)
   private val classPath = "classPath"
   private val jobInfo = JobInfo(jobId, contextName, jarInfo, classPath, DateTime.now, None, None)
-  private val jobConfig = ConfigFactory.empty()
 
   override def afterAll() {
     ooyala.common.akka.AkkaTestUtils.shutdownAndWait(JobStatusActorSpec.system)
diff --git a/job-server/test/spark.jobserver/JobWithNamedRddsSpec.scala b/job-server/test/spark.jobserver/JobWithNamedRddsSpec.scala
@@ -1,19 +1,14 @@
 package spark.jobserver
 
-import akka.actor.{ ActorRef, ActorSystem, Props }
-import akka.testkit.{ ImplicitSender, TestKit }
-import org.apache.spark.{ SparkContext, SparkConf }
+import org.apache.spark.{SparkConf, SparkContext}
 import org.apache.spark.storage.StorageLevel
-import org.scalatest.{ FunSpecLike, FunSpec, BeforeAndAfterAll, BeforeAndAfter }
-import com.typesafe.config.Config
-import com.typesafe.config.ConfigFactory
-import spark.jobserver.CommonMessages.{ JobErroredOut, JobResult }
 import java.util.concurrent.TimeoutException
 import scala.concurrent.duration._
 import org.apache.spark.rdd.RDD
-class JobWithNamedRddsSpec extends JobSpecBase(JobManagerSpec.getNewSystem) {
 
-  private val emptyConfig = ConfigFactory.parseString("spark.jobserver.named-object-creation-timeout = 60 s")
+import com.typesafe.config.Config
+
+class JobWithNamedRddsSpec extends JobSpecBase(JobManagerSpec.getNewSystem) {
 
   val sc = new SparkContext("local[4]", getClass.getSimpleName, new SparkConf)
 
@@ -42,7 +37,7 @@ class JobWithNamedRddsSpec extends JobSpecBase(JobManagerSpec.getNewSystem) {
 
   describe("NamedRdds") {
     it("get() should return None when RDD does not exist") {
-      namedTestRdds.getNames.foreach { rddName => namedTestRdds.destroy(rddName) }
+      namedTestRdds.getNames().foreach { rddName => namedTestRdds.destroy(rddName) }
       namedTestRdds.get[Int]("No such RDD") should equal(None)
     }
 
@@ -60,7 +55,7 @@ class JobWithNamedRddsSpec extends JobSpecBase(JobManagerSpec.getNewSystem) {
 
       var rdd : Option[RDD[Int]] = None
       val thread = new Thread {
-        override def run {
+        override def run() {
           namedTestRdds.getOrElseCreate("rdd-sleep", {
             val t1 = System.currentTimeMillis()
             var x = 0d
@@ -73,7 +68,7 @@ class JobWithNamedRddsSpec extends JobSpecBase(JobManagerSpec.getNewSystem) {
           })(1.milliseconds)
         }
       }
-      thread.start
+      thread.start()
       Thread.sleep(11)
       //don't wait
       val err = intercept[TimeoutException] { namedTestRdds.get[Int]("rdd-sleep")(1.milliseconds) }
@@ -147,7 +142,7 @@ class JobWithNamedRddsSpec extends JobSpecBase(JobManagerSpec.getNewSystem) {
     }
 
     it("should include underlying exception when error occurs") {
-      def errorFunc = {
+      def errorFunc: RDD[Int] = {
         throw new IllegalArgumentException("boo!")
         sc.parallelize(Seq(1, 2))
       }
diff --git a/job-server/test/spark.jobserver/LocalContextSupervisorSpec.scala b/job-server/test/spark.jobserver/LocalContextSupervisorSpec.scala
@@ -3,8 +3,6 @@ package spark.jobserver
 import akka.actor._
 import akka.testkit.{ImplicitSender, TestKit}
 import com.typesafe.config.ConfigFactory
-import org.apache.hadoop.conf.Configuration
-import org.apache.spark.SparkConf
 import spark.jobserver.io.{JobDAO, JobDAOActor}
 import org.scalatest.{BeforeAndAfter, BeforeAndAfterAll, FunSpecLike, Matchers}
 
@@ -102,21 +100,20 @@ class LocalContextSupervisorSpec extends TestKit(LocalContextSupervisorSpec.syst
       expectMsg(Seq("c1", "c2"))
       supervisor ! GetResultActor("c1")
       val rActor = expectMsgClass(classOf[ActorRef])
-      rActor.path.toString should not include ("global")
+      rActor.path.toString should not include "global"
     }
 
     it("should be able to get context configs") {
       supervisor ! AddContext("c1", contextConfig)
       expectMsg(ContextInitialized)
       supervisor ! GetContext("c1")
       expectMsgPF(5 seconds, "I can't find that context :'-(") {
-        case (contextActor: ActorRef, resultActor: ActorRef) => {
+        case (contextActor: ActorRef, resultActor: ActorRef) =>
           contextActor ! GetContextConfig
           val cc = expectMsgClass(classOf[ContextConfig])
           cc.contextName shouldBe "c1"
           cc.contextConfig.get("spark.ui.enabled") shouldBe "false"
           cc.hadoopConfig.get("mapreduce.framework.name") shouldBe "ayylmao"
-        }
       }
     }
 
diff --git a/job-server/test/spark.jobserver/NamedObjectsRDDsOnlySpec.scala b/job-server/test/spark.jobserver/NamedObjectsRDDsOnlySpec.scala
@@ -1,11 +1,11 @@
 package spark.jobserver
 
-import akka.actor.{ ActorRef, ActorSystem, Props }
+import akka.actor.ActorSystem
 import akka.testkit.{ ImplicitSender, TestKit }
 import org.apache.spark.{ SparkContext, SparkConf }
-import org.apache.spark.rdd.RDD
+
 import org.apache.spark.storage.StorageLevel
-import org.scalatest.{ Matchers, FunSpecLike, FunSpec, BeforeAndAfterAll, BeforeAndAfter }
+import org.scalatest.{ Matchers, FunSpecLike, BeforeAndAfterAll, BeforeAndAfter }
 
 /**
  * please note that this test only uses RDDs as named objects, there exists another test class
@@ -17,10 +17,10 @@ class NamedObjectsRDDsOnlySpec extends TestKit(ActorSystem("NamedObjectsSpec"))
   val sc = new SparkContext("local[2]", getClass.getSimpleName, new SparkConf)
   val namedObjects: NamedObjects = new JobServerNamedObjects(system)
 
-  implicit def rddPersister[T] = new RDDPersister[T]
+  implicit def rddPersister[T]: RDDPersister[T] = new RDDPersister[T]
 
   before {
-    namedObjects.getNames.foreach { namedObjects.forget(_) }
+    namedObjects.getNames().foreach { namedObjects.forget }
   }
 
   override def afterAll() {
@@ -106,7 +106,7 @@ class NamedObjectsRDDsOnlySpec extends TestKit(ActorSystem("NamedObjectsSpec"))
     }
 
     it("should include underlying exception when error occurs") {
-      def errorFunc = {
+      def errorFunc: NamedRDD[Int] = {
         throw new IllegalArgumentException("boo!")
         NamedRDD(sc.parallelize(Seq(1, 2)), true, StorageLevel.MEMORY_ONLY)
       }
diff --git a/job-server/test/spark.jobserver/SparkJobSpec.scala b/job-server/test/spark.jobserver/SparkJobSpec.scala
@@ -9,15 +9,15 @@ class SparkJobSpec extends FunSpec with Matchers {
 
   describe("Sample tests for default validation && method") {
     it("should return valid") {
-      validSparkValidation && (validSparkValidation) should equal (SparkJobValid)
+      validSparkValidation && validSparkValidation should equal (SparkJobValid)
     }
 
     it("should return invalid if one of them is invalid") {
-      validSparkValidation && (invalidSparkValidation) should equal (SparkJobInvalid("Sample message 1"))
+      validSparkValidation && invalidSparkValidation should equal (SparkJobInvalid("Sample message 1"))
     }
 
     it ("should return invalid if both of them are invalid with the first message") {
-      invalidSparkValidation2 && (invalidSparkValidation) should equal (SparkJobInvalid("Sample message 2"))
+      invalidSparkValidation2 && invalidSparkValidation should equal (SparkJobInvalid("Sample message 2"))
     }
   }
 }
diff --git a/job-server/test/spark.jobserver/TestJarFinder.scala b/job-server/test/spark.jobserver/TestJarFinder.scala
@@ -19,9 +19,9 @@ trait TestJarFinder {
     */
   def getBaseDir(pkg: String): String ={
     // Current directory.  Find out if we are in project root, and need to go up a level.
-    val cwd = Paths.get(".").toAbsolutePath().normalize().toString()
-    val dotdot = if (Paths.get(cwd + s"/${pkg}").toFile.isDirectory) "" else "../"
-    s"$cwd/${dotdot}"
+    val cwd = Paths.get(".").toAbsolutePath.normalize().toString
+    val dotdot = if (Paths.get(cwd + s"/$pkg").toFile.isDirectory) "" else "../"
+    s"$cwd/$dotdot"
   }
 
   /**
diff --git a/job-server/test/spark.jobserver/WebApiSpec.scala b/job-server/test/spark.jobserver/WebApiSpec.scala
diff --git a/job-server/test/spark.jobserver/io/JobSqlDAOSpec.scala b/job-server/test/spark.jobserver/io/JobSqlDAOSpec.scala
diff --git a/project/Build.scala b/project/Build.scala
diff --git a/project/Dependencies.scala b/project/Dependencies.scala
diff --git a/project/build.properties b/project/build.properties
diff --git a/project/plugins.sbt b/project/plugins.sbt

Original file line number	Diff line number	Diff line change
`@@ -184,6 +184,9 @@ class AkkaClusterSupervisorActor(daoActor: ActorRef) extends InstrumentedActor {`
`184`	`184`	`contexts(ctxName) = (ref, resActor)`
`185`	`185`	`context.watch(ref)`
`186`	`186`	`successFunc(ref)`
	`187`	`+ case _ => logger.info("Failed for unknown reason.")`
	`188`	`+ ref ! PoisonPill`
	`189`	`+ failureFunc(new RuntimeException("Failed for unknown reason."))`
`187`	`190`	`}`
`188`	`191`	`}`
`189`	`192`
Original file line number	Diff line number	Diff line change
`@@ -16,12 +16,12 @@ class InMemoryDAO extends JobDAO {`
`16`	`16`	`jars((appName, uploadTime)) = jarBytes`
`17`	`17`	`}`
`18`	`18`
`19`		`- def getApps(): Map[String, DateTime] = {`
	`19`	`+ def getApps: Map[String, DateTime] = {`
`20`	`20`	`jars.keys`
`21`	`21`	`.groupBy(_._1)`
`22`	`22`	`.map { case (appName, appUploadTimeTuples) =>`
`23`	`23`	`appName -> appUploadTimeTuples.map(_._2).toSeq.head`
`24`		`- }.toMap`
	`24`	`+ }`
`25`	`25`	`}`
`26`	`26`
`27`	`27`	`def retrieveJarFile(appName: String, uploadTime: DateTime): String = {`
`@@ -34,7 +34,7 @@ class InMemoryDAO extends JobDAO {`
`34`	`34`	`} finally {`
`35`	`35`	`bos.close()`
`36`	`36`	`}`
`37`		`- outFile.getAbsolutePath()`
	`37`	`+ outFile.getAbsolutePath`
`38`	`38`	`}`
`39`	`39`
`40`	`40`	`val jobInfos = mutable.HashMap.empty[String, JobInfo]`
`@@ -49,5 +49,5 @@ class InMemoryDAO extends JobDAO {`
`49`	`49`
`50`	`50`	`def saveJobConfig(jobId: String, jobConfig: Config) { jobConfigs(jobId) = jobConfig }`
`51`	`51`
`52`		`- def getJobConfigs(): Map[String, Config] = jobConfigs.toMap`
	`52`	`+ def getJobConfigs: Map[String, Config] = jobConfigs.toMap`
`53`	`53`	`}`