refact some duplicated code, style and comments

Author: WangTaoTheTonic

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

@@ -17,6 +17,7 @@
 
 package org.apache.spark.deploy
 
+import scala.collection.mutable.HashSet
 import scala.concurrent._
 
 import akka.actor._
@@ -28,19 +29,21 @@ import org.apache.spark.{Logging, SecurityManager, SparkConf}
 import org.apache.spark.deploy.DeployMessages._
 import org.apache.spark.deploy.master.{DriverState, Master}
 import org.apache.spark.util.{ActorLogReceive, AkkaUtils, Utils}
-import scala.collection.mutable.HashSet
 
 /**
  * Proxy that relays messages to the driver.
+ *
+ * Now we don't support retry in case submission failed. In HA mode, client will submit request to
+ * all masters and see which one could handle it.
  */
 private class ClientActor(driverArgs: ClientArguments, conf: SparkConf)
   extends Actor with ActorLogReceive with Logging {
 
-  val mastersActor = driverArgs.masters.map { m =>
+  private val masterActors = driverArgs.masters.map { m =>
     context.actorSelection(Master.toAkkaUrl(m, AkkaUtils.protocol(context.system)))
   }
-  val lostMasters = new HashSet[Address]
-  var activeMasterActor: ActorSelection = null
+  private val lostMasters = new HashSet[Address]
+  private var activeMasterActor: ActorSelection = null
 
   val timeout = AkkaUtils.askTimeout(conf)
 
@@ -80,15 +83,17 @@ private class ClientActor(driverArgs: ClientArguments, conf: SparkConf)
           driverArgs.supervise,
           command)
 
-      for (masterActor <- mastersActor) {
-        masterActor ! RequestSubmitDriver(driverDescription)
-      }
+        // This assumes only one Master is active at a time
+        for (masterActor <- masterActors) {
+          masterActor ! RequestSubmitDriver(driverDescription)
+        }
 
       case "kill" =>
         val driverId = driverArgs.driverId
-      for (masterActor <- mastersActor) {
-        masterActor ! RequestKillDriver(driverId)
-      }
+        // This assumes only one Master is active at a time
+        for (masterActor <- masterActors) {
+          masterActor ! RequestKillDriver(driverId)
+        }
     }
   }
 
@@ -121,7 +126,7 @@ private class ClientActor(driverArgs: ClientArguments, conf: SparkConf)
           System.exit(-1)
         }
         System.exit(0)
-      }
+    }
   }
 
   override def receiveWithLogging: PartialFunction[Any, Unit] = {
@@ -131,7 +136,7 @@ private class ClientActor(driverArgs: ClientArguments, conf: SparkConf)
       if (success) {
         activeMasterActor = context.actorSelection(sender.path)
         pollAndReportStatus(driverId.get)
-      } else if (!message.contains("Can only")) {
+      } else if (!message.startsWith(Utils.MASTER_NOT_ALIVE_STRING)) {
         System.exit(-1)
       }
 
@@ -141,16 +146,16 @@ private class ClientActor(driverArgs: ClientArguments, conf: SparkConf)
       if (success) {
         activeMasterActor = context.actorSelection(sender.path)
         pollAndReportStatus(driverId)
-      } else if (!message.contains("Can only")) {
+      } else if (!message.startsWith(Utils.MASTER_NOT_ALIVE_STRING)) {
         System.exit(-1)
       }
 
     case DisassociatedEvent(_, remoteAddress, _) =>
       if (!lostMasters.contains(remoteAddress)) {
         println(s"Error connecting to master $remoteAddress.")
         lostMasters += remoteAddress
-        if (lostMasters.size >= mastersActor.size) {
-          println(s"No master is available, exiting.")
+        if (lostMasters.size >= masterActors.size) {
+          println("No master is available, exiting.")
           System.exit(-1)
         }
       }
@@ -160,8 +165,8 @@ private class ClientActor(driverArgs: ClientArguments, conf: SparkConf)
         println(s"Error connecting to master ($remoteAddress).")
         println(s"Cause was: $cause")
         lostMasters += remoteAddress
-        if (lostMasters.size >= mastersActor.size) {
-          println(s"No master is available, exiting.")
+        if (lostMasters.size >= masterActors.size) {
+          println("No master is available, exiting.")
           System.exit(-1)
         }
       }

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

@@ -22,8 +22,7 @@ import java.net.{URI, URISyntaxException}
 import scala.collection.mutable.ListBuffer
 
 import org.apache.log4j.Level
-
-import org.apache.spark.util.{IntParam, MemoryParam}
+import org.apache.spark.util.{IntParam, MemoryParam, Utils}
 
 /**
  * Command-line parser for the driver client.
@@ -80,13 +79,13 @@ private[deploy] class ClientArguments(args: Array[String]) {
       }
 
       jarUrl = _jarUrl
-      masters = _master.stripPrefix("spark://").split(",").map("spark://" + _)
+      masters = Utils.splitMasterAdress(_master)
       mainClass = _mainClass
       _driverOptions ++= tail
 
     case "kill" :: _master :: _driverId :: tail =>
       cmd = "kill"
-      masters = _master.stripPrefix("spark://").split(",").map("spark://" + _)
+      masters = Utils.splitMasterAdress(_master)
       driverId = _driverId
 
     case _ =>

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

@@ -254,7 +254,8 @@ private[master] class Master(
 
     case RequestSubmitDriver(description) => {
       if (state != RecoveryState.ALIVE) {
-        val msg = s"Can only accept driver submissions in ALIVE state. Current state: $state."
+        val msg = s"${Utils.MASTER_NOT_ALIVE_STRING}$state. " +
+          "Can only accept driver submissions in ALIVE state."
         sender ! SubmitDriverResponse(false, None, msg)
       } else {
         logInfo("Driver submitted " + description.command.mainClass)
@@ -274,7 +275,7 @@ private[master] class Master(
 
     case RequestKillDriver(driverId) => {
       if (state != RecoveryState.ALIVE) {
-        val msg = s"Can only kill drivers in ALIVE state. Current state: $state."
+        val msg = s"${Utils.MASTER_NOT_ALIVE_STRING}$state. Can only kill drivers in ALIVE state."
         sender ! KillDriverResponse(driverId, success = false, msg)
       } else {
         logInfo("Asked to kill driver " + driverId)
@@ -306,7 +307,8 @@ private[master] class Master(
 
     case RequestDriverStatus(driverId) => {
       if (state != RecoveryState.ALIVE) {
-        val msg = s"Can only request driver status in ALIVE state. Current state: $state."
+        val msg = s"${Utils.MASTER_NOT_ALIVE_STRING}$state. " +
+          "Can only request driver status in ALIVE state."
         sender ! DriverStatusResponse(found = false, None, None, None, Some(new Exception(msg)))
       } else {
         (drivers ++ completedDrivers).find(_.id == driverId) match {

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

@@ -21,13 +21,14 @@ import java.io.{DataOutputStream, FileNotFoundException}
 import java.net.{ConnectException, HttpURLConnection, SocketException, URL}
 import javax.servlet.http.HttpServletResponse
 
+import scala.collection.mutable
 import scala.io.Source
 
 import com.fasterxml.jackson.core.JsonProcessingException
 import com.google.common.base.Charsets
 
 import org.apache.spark.{Logging, SparkConf, SPARK_VERSION => sparkVersion}
-import scala.collection.mutable
+import org.apache.spark.util.Utils
 
 /**
  * A client that submits applications to the standalone Master using a REST protocol.
@@ -52,11 +53,14 @@ import scala.collection.mutable
  * is a mismatch, the server will respond with the highest protocol version it supports. A future
  * implementation of this client can use that information to retry using the version specified
  * by the server.
+ *
+ * Now we don't support retry in case submission failed. In HA mode, client will submit request to
+ * all masters and see which one could handle it.
  */
 private[deploy] class StandaloneRestClient(master: String) extends Logging {
   import StandaloneRestClient._
 
-  val masters: Array[String] = master.stripPrefix("spark://").split(",").map("spark://" + _)
+  private val masters: Array[String] = Utils.splitMasterAdress(master)
 
   private val lostMasters = new mutable.HashSet[String]
 
@@ -87,10 +91,15 @@ private[deploy] class StandaloneRestClient(master: String) extends Logging {
             handleUnexpectedRestResponse(unexpected)
         }
       } catch {
-        case e @ (_: SubmitRestConnectionException | _: ConnectException) =>
-          if(handleSubmitRestConnectionException(m)) {
+        case unreachable @ (_: FileNotFoundException | _: SocketException | _: ConnectException) =>
+          if(handleConnectionException(m)) {
             throw new SubmitRestConnectionException(
-              "No master is available for createSubmission.", new Throwable(""))
+              s"Unable to connect to server", unreachable)
+          }
+        case malformed @ (_: JsonProcessingException | _: SubmitRestProtocolException) =>
+          if(handleConnectionException(m)) {
+            throw new SubmitRestProtocolException(
+              "Malformed response received from server", malformed)
           }
       }
     }
@@ -109,18 +118,23 @@ private[deploy] class StandaloneRestClient(master: String) extends Logging {
         response = post(url)
         response match {
           case k: KillSubmissionResponse =>
-            if (!k.message.contains("Can only")) {
+            if (!k.message.startsWith(Utils.MASTER_NOT_ALIVE_STRING)) {
               handleRestResponse(k)
               suc = true
             }
           case unexpected =>
             handleUnexpectedRestResponse(unexpected)
         }
       } catch {
-        case e @ (_: SubmitRestConnectionException | _: ConnectException) =>
-          if(handleSubmitRestConnectionException(m)) {
+        case unreachable @ (_: FileNotFoundException | _: SocketException | _: ConnectException) =>
+          if(handleConnectionException(m)) {
             throw new SubmitRestConnectionException(
-              "No master is available for killSubmission.", new Throwable(""))
+              s"Unable to connect to server", unreachable)
+          }
+        case malformed @ (_: JsonProcessingException | _: SubmitRestProtocolException) =>
+          if(handleConnectionException(m)) {
+            throw new SubmitRestProtocolException(
+              "Malformed response received from server", malformed)
           }
       }
     }
@@ -149,10 +163,15 @@ private[deploy] class StandaloneRestClient(master: String) extends Logging {
             handleUnexpectedRestResponse(unexpected)
         }
       } catch {
-        case e @ (_: SubmitRestConnectionException | _: ConnectException) =>
-          if(handleSubmitRestConnectionException(m)) {
+        case unreachable @ (_: FileNotFoundException | _: SocketException | _: ConnectException) =>
+          if(handleConnectionException(m)) {
             throw new SubmitRestConnectionException(
-              "No master is available for requestSubmissionStatus.", new Throwable(""))
+              s"Unable to connect to server", unreachable)
+          }
+        case malformed @ (_: JsonProcessingException | _: SubmitRestProtocolException) =>
+          if(handleConnectionException(m)) {
+            throw new SubmitRestProtocolException(
+              "Malformed response received from server", malformed)
           }
       }
     }
@@ -213,39 +232,30 @@ private[deploy] class StandaloneRestClient(master: String) extends Logging {
    * Exposed for testing.
    */
   private[rest] def readResponse(connection: HttpURLConnection): SubmitRestProtocolResponse = {
-    try {
-      val dataStream =
-        if (connection.getResponseCode == HttpServletResponse.SC_OK) {
-          connection.getInputStream
-        } else {
-          connection.getErrorStream
-        }
-      // If the server threw an exception while writing a response, it will not have a body
-      if (dataStream == null) {
-        throw new SubmitRestProtocolException("Server returned empty body")
-      }
-      val responseJson = Source.fromInputStream(dataStream).mkString
-      logDebug(s"Response from the server:\n$responseJson")
-      val response = SubmitRestProtocolMessage.fromJson(responseJson)
-      response.validate()
-      response match {
-        // If the response is an error, log the message
-        case error: ErrorResponse =>
-          logError(s"Server responded with error:\n${error.message}")
-          error
-        // Otherwise, simply return the response
-        case response: SubmitRestProtocolResponse => response
-        case unexpected =>
-          throw new SubmitRestProtocolException(
-            s"Message received from server was not a response:\n${unexpected.toJson}")
+    val dataStream =
+      if (connection.getResponseCode == HttpServletResponse.SC_OK) {
+        connection.getInputStream
+      } else {
+        connection.getErrorStream
       }
-    } catch {
-      case unreachable @ (_: FileNotFoundException | _: SocketException) =>
-        throw new SubmitRestConnectionException(
-          s"Unable to connect to server ${connection.getURL}", unreachable)
-      case malformed @ (_: JsonProcessingException | _: SubmitRestProtocolException) =>
+    // If the server threw an exception while writing a response, it will not have a body
+    if (dataStream == null) {
+      throw new SubmitRestProtocolException("Server returned empty body")
+    }
+    val responseJson = Source.fromInputStream(dataStream).mkString
+    logDebug(s"Response from the server:\n$responseJson")
+    val response = SubmitRestProtocolMessage.fromJson(responseJson)
+    response.validate()
+    response match {
+      // If the response is an error, log the message
+      case error: ErrorResponse =>
+        logError(s"Server responded with error:\n${error.message}")
+        error
+      // Otherwise, simply return the response
+      case response: SubmitRestProtocolResponse => response
+      case unexpected =>
         throw new SubmitRestProtocolException(
-          "Malformed response received from server", malformed)
+          s"Message received from server was not a response:\n${unexpected.toJson}")
     }
   }
 
@@ -338,10 +348,13 @@ private[deploy] class StandaloneRestClient(master: String) extends Logging {
     logError(s"Error: Server responded with message of unexpected type ${unexpected.messageType}.")
   }
 
-  private def handleSubmitRestConnectionException(url: String): Boolean = {
-    if (!lostMasters.contains(url)) {
-      logWarning(s"Unable to connect to server ${url}.")
-      lostMasters += url
+  /**
+   * When a connection exception was caught, we see whether all masters are lost.
+   */
+  private def handleConnectionException(masterUrl: String): Boolean = {
+    if (!lostMasters.contains(masterUrl)) {
+      logWarning(s"Unable to connect to server ${masterUrl}.")
+      lostMasters += masterUrl
     }
     lostMasters.size >= masters.size
   }

core/src/main/scala/org/apache/spark/deploy/worker/WorkerArguments.scala

@@ -105,7 +105,7 @@ private[worker] class WorkerArguments(args: Array[String], conf: SparkConf) {
       if (masters != null) {  // Two positional arguments were given
         printUsageAndExit(1)
       }
-      masters = value.stripPrefix("spark://").split(",").map("spark://" + _)
+      masters = Utils.splitMasterAdress(value)
       parse(tail)
 
     case Nil =>

core/src/main/scala/org/apache/spark/util/Utils.scala

@@ -2058,6 +2058,11 @@ private[spark] object Utils extends Logging {
       .getOrElse(UserGroupInformation.getCurrentUser().getUserName())
   }
 
+  def splitMasterAdress(masterAddr: String): Array[String] = {
+    masterAddr.stripPrefix("spark://").split(",").map("spark://" + _)
+  }
+
+  val MASTER_NOT_ALIVE_STRING = "Current state is not alive: "
 }
 
 /**