ODIn/spark-instrumented-optimizer
1
0
Fork 0
Code Issues Pull requests Projects Releases Wiki Activity

[SPARK-25934][MESOS] Don't propagate SPARK_CONF_DIR from spark submit

Browse source 
## What changes were proposed in this pull request?

Don't propagate SPARK_CONF_DIR to the driver in mesos cluster mode.

## How was this patch tested?

I built the 2.3.2 tag with this patch added and deployed a test job to a mesos cluster to confirm that the incorrect SPARK_CONF_DIR was no longer passed from the submit command.

Closes #22937 from mpmolek/fix-conf-dir.

Authored-by: Matt Molek <mpmolek@gmail.com>
Signed-off-by: Sean Owen <sean.owen@databricks.com>
This commit is contained in:
Matt Molek 2018-11-16 10:00:21 -06:00 committed by Sean Owen
parent 2aef79a65a
commit 696b75a810
2 changed files with 17 additions and 3 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

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

@ -408,6 +408,10 @@ private[spark] class RestSubmissionClient(master: String) extends Logging {
} }
private[spark] object RestSubmissionClient { private[spark] object RestSubmissionClient {
// SPARK_HOME and SPARK_CONF_DIR are filtered out because they are usually wrong
// on the remote machine (SPARK-12345) (SPARK-25934)
private val BLACKLISTED_SPARK_ENV_VARS = Set("SPARK_ENV_LOADED", "SPARK_HOME", "SPARK_CONF_DIR")
private val REPORT_DRIVER_STATUS_INTERVAL = 1000 private val REPORT_DRIVER_STATUS_INTERVAL = 1000
private val REPORT_DRIVER_STATUS_MAX_TRIES = 10 private val REPORT_DRIVER_STATUS_MAX_TRIES = 10
val PROTOCOL_VERSION = "v1" val PROTOCOL_VERSION = "v1"
@ -417,9 +421,7 @@ private[spark] object RestSubmissionClient {
*/ */
private[rest] def filterSystemEnvironment(env: Map[String, String]): Map[String, String] = { private[rest] def filterSystemEnvironment(env: Map[String, String]): Map[String, String] = {
env.filterKeys { k => env.filterKeys { k =>
// SPARK_HOME is filtered out because it is usually wrong on the remote machine (SPARK-12345) (k.startsWith("SPARK_") && !BLACKLISTED_SPARK_ENV_VARS.contains(k)) || k.startsWith("MESOS_")
(k.startsWith("SPARK_") && k != "SPARK_ENV_LOADED" && k != "SPARK_HOME") ||
k.startsWith("MESOS_")
} }
} }
} }

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

@ -396,6 +396,18 @@ class StandaloneRestSubmitSuite extends SparkFunSuite with BeforeAndAfterEach {
assert(filteredVariables == Map("SPARK_VAR" -> "1")) assert(filteredVariables == Map("SPARK_VAR" -> "1"))
} }
test("client does not send 'SPARK_HOME' env var by default") {
val environmentVariables = Map("SPARK_VAR" -> "1", "SPARK_HOME" -> "1")
val filteredVariables = RestSubmissionClient.filterSystemEnvironment(environmentVariables)
assert(filteredVariables == Map("SPARK_VAR" -> "1"))
}
test("client does not send 'SPARK_CONF_DIR' env var by default") {
val environmentVariables = Map("SPARK_VAR" -> "1", "SPARK_CONF_DIR" -> "1")
val filteredVariables = RestSubmissionClient.filterSystemEnvironment(environmentVariables)
assert(filteredVariables == Map("SPARK_VAR" -> "1"))
}
test("client includes mesos env vars") { test("client includes mesos env vars") {
val environmentVariables = Map("SPARK_VAR" -> "1", "MESOS_VAR" -> "1", "OTHER_VAR" -> "1") val environmentVariables = Map("SPARK_VAR" -> "1", "MESOS_VAR" -> "1", "OTHER_VAR" -> "1")
val filteredVariables = RestSubmissionClient.filterSystemEnvironment(environmentVariables) val filteredVariables = RestSubmissionClient.filterSystemEnvironment(environmentVariables)