Pass spark properties to mesos cluster tasks.

Author: tnachen

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

@@ -33,6 +33,7 @@ import org.apache.spark.deploy.Command
  * @param schedulerProperties Extra properties to pass the Mesos scheduler
  */
 private[spark] case class MesosDriverDescription(
+    val name: String,
     val jarUrl: String,
     val mem: Int,
     val cores: Double,

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

@@ -27,8 +27,6 @@ import org.apache.spark.scheduler.cluster.mesos.MesosClusterScheduler
 import org.apache.spark.util.Utils
 import org.apache.spark.{SPARK_VERSION => sparkVersion, SparkConf}
 
-import scala.collection.mutable
-
 
 /**
  * A server that responds to requests submitted by the [[RestClient]].
@@ -87,22 +85,6 @@ private[mesos] class MesosSubmitRequestServlet(
     val driverCores = sparkProperties.get("spark.driver.cores")
     val appArgs = request.appArgs
     val environmentVariables = request.environmentVariables
-    val schedulerProperties = new mutable.HashMap[String, String]
-    // Store Spark submit specific arguments here to pass to the scheduler.
-    schedulerProperties("spark.app.name") = sparkProperties.getOrElse("spark.app.name", mainClass)
-
-    sparkProperties.get("spark.executor.memory").foreach { v =>
-      schedulerProperties("spark.executor.memory") = v
-    }
-    sparkProperties.get("spark.cores.max").foreach { v =>
-      schedulerProperties("spark.cores.max") = v
-    }
-    sparkProperties.get("spark.executor.uri").foreach { v =>
-      schedulerProperties("spark.executor.uri") = v
-    }
-    sparkProperties.get("spark.mesos.executor.home").foreach { v =>
-      schedulerProperties("spark.mesos.executor.home") = v
-    }
 
     // Construct driver description
     val conf = new SparkConf(false)
@@ -119,8 +101,9 @@ private[mesos] class MesosSubmitRequestServlet(
     val actualDriverCores = driverCores.map(_.toDouble).getOrElse(DEFAULT_CORES)
 
     new MesosDriverDescription(
+      request.sparkProperties.get("spark.app.name").getOrElse(mainClass),
       appResource, actualDriverMemory, actualDriverCores,
-      actualSuperviseDriver, command, schedulerProperties.toMap)
+      actualSuperviseDriver, command, request.sparkProperties)
   }
 
   protected override def handleSubmit(

core/src/main/scala/org/apache/spark/scheduler/cluster/mesos/MesosClusterScheduler.scala

@@ -287,16 +287,19 @@ private[spark] class MesosClusterSchedulerDriver(
     desc.command.environment.foreach { case (k, v) =>
       envBuilder.addVariables(Variable.newBuilder().setName(k).setValue(v).build())
     }
-    builder.setEnvironment(envBuilder.build())
+    val executorOpts = desc.schedulerProperties.map { case (k, v) => s"-D$k=$v"}.mkString(" ")
+    // Pass all spark properties to executor.
+    envBuilder.addVariables(
+      Variable.newBuilder().setName("SPARK_EXECUTOR_OPTS").setValue(executorOpts))
     val cmdOptions = generateCmdOption(desc)
     val executorUri = desc.schedulerProperties.get("spark.executor.uri")
       .orElse(desc.command.environment.get("SPARK_EXECUTOR_URI"))
+    val appArguments = desc.command.arguments.mkString(" ")
     val cmd = if (executorUri.isDefined) {
       builder.addUris(CommandInfo.URI.newBuilder().setValue(executorUri.get).build())
       val folderBasename = executorUri.get.split('/').last.split('.').head
       val cmdExecutable = s"cd $folderBasename*; $prefixEnv bin/spark-submit"
       val cmdJar = s"../${desc.jarUrl.split("/").last}"
-      val appArguments = desc.command.arguments.mkString(" ")
       s"$cmdExecutable ${cmdOptions.mkString(" ")} $cmdJar $appArguments"
     } else {
       val executorSparkHome = desc.schedulerProperties.get("spark.mesos.executor.home")
@@ -307,9 +310,10 @@ private[spark] class MesosClusterSchedulerDriver(
         }
       val cmdExecutable = new File(executorSparkHome, "./bin/spark-submit").getCanonicalPath
       val cmdJar = desc.jarUrl.split("/").last
-      s"$cmdExecutable ${cmdOptions.mkString(" ")} $cmdJar"
+      s"$cmdExecutable ${cmdOptions.mkString(" ")} $cmdJar $appArguments"
     }
     builder.setValue(cmd)
+    builder.setEnvironment(envBuilder.build())
     builder.build
   }
 

core/src/test/scala/org/apache/spark/scheduler/mesos/MesosClusterSchedulerSuite.scala

@@ -39,15 +39,13 @@ class MesosClusterSchedulerSuite extends FunSuite with LocalSparkContext with Mo
     scheduler.recoverState
     val response =
       scheduler.submitDriver(
-        new MesosDriverDescription("jar", 1000, 1, true,
+        new MesosDriverDescription("d1", "jar", 1000, 1, true,
           createCommand, Map[String, String]()))
     assert(response.success)
-
     val response2 =
       scheduler.submitDriver(new MesosDriverDescription(
-        "jar", 1000, 1, true, createCommand, Map[String, String]()))
+        "d1", "jar", 1000, 1, true, createCommand, Map[String, String]()))
     assert(response2.success)
-
     val state = scheduler.getState()
     assert(state.queuedDrivers.exists(d => d.submissionId.get == response.submissionId))
     assert(state.queuedDrivers.exists(d => d.submissionId.get == response2.submissionId))
@@ -62,12 +60,11 @@ class MesosClusterSchedulerSuite extends FunSuite with LocalSparkContext with Mo
     scheduler.recoverState
     val response =
       scheduler.submitDriver(
-        new MesosDriverDescription("jar", 1000, 1, true, createCommand, Map[String, String]()))
+        new MesosDriverDescription("d1", "jar", 1000, 1, true,
+          createCommand, Map[String, String]()))
     assert(response.success)
-
     val killResponse = scheduler.killDriver(response.submissionId)
     assert(killResponse.success)
-
     val state = scheduler.getState()
     assert(state.queuedDrivers.isEmpty)
   }