diff options
author | Prashant Sharma <prashant.s@imaginea.com> | 2013-12-09 12:55:19 +0530 |
---|---|---|
committer | Prashant Sharma <prashant.s@imaginea.com> | 2013-12-09 13:00:50 +0530 |
commit | c1201f47e0d44e92da42adb23d27f60d9d494642 (patch) | |
tree | 34fcbd11eceee133b822ddf110b30899702c4d86 | |
parent | 7ad6921ae0657ca806704f859d5b8c9ff26633e4 (diff) | |
download | spark-c1201f47e0d44e92da42adb23d27f60d9d494642.tar.gz spark-c1201f47e0d44e92da42adb23d27f60d9d494642.tar.bz2 spark-c1201f47e0d44e92da42adb23d27f60d9d494642.zip |
fixed yarn build
-rw-r--r-- | yarn/src/main/scala/org/apache/spark/deploy/yarn/WorkerLauncher.scala | 15 | ||||
-rw-r--r-- | yarn/src/main/scala/org/apache/spark/deploy/yarn/YarnAllocationHandler.scala | 4 |
2 files changed, 8 insertions, 11 deletions
diff --git a/yarn/src/main/scala/org/apache/spark/deploy/yarn/WorkerLauncher.scala b/yarn/src/main/scala/org/apache/spark/deploy/yarn/WorkerLauncher.scala index b67e068844..69038844bb 100644 --- a/yarn/src/main/scala/org/apache/spark/deploy/yarn/WorkerLauncher.scala +++ b/yarn/src/main/scala/org/apache/spark/deploy/yarn/WorkerLauncher.scala @@ -27,10 +27,8 @@ import org.apache.hadoop.yarn.conf.YarnConfiguration import org.apache.hadoop.yarn.ipc.YarnRPC import org.apache.hadoop.yarn.util.{ConverterUtils, Records} import akka.actor._ -import akka.remote.{RemoteClientShutdown, RemoteClientDisconnected, RemoteClientLifeCycleEvent} -import akka.remote.RemoteClientShutdown +import akka.remote._ import akka.actor.Terminated -import akka.remote.RemoteClientDisconnected import org.apache.spark.{SparkContext, Logging} import org.apache.spark.util.{Utils, AkkaUtils} import org.apache.spark.scheduler.cluster.CoarseGrainedSchedulerBackend @@ -55,19 +53,18 @@ class WorkerLauncher(args: ApplicationMasterArguments, conf: Configuration) exte // This actor just working as a monitor to watch on Driver Actor. class MonitorActor(driverUrl: String) extends Actor { - var driver: ActorRef = null + var driver: ActorSelection = null override def preStart() { logInfo("Listen to driver: " + driverUrl) - driver = context.actorFor(driverUrl) + driver = context.actorSelection(driverUrl) driver ! "hello" - context.system.eventStream.subscribe(self, classOf[RemoteClientLifeCycleEvent]) - context.watch(driver) // Doesn't work with remote actors, but useful for testing + context.system.eventStream.subscribe(self, classOf[RemotingLifecycleEvent]) } override def receive = { - case Terminated(_) | RemoteClientDisconnected(_, _) | RemoteClientShutdown(_, _) => - logInfo("Driver terminated or disconnected! Shutting down.") + case x: DisassociatedEvent => + logInfo(s"Driver terminated or disconnected! Shutting down. $x") driverClosed = true } } diff --git a/yarn/src/main/scala/org/apache/spark/deploy/yarn/YarnAllocationHandler.scala b/yarn/src/main/scala/org/apache/spark/deploy/yarn/YarnAllocationHandler.scala index baa030b4a4..a6ce1b60a7 100644 --- a/yarn/src/main/scala/org/apache/spark/deploy/yarn/YarnAllocationHandler.scala +++ b/yarn/src/main/scala/org/apache/spark/deploy/yarn/YarnAllocationHandler.scala @@ -34,7 +34,7 @@ import org.apache.hadoop.conf.Configuration import java.util.{Collections, Set => JSet} import java.lang.{Boolean => JBoolean} -object AllocationType extends Enumeration ("HOST", "RACK", "ANY") { +object AllocationType extends Enumeration { type AllocationType = Value val HOST, RACK, ANY = Value } @@ -370,7 +370,7 @@ private[yarn] class YarnAllocationHandler(val conf: Configuration, val resourceM createResourceRequest(AllocationType.ANY, null, numWorkers, YarnAllocationHandler.PRIORITY) val containerRequests: ArrayBuffer[ResourceRequest] = - new ArrayBuffer[ResourceRequest](hostContainerRequests.size() + rackContainerRequests.size() + 1) + new ArrayBuffer[ResourceRequest](hostContainerRequests.size + rackContainerRequests.size + 1) containerRequests ++= hostContainerRequests containerRequests ++= rackContainerRequests |