diff options
author | Patrick Wendell <pwendell@gmail.com> | 2013-12-15 20:30:21 -0800 |
---|---|---|
committer | Patrick Wendell <pwendell@gmail.com> | 2013-12-16 21:53:51 -0800 |
commit | c6f95e603e2c58b2fab6255c824a839807f03026 (patch) | |
tree | 7b11d9a3374b1b86dbf318832ef11b6bfe875e44 /new-yarn/src/main/scala/org/apache | |
parent | 964a3b6971716823f473f672611951d1e489a552 (diff) | |
download | spark-c6f95e603e2c58b2fab6255c824a839807f03026.tar.gz spark-c6f95e603e2c58b2fab6255c824a839807f03026.tar.bz2 spark-c6f95e603e2c58b2fab6255c824a839807f03026.zip |
Attempt with extra repositories
Diffstat (limited to 'new-yarn/src/main/scala/org/apache')
-rw-r--r-- | new-yarn/src/main/scala/org/apache/spark/deploy/yarn/WorkerLauncher.scala | 10 | ||||
-rw-r--r-- | new-yarn/src/main/scala/org/apache/spark/deploy/yarn/YarnAllocationHandler.scala | 2 |
2 files changed, 6 insertions, 6 deletions
diff --git a/new-yarn/src/main/scala/org/apache/spark/deploy/yarn/WorkerLauncher.scala b/new-yarn/src/main/scala/org/apache/spark/deploy/yarn/WorkerLauncher.scala index c38f33e212..11da1c4e73 100644 --- a/new-yarn/src/main/scala/org/apache/spark/deploy/yarn/WorkerLauncher.scala +++ b/new-yarn/src/main/scala/org/apache/spark/deploy/yarn/WorkerLauncher.scala @@ -26,7 +26,7 @@ import org.apache.hadoop.yarn.api.protocolrecords._ import org.apache.hadoop.yarn.conf.YarnConfiguration import org.apache.hadoop.yarn.util.{ConverterUtils, Records} import akka.actor._ -import akka.remote.{RemoteClientShutdown, RemoteClientDisconnected, RemoteClientLifeCycleEvent} +import akka.remote._ import akka.actor.Terminated import org.apache.spark.{SparkContext, Logging} import org.apache.spark.util.{Utils, AkkaUtils} @@ -59,12 +59,12 @@ class WorkerLauncher(args: ApplicationMasterArguments, conf: Configuration) exte override def preStart() { logInfo("Listen to driver: " + driverUrl) driver = context.actorFor(driverUrl) - context.system.eventStream.subscribe(self, classOf[RemoteClientLifeCycleEvent]) - context.watch(driver) // Doesn't work with remote actors, but useful for testing + driver ! "hello" + context.system.eventStream.subscribe(self, classOf[RemotingLifecycleEvent]) } override def receive = { - case Terminated(_) | RemoteClientDisconnected(_, _) | RemoteClientShutdown(_, _) => + case x: DisassociatedEvent => logInfo("Driver terminated or disconnected! Shutting down.") driverClosed = true } @@ -140,7 +140,7 @@ class WorkerLauncher(args: ApplicationMasterArguments, conf: Configuration) exte System.setProperty("spark.driver.host", driverHost) System.setProperty("spark.driver.port", driverPort.toString) - val driverUrl = "akka://spark@%s:%s/user/%s".format( + val driverUrl = "akka.tcp://spark@%s:%s/user/%s".format( driverHost, driverPort.toString, CoarseGrainedSchedulerBackend.ACTOR_NAME) actor = actorSystem.actorOf(Props(new MonitorActor(driverUrl)), name = "YarnAM") diff --git a/new-yarn/src/main/scala/org/apache/spark/deploy/yarn/YarnAllocationHandler.scala b/new-yarn/src/main/scala/org/apache/spark/deploy/yarn/YarnAllocationHandler.scala index dba0f7640e..c27257cda4 100644 --- a/new-yarn/src/main/scala/org/apache/spark/deploy/yarn/YarnAllocationHandler.scala +++ b/new-yarn/src/main/scala/org/apache/spark/deploy/yarn/YarnAllocationHandler.scala @@ -253,7 +253,7 @@ private[yarn] class YarnAllocationHandler( numWorkersRunning.decrementAndGet() } else { val workerId = workerIdCounter.incrementAndGet().toString - val driverUrl = "akka://spark@%s:%s/user/%s".format( + val driverUrl = "akka.tcp://spark@%s:%s/user/%s".format( System.getProperty("spark.driver.host"), System.getProperty("spark.driver.port"), CoarseGrainedSchedulerBackend.ACTOR_NAME) |