[SPARK-9672][MESOS] Don’t filter out MESOS_* when passing env vars

Author: pashields

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

@@ -420,7 +420,7 @@ private[spark] object RestSubmissionClient {
    * Filter non-spark environment variables from any environment.
    */
   def filterSystemEnvironment(env: Map[String, String]): Map[String, String] = {
-    val sparkVars = env.filter { case (k, _) => k.startsWith("SPARK_") }
+    val sparkVars = env.filter { case (k, _) => k.startsWith("SPARK_") || k.startsWith("MESOS_") }
     sparkVars - "SPARK_ENV_LOADED"
   }
 }

core/src/test/scala/org/apache/spark/deploy/rest/StandaloneRestSubmitSuite.scala

@@ -372,6 +372,12 @@ class StandaloneRestSubmitSuite extends SparkFunSuite with BeforeAndAfterEach {
     assert(filteredVariables == Map("SPARK_VAR" -> "1"))
   }
 
+  test("client includes mesos env vars") {
+    val environmentVariables = Map("SPARK_VAR" -> "1", "MESOS_VAR" -> "1", "OTHER_VAR" -> "1")
+    val filteredVariables = RestSubmissionClient.filterSystemEnvironment(environmentVariables)
+    assert(filteredVariables == Map("SPARK_VAR" -> "1", "MESOS_VAR" -> "1"))
+  }
+
   /* --------------------- *
    |     Helper methods    |
    * --------------------- */