This is an automated email from the ASF dual-hosted git repository.
jerryshao pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/gravitino.git
The following commit(s) were added to refs/heads/main by this push:
new 7555479b9d [#13617] feat(k8s-job-executor): Pass the environment
variables of a job template to the Spark driver and executors (#13948)
7555479b9d is described below
commit 7555479b9d38cf3f2294ebe4fe4fe3e84dd1dd48
Author: Jerry Shao <[email protected]>
AuthorDate: Tue Oct 20 11:47:26 2026 +0800
[#13617] feat(k8s-job-executor): Pass the environment variables of a job
template to the Spark driver and executors (#13948)
### What changes were proposed in this pull request?
The `k8s` job executor rejected Spark job templates with environment
variables. It now passes them to the Spark driver and executors.
- Each environment variable `X` of the job template becomes the Spark
configurations `spark.kubernetes.driverEnv.X` and `spark.executorEnv.X`.
- They override the default Spark configurations of the executor
(`gravitino.jobExecutor.k8s.spark.conf.*`), and the Spark configurations
of the job template override them, so a template can still set a
different value for the driver or the executors.
- A name must match `[-._a-zA-Z][-._a-zA-Z0-9]*`, must not be `.` and
must not start with `..`, otherwise the job is rejected when it is
submitted. Spark 3.5 silently drops any other `spark.executorEnv.*` name
for the executors while the driver still gets it, and Kubernetes without
the relaxed environment variable validation rejects such a name when the
operator creates the driver pod. The error message names the variable,
never its value.
The values are readable in the SparkApplication and in its pods. A
sensitive value can come from a Secret created beforehand, referenced in
the Spark configurations with
`spark.kubernetes.driver.secretKeyRef.<name>` and
`spark.kubernetes.executor.secretKeyRef.<name>`, which needs no change
in Gravitino. #13618 documents both.
### Why are the changes needed?
A job template with environment variables runs on the `local` job
executor but was rejected by the `k8s` one.
Fix: #13617
### Does this PR introduce _any_ user-facing change?
Yes. Spark job templates with environment variables run on the `k8s` job
executor, and an invalid variable name is rejected with an error. No new
configuration.
### How was this patch tested?
- Unit tests for the mapping, the precedence between the executor
defaults, the environment variables and the template configurations, and
the rejected names.
- Manually, with a Gravitino server started from the distribution
package of this branch against a local Kubernetes cluster (OrbStack)
with spark-kubernetes-operator 0.9.0, through the REST API: a Spark Pi
template with environment variables, one of them from a placeholder,
succeeded. The values were checked in the pod specs and in the processes
of the driver and the executor, including a name with a dot, a server
default overridden by the template, a template configuration overriding
a variable for the executors only, and a variable from a Secret through
`secretKeyRef` in the template configurations. A template with the
variable name `MY VAR` was rejected with a 400.
🤖 Generated with [Claude Code](https://claude.com/claude-code)
---------
Co-authored-by: Claude Opus 5.5 <[email protected]>
---
.../job/k8s/spark/SparkApplicationUtils.java | 46 +++++++++++---
.../job/k8s/spark/TestSparkApplicationUtils.java | 72 +++++++++++++++++++---
2 files changed, 101 insertions(+), 17 deletions(-)
diff --git
a/plugins/k8s-job-executor/src/main/java/org/apache/gravitino/job/k8s/spark/SparkApplicationUtils.java
b/plugins/k8s-job-executor/src/main/java/org/apache/gravitino/job/k8s/spark/SparkApplicationUtils.java
index 1a10510520..8ce2a8a6d2 100644
---
a/plugins/k8s-job-executor/src/main/java/org/apache/gravitino/job/k8s/spark/SparkApplicationUtils.java
+++
b/plugins/k8s-job-executor/src/main/java/org/apache/gravitino/job/k8s/spark/SparkApplicationUtils.java
@@ -86,6 +86,15 @@ public final class SparkApplicationUtils {
private static final String SPARK_JARS = "spark.jars";
private static final String SPARK_FILES = "spark.files";
private static final String SPARK_ARCHIVES = "spark.archives";
+ private static final String SPARK_DRIVER_ENV_PREFIX =
"spark.kubernetes.driverEnv.";
+ private static final String SPARK_EXECUTOR_ENV_PREFIX = "spark.executorEnv.";
+
+ // The environment variable names that Spark passes to the executors: Spark
3.5 drops any other
+ // spark.executorEnv.* key with only a warning, while the driver would still
get it. Kubernetes
+ // without the relaxed environment variable validation accepts the same
names, except '.' and
+ // the names starting with '..', which are rejected separately.
+ private static final Pattern ENVIRONMENT_NAME_PATTERN =
+ Pattern.compile("[-._a-zA-Z][-._a-zA-Z0-9]*");
private static final Pattern URI_USER_INFO =
Pattern.compile("(?<=://)[^/?#@]*@");
@@ -98,6 +107,12 @@ public final class SparkApplicationUtils {
* the Gravitino server, runs the jar of {@link
K8sJobExecutorConfigs#SPARK_BUILTIN_JOBS_JAR}
* instead.
*
+ * <p>The environment variables of the job template are set for the driver
and the executors
+ * through the Spark configurations {@code spark.kubernetes.driverEnv.*} and
{@code
+ * spark.executorEnv.*}, which means their values are visible in the
SparkApplication and in its
+ * pods. They override the default Spark configurations of the job executor,
and the Spark
+ * configurations of the job template override them.
+ *
* @param context the context of the job run
* @param template the runtime job template, whose resources are the URIs
given in the template
* @param configs the job executor configurations
@@ -106,15 +121,6 @@ public final class SparkApplicationUtils {
*/
public static GenericKubernetesResource buildSparkApplication(
JobContext context, SparkJobTemplate template, K8sJobExecutorConfigs
configs) {
- if (!template.environments().isEmpty()) {
- throw new IllegalArgumentException(
- String.format(
- "Job template %s sets environment variables %s, which the k8s
job executor doesn't"
- + " support yet. Pass them as Spark configurations instead,
for example"
- + " spark.kubernetes.driverEnv.<name> and
spark.executorEnv.<name>",
- template.name(), template.environments().keySet()));
- }
-
String executable = getExecutableUri(context, template, configs);
checkClusterReachable(executable);
template.jars().forEach(SparkApplicationUtils::checkClusterReachable);
@@ -122,6 +128,14 @@ public final class SparkApplicationUtils {
template.archives().forEach(SparkApplicationUtils::checkClusterReachable);
Map<String, String> sparkConf = new TreeMap<>(configs.sparkConf());
+ template
+ .environments()
+ .forEach(
+ (name, value) -> {
+ checkEnvironmentName(template, name);
+ sparkConf.put(SPARK_DRIVER_ENV_PREFIX + name, value);
+ sparkConf.put(SPARK_EXECUTOR_ENV_PREFIX + name, value);
+ });
sparkConf.putAll(template.configs());
// The operator only supports cluster mode, so a job that asks for client
mode is rejected
// rather than silently run in cluster mode.
@@ -224,6 +238,20 @@ public final class SparkApplicationUtils {
return template.executable();
}
+ private static void checkEnvironmentName(SparkJobTemplate template, String
name) {
+ // Checked here, as an invalid name would otherwise only miss the
executors, or fail the job
+ // when the operator creates the driver pod.
+ if (!ENVIRONMENT_NAME_PATTERN.matcher(name).matches()
+ || ".".equals(name)
+ || name.startsWith("..")) {
+ throw new IllegalArgumentException(
+ String.format(
+ "Environment variable name '%s' of job template %s is invalid:
it must match %s,"
+ + " must not be '.' and must not start with '..'",
+ name, template.name(), ENVIRONMENT_NAME_PATTERN.pattern()));
+ }
+ }
+
/**
* Checks that a resource of the job, or each of them if it is a
comma-separated list as Spark
* accepts, is a URI that the cluster can reach rather than a path on the
Gravitino server.
diff --git
a/plugins/k8s-job-executor/src/test/java/org/apache/gravitino/job/k8s/spark/TestSparkApplicationUtils.java
b/plugins/k8s-job-executor/src/test/java/org/apache/gravitino/job/k8s/spark/TestSparkApplicationUtils.java
index fdb6b2e65b..39da670461 100644
---
a/plugins/k8s-job-executor/src/test/java/org/apache/gravitino/job/k8s/spark/TestSparkApplicationUtils.java
+++
b/plugins/k8s-job-executor/src/test/java/org/apache/gravitino/job/k8s/spark/TestSparkApplicationUtils.java
@@ -514,20 +514,76 @@ public class TestSparkApplicationUtils {
}
@Test
- public void testRejectEnvironments() {
+ @SuppressWarnings("unchecked")
+ public void testEnvironments() {
+ K8sJobExecutorConfigs withDefaults =
+ newConfigs(
+ ImmutableMap.of(
+ "spark.conf.spark.kubernetes.driverEnv.REGION",
+ "default-region",
+ "spark.conf.spark.executorEnv.STAGE",
+ "default-stage"));
SparkJobTemplate template =
SparkJobTemplate.builder()
.withName("user-job")
.withExecutable("https://repo/app.jar";)
.withClassName("org.example.App")
- .withEnvironments(ImmutableMap.of("TOKEN", "secret"))
+ .withEnvironments(ImmutableMap.of("REGION", "us-west",
"app.mode-1", "", "TZ", "UTC"))
+ .withConfigs(ImmutableMap.of("spark.executorEnv.TZ",
"Asia/Shanghai"))
.build();
- IllegalArgumentException e =
- Assertions.assertThrows(
- IllegalArgumentException.class,
- () -> SparkApplicationUtils.buildSparkApplication(CONTEXT,
template, configs));
- Assertions.assertTrue(e.getMessage().contains("TOKEN"));
- Assertions.assertFalse(e.getMessage().contains("secret"));
+
+ Map<String, String> sparkConf =
+ (Map<String, String>) getSpec(template, withDefaults).get("sparkConf");
+
+ // Set for the driver and the executors, in place of the defaults of the
job executor.
+ Assertions.assertEquals("us-west",
sparkConf.get("spark.kubernetes.driverEnv.REGION"));
+ Assertions.assertEquals("us-west",
sparkConf.get("spark.executorEnv.REGION"));
+ Assertions.assertEquals("",
sparkConf.get("spark.kubernetes.driverEnv.app.mode-1"));
+ Assertions.assertEquals("", sparkConf.get("spark.executorEnv.app.mode-1"));
+ // The Spark configurations of the job template take precedence.
+ Assertions.assertEquals("UTC",
sparkConf.get("spark.kubernetes.driverEnv.TZ"));
+ Assertions.assertEquals("Asia/Shanghai",
sparkConf.get("spark.executorEnv.TZ"));
+ // A default that the job template doesn't set is kept.
+ Assertions.assertEquals("default-stage",
sparkConf.get("spark.executorEnv.STAGE"));
+ }
+
+ @Test
+ @SuppressWarnings("unchecked")
+ public void testAcceptEnvironmentNames() {
+ for (String valid : new String[] {"-X", ".X", "_X", "a.b-c_1", ".a.."}) {
+ SparkJobTemplate template =
+ SparkJobTemplate.builder()
+ .withName("user-job")
+ .withExecutable("https://repo/app.jar";)
+ .withClassName("org.example.App")
+ .withEnvironments(ImmutableMap.of(valid, "v"))
+ .build();
+ Map<String, String> sparkConf =
+ (Map<String, String>) getSpec(template, configs).get("sparkConf");
+ Assertions.assertEquals("v", sparkConf.get("spark.kubernetes.driverEnv."
+ valid), valid);
+ Assertions.assertEquals("v", sparkConf.get("spark.executorEnv." +
valid), valid);
+ }
+ }
+
+ @Test
+ public void testRejectInvalidEnvironmentName() {
+ for (String invalid : new String[] {"", "MY VAR", "A=B", "1ST", "PATH/X",
".", "..", "..CFG"}) {
+ SparkJobTemplate template =
+ SparkJobTemplate.builder()
+ .withName("user-job")
+ .withExecutable("https://repo/app.jar";)
+ .withClassName("org.example.App")
+ .withEnvironments(ImmutableMap.of(invalid, "secret-value"))
+ .build();
+ IllegalArgumentException e =
+ Assertions.assertThrows(
+ IllegalArgumentException.class,
+ () -> SparkApplicationUtils.buildSparkApplication(CONTEXT,
template, configs));
+ Assertions.assertTrue(e.getMessage().contains("user-job"),
e.getMessage());
+ Assertions.assertTrue(e.getMessage().contains("'" + invalid + "'"),
e.getMessage());
+ // The value of an environment variable never goes into an error message.
+ Assertions.assertFalse(e.getMessage().contains("secret-value"),
e.getMessage());
+ }
}
public static K8sJobExecutorConfigs newConfigs(Map<String, String> extra) {
