diff --git a/core/src/main/scala/org/apache/spark/deploy/Client.scala b/core/src/main/scala/org/apache/spark/deploy/Client.scala index 2891247f20104..99ab5dadf3746 100644 --- a/core/src/main/scala/org/apache/spark/deploy/Client.scala +++ b/core/src/main/scala/org/apache/spark/deploy/Client.scala @@ -31,6 +31,7 @@ import org.apache.spark.{SecurityManager, SparkConf} import org.apache.spark.deploy.DeployMessages._ import org.apache.spark.deploy.master.{DriverState, Master} import org.apache.spark.deploy.master.DriverState.DriverState +import org.apache.spark.deploy.rest.RestSubmissionClient import org.apache.spark.internal.{config, Logging} import org.apache.spark.internal.LogKeys._ import org.apache.spark.internal.config.Network.RPC_ASK_TIMEOUT @@ -101,9 +102,10 @@ private class ClientEndpoint( val sparkJavaOpts = Utils.sparkJavaOpts(conf) val javaOpts = sparkJavaOpts ++ extraJavaOpts + val driverEnv = Client.driverEnvironment(conf, sys.env) val command = new Command(mainClass, Seq("{{WORKER_URL}}", "{{USER_JAR}}", driverArgs.mainClass) ++ driverArgs.driverOptions, - sys.env, classPathEntries, libraryPathEntries, javaOpts) + driverEnv, classPathEntries, libraryPathEntries, javaOpts) val driverResourceReqs = ResourceUtils.parseResourceRequirements(conf, config.SPARK_DRIVER_PREFIX) val driverDescription = new DriverDescription( @@ -275,6 +277,21 @@ object Client { // scalastyle:on println new ClientApp().start(args, new SparkConf()) } + + /** + * Environment variables to forward to the driver. Only Spark-related variables are forwarded, + * matching the REST submission client, unless `spark.standalone.submit.filterEnvironment` is + * disabled, in which case the full environment of the submitting process is forwarded. + */ + private[deploy] def driverEnvironment( + conf: SparkConf, + env: Map[String, String]): Map[String, String] = { + if (conf.get(config.STANDALONE_SUBMIT_FILTER_ENVIRONMENT)) { + RestSubmissionClient.filterSystemEnvironment(env) + } else { + env + } + } } private[spark] class ClientApp extends SparkApplication { diff --git a/core/src/main/scala/org/apache/spark/deploy/rest/RestSubmissionClient.scala b/core/src/main/scala/org/apache/spark/deploy/rest/RestSubmissionClient.scala index 6dc422515dbdb..6d11e54e616ec 100644 --- a/core/src/main/scala/org/apache/spark/deploy/rest/RestSubmissionClient.scala +++ b/core/src/main/scala/org/apache/spark/deploy/rest/RestSubmissionClient.scala @@ -540,7 +540,7 @@ private[spark] object RestSubmissionClient { /** * Filter non-spark environment variables from any environment. */ - private[rest] def filterSystemEnvironment(env: Map[String, String]): Map[String, String] = { + private[spark] def filterSystemEnvironment(env: Map[String, String]): Map[String, String] = { env.filter { case (k, _) => k.startsWith("SPARK_") && !EXCLUDED_SPARK_ENV_VARS.contains(k) } diff --git a/core/src/main/scala/org/apache/spark/internal/config/package.scala b/core/src/main/scala/org/apache/spark/internal/config/package.scala index 64a05fece570c..05fc3c49aea67 100644 --- a/core/src/main/scala/org/apache/spark/internal/config/package.scala +++ b/core/src/main/scala/org/apache/spark/internal/config/package.scala @@ -2787,6 +2787,16 @@ package object config { .booleanConf .createWithDefault(false) + private[spark] val STANDALONE_SUBMIT_FILTER_ENVIRONMENT = + ConfigBuilder("spark.standalone.submit.filterEnvironment") + .doc("In standalone cluster mode, controls whether the client forwards only " + + "Spark-related environment variables to the driver, matching the REST submission " + + "client. If set to false, the full environment of the submitting process is " + + "forwarded to the driver.") + .version("4.4.0") + .booleanConf + .createWithDefault(true) + private[spark] val EXECUTOR_ALLOW_SPARK_CONTEXT = ConfigBuilder("spark.executor.allowSparkContext") .doc("If set to true, SparkContext can be created in executors.") diff --git a/core/src/test/scala/org/apache/spark/deploy/ClientSuite.scala b/core/src/test/scala/org/apache/spark/deploy/ClientSuite.scala index 792168834dea2..da53a82dfff89 100644 --- a/core/src/test/scala/org/apache/spark/deploy/ClientSuite.scala +++ b/core/src/test/scala/org/apache/spark/deploy/ClientSuite.scala @@ -20,7 +20,8 @@ package org.apache.spark.deploy import org.scalatest.matchers.must.Matchers import org.scalatest.matchers.should.Matchers._ -import org.apache.spark.SparkFunSuite +import org.apache.spark.{SparkConf, SparkFunSuite} +import org.apache.spark.internal.config.STANDALONE_SUBMIT_FILTER_ENVIRONMENT class ClientSuite extends SparkFunSuite with Matchers { test("correctly validates driver jar URL's") { @@ -48,4 +49,19 @@ class ClientSuite extends SparkFunSuite with Matchers { // Invalid syntax. ClientArguments.isValidJarUrl("hdfs:") should be (false) } + + test("SPARK-59404: forward only Spark-related environment variables to the driver") { + val env = Map( + "SPARK_LOCAL_IP" -> "127.0.0.1", + "SPARK_HOME" -> "/opt/spark", + "PATH" -> "/usr/bin", + "SECRET_TOKEN" -> "hunter2") + Client.driverEnvironment(new SparkConf(), env) should be (Map("SPARK_LOCAL_IP" -> "127.0.0.1")) + } + + test("SPARK-59404: forward the full environment when filtering is disabled") { + val env = Map("SPARK_LOCAL_IP" -> "127.0.0.1", "PATH" -> "/usr/bin") + val conf = new SparkConf().set(STANDALONE_SUBMIT_FILTER_ENVIRONMENT, false) + Client.driverEnvironment(conf, env) should be (env) + } } diff --git a/docs/spark-standalone.md b/docs/spark-standalone.md index 149c79105a581..511cbdc7f9ef8 100644 --- a/docs/spark-standalone.md +++ b/docs/spark-standalone.md @@ -575,6 +575,16 @@ Spark applications supports the following configuration properties specific to s
spark.standalone.submit.filterEnvironmenttruefalse,
+ the full environment of the submitting process is forwarded to the driver.
+