Address review comments.

Author: tnachen

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

@@ -22,6 +22,8 @@ import java.lang.reflect.{InvocationTargetException, Modifier, UndeclaredThrowab
 import java.net.URL
 import java.security.PrivilegedExceptionAction
 
+import scala.collection.mutable.{ArrayBuffer, HashMap, Map}
+
 import org.apache.hadoop.fs.Path
 import org.apache.hadoop.security.UserGroupInformation
 import org.apache.ivy.Ivy
@@ -38,7 +40,6 @@ import org.apache.spark.SPARK_VERSION
 import org.apache.spark.deploy.rest._
 import org.apache.spark.util.{ChildFirstURLClassLoader, MutableURLClassLoader, Utils}
 
-import scala.collection.mutable.{ArrayBuffer, HashMap, Map}
 
 /**
  * Whether to submit, kill, or request the status of an application.
@@ -380,10 +381,6 @@ object SparkSubmit {
       OptionAssigner(args.driverExtraLibraryPath, ALL_CLUSTER_MGRS, ALL_DEPLOY_MODES,
         sysProp = "spark.driver.extraLibraryPath"),
 
-      // Standalone cluster only
-      // Do not set CL arguments here because there are multiple possibilities for the main class
-      OptionAssigner(args.ivyRepoPath, STANDALONE, CLUSTER, sysProp = "spark.jars.ivy"),
-
       // Yarn client only
       OptionAssigner(args.queue, YARN, CLIENT, sysProp = "spark.yarn.queue"),
       OptionAssigner(args.numExecutors, YARN, CLIENT, sysProp = "spark.executor.instances"),
@@ -418,7 +415,8 @@ object SparkSubmit {
       OptionAssigner(args.driverCores, STANDALONE | MESOS, CLUSTER,
         sysProp = "spark.driver.cores"),
       OptionAssigner(args.supervise.toString, STANDALONE | MESOS, CLUSTER,
-        sysProp = "spark.driver.supervise")
+        sysProp = "spark.driver.supervise"),
+      OptionAssigner(args.ivyRepoPath, STANDALONE, CLUSTER, sysProp = "spark.jars.ivy")
     )
 
     // In client mode, launch the application main class directly
@@ -457,7 +455,7 @@ object SparkSubmit {
     // 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.RestClient"
+        childMainClass = "org.apache.spark.deploy.rest.RestSubmissionClient"
         childArgs += (args.primaryResource, args.mainClass)
       } else {
         // In legacy standalone cluster mode, use Client as a wrapper around the user class
@@ -503,7 +501,7 @@ object SparkSubmit {
 
     if (isMesosCluster) {
       assert(args.useRest, "Mesos cluster mode is only supported through the REST submission API")
-      childMainClass = "org.apache.spark.deploy.rest.RestClient"
+      childMainClass = "org.apache.spark.deploy.rest.RestSubmissionClient"
       childArgs += (args.primaryResource, args.mainClass)
       if (args.childArgs != null) {
         childArgs ++= args.childArgs

core/src/main/scala/org/apache/spark/deploy/mesos/MesosClusterDispatcher.scala

@@ -19,7 +19,6 @@ package org.apache.spark.deploy.mesos
 
 import java.util.concurrent.CountDownLatch
 
-import org.apache.spark
 import org.apache.spark.deploy.mesos.ui.MesosClusterUI
 import org.apache.spark.deploy.rest.mesos.MesosRestServer
 import org.apache.spark.scheduler.cluster.mesos._
@@ -50,17 +49,14 @@ private[mesos] class MesosClusterDispatcher(
     conf: SparkConf)
   extends Logging {
 
-  private def publicAddress(conf: SparkConf, defaultAddress: String): String = {
-    val envVar = conf.getenv("SPARK_PUBLIC_DNS")
-    if (envVar != null) envVar else defaultAddress
-  }
-
+  private val publicAddress = Option(conf.getenv("SPARK_PUBLIC_DNS")).getOrElse(args.host)
   private val recoveryMode = conf.get("spark.mesos.deploy.recoveryMode", "NONE").toUpperCase()
   logInfo("Recovery mode in Mesos dispatcher set to: " + recoveryMode)
 
   private val engineFactory = recoveryMode match {
     case "NONE" => new BlackHoleMesosClusterPersistenceEngineFactory
     case "ZOOKEEPER" => new ZookeeperMesosClusterPersistenceEngineFactory(conf)
+    case _ => throw new IllegalArgumentException("Unsupported recovery mode: " + recoveryMode)
   }
 
   private val scheduler = new MesosClusterScheduler(engineFactory, conf)
@@ -70,7 +66,7 @@ private[mesos] class MesosClusterDispatcher(
     new SecurityManager(conf),
     args.webUiPort,
     conf,
-    publicAddress(conf, args.host),
+    publicAddress,
     scheduler)
 
   private val shutdownLatch = new CountDownLatch(1)
@@ -94,7 +90,7 @@ private[mesos] class MesosClusterDispatcher(
   }
 }
 
-private[mesos] object MesosClusterDispatcher extends spark.Logging {
+private[mesos] object MesosClusterDispatcher extends Logging {
   def main(args: Array[String]) {
     SignalLogger.register(log)
     val conf = new SparkConf
@@ -105,9 +101,7 @@ private[mesos] object MesosClusterDispatcher extends spark.Logging {
       conf.set("spark.mesos.deploy.recoveryMode", "ZOOKEEPER")
       conf.set("spark.mesos.deploy.zookeeper.url", z)
     }
-    val dispatcher = new MesosClusterDispatcher(
-      dispatcherArgs,
-      conf)
+    val dispatcher = new MesosClusterDispatcher(dispatcherArgs, conf)
     dispatcher.start()
     val shutdownHook = new Thread() {
       override def run() {

core/src/main/scala/org/apache/spark/deploy/mesos/MesosClusterDispatcherArguments.scala

@@ -74,7 +74,7 @@ private[mesos] class MesosClusterDispatcherArguments(args: Array[String], conf:
     case Nil => {
       if (masterUrl == null) {
         System.err.println("--master is required")
-        System.exit(1)
+        printUsageAndExit(1)
       }
     }
 

core/src/main/scala/org/apache/spark/deploy/mesos/MesosDriverDescription.scala

@@ -20,15 +20,15 @@ package org.apache.spark.deploy.mesos
 import java.util.Date
 
 import org.apache.spark.deploy.Command
-import org.apache.spark.scheduler.cluster.mesos.RetryState
+import org.apache.spark.scheduler.cluster.mesos.MesosClusterRetryState
 
 /**
  * Describes a Spark driver that is submitted from the
  * [[org.apache.spark.deploy.rest.mesos.MesosRestServer]], to be launched by
  * [[org.apache.spark.scheduler.cluster.mesos.MesosClusterScheduler]].
  * @param jarUrl URL to the application jar
  * @param mem Amount of memory for the driver
- * @param cores Amount of cores for the driver
+ * @param cores Number of cores for the driver
  * @param supervise Supervise the driver for long running app
  * @param command The command to launch the driver.
  * @param schedulerProperties Extra properties to pass the Mesos scheduler
@@ -43,8 +43,9 @@ private[spark] class MesosDriverDescription(
     val schedulerProperties: Map[String, String],
     val submissionId: String,
     val submissionDate: Date,
-    val retryState: Option[RetryState] = None)
+    val retryState: Option[MesosClusterRetryState] = None)
   extends Serializable {
+
   def copy(
       name: String = name,
       jarUrl: String = jarUrl,
@@ -53,11 +54,12 @@ private[spark] class MesosDriverDescription(
       supervise: Boolean = supervise,
       command: Command = command,
       schedulerProperties: Map[String, String] = schedulerProperties,
-      retryState: Option[RetryState] = retryState,
       submissionId: String = submissionId,
-      submissionDate: Date = submissionDate): MesosDriverDescription = {
+      submissionDate: Date = submissionDate,
+      retryState: Option[MesosClusterRetryState] = retryState): MesosDriverDescription = {
     new MesosDriverDescription(name, jarUrl, mem, cores, supervise, command, schedulerProperties,
       submissionId, submissionDate, retryState)
   }
+
   override def toString: String = s"MesosDriverDescription (${command.mainClass})"
 }

core/src/main/scala/org/apache/spark/deploy/mesos/ui/MesosClusterPage.scala

@@ -19,16 +19,16 @@ package org.apache.spark.deploy.mesos.ui
 
 import javax.servlet.http.HttpServletRequest
 
+import scala.xml.Node
+
 import org.apache.mesos.Protos.TaskStatus
 import org.apache.spark.deploy.mesos.MesosDriverDescription
-import org.apache.spark.scheduler.cluster.mesos.MesosClusterTaskState
+import org.apache.spark.scheduler.cluster.mesos.MesosClusterSubmissionState
 import org.apache.spark.ui.{UIUtils, WebUIPage}
 
-import scala.xml.Node
-
 private[mesos] class MesosClusterPage(parent: MesosClusterUI) extends WebUIPage("") {
   def render(request: HttpServletRequest): Seq[Node] = {
-    val state = parent.scheduler.getState()
+    val state = parent.scheduler.getSchedulerState()
     val queuedHeaders = Seq("Driver ID", "Submit Date", "Main Class", "Driver Resources")
     val driverHeaders = queuedHeaders ++
       Seq("Start Date", "Mesos Slave ID", "State")
@@ -37,7 +37,7 @@ private[mesos] class MesosClusterPage(parent: MesosClusterUI) extends WebUIPage(
     val queuedTable = UIUtils.listingTable(queuedHeaders, queuedRow, state.queuedDrivers)
     val launchedTable = UIUtils.listingTable(driverHeaders, driverRow, state.launchedDrivers)
     val finishedTable = UIUtils.listingTable(driverHeaders, driverRow, state.finishedDrivers)
-    val retryTable = UIUtils.listingTable(retryHeaders, retryRow, state.retryList)
+    val retryTable = UIUtils.listingTable(retryHeaders, retryRow, state.pendingRetryDrivers)
     val content =
       <p>Mesos Framework ID: {state.frameworkId}</p>
       <div class="row-fluid">
@@ -64,15 +64,15 @@ private[mesos] class MesosClusterPage(parent: MesosClusterUI) extends WebUIPage(
     </tr>
   }
 
-  private def driverRow(state: MesosClusterTaskState): Seq[Node] = {
+  private def driverRow(state: MesosClusterSubmissionState): Seq[Node] = {
     <tr>
-      <td>{state.submission.submissionId}</td>
-      <td>{state.submission.submissionDate}</td>
-      <td>{state.submission.command.mainClass}</td>
-      <td>cpus: {state.submission.cores}, mem: {state.submission.mem}</td>
+      <td>{state.driverDescription.submissionId}</td>
+      <td>{state.driverDescription.submissionDate}</td>
+      <td>{state.driverDescription.command.mainClass}</td>
+      <td>cpus: {state.driverDescription.cores}, mem: {state.driverDescription.mem}</td>
       <td>{state.startDate}</td>
       <td>{state.slaveId.getValue}</td>
-      <td>{stateString(state.taskState)}</td>
+      <td>{stateString(state.mesosTaskStatus)}</td>
     </tr>
   }
 

core/src/main/scala/org/apache/spark/deploy/mesos/ui/MesosClusterUI.scala

@@ -25,7 +25,7 @@ import org.apache.spark.ui.{SparkUI, WebUI}
 /**
  * UI that displays driver results from the [[org.apache.spark.deploy.mesos.MesosClusterDispatcher]]
  */
-private [spark] class MesosClusterUI(
+private[spark] class MesosClusterUI(
     securityManager: SecurityManager,
     port: Int,
     conf: SparkConf,
@@ -43,6 +43,6 @@ private [spark] class MesosClusterUI(
   }
 }
 
-private[spark] object MesosClusterUI {
+private object MesosClusterUI {
   val STATIC_RESOURCE_DIR = SparkUI.STATIC_RESOURCE_DIR
 }

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

@@ -332,7 +332,7 @@ private[spark] object RestSubmissionClient {
 
   def main(args: Array[String]): Unit = {
     if (args.size < 2) {
-      sys.error("Usage: RestClient [app resource] [main class] [app args*]")
+      sys.error("Usage: RestSubmissionClient [app resource] [main class] [app args*]")
       sys.exit(1)
     }
     val appResource = args(0)

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

@@ -47,9 +47,9 @@ import org.apache.spark.util.Utils
  * server error.
  */
 private[spark] abstract class RestSubmissionServer(
-  val host: String,
-  val requestedPort: Int,
-  val masterConf: SparkConf) extends Logging {
+    val host: String,
+    val requestedPort: Int,
+    val masterConf: SparkConf) extends Logging {
   protected val submitRequestServlet: SubmitRequestServlet
   protected val killRequestServlet: KillRequestServlet
   protected val statusRequestServlet: StatusRequestServlet
@@ -190,8 +190,7 @@ private[rest] abstract class RestServlet extends HttpServlet with Logging {
 /**
  * A servlet for handling kill requests passed to the [[RestSubmissionServer]].
  */
-private[rest] abstract class KillRequestServlet
-  extends RestServlet {
+private[rest] abstract class KillRequestServlet extends RestServlet {
 
   /**
    * If a submission ID is specified in the URL, have the Master kill the corresponding

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

@@ -56,9 +56,13 @@ private[deploy] class StandaloneRestServer(
     masterActor: ActorRef,
     masterUrl: String)
   extends RestSubmissionServer(host, requestedPort, masterConf) {
-  val submitRequestServlet = new StandaloneSubmitRequestServlet(masterActor, masterUrl, masterConf)
-  val killRequestServlet = new StandaloneKillRequestServlet(masterActor, masterConf)
-  val statusRequestServlet = new StandaloneStatusRequestServlet(masterActor, masterConf)
+
+  protected override val submitRequestServlet =
+    new StandaloneSubmitRequestServlet(masterActor, masterUrl, masterConf)
+  protected override val killRequestServlet =
+    new StandaloneKillRequestServlet(masterActor, masterConf)
+  protected override val statusRequestServlet =
+    new StandaloneStatusRequestServlet(masterActor, masterConf)
 }
 
 /**

core/src/main/scala/org/apache/spark/deploy/rest/mesos/MesosRestServer.scala

@@ -36,18 +36,21 @@ import org.apache.spark.{SPARK_VERSION => sparkVersion, SparkConf}
  * All requests are forwarded to
  * [[org.apache.spark.scheduler.cluster.mesos.MesosClusterScheduler]].
  * This is intended to be used in Mesos cluster mode only.
- * For more details about the RestServer Spark protocol and status codes please refer to
- * [[RestSubmissionServer]] javadocs.
+ * For more details about the REST submission please refer to [[RestSubmissionServer]] javadocs.
  */
 private[spark] class MesosRestServer(
     host: String,
     requestedPort: Int,
     masterConf: SparkConf,
     scheduler: MesosClusterScheduler)
   extends RestSubmissionServer(host, requestedPort, masterConf) {
-  protected val submitRequestServlet = new MesosSubmitRequestServlet(scheduler, masterConf)
-  protected val killRequestServlet = new MesosKillRequestServlet(scheduler, masterConf)
-  protected val statusRequestServlet = new MesosStatusRequestServlet(scheduler, masterConf)
+
+  protected override val submitRequestServlet =
+    new MesosSubmitRequestServlet(scheduler, masterConf)
+  protected override val killRequestServlet =
+    new MesosKillRequestServlet(scheduler, masterConf)
+  protected override val statusRequestServlet =
+    new MesosStatusRequestServlet(scheduler, masterConf)
 }
 
 private[deploy] class MesosSubmitRequestServlet(
@@ -59,7 +62,7 @@ private[deploy] class MesosSubmitRequestServlet(
   private val DEFAULT_MEMORY = 512 // mb
   private val DEFAULT_CORES = 1.0
 
-  private val nextDriverNumber: AtomicLong = new AtomicLong(0)
+  private val nextDriverNumber = new AtomicLong(0)
   private def createDateFormat = new SimpleDateFormat("yyyyMMddHHmmss")  // For application IDs
   private def newDriverId(submitDate: Date): String = {
     "driver-%s-%04d".format(
@@ -93,9 +96,9 @@ private[deploy] class MesosSubmitRequestServlet(
     val appArgs = request.appArgs
     val environmentVariables = request.environmentVariables
     val name = request.sparkProperties.get("spark.app.name").getOrElse(mainClass)
+
     // Construct driver description
-    val conf = new SparkConf(false)
-      .setAll(sparkProperties)
+    val conf = new SparkConf(false).setAll(sparkProperties)
     val extraClassPath = driverExtraClassPath.toSeq.flatMap(_.split(File.pathSeparator))
     val extraLibraryPath = driverExtraLibraryPath.toSeq.flatMap(_.split(File.pathSeparator))
     val extraJavaOpts = driverExtraJavaOptions.map(Utils.splitCommandString).getOrElse(Seq.empty)
@@ -106,7 +109,6 @@ private[deploy] class MesosSubmitRequestServlet(
     val actualSuperviseDriver = superviseDriver.map(_.toBoolean).getOrElse(DEFAULT_SUPERVISE)
     val actualDriverMemory = driverMemory.map(Utils.memoryStringToMb).getOrElse(DEFAULT_MEMORY)
     val actualDriverCores = driverCores.map(_.toDouble).getOrElse(DEFAULT_CORES)
-
     val submitDate = new Date()
     val submissionId = newDriverId(submitDate)
 
@@ -149,7 +151,7 @@ private[deploy] class MesosKillRequestServlet(scheduler: MesosClusterScheduler,
 private[deploy] class MesosStatusRequestServlet(scheduler: MesosClusterScheduler, conf: SparkConf)
   extends StatusRequestServlet {
   protected override def handleStatus(submissionId: String): SubmissionStatusResponse = {
-    val d = scheduler.getStatus(submissionId)
+    val d = scheduler.getDriverStatus(submissionId)
     d.serverSparkVersion = sparkVersion
     d
   }