Comment changes + minor code refactoring across the board

This is a comprehensive pass across the entire patch. The most
semantically significant change included in this commit is the
correction of server behavior when an error occurs. Previously
we will attempt to find the unknown fields of a faulty request,
which would cause an exception on the server. This is, however,
a client error not a server error. This behavior is now fixed.

Author: Andrew Or

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

@@ -101,20 +101,19 @@ object SparkSubmit {
     }
   }
 
-  /**
-   * Kill an existing driver using the REST application submission protocol.
-   * Standalone cluster mode only.
-   */
+  /** Kill an existing submission using the REST protocol. Standalone cluster mode only. */
   private def kill(args: SparkSubmitArguments): Unit = {
-    new StandaloneRestClient().killSubmission(args.master, args.driverToKill)
+    new StandaloneRestClient()
+      .killSubmission(args.master, args.submissionToKill)
   }
 
   /**
-   * Request the status of an existing driver using the REST application submission protocol.
+   * Request the status of an existing submission using the REST protocol.
    * Standalone cluster mode only.
    */
   private def requestStatus(args: SparkSubmitArguments): Unit = {
-    new StandaloneRestClient().requestSubmissionStatus(args.master, args.driverToRequestStatusFor)
+    new StandaloneRestClient()
+      .requestSubmissionStatus(args.master, args.submissionToRequestStatusFor)
   }
 
   /**
@@ -300,6 +299,7 @@ object SparkSubmit {
         sysProp = "spark.driver.extraLibraryPath"),
 
       // Standalone cluster only
+      // Do not set CL arguments here because there are multiple possibilities for the main class
       OptionAssigner(args.jars, STANDALONE, CLUSTER, sysProp = "spark.jars"),
       OptionAssigner(args.ivyRepoPath, STANDALONE, CLUSTER, sysProp = "spark.jars.ivy"),
       OptionAssigner(args.driverMemory, STANDALONE, CLUSTER, sysProp = "spark.driver.memory"),
@@ -368,8 +368,7 @@ object SparkSubmit {
     }
 
     // In standalone cluster mode, use the REST client to submit the application (Spark 1.3+).
-    // All parameters except application arguments are expected to be passed to the main class
-    // through system properties.
+    // All Spark parameters are expected to be passed to the client through system properties.
     if (args.isStandaloneCluster) {
       if (args.useRest) {
         childMainClass = "org.apache.spark.deploy.rest.StandaloneRestClient"
@@ -383,7 +382,6 @@ object SparkSubmit {
         childArgs += "launch"
         childArgs += (args.master, args.primaryResource, args.mainClass)
       }
-      // Whether or not we use REST, pass application arguments through the command line
       if (args.childArgs != null) {
         childArgs ++= args.childArgs
       }

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

@@ -62,8 +62,8 @@ private[spark] class SparkSubmitArguments(args: Seq[String], env: Map[String, St
   var useRest: Boolean = true
   var supervise: Boolean = false
   var driverCores: String = null
-  var driverToKill: String = null
-  var driverToRequestStatusFor: String = null
+  var submissionToKill: String = null
+  var submissionToRequestStatusFor: String = null
 
   /** Default properties present in the currently defined defaults file. */
   lazy val defaultSparkProperties: HashMap[String, String] = {
@@ -220,21 +220,21 @@ private[spark] class SparkSubmitArguments(args: Seq[String], env: Map[String, St
   }
 
   private def validateKillArguments(): Unit = {
-    if (!isStandaloneCluster) {
-      SparkSubmit.printErrorAndExit("Killing drivers is only supported in standalone cluster mode")
+    if (!master.startsWith("spark://")) {
+      SparkSubmit.printErrorAndExit("Killing submissions is only supported in standalone mode!")
     }
-    if (driverToKill == null) {
-      SparkSubmit.printErrorAndExit("Please specify a driver to kill")
+    if (submissionToKill == null) {
+      SparkSubmit.printErrorAndExit("Please specify a submission to kill.")
     }
   }
 
   private def validateStatusRequestArguments(): Unit = {
-    if (!isStandaloneCluster) {
+    if (!master.startsWith("spark://")) {
       SparkSubmit.printErrorAndExit(
-        "Requesting driver statuses is only supported in standalone cluster mode")
+        "Requesting submission statuses is only supported in standalone mode!")
     }
-    if (driverToRequestStatusFor == null) {
-      SparkSubmit.printErrorAndExit("Please specify a driver to request status for")
+    if (submissionToRequestStatusFor == null) {
+      SparkSubmit.printErrorAndExit("Please specify a submission to request status for.")
     }
   }
 
@@ -351,15 +351,15 @@ private[spark] class SparkSubmitArguments(args: Seq[String], env: Map[String, St
         parse(tail)
 
       case ("--kill") :: value :: tail =>
-        driverToKill = value
+        submissionToKill = value
         if (action != null) {
           SparkSubmit.printErrorAndExit(s"Action cannot be both $action and $KILL.")
         }
         action = KILL
         parse(tail)
 
       case ("--status") :: value :: tail =>
-        driverToRequestStatusFor = value
+        submissionToRequestStatusFor = value
         if (action != null) {
           SparkSubmit.printErrorAndExit(s"Action cannot be both $action and $REQUEST_STATUS.")
         }
@@ -439,6 +439,9 @@ private[spark] class SparkSubmitArguments(args: Seq[String], env: Map[String, St
     }
     outStream.println(
       """Usage: spark-submit [options] <app jar | python file> [app arguments]
+        |Usage: spark-submit --kill [submission ID] --master [spark://...]
+        |Usage: spark-submit --status [submission ID] --master [spark://...]
+        |
         |Options:
         |  --master MASTER_URL         spark://host:port, mesos://host:port, yarn, or local.
         |  --deploy-mode DEPLOY_MODE   Whether to launch the driver program locally ("client") or

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

@@ -127,7 +127,7 @@ private[spark] class Master(
   private val restServer =
     if (restServerEnabled) {
       val port = conf.getInt("spark.master.rest.port", 6066)
-      Some(new StandaloneRestServer(this, host, port))
+      Some(new StandaloneRestServer(host, port, self, masterUrl, conf))
     } else {
       None
     }

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

@@ -22,6 +22,7 @@ import java.net.{HttpURLConnection, SocketException, URL}
 
 import scala.io.Source
 
+import com.fasterxml.jackson.databind.JsonMappingException
 import com.google.common.base.Charsets
 
 import org.apache.spark.{Logging, SparkConf, SPARK_VERSION => sparkVersion}
@@ -54,15 +55,15 @@ private[spark] class StandaloneRestClient extends Logging {
   import StandaloneRestClient._
 
   /**
-   * Submit an application specified by the provided arguments.
+   * Submit an application specified by the parameters in the provided request.
    *
    * If the submission was successful, poll the status of the submission and report
    * it to the user. Otherwise, report the error message provided by the server.
    */
   def createSubmission(
       master: String,
       request: CreateSubmissionRequest): SubmitRestProtocolResponse = {
-    logInfo(s"Submitting a request to launch a driver in $master.")
+    logInfo(s"Submitting a request to launch an application in $master.")
     validateMaster(master)
     val url = getSubmitUrl(master)
     val response = postJson(url, request.toJson)
@@ -103,6 +104,24 @@ private[spark] class StandaloneRestClient extends Logging {
     response
   }
 
+  /** Construct a message that captures the specified parameters for submitting an application. */
+  def constructSubmitRequest(
+      appResource: String,
+      mainClass: String,
+      appArgs: Array[String],
+      sparkProperties: Map[String, String],
+      environmentVariables: Map[String, String]): CreateSubmissionRequest = {
+    val message = new CreateSubmissionRequest
+    message.clientSparkVersion = sparkVersion
+    message.appResource = appResource
+    message.mainClass = mainClass
+    message.appArgs = appArgs
+    message.sparkProperties = sparkProperties
+    message.environmentVariables = environmentVariables
+    message.validate()
+    message
+  }
+
   /** Send a GET request to the specified URL. */
   private def get(url: URL): SubmitRestProtocolResponse = {
     logDebug(s"Sending GET request to server at $url.")
@@ -134,40 +153,33 @@ private[spark] class StandaloneRestClient extends Logging {
   }
 
   /**
-   * Read the response from the given connection.
-   *
-   * The response is expected to represent a [[SubmitRestProtocolResponse]] in the form of JSON.
-   * Additionally, this validates the response to ensure that it is properly constructed.
-   * If the response represents an error, report the message from the server.
+   * Read the response from the server and return it as a validated [[SubmitRestProtocolResponse]].
+   * If the response represents an error, report the embedded message to the user.
    */
   private def readResponse(connection: HttpURLConnection): SubmitRestProtocolResponse = {
     try {
       val responseJson = Source.fromInputStream(connection.getInputStream).mkString
       logDebug(s"Response from the server:\n$responseJson")
       val response = SubmitRestProtocolMessage.fromJson(responseJson)
-      // The response should have already been validated on the server.
-      // In case this is not true, validate it ourselves to avoid potential NPEs.
-      try {
-        response.validate()
-      } catch {
-        case e: SubmitRestProtocolException =>
-          throw new SubmitRestProtocolException("Malformed response received from server", e)
-      }
-      // If the response is an error, log the message
-      // Otherwise, simply return the response
+      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}")
       }
     } catch {
-      case e @ (_: FileNotFoundException | _: SocketException) =>
+      case unreachable @ (_: FileNotFoundException | _: SocketException) =>
         throw new SubmitRestConnectionException(
-          s"Unable to connect to server ${connection.getURL}", e)
+          s"Unable to connect to server ${connection.getURL}", unreachable)
+      case malformed @ (_: SubmitRestProtocolException | _: JsonMappingException) =>
+        throw new SubmitRestProtocolException(
+          "Malformed response received from server", malformed)
     }
   }
 
@@ -202,24 +214,6 @@ private[spark] class StandaloneRestClient extends Logging {
     }
   }
 
-  /** Construct a message that captures the specified parameters for submitting an application. */
-  def constructSubmitRequest(
-      appResource: String,
-      mainClass: String,
-      appArgs: Array[String],
-      sparkProperties: Map[String, String],
-      environmentVariables: Map[String, String]): CreateSubmissionRequest = {
-    val message = new CreateSubmissionRequest
-    message.clientSparkVersion = sparkVersion
-    message.appResource = appResource
-    message.mainClass = mainClass
-    message.appArgs = appArgs
-    message.sparkProperties = sparkProperties
-    message.environmentVariables = environmentVariables
-    message.validate()
-    message
-  }
-
   /** Report the status of a newly created submission. */
   private def reportSubmissionStatus(
       master: String,
@@ -292,7 +286,7 @@ private[spark] object StandaloneRestClient {
   val REPORT_DRIVER_STATUS_MAX_TRIES = 10
   val PROTOCOL_VERSION = "v1"
 
-  /** Submit an application, assuming parameters are specified through system properties. */
+  /** Submit an application, assuming Spark parameters are specified through system properties. */
   def main(args: Array[String]): Unit = {
     if (args.size < 2) {
       sys.error("Usage: StandaloneRestClient [app resource] [main class] [app args*]")
@@ -301,12 +295,13 @@ private[spark] object StandaloneRestClient {
     val appResource = args(0)
     val mainClass = args(1)
     val appArgs = args.slice(2, args.size)
-    val client = new StandaloneRestClient
-    val master = sys.props.get("spark.master").getOrElse {
+    val conf = new SparkConf
+    val master = conf.getOption("spark.master").getOrElse {
       throw new IllegalArgumentException("'spark.master' must be set.")
     }
-    val sparkProperties = new SparkConf().getAll.toMap
+    val sparkProperties = conf.getAll.toMap
     val environmentVariables = sys.env.filter { case (k, _) => k.startsWith("SPARK_") }
+    val client = new StandaloneRestClient
     val submitRequest = client.constructSubmitRequest(
       appResource, mainClass, appArgs, sparkProperties, environmentVariables)
     client.createSubmission(master, submitRequest)