Skip to content
Snippets Groups Projects
Commit cf7b1154 authored by Ali Ghodsi's avatar Ali Ghodsi
Browse files

Enabling getting the actual WEBUI port

parent 636fc0c8
No related branches found
No related tags found
No related merge requests found
......@@ -127,4 +127,8 @@ private[deploy] object DeployMessages {
case object CheckForWorkerTimeOut
case object RequestWebUIPort
case class WebUIPortResponse(boundedPort: Int) {}
}
......@@ -43,7 +43,7 @@ class LocalSparkCluster(numWorkers: Int, coresPerWorker: Int, memoryPerWorker: I
logInfo("Starting a local Spark cluster with " + numWorkers + " workers.")
/* Start the Master */
val (masterSystem, masterPort) = Master.startSystemAndActor(localHostname, 0, 0)
val (masterSystem, masterPort, _) = Master.startSystemAndActor(localHostname, 0, 0)
masterActorSystems += masterSystem
val masterUrl = "spark://" + localHostname + ":" + masterPort
......
......@@ -26,6 +26,8 @@ import akka.actor._
import akka.actor.Terminated
import akka.remote.{RemoteClientLifeCycleEvent, RemoteClientDisconnected, RemoteClientShutdown}
import akka.util.duration._
import akka.dispatch.Await
import akka.pattern.ask
import org.apache.spark.{Logging, SparkException}
import org.apache.spark.deploy.{ApplicationDescription, ExecutorState}
......@@ -33,6 +35,8 @@ import org.apache.spark.deploy.DeployMessages._
import org.apache.spark.deploy.master.ui.MasterWebUI
import org.apache.spark.metrics.MetricsSystem
import org.apache.spark.util.{Utils, AkkaUtils}
import akka.util.Timeout
import java.util.concurrent.TimeUnit
private[spark] class Master(host: String, port: Int, webUiPort: Int) extends Actor with Logging {
......@@ -180,6 +184,10 @@ private[spark] class Master(host: String, port: Int, webUiPort: Int) extends Act
case CheckForWorkerTimeOut => {
timeOutDeadWorkers()
}
case RequestWebUIPort => {
sender ! WebUIPortResponse(webUi.boundPort.getOrElse(-1))
}
}
/**
......@@ -364,7 +372,7 @@ private[spark] object Master {
def main(argStrings: Array[String]) {
val args = new MasterArguments(argStrings)
val (actorSystem, _) = startSystemAndActor(args.host, args.port, args.webUiPort)
val (actorSystem, _, _) = startSystemAndActor(args.host, args.port, args.webUiPort)
actorSystem.awaitTermination()
}
......@@ -378,9 +386,12 @@ private[spark] object Master {
}
}
def startSystemAndActor(host: String, port: Int, webUiPort: Int): (ActorSystem, Int) = {
def startSystemAndActor(host: String, port: Int, webUiPort: Int): (ActorSystem, Int, Int) = {
val (actorSystem, boundPort) = AkkaUtils.createActorSystem(systemName, host, port)
val actor = actorSystem.actorOf(Props(new Master(host, boundPort, webUiPort)), name = actorName)
(actorSystem, boundPort)
implicit val timeout = Timeout(1 seconds)
val respFuture = actor ? RequestWebUIPort
val resp = Await.result(respFuture, timeout.duration).asInstanceOf[WebUIPortResponse]
(actorSystem, boundPort, resp.boundedPort)
}
}
0% Loading or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment