Merge branch 'master' into configTimeout-6980

Conflicts:
	core/src/main/scala/org/apache/spark/deploy/Client.scala
	core/src/main/scala/org/apache/spark/deploy/client/AppClient.scala
	core/src/main/scala/org/apache/spark/deploy/master/Master.scala
	core/src/main/scala/org/apache/spark/deploy/master/ui/ApplicationPage.scala
	core/src/main/scala/org/apache/spark/deploy/master/ui/MasterPage.scala
	core/src/main/scala/org/apache/spark/deploy/master/ui/MasterWebUI.scala
	core/src/main/scala/org/apache/spark/deploy/rest/StandaloneRestServer.scala
	core/src/main/scala/org/apache/spark/deploy/worker/ui/WorkerPage.scala
	core/src/test/scala/org/apache/spark/rpc/akka/AkkaRpcEnvSuite.scala

core/src/main/scala/org/apache/spark/deploy/Client.scala

@@ -56,10 +56,8 @@ private class ClientEndpoint(
           System.exit(SparkExitCode.UNCAUGHT_EXCEPTION)
       })
 
-  val timeout = RpcUtils.askRpcTimeout(conf)
-
-  override def preStart(): Unit = {
-    context.system.eventStream.subscribe(self, classOf[RemotingLifecycleEvent])
+   private val lostMasters = new HashSet[RpcAddress]
+   private var activeMasterEndpoint: RpcEndpointRef = null
 
   override def onStart(): Unit = {
     driverArgs.cmd match {
@@ -123,9 +121,8 @@ private class ClientEndpoint(
     println("... waiting before polling master for driver state")
     Thread.sleep(5000)
     println("... polling master for driver state")
-    val statusFuture = (activeMasterActor ? RequestDriverStatus(driverId))(timeout.duration)
-      .mapTo[DriverStatusResponse]
-    val statusResponse = timeout.awaitResult(statusFuture)
+    val statusResponse =
+      activeMasterEndpoint.askWithRetry[DriverStatusResponse](RequestDriverStatus(driverId))
     statusResponse.found match {
       case false =>
         println(s"ERROR: Cluster master did not recognize $driverId")

core/src/main/scala/org/apache/spark/deploy/client/AppClient.scala

@@ -248,9 +248,7 @@ private[spark] class AppClient(
   def stop() {
     if (endpoint != null) {
       try {
-        val timeout = RpcUtils.askRpcTimeout(conf)
-        val future = actor.ask(StopAppClient)(timeout.duration)
-        timeout.awaitResult(future)
+        endpoint.askWithRetry[Boolean](StopAppClient)
       } catch {
         case e: TimeoutException =>
           logInfo("Stop request to Master timed out; it may already be shut down.")

core/src/main/scala/org/apache/spark/deploy/master/Master.scala

@@ -24,7 +24,6 @@ import java.util.Date
 import java.util.concurrent.{ScheduledFuture, TimeUnit}
 
 import scala.collection.mutable.{ArrayBuffer, HashMap, HashSet}
-import scala.concurrent.duration._
 import scala.language.postfixOps
 import scala.util.Random
 
@@ -928,13 +927,10 @@ private[deploy] object Master extends Logging {
       webUiPort: Int,
       conf: SparkConf): (RpcEnv, Int, Option[Int]) = {
     val securityMgr = new SecurityManager(conf)
-    val (actorSystem, boundPort) = AkkaUtils.createActorSystem(systemName, host, port, conf = conf,
-      securityManager = securityMgr)
-    val actor = actorSystem.actorOf(
-      Props(classOf[Master], host, boundPort, webUiPort, securityMgr, conf), actorName)
-    val timeout = RpcUtils.askRpcTimeout(conf)
-    val portsRequest = actor.ask(BoundPortsRequest)(timeout.duration)
-    val portsResponse = timeout.awaitResult(portsRequest).asInstanceOf[BoundPortsResponse]
-    (actorSystem, boundPort, portsResponse.webUIPort, portsResponse.restPort)
+    val rpcEnv = RpcEnv.create(SYSTEM_NAME, host, port, conf, securityMgr)
+    val masterEndpoint = rpcEnv.setupEndpoint(ENDPOINT_NAME,
+      new Master(rpcEnv, rpcEnv.address, webUiPort, securityMgr, conf))
+    val portsResponse = masterEndpoint.askWithRetry[BoundPortsResponse](BoundPortsRequest)
+    (rpcEnv, portsResponse.webUIPort, portsResponse.restPort)
   }
 }

core/src/main/scala/org/apache/spark/deploy/master/ui/ApplicationPage.scala

@@ -34,8 +34,7 @@ private[ui] class ApplicationPage(parent: MasterWebUI) extends WebUIPage("app")
   /** Executor details for a particular application */
   def render(request: HttpServletRequest): Seq[Node] = {
     val appId = request.getParameter("appId")
-    val stateFuture = (master ? RequestMasterState)(timeout.duration).mapTo[MasterStateResponse]
-    val state = timeout.awaitResult(stateFuture)
+    val state = master.askWithRetry[MasterStateResponse](RequestMasterState)
     val app = state.activeApps.find(_.id == appId).getOrElse({
       state.completedApps.find(_.id == appId).getOrElse(null)
     })

core/src/main/scala/org/apache/spark/deploy/master/ui/MasterPage.scala

@@ -33,8 +33,7 @@ private[ui] class MasterPage(parent: MasterWebUI) extends WebUIPage("") {
   private val master = parent.masterEndpointRef
 
   def getMasterState: MasterStateResponse = {
-    val stateFuture = (master ? RequestMasterState)(timeout.duration).mapTo[MasterStateResponse]
-    timeout.awaitResult(stateFuture)
+    master.askWithRetry[MasterStateResponse](RequestMasterState)
   }
 
   override def renderJson(request: HttpServletRequest): JValue = {

core/src/main/scala/org/apache/spark/deploy/master/ui/MasterWebUI.scala

@@ -32,8 +32,7 @@ class MasterWebUI(val master: Master, requestedPort: Int)
   extends WebUI(master.securityMgr, requestedPort, master.conf, name = "MasterUI") with Logging
   with UIRoot {
 
-  val masterActorRef = master.self
-  val timeout = RpcUtils.askRpcTimeout(master.conf)
+  val masterEndpointRef = master.self
   val killEnabled = master.conf.getBoolean("spark.ui.killEnabled", true)
 
   val masterPage = new MasterPage(this)

core/src/main/scala/org/apache/spark/deploy/rest/StandaloneRestServer.scala

@@ -71,9 +71,8 @@ private[rest] class StandaloneKillRequestServlet(masterEndpoint: RpcEndpointRef,
   extends KillRequestServlet {
 
   protected def handleKill(submissionId: String): KillSubmissionResponse = {
-    val askTimeout = RpcUtils.askRpcTimeout(conf)
-    val response = AkkaUtils.askWithReply[DeployMessages.KillDriverResponse](
-      DeployMessages.RequestKillDriver(submissionId), masterActor, askTimeout)
+    val response = masterEndpoint.askWithRetry[DeployMessages.KillDriverResponse](
+      DeployMessages.RequestKillDriver(submissionId))
     val k = new KillSubmissionResponse
     k.serverSparkVersion = sparkVersion
     k.message = response.message
@@ -90,9 +89,8 @@ private[rest] class StandaloneStatusRequestServlet(masterEndpoint: RpcEndpointRe
   extends StatusRequestServlet {
 
   protected def handleStatus(submissionId: String): SubmissionStatusResponse = {
-    val askTimeout = RpcUtils.askRpcTimeout(conf)
-    val response = AkkaUtils.askWithReply[DeployMessages.DriverStatusResponse](
-      DeployMessages.RequestDriverStatus(submissionId), masterActor, askTimeout)
+    val response = masterEndpoint.askWithRetry[DeployMessages.DriverStatusResponse](
+      DeployMessages.RequestDriverStatus(submissionId))
     val message = response.exception.map { s"Exception from the cluster:\n" + formatException(_) }
     val d = new SubmissionStatusResponse
     d.serverSparkVersion = sparkVersion
@@ -175,7 +173,6 @@ private[rest] class StandaloneSubmitRequestServlet(
       responseServlet: HttpServletResponse): SubmitRestProtocolResponse = {
     requestMessage match {
       case submitRequest: CreateSubmissionRequest =>
-        val askTimeout = RpcUtils.askRpcTimeout(conf)
         val driverDescription = buildDriverDescription(submitRequest)
         val response = masterEndpoint.askWithRetry[DeployMessages.SubmitDriverResponse](
           DeployMessages.RequestSubmitDriver(driverDescription))

core/src/main/scala/org/apache/spark/deploy/worker/ui/WorkerPage.scala

@@ -33,16 +33,12 @@ private[ui] class WorkerPage(parent: WorkerWebUI) extends WebUIPage("") {
   private val workerEndpoint = parent.worker.self
 
   override def renderJson(request: HttpServletRequest): JValue = {
-    val stateFuture = (workerActor ? RequestWorkerState)(timeout.duration).
-      mapTo[WorkerStateResponse]
-    val workerState = timeout.awaitResult(stateFuture)
+    val workerState = workerEndpoint.askWithRetry[WorkerStateResponse](RequestWorkerState)
     JsonProtocol.writeWorkerState(workerState)
   }
 
   def render(request: HttpServletRequest): Seq[Node] = {
-    val stateFuture = (workerActor ? RequestWorkerState)(timeout.duration).
-      mapTo[WorkerStateResponse]
-    val workerState = timeout.awaitResult(stateFuture)
+    val workerState = workerEndpoint.askWithRetry[WorkerStateResponse](RequestWorkerState)
 
     val executorHeaders = Seq("ExecutorID", "Cores", "State", "Memory", "Job Details", "Logs")
     val runningExecutors = workerState.executors

core/src/test/scala/org/apache/spark/rpc/akka/AkkaRpcEnvSuite.scala

@@ -17,8 +17,8 @@
 
 package org.apache.spark.rpc.akka
 
-import org.apache.spark.{SecurityManager, SparkConf}
 import org.apache.spark.rpc._
+import org.apache.spark.{SSLSampleConfigs, SecurityManager, SparkConf}
 
 class AkkaRpcEnvSuite extends RpcEnvSuite {