{"action":"created","comment":{"url":"https://api.github.com/repos/apache/spark/pulls/comments/28465329","id":28465329,"diff_hunk":"@@ -0,0 +1,614 @@\n+/*\n+ * Licensed to the Apache Software Foundation (ASF) under one or more\n+ * contributor license agreements.  See the NOTICE file distributed with\n+ * this work for additional information regarding copyright ownership.\n+ * The ASF licenses this file to You under the Apache License, Version 2.0\n+ * (the \"License\"); you may not use this file except in compliance with\n+ * the License.  You may obtain a copy of the License at\n+ *\n+ *    http://www.apache.org/licenses/LICENSE-2.0\n+ *\n+ * Unless required by applicable law or agreed to in writing, software\n+ * distributed under the License is distributed on an \"AS IS\" BASIS,\n+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n+ * See the License for the specific language governing permissions and\n+ * limitations under the License.\n+ */\n+\n+package org.apache.spark.scheduler.cluster.mesos\n+\n+import java.io.File\n+import java.util.concurrent.locks.ReentrantLock\n+import java.util.{Collections, Date, List => JList}\n+\n+import org.apache.mesos.Protos.Environment.Variable\n+import org.apache.mesos.Protos.TaskStatus.Reason\n+import org.apache.mesos.Protos.{TaskState => MesosTaskState, _}\n+import org.apache.mesos.{Scheduler, SchedulerDriver}\n+import org.apache.spark.deploy.mesos.MesosDriverDescription\n+import org.apache.spark.deploy.rest.{CreateSubmissionResponse, KillSubmissionResponse, SubmissionStatusResponse}\n+import org.apache.spark.metrics.MetricsSystem\n+import org.apache.spark.util.Utils\n+import org.apache.spark.{SecurityManager, SparkConf, SparkException, TaskState}\n+\n+import scala.collection.JavaConversions._\n+import scala.collection.mutable\n+import scala.collection.mutable.ArrayBuffer\n+\n+\n+/**\n+ * Tracks the current state of a Mesos Task that runs a Spark driver.\n+ * @param submission Submitted driver description from\n+ *                   [[org.apache.spark.deploy.rest.mesos.MesosRestServer]]\n+ * @param taskId Mesos TaskID generated for the task\n+ * @param slaveId Slave ID that the task is assigned to\n+ * @param taskState The last known task status update.\n+ * @param startDate The date the task was launched\n+ */\n+private[spark] class MesosClusterTaskState(\n+    val submission: MesosDriverDescription,\n+    val taskId: TaskID,\n+    val slaveId: SlaveID,\n+    var taskState: Option[TaskStatus],\n+    var startDate: Date)\n+  extends Serializable {\n+\n+  def copy(): MesosClusterTaskState = {\n+    new MesosClusterTaskState(\n+      submission, taskId, slaveId, taskState, startDate)\n+  }\n+}\n+\n+/**\n+ * Tracks the retry state of a driver, which includes the next time it should be scheduled\n+ * and necessary information to do exponential backoff.\n+ * This class is not thread-safe, and we expect the caller to handle synchronizing state.\n+ * @param lastFailureStatus Last Task status when it failed.\n+ * @param retries Number of times it has retried.\n+ * @param nextRetry Next retry time to be scheduled.\n+ * @param waitTime The amount of time driver is scheduled to wait until next retry.\n+ */\n+private[spark] class RetryState(\n+    val lastFailureStatus: TaskStatus,\n+    val retries: Int,\n+    val nextRetry: Date,\n+    val waitTime: Int) extends Serializable {\n+  def copy(): RetryState =\n+    new RetryState(lastFailureStatus, retries, nextRetry, waitTime)\n+}\n+\n+/**\n+ * The full state of the cluster scheduler, currently being used for displaying\n+ * information on the UI.\n+ * @param frameworkId Mesos Framework id for the cluster scheduler.\n+ * @param masterUrl The Mesos master url\n+ * @param queuedDrivers All drivers queued to be launched\n+ * @param launchedDrivers All launched or running drivers\n+ * @param finishedDrivers All terminated drivers\n+ * @param retryList All drivers pending to be retried\n+ */\n+private[spark] class MesosClusterSchedulerState(\n+    val frameworkId: String,\n+    val masterUrl: Option[String],\n+    val queuedDrivers: Iterable[MesosDriverDescription],\n+    val launchedDrivers: Iterable[MesosClusterTaskState],\n+    val finishedDrivers: Iterable[MesosClusterTaskState],\n+    val retryList: Iterable[MesosDriverDescription])\n+\n+/**\n+ * A Mesos scheduler that is responsible for launching submitted Spark drivers in cluster mode\n+ * as Mesos tasks in a Mesos cluster.\n+ * All drivers are launched asynchronously by the framework, which will eventually be launched\n+ * by one of the slaves in the cluster. The results of the driver will be stored in slave's task\n+ * sandbox which is accessible by visiting the Mesos UI.\n+ * This scheduler supports recovery by persisting all its state and performs task reconciliation\n+ * on recover, which gets all the latest state for all the drivers from Mesos master.\n+ */\n+private[spark] class MesosClusterScheduler(\n+    engineFactory: MesosClusterPersistenceEngineFactory,\n+    conf: SparkConf)\n+  extends Scheduler with MesosSchedulerUtils {\n+  var frameworkUrl: String = _\n+\n+  private val metricsSystem =\n+    MetricsSystem.createMetricsSystem(\"mesos_cluster\", conf, new SecurityManager(conf))\n+  private val master = conf.get(\"spark.master\")\n+  private val appName = conf.get(\"spark.app.name\")\n+  private val queuedCapacity = conf.getInt(\"spark.deploy.mesos.queuedDrivers\", 200)\n+  private val retainedDrivers = conf.getInt(\"spark.deploy.retainedDrivers\", 200)\n+  private val maxRetryWaitTime = conf.getInt(\"spark.mesos.cluster.retry.wait.max\", 60) // 1 minute\n+  private val schedulerState = engineFactory.createEngine(\"scheduler\")\n+  private val stateLock = new ReentrantLock()\n+  private val finishedDrivers = new mutable.ArrayBuffer[MesosClusterTaskState](retainedDrivers)\n+  private var frameworkId: String = null\n+  val launchedDrivers = new mutable.HashMap[String, MesosClusterTaskState]()\n+  // Holds the list of tasks that needs to reconcile with Mesos master.\n+  // All states that are loaded after failover are added here.\n+  private val pendingRecover = new mutable.HashMap[String, SlaveID]()\n+  // A queue that stores all the submitted drivers that hasn't been launched.\n+  val queuedDrivers = new ArrayBuffer[MesosDriverDescription]()\n+  // All supervised drivers that are waiting to retry after termination.\n+  val pendingRetryDrivers = new ArrayBuffer[MesosDriverDescription]()\n+  private val queuedDriversState = engineFactory.createEngine(\"driverQueue\")\n+  private val launchedDriversState = engineFactory.createEngine(\"launchedDrivers\")\n+  private val pendingRetryDriversState = engineFactory.createEngine(\"retryList\")\n+  // Flag to mark if the scheduler is ready to be called, which is until the scheduler\n+  // is registered with Mesos master.\n+  protected var ready = false\n+  private var masterInfo: Option[MasterInfo] = None\n+\n+  private def isQueueFull(): Boolean = launchedDrivers.size >= queuedCapacity\n+\n+  def submitDriver(desc: MesosDriverDescription): CreateSubmissionResponse = {\n+    val c = new CreateSubmissionResponse\n+    if (!ready) {\n+      c.success = false\n+      c.message = \"Scheduler is not ready to take requests\"\n+      return c\n+    }\n+\n+    stateLock.synchronized {\n+      if (isQueueFull()) {\n+        c.success = false\n+        c.message = \"Already reached maximum submission size\"\n+      }\n+      c.submissionId = desc.submissionId\n+      queuedDrivers += desc\n+      c.success = true\n+    }\n+    c\n+  }\n+\n+  private def removeFromQueuedDrivers(id: String): Boolean = {\n+    val index = queuedDrivers.indexWhere(_.submissionId.equals(id))\n+    if (index != -1) {\n+      queuedDrivers.remove(index)\n+      queuedDriversState.expunge(id)\n+      true\n+    } else {\n+      false\n+    }\n+  }\n+\n+  private def removeFromLaunchedDrivers(id: String): Boolean = {\n+    if (launchedDrivers.remove(id).isDefined) {\n+      launchedDriversState.expunge(id)\n+      true\n+    } else {\n+      false\n+    }\n+  }\n+\n+  private def removeFromPendingRetryDrivers(id: String): Boolean = {\n+    val index = pendingRetryDrivers.indexWhere(_.submissionId.equals(id))\n+    if (index != -1) {\n+      pendingRetryDrivers.remove(index)\n+      pendingRetryDriversState.expunge(id)\n+      true\n+    } else {\n+      false\n+    }\n+  }\n+\n+  def killDriver(submissionId: String): KillSubmissionResponse = {\n+    val k = new KillSubmissionResponse\n+    if (!ready) {\n+      k.success = false\n+      k.message = \"Scheduler is not ready to take requests\"\n+      return k\n+    }\n+    k.submissionId = submissionId\n+    stateLock.synchronized {\n+      // We look for the requested driver in the following places:\n+      // 1. Check if submission is running or launched.\n+      // 2. Check if it's still queued.\n+      // 3. Check if it's in the retry list.\n+      // 4. Check if it has already completed.\n+      if (launchedDrivers.contains(submissionId)) {\n+        val task = launchedDrivers(submissionId)\n+        driver.killTask(task.taskId)\n+        k.success = true\n+        k.message = \"Killing running driver\"\n+      } else if (removeFromQueuedDrivers(submissionId)) {\n+        k.success = true\n+        k.message = \"Removed driver while it's still pending\"\n+      } else if (removeFromPendingRetryDrivers(submissionId)) {\n+        k.success = true\n+        k.message = \"Removed driver while it's retrying\"\n+      } else if (finishedDrivers.exists(_.submission.submissionId.equals(submissionId))) {\n+        k.success = false\n+        k.message = \"Driver already terminated\"\n+      } else {\n+        k.success = false\n+        k.message = \"Cannot find driver\"\n+      }\n+    }\n+    k\n+  }\n+\n+  /**\n+   * Recover scheduler state that is persisted.\n+   * We still need to do task reconciliation to be up to date of the latest task states\n+   * as it might have changed while the scheduler is failing over.\n+   */\n+  private def recoverState(): Unit = {\n+    stateLock.synchronized {\n+      launchedDriversState.fetchAll[MesosClusterTaskState]().foreach { state =>\n+        launchedDrivers(state.taskId.getValue) = state\n+        pendingRecover(state.taskId.getValue) = state.slaveId\n+      }\n+      queuedDriversState.fetchAll[MesosDriverDescription]().foreach(d => queuedDrivers += d)\n+      // There is potential timing issue where a queued driver might have been launched\n+      // but the scheduler shuts down before the queued driver was able to be removed\n+      // from the queue. We try to mitigate this issue by walking through all queued drivers\n+      // and remove if they're already launched.\n+      queuedDrivers\n+        .filter(d => launchedDrivers.contains(d.submissionId))\n+        .toSeq\n+        .foreach(d => removeFromQueuedDrivers(d.submissionId))\n+      pendingRetryDriversState.fetchAll[MesosDriverDescription]()\n+        .foreach(s => pendingRetryDrivers += s)\n+      // TODO: Consider storing finished drivers so we can show them on the UI after\n+      // failover. For now we clear the history on each recovery.\n+      finishedDrivers.clear()\n+    }\n+  }\n+\n+  /**\n+   * Starts the cluster scheduler and wait until the scheduler is registered.\n+   * This also marks the scheduler to be ready for requests.\n+   */\n+  def start(): Unit = {\n+    // TODO: Implement leader election to make sure only one framework running in the cluster.\n+    val fwId = schedulerState.fetch[String](\"frameworkId\")\n+    val builder = FrameworkInfo.newBuilder()\n+      .setUser(Utils.getCurrentUserName())\n+      .setName(appName)\n+      .setWebuiUrl(frameworkUrl)\n+      .setCheckpoint(true)\n+      .setFailoverTimeout(Integer.MAX_VALUE) // Setting to max so tasks keep running on crash\n+    fwId.foreach { id =>\n+      builder.setId(FrameworkID.newBuilder().setValue(id).build())\n+      frameworkId = id\n+    }\n+    recoverState()\n+    metricsSystem.registerSource(new MesosClusterSchedulerSource(this))\n+    metricsSystem.start()\n+    startScheduler(\n+      \"MesosClusterScheduler\", master, MesosClusterScheduler.this, builder.build())\n+    ready = true\n+  }\n+\n+  def stop(): Unit = {\n+    ready = false\n+    metricsSystem.report()\n+    metricsSystem.stop()\n+    driver.stop(true)\n+  }\n+\n+  override def registered(\n+      driver: SchedulerDriver,\n+      newFrameworkId: FrameworkID,\n+      masterInfo: MasterInfo): Unit = {\n+    logInfo(\"Registered as framework ID \" + newFrameworkId.getValue)\n+    if (newFrameworkId.getValue != frameworkId) {\n+      frameworkId = newFrameworkId.getValue\n+      schedulerState.persist(\"frameworkId\", frameworkId)\n+    }\n+    markRegistered()\n+\n+    stateLock.synchronized {\n+      this.masterInfo = Some(masterInfo)\n+      if (!pendingRecover.isEmpty) {\n+        // Start task reconciliation if we need to recover.\n+        val statuses = pendingRecover.collect {\n+          case (taskId, slaveId) =>\n+            val newStatus = TaskStatus.newBuilder()\n+              .setTaskId(TaskID.newBuilder().setValue(taskId).build())\n+              .setSlaveId(slaveId)\n+              .setState(MesosTaskState.TASK_STAGING)\n+              .build()\n+            launchedDrivers.get(taskId).map(_.taskState.getOrElse(newStatus))\n+              .getOrElse(newStatus)\n+        }\n+        // TODO: Page the status updates to avoid trying to reconcile\n+        // a large amount of tasks at once.\n+        driver.reconcileTasks(statuses)\n+      }\n+    }\n+  }\n+\n+  private def buildDriverCommand(desc: MesosDriverDescription): CommandInfo = {\n+    val appJar = CommandInfo.URI.newBuilder()\n+      .setValue(desc.jarUrl.stripPrefix(\"file:\").stripPrefix(\"local:\")).build()\n+    val builder = CommandInfo.newBuilder().addUris(appJar)\n+    val entries =\n+      (conf.getOption(\"spark.executor.extraLibraryPath\").toList ++\n+        desc.command.libraryPathEntries)\n+    val prefixEnv = if (!entries.isEmpty) {\n+      Utils.libraryPathEnvPrefix(entries)\n+    } else {\n+      \"\"\n+    }\n+    val envBuilder = Environment.newBuilder()\n+    desc.command.environment.foreach { case (k, v) =>\n+      envBuilder.addVariables(Variable.newBuilder().setName(k).setValue(v).build())\n+    }\n+    // Pass all spark properties to executor.\n+    val executorOpts = desc.schedulerProperties.map { case (k, v) => s\"-D$k=$v\" }.mkString(\" \")\n+    envBuilder.addVariables(\n+      Variable.newBuilder().setName(\"SPARK_EXECUTOR_OPTS\").setValue(executorOpts))\n+    val cmdOptions = generateCmdOption(desc)\n+    val executorUri = desc.schedulerProperties.get(\"spark.executor.uri\")\n+      .orElse(desc.command.environment.get(\"SPARK_EXECUTOR_URI\"))\n+    val appArguments = desc.command.arguments.mkString(\" \")\n+    val cmd = if (executorUri.isDefined) {\n+      builder.addUris(CommandInfo.URI.newBuilder().setValue(executorUri.get).build())\n+      val folderBasename = executorUri.get.split('/').last.split('.').head\n+      val cmdExecutable = s\"cd $folderBasename*; $prefixEnv bin/spark-submit\"\n+      val cmdJar = s\"../${desc.jarUrl.split(\"/\").last}\"\n+      s\"$cmdExecutable ${cmdOptions.mkString(\" \")} $cmdJar $appArguments\"\n+    } else {\n+      val executorSparkHome = desc.schedulerProperties.get(\"spark.mesos.executor.home\")\n+        .orElse(conf.getOption(\"spark.home\"))\n+        .orElse(Option(System.getenv(\"SPARK_HOME\")))\n+        .getOrElse {\n+          throw new SparkException(\"Executor Spark home `spark.mesos.executor.home` is not set!\")\n+        }\n+      val cmdExecutable = new File(executorSparkHome, \"./bin/spark-submit\").getCanonicalPath\n+      val cmdJar = desc.jarUrl.split(\"/\").last\n+      s\"$cmdExecutable ${cmdOptions.mkString(\" \")} $cmdJar $appArguments\"\n+    }\n+    builder.setValue(cmd)\n+    builder.setEnvironment(envBuilder.build())\n+    builder.build()\n+  }\n+\n+  private def generateCmdOption(desc: MesosDriverDescription): Seq[String] = {\n+    var options = Seq(\n+      \"--name\", desc.schedulerProperties(\"spark.app.name\"),\n+      \"--class\", desc.command.mainClass,\n+      \"--master\", s\"mesos://${conf.get(\"spark.master\")}\",\n+      \"--driver-cores\", desc.cores.toString,\n+      \"--driver-memory\", s\"${desc.mem}M\")\n+    desc.schedulerProperties.get(\"spark.executor.memory\").map { v =>\n+      options ++= Seq(\"--executor-memory\", v)\n+    }\n+    desc.schedulerProperties.get(\"spark.cores.max\").map { v =>\n+      options ++= Seq(\"--total-executor-cores\", v)\n+    }\n+    options\n+  }\n+\n+  private class ResourceOffer(val offer: Offer, var cpu: Double, var mem: Double) {\n+    override def toString(): String = {\n+      s\"Offer id: ${offer.getId.getValue}, cpu: $cpu, mem: $mem\"\n+    }\n+  }\n+\n+  /**\n+   * This method takes all the possible candidates provided by the tasksFunc\n+   * and attempt to schedule them with Mesos offers.\n+   * Every time a new task is scheduled, the scheduledCallback is called to\n+   * perform post scheduled logic on each task.\n+   */\n+  private def scheduleTasks(\n+      candidates: Seq[MesosDriverDescription],\n+      scheduledCallback: (String) => Boolean,\n+      currentOffers: List[ResourceOffer],\n+      tasks: mutable.HashMap[OfferID, ArrayBuffer[TaskInfo]],\n+      currentTime: Date): Unit = {\n+    for (submission <- candidates) {\n+      val driverCpu = submission.cores\n+      val driverMem = submission.mem\n+      logTrace(s\"Finding offer to launch driver with cpu: $driverCpu, mem: $driverMem\")\n+      val offerOption = currentOffers.find { o =>\n+        o.cpu >= driverCpu && o.mem >= driverMem\n+      }\n+      if (offerOption.isEmpty) {\n+        logDebug(s\"Unable to find offer to launch driver id: ${submission.submissionId},\" +\n+          s\"cpu: $driverCpu, mem: $driverMem\")\n+      } else {\n+        val offer = offerOption.get\n+        offer.cpu -= driverCpu\n+        offer.mem -= driverMem\n+        val taskId = TaskID.newBuilder().setValue(submission.submissionId).build()\n+        val cpuResource = Resource.newBuilder()\n+          .setName(\"cpus\").setType(Value.Type.SCALAR)\n+          .setScalar(Value.Scalar.newBuilder().setValue(driverCpu)).build()\n+        val memResource = Resource.newBuilder()\n+          .setName(\"mem\").setType(Value.Type.SCALAR)\n+          .setScalar(Value.Scalar.newBuilder().setValue(driverMem)).build()\n+        val commandInfo = buildDriverCommand(submission)\n+        val appName = submission.schedulerProperties(\"spark.app.name\")\n+        val taskInfo = TaskInfo.newBuilder()\n+          .setTaskId(taskId)\n+          .setName(s\"Driver for $appName\")\n+          .setSlaveId(offer.offer.getSlaveId)\n+          .setCommand(commandInfo)\n+          .addResources(cpuResource)\n+          .addResources(memResource)\n+          .build()\n+        val queuedTasks = if (!tasks.contains(offer.offer.getId)) {\n+          val buffer = new ArrayBuffer[TaskInfo]\n+          tasks(offer.offer.getId) = buffer\n+          buffer\n+        } else {\n+          tasks(offer.offer.getId)\n+        }\n+        queuedTasks += taskInfo\n+        logTrace(s\"Using offer ${offer.offer.getId.getValue} to launch driver \" +\n+          submission.submissionId)\n+        val newState = new MesosClusterTaskState(submission, taskId, offer.offer.getSlaveId,\n+          None, new Date())\n+        launchedDrivers(submission.submissionId) = newState\n+        launchedDriversState.persist(submission.submissionId, newState)\n+        scheduledCallback(submission.submissionId)\n+      }\n+    }\n+  }\n+\n+  override def resourceOffers(driver: SchedulerDriver, offers: JList[Offer]): Unit = {\n+    /**\n+     * Prints the list of Mesos offers for logging purpose.\n+     */\n+    def printOffers(offers: Iterable[ResourceOffer]): String = {\n+      val builder = new StringBuilder()\n+      offers.foreach { o =>\n+        builder.append(o).append(\"\\n\")\n+      }\n+      builder.toString()\n+    }\n+    val currentOffers = offers.map { o =>\n+      new ResourceOffer(\n+        o, getResource(o.getResourcesList, \"cpus\"), getResource(o.getResourcesList, \"mem\"))\n+    }.toList\n+    logTrace(s\"Received offers from Mesos: \\n${printOffers(currentOffers)}\")\n+    val tasks = new mutable.HashMap[OfferID, ArrayBuffer[TaskInfo]]()\n+    val currentTime = new Date()\n+\n+    stateLock.synchronized {\n+      // We first schedule all the supervised drivers that are ready to retry.\n+      // This list will be empty if none of the drivers are marked as supervise.\n+      scheduleTasks(pendingRetryDrivers\n+        .filter(d => d.retryState.get.nextRetry.before(currentTime)),\n+        removeFromPendingRetryDrivers,\n+        currentOffers, tasks, currentTime)\n+      // Then we walk through the queued drivers and try to schedule them.\n+      scheduleTasks(queuedDrivers,\n+        removeFromQueuedDrivers,\n+        currentOffers, tasks, currentTime)\n+    }\n+    tasks.foreach { case (offerId, tasks) =>\n+      driver.launchTasks(Collections.singleton(offerId), tasks)\n+    }\n+    offers\n+      .filter(o => !tasks.keySet.contains(o.getId))\n+      .foreach(o => driver.declineOffer(o.getId))\n+  }\n+\n+  def getState(): MesosClusterSchedulerState = {\n+    def copyBuffer(\n+      buffer: ArrayBuffer[MesosDriverDescription]): ArrayBuffer[MesosDriverDescription] = {","path":"core/src/main/scala/org/apache/spark/scheduler/cluster/mesos/MesosClusterScheduler.scala","position":492,"original_position":492,"commit_id":"43b66a3ddea2d363d4876c41ba1fe1c5471279b6","original_commit_id":"43b66a3ddea2d363d4876c41ba1fe1c5471279b6","user":{"login":"andrewor14","id":2133137,"avatar_url":"https://avatars.githubusercontent.com/u/2133137?v=3","gravatar_id":"","url":"https://api.github.com/users/andrewor14","html_url":"https://github.com/andrewor14","followers_url":"https://api.github.com/users/andrewor14/followers","following_url":"https://api.github.com/users/andrewor14/following{/other_user}","gists_url":"https://api.github.com/users/andrewor14/gists{/gist_id}","starred_url":"https://api.github.com/users/andrewor14/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/andrewor14/subscriptions","organizations_url":"https://api.github.com/users/andrewor14/orgs","repos_url":"https://api.github.com/users/andrewor14/repos","events_url":"https://api.github.com/users/andrewor14/events{/privacy}","received_events_url":"https://api.github.com/users/andrewor14/received_events","type":"User","site_admin":false},"body":"indent by 2","created_at":"2015-04-15T21:34:02Z","updated_at":"2015-04-15T21:34:02Z","html_url":"https://github.com/apache/spark/pull/5144#discussion_r28465329","pull_request_url":"https://api.github.com/repos/apache/spark/pulls/5144","_links":{"self":{"href":"https://api.github.com/repos/apache/spark/pulls/comments/28465329"},"html":{"href":"https://github.com/apache/spark/pull/5144#discussion_r28465329"},"pull_request":{"href":"https://api.github.com/repos/apache/spark/pulls/5144"}}},"pull_request":{"url":"https://api.github.com/repos/apache/spark/pulls/5144","id":31778104,"html_url":"https://github.com/apache/spark/pull/5144","diff_url":"https://github.com/apache/spark/pull/5144.diff","patch_url":"https://github.com/apache/spark/pull/5144.patch","issue_url":"https://api.github.com/repos/apache/spark/issues/5144","number":5144,"state":"open","locked":false,"title":"[SPARK-5338][MESOS] Add cluster mode support for Mesos","user":{"login":"tnachen","id":67244,"avatar_url":"https://avatars.githubusercontent.com/u/67244?v=3","gravatar_id":"","url":"https://api.github.com/users/tnachen","html_url":"https://github.com/tnachen","followers_url":"https://api.github.com/users/tnachen/followers","following_url":"https://api.github.com/users/tnachen/following{/other_user}","gists_url":"https://api.github.com/users/tnachen/gists{/gist_id}","starred_url":"https://api.github.com/users/tnachen/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/tnachen/subscriptions","organizations_url":"https://api.github.com/users/tnachen/orgs","repos_url":"https://api.github.com/users/tnachen/repos","events_url":"https://api.github.com/users/tnachen/events{/privacy}","received_events_url":"https://api.github.com/users/tnachen/received_events","type":"User","site_admin":false},"body":"This patch adds the support for cluster mode to run on Mesos.\r\nIt introduces a new Mesos framework dedicated to launch new apps/drivers, and can be called with the spark-submit script and specifying --master flag to the cluster mode REST interface instead of Mesos master.\r\n\r\nExample:\r\n./bin/spark-submit --deploy-mode cluster --class org.apache.spark.examples.SparkPi --master mesos://10.0.0.206:8077 --executor-memory 1G --total-executor-cores 100 examples/target/spark-examples_2.10-1.3.0-SNAPSHOT.jar 30\r\n\r\nPart of this patch is also to abstract the StandaloneRestServer so it can have different implementations of the REST endpoints.\r\n\r\nFeatures of the cluster mode in this PR:\r\n- Supports supervise mode where scheduler will keep trying to reschedule exited job.\r\n- Adds a new UI for the cluster mode scheduler to see all the running jobs, finished jobs, and supervise jobs waiting to be retried\r\n- Supports state persistence to ZK, so when the cluster scheduler fails over it can pick up all the queued and running jobs\r\n","created_at":"2015-03-23T21:54:56Z","updated_at":"2015-04-15T21:34:02Z","closed_at":null,"merged_at":null,"merge_commit_sha":"0e396af37f89d6045cb985985cfcdf7913195c48","assignee":null,"milestone":null,"commits_url":"https://api.github.com/repos/apache/spark/pulls/5144/commits","review_comments_url":"https://api.github.com/repos/apache/spark/pulls/5144/comments","review_comment_url":"https://api.github.com/repos/apache/spark/pulls/comments{/number}","comments_url":"https://api.github.com/repos/apache/spark/issues/5144/comments","statuses_url":"https://api.github.com/repos/apache/spark/statuses/43b66a3ddea2d363d4876c41ba1fe1c5471279b6","head":{"label":"tnachen:mesos_cluster_mode","ref":"mesos_cluster_mode","sha":"43b66a3ddea2d363d4876c41ba1fe1c5471279b6","user":{"login":"tnachen","id":67244,"avatar_url":"https://avatars.githubusercontent.com/u/67244?v=3","gravatar_id":"","url":"https://api.github.com/users/tnachen","html_url":"https://github.com/tnachen","followers_url":"https://api.github.com/users/tnachen/followers","following_url":"https://api.github.com/users/tnachen/following{/other_user}","gists_url":"https://api.github.com/users/tnachen/gists{/gist_id}","starred_url":"https://api.github.com/users/tnachen/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/tnachen/subscriptions","organizations_url":"https://api.github.com/users/tnachen/orgs","repos_url":"https://api.github.com/users/tnachen/repos","events_url":"https://api.github.com/users/tnachen/events{/privacy}","received_events_url":"https://api.github.com/users/tnachen/received_events","type":"User","site_admin":false},"repo":{"id":21994538,"name":"spark","full_name":"tnachen/spark","owner":{"login":"tnachen","id":67244,"avatar_url":"https://avatars.githubusercontent.com/u/67244?v=3","gravatar_id":"","url":"https://api.github.com/users/tnachen","html_url":"https://github.com/tnachen","followers_url":"https://api.github.com/users/tnachen/followers","following_url":"https://api.github.com/users/tnachen/following{/other_user}","gists_url":"https://api.github.com/users/tnachen/gists{/gist_id}","starred_url":"https://api.github.com/users/tnachen/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/tnachen/subscriptions","organizations_url":"https://api.github.com/users/tnachen/orgs","repos_url":"https://api.github.com/users/tnachen/repos","events_url":"https://api.github.com/users/tnachen/events{/privacy}","received_events_url":"https://api.github.com/users/tnachen/received_events","type":"User","site_admin":false},"private":false,"html_url":"https://github.com/tnachen/spark","description":"Mirror of Apache Spark","fork":true,"url":"https://api.github.com/repos/tnachen/spark","forks_url":"https://api.github.com/repos/tnachen/spark/forks","keys_url":"https://api.github.com/repos/tnachen/spark/keys{/key_id}","collaborators_url":"https://api.github.com/repos/tnachen/spark/collaborators{/collaborator}","teams_url":"https://api.github.com/repos/tnachen/spark/teams","hooks_url":"https://api.github.com/repos/tnachen/spark/hooks","issue_events_url":"https://api.github.com/repos/tnachen/spark/issues/events{/number}","events_url":"https://api.github.com/repos/tnachen/spark/events","assignees_url":"https://api.github.com/repos/tnachen/spark/assignees{/user}","branches_url":"https://api.github.com/repos/tnachen/spark/branches{/branch}","tags_url":"https://api.github.com/repos/tnachen/spark/tags","blobs_url":"https://api.github.com/repos/tnachen/spark/git/blobs{/sha}","git_tags_url":"https://api.github.com/repos/tnachen/spark/git/tags{/sha}","git_refs_url":"https://api.github.com/repos/tnachen/spark/git/refs{/sha}","trees_url":"https://api.github.com/repos/tnachen/spark/git/trees{/sha}","statuses_url":"https://api.github.com/repos/tnachen/spark/statuses/{sha}","languages_url":"https://api.github.com/repos/tnachen/spark/languages","stargazers_url":"https://api.github.com/repos/tnachen/spark/stargazers","contributors_url":"https://api.github.com/repos/tnachen/spark/contributors","subscribers_url":"https://api.github.com/repos/tnachen/spark/subscribers","subscription_url":"https://api.github.com/repos/tnachen/spark/subscription","commits_url":"https://api.github.com/repos/tnachen/spark/commits{/sha}","git_commits_url":"https://api.github.com/repos/tnachen/spark/git/commits{/sha}","comments_url":"https://api.github.com/repos/tnachen/spark/comments{/number}","issue_comment_url":"https://api.github.com/repos/tnachen/spark/issues/comments{/number}","contents_url":"https://api.github.com/repos/tnachen/spark/contents/{+path}","compare_url":"https://api.github.com/repos/tnachen/spark/compare/{base}...{head}","merges_url":"https://api.github.com/repos/tnachen/spark/merges","archive_url":"https://api.github.com/repos/tnachen/spark/{archive_format}{/ref}","downloads_url":"https://api.github.com/repos/tnachen/spark/downloads","issues_url":"https://api.github.com/repos/tnachen/spark/issues{/number}","pulls_url":"https://api.github.com/repos/tnachen/spark/pulls{/number}","milestones_url":"https://api.github.com/repos/tnachen/spark/milestones{/number}","notifications_url":"https://api.github.com/repos/tnachen/spark/notifications{?since,all,participating}","labels_url":"https://api.github.com/repos/tnachen/spark/labels{/name}","releases_url":"https://api.github.com/repos/tnachen/spark/releases{/id}","created_at":"2014-07-18T21:08:12Z","updated_at":"2015-03-12T16:55:55Z","pushed_at":"2015-04-11T21:39:57Z","git_url":"git://github.com/tnachen/spark.git","ssh_url":"git@github.com:tnachen/spark.git","clone_url":"https://github.com/tnachen/spark.git","svn_url":"https://github.com/tnachen/spark","homepage":null,"size":95563,"stargazers_count":1,"watchers_count":1,"language":"Scala","has_issues":false,"has_downloads":true,"has_wiki":false,"has_pages":false,"forks_count":2,"mirror_url":null,"open_issues_count":0,"forks":2,"open_issues":0,"watchers":1,"default_branch":"master"}},"base":{"label":"apache:master","ref":"master","sha":"3ceb810aa8e69bc4abb69cbe713a624cb351cb35","user":{"login":"apache","id":47359,"avatar_url":"https://avatars.githubusercontent.com/u/47359?v=3","gravatar_id":"","url":"https://api.github.com/users/apache","html_url":"https://github.com/apache","followers_url":"https://api.github.com/users/apache/followers","following_url":"https://api.github.com/users/apache/following{/other_user}","gists_url":"https://api.github.com/users/apache/gists{/gist_id}","starred_url":"https://api.github.com/users/apache/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/apache/subscriptions","organizations_url":"https://api.github.com/users/apache/orgs","repos_url":"https://api.github.com/users/apache/repos","events_url":"https://api.github.com/users/apache/events{/privacy}","received_events_url":"https://api.github.com/users/apache/received_events","type":"Organization","site_admin":false},"repo":{"id":17165658,"name":"spark","full_name":"apache/spark","owner":{"login":"apache","id":47359,"avatar_url":"https://avatars.githubusercontent.com/u/47359?v=3","gravatar_id":"","url":"https://api.github.com/users/apache","html_url":"https://github.com/apache","followers_url":"https://api.github.com/users/apache/followers","following_url":"https://api.github.com/users/apache/following{/other_user}","gists_url":"https://api.github.com/users/apache/gists{/gist_id}","starred_url":"https://api.github.com/users/apache/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/apache/subscriptions","organizations_url":"https://api.github.com/users/apache/orgs","repos_url":"https://api.github.com/users/apache/repos","events_url":"https://api.github.com/users/apache/events{/privacy}","received_events_url":"https://api.github.com/users/apache/received_events","type":"Organization","site_admin":false},"private":false,"html_url":"https://github.com/apache/spark","description":"Mirror of Apache Spark","fork":false,"url":"https://api.github.com/repos/apache/spark","forks_url":"https://api.github.com/repos/apache/spark/forks","keys_url":"https://api.github.com/repos/apache/spark/keys{/key_id}","collaborators_url":"https://api.github.com/repos/apache/spark/collaborators{/collaborator}","teams_url":"https://api.github.com/repos/apache/spark/teams","hooks_url":"https://api.github.com/repos/apache/spark/hooks","issue_events_url":"https://api.github.com/repos/apache/spark/issues/events{/number}","events_url":"https://api.github.com/repos/apache/spark/events","assignees_url":"https://api.github.com/repos/apache/spark/assignees{/user}","branches_url":"https://api.github.com/repos/apache/spark/branches{/branch}","tags_url":"https://api.github.com/repos/apache/spark/tags","blobs_url":"https://api.github.com/repos/apache/spark/git/blobs{/sha}","git_tags_url":"https://api.github.com/repos/apache/spark/git/tags{/sha}","git_refs_url":"https://api.github.com/repos/apache/spark/git/refs{/sha}","trees_url":"https://api.github.com/repos/apache/spark/git/trees{/sha}","statuses_url":"https://api.github.com/repos/apache/spark/statuses/{sha}","languages_url":"https://api.github.com/repos/apache/spark/languages","stargazers_url":"https://api.github.com/repos/apache/spark/stargazers","contributors_url":"https://api.github.com/repos/apache/spark/contributors","subscribers_url":"https://api.github.com/repos/apache/spark/subscribers","subscription_url":"https://api.github.com/repos/apache/spark/subscription","commits_url":"https://api.github.com/repos/apache/spark/commits{/sha}","git_commits_url":"https://api.github.com/repos/apache/spark/git/commits{/sha}","comments_url":"https://api.github.com/repos/apache/spark/comments{/number}","issue_comment_url":"https://api.github.com/repos/apache/spark/issues/comments{/number}","contents_url":"https://api.github.com/repos/apache/spark/contents/{+path}","compare_url":"https://api.github.com/repos/apache/spark/compare/{base}...{head}","merges_url":"https://api.github.com/repos/apache/spark/merges","archive_url":"https://api.github.com/repos/apache/spark/{archive_format}{/ref}","downloads_url":"https://api.github.com/repos/apache/spark/downloads","issues_url":"https://api.github.com/repos/apache/spark/issues{/number}","pulls_url":"https://api.github.com/repos/apache/spark/pulls{/number}","milestones_url":"https://api.github.com/repos/apache/spark/milestones{/number}","notifications_url":"https://api.github.com/repos/apache/spark/notifications{?since,all,participating}","labels_url":"https://api.github.com/repos/apache/spark/labels{/name}","releases_url":"https://api.github.com/repos/apache/spark/releases{/id}","created_at":"2014-02-25T08:00:08Z","updated_at":"2015-04-15T21:17:34Z","pushed_at":"2015-04-15T21:31:22Z","git_url":"git://github.com/apache/spark.git","ssh_url":"git@github.com:apache/spark.git","clone_url":"https://github.com/apache/spark.git","svn_url":"https://github.com/apache/spark","homepage":null,"size":1499677,"stargazers_count":3621,"watchers_count":3621,"language":"Scala","has_issues":false,"has_downloads":true,"has_wiki":false,"has_pages":false,"forks_count":3126,"mirror_url":"git://git.apache.org/spark.git","open_issues_count":288,"forks":3126,"open_issues":288,"watchers":3621,"default_branch":"master"}},"_links":{"self":{"href":"https://api.github.com/repos/apache/spark/pulls/5144"},"html":{"href":"https://github.com/apache/spark/pull/5144"},"issue":{"href":"https://api.github.com/repos/apache/spark/issues/5144"},"comments":{"href":"https://api.github.com/repos/apache/spark/issues/5144/comments"},"review_comments":{"href":"https://api.github.com/repos/apache/spark/pulls/5144/comments"},"review_comment":{"href":"https://api.github.com/repos/apache/spark/pulls/comments{/number}"},"commits":{"href":"https://api.github.com/repos/apache/spark/pulls/5144/commits"},"statuses":{"href":"https://api.github.com/repos/apache/spark/statuses/43b66a3ddea2d363d4876c41ba1fe1c5471279b6"}}}}