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 51a316e320 [#13613] feat(k8s-job-executor): Build the SparkApplication
of a Spark job and map its state to the job status (part 2/3) (#13725)
51a316e320 is described below
commit 51a316e320cdeced1dcaf612832f47b9f64db24f
Author: Jerry Shao <[email protected]>
AuthorDate: Wed Oct 14 13:48:40 2026 +0800
[#13613] feat(k8s-job-executor): Build the SparkApplication of a Spark job
and map its state to the job status (part 2/3) (#13725)
### What changes were proposed in this pull request?
This is part 2 of 3 of #13613, which adds a job executor named `k8s`
that runs Spark jobs on Kubernetes. Part 1 (#13724) is merged.
This part adds how the executor talks to
[apache/spark-kubernetes-operator](https://github.com/apache/spark-kubernetes-operator)
0.9.x. Both classes are plain functions that don't call Kubernetes:
- **`SparkApplicationUtils`** builds the `SparkApplication` of a Spark
job template.
- The executable, main class, arguments, jars, files, archives and Spark
configs go into the SparkApplication.
- `spark.master` and `spark.submit.deployMode` are removed, as the
operator owns them.
- The image, driver service account and namespace come from the executor
config as defaults, and the template can override them.
- Resources must be URIs that the cluster can reach: paths on the
Gravitino server are rejected.
- Templates with environment variables are rejected until #13617 passes
them through Secrets.
- **`SparkApplicationStatusUtils`** maps the state of a SparkApplication
to the job status.
- The final states `ResourceReleased` and
`TerminatedWithoutReleaseResources` walk back through the state history
to the last outcome.
- A SparkApplication without any status after a timeout fails.
- An unknown state, of a newer operator, keeps the job unfinished.
- The started and finished times come from the driver container, because
the operator records a state some time after it happens.
### Why are the changes needed?
See #13554. The executor in part 3 (#13700) uses these to submit jobs
and to report their status.
Part of: #13613
### Does this PR introduce _any_ user-facing change?
No.
### How was this patch tested?
Unit tests. The state mapping was also checked against operator 0.9.0 on
a local Kubernetes cluster, as part of the manual tests of #13700.
🤖 Generated with [Claude Code](https://claude.com/claude-code)
---------
Co-authored-by: Claude Opus 5.5 <[email protected]>
---
.../job/k8s/spark/SparkApplicationStatusUtils.java | 480 +++++++++++++
.../job/k8s/spark/SparkApplicationUtils.java | 245 +++++++
.../k8s/spark/TestSparkApplicationStatusUtils.java | 767 +++++++++++++++++++++
.../job/k8s/spark/TestSparkApplicationUtils.java | 471 +++++++++++++
4 files changed, 1963 insertions(+)
diff --git
a/plugins/k8s-job-executor/src/main/java/org/apache/gravitino/job/k8s/spark/SparkApplicationStatusUtils.java
b/plugins/k8s-job-executor/src/main/java/org/apache/gravitino/job/k8s/spark/SparkApplicationStatusUtils.java
new file mode 100644
index 0000000000..90c17a664e
--- /dev/null
+++
b/plugins/k8s-job-executor/src/main/java/org/apache/gravitino/job/k8s/spark/SparkApplicationStatusUtils.java
@@ -0,0 +1,480 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.gravitino.job.k8s.spark;
+
+import com.google.common.collect.ImmutableSet;
+import io.fabric8.kubernetes.api.model.ContainerState;
+import io.fabric8.kubernetes.api.model.ContainerStatus;
+import io.fabric8.kubernetes.api.model.GenericKubernetesResource;
+import io.fabric8.kubernetes.api.model.ObjectMeta;
+import io.fabric8.kubernetes.api.model.Pod;
+import java.time.Instant;
+import java.time.format.DateTimeParseException;
+import java.util.ArrayList;
+import java.util.Comparator;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Set;
+import javax.annotation.Nullable;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.gravitino.connector.job.JobExecutionInfo;
+import org.apache.gravitino.job.JobHandle;
+import org.apache.gravitino.job.k8s.K8sJobResourceUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Maps the state of a SparkApplication of apache/spark-kubernetes-operator to
the Gravitino job
+ * execution info. The state names are the operator's {@code
ApplicationStateSummary}.
+ */
+public final class SparkApplicationStatusUtils {
+
+ /**
+ * A state of the SparkApplication, an entry of its state transition
history. It may carry the
+ * status of the driver pod the operator observed when entering the state.
+ */
+ private static final class State {
+
+ private final long id;
+
+ private final String summary;
+
+ @Nullable private final Instant time;
+
+ @Nullable private final Instant driverStartedAt;
+
+ @Nullable private final Instant driverFinishedAt;
+
+ private State(
+ long id,
+ String summary,
+ @Nullable Instant time,
+ @Nullable Instant driverStartedAt,
+ @Nullable Instant driverFinishedAt) {
+ this.id = id;
+ this.summary = summary;
+ this.time = time;
+ this.driverStartedAt = driverStartedAt;
+ this.driverFinishedAt = driverFinishedAt;
+ }
+ }
+
+ private static final Logger LOG =
LoggerFactory.getLogger(SparkApplicationStatusUtils.class);
+
+ private static final Set<String> QUEUED_STATES =
+ ImmutableSet.of("Submitted", "ScheduledToRestart", "DriverRequested");
+
+ // The operator enters DriverStarted when the driver container is ready, so
the job is running
+ // from then on. DriverReady only adds that the whole driver pod is ready,
which a sidecar that
+ // isn't ready can hold back.
+ private static final Set<String> STARTED_STATES =
+ ImmutableSet.of(
+ "DriverStarted",
+ "DriverReady",
+ "InitializedBelowThresholdExecutors",
+ "RunningHealthy",
+ "RunningWithPartialCapacity",
+ "RunningWithBelowThresholdExecutors");
+
+ private static final String SUCCEEDED_STATE = "Succeeded";
+
+ private static final Set<String> FAILED_STATES =
+ ImmutableSet.of(
+ "Failed",
+ "SchedulingFailure",
+ "DriverEvicted",
+ "DriverStartTimedOut",
+ "ExecutorsStartTimedOut",
+ "DriverReadyTimedOut");
+
+ // The last states of every application, which don't tell success from
failure.
+ private static final Set<String> STOPPED_STATES =
+ ImmutableSet.of("ResourceReleased", "TerminatedWithoutReleaseResources");
+
+ private SparkApplicationStatusUtils() {}
+
+ /**
+ * Maps the SparkApplication to the job execution info.
+ *
+ * <ul>
+ * <li>A SparkApplication that Gravitino requested to cancel is cancelled
once it is being
+ * deleted or has stopped.
+ * <li>A SparkApplication being deleted by someone else, for example by
its TTL, keeps the
+ * outcome it reached, or fails if it has none.
+ * <li>The stopped states walk back through the state transition history
to the last outcome.
+ * <li>A SparkApplication without any status after the timeout fails, for
example if its
+ * namespace isn't watched by the operator.
+ * <li>An unknown state, of a newer operator, after an outcome is a
stopped state too. Before
+ * any outcome it keeps the job unfinished: started if the application
has reached a started
+ * state, queued otherwise.
+ * </ul>
+ *
+ * <p>The operator records a state when it observes it, which can be a while
after the driver
+ * actually started or finished. So the times are the ones Kubernetes
reports for the driver
+ * container, in seconds, taken from the given driver pod or from the driver
pod status the
+ * operator keeps with some states. Without them, the job started when the
application first
+ * reached a started state, and finished when it reached its outcome. As the
driver container
+ * status is only known once the job finished, the started time of a running
job is refined to the
+ * driver's when it finishes. A job that stopped without an outcome finished
when its driver
+ * container terminated if that is known, else when it was deleted, or else
when it entered its
+ * last state.
+ *
+ * @param app the SparkApplication
+ * @param driver the driver pod, or null if unknown
+ * @param now the current time
+ * @param noStatusTimeoutMs the time after which a SparkApplication without
status fails
+ * @return the job execution info
+ */
+ public static JobExecutionInfo executionInfoOf(
+ GenericKubernetesResource app, @Nullable Pod driver, Instant now, long
noStatusTimeoutMs) {
+ List<State> history = stateHistory(app);
+ State current = history.isEmpty() ? null : history.get(history.size() - 1);
+ State outcome = lastOutcome(history);
+ Instant createdAt = parseTime(app.getMetadata().getCreationTimestamp());
+ Instant deletedAt = parseTime(app.getMetadata().getDeletionTimestamp());
+
+ ContainerState driverState =
+ driver == null ? null : driverContainerState(driver,
driverContainerName(app));
+ Instant driverStartedAt = null;
+ Instant driverFinishedAt = null;
+ if (driverState != null && driverState.getTerminated() != null) {
+ driverStartedAt = parseTime(driverState.getTerminated().getStartedAt());
+ driverFinishedAt =
parseTime(driverState.getTerminated().getFinishedAt());
+ } else if (driverState != null && driverState.getRunning() != null) {
+ driverStartedAt = parseTime(driverState.getRunning().getStartedAt());
+ }
+ Instant startedAt = startedAt(history, driverStartedAt, createdAt);
+ // When the application stopped: when the driver container terminated if
that is known, even
+ // before the operator records an outcome, else when the application
reached its outcome, else
+ // when it was deleted, else when it entered its current state. A deletion
time is in whole
+ // seconds, so it may read earlier than a started time of the same second,
which Gravitino
+ // would then drop.
+ Instant stoppedAt;
+ if (driverFinishedAt != null) {
+ stoppedAt = driverFinishedAt;
+ } else if (outcome != null) {
+ stoppedAt = outcome.driverFinishedAt != null ? outcome.driverFinishedAt
: outcome.time;
+ } else if (deletedAt != null) {
+ stoppedAt = startedAt != null && deletedAt.isBefore(startedAt) ?
startedAt : deletedAt;
+ } else {
+ stoppedAt = current == null ? null : current.time;
+ }
+
+ JobHandle.Status status;
+ Instant finishedAt = null;
+ if (K8sJobResourceUtils.isCancelRequested(app)
+ && (deletedAt != null || isStopped(current, outcome))) {
+ status = JobHandle.Status.CANCELLED;
+ finishedAt = stoppedAt;
+ } else if (deletedAt != null || isStopped(current, outcome)) {
+ // Stopped, or deleted by someone else than Gravitino, for example by
its TTL or kubectl.
+ status = outcome == null ? JobHandle.Status.FAILED : statusOf(outcome);
+ finishedAt = stoppedAt;
+ } else if (current == null) {
+ if (createdAt != null &&
createdAt.plusMillis(noStatusTimeoutMs).isBefore(now)) {
+ LOG.warn(
+ "SparkApplication {} has no status after {} ms, is its namespace
watched by the Spark"
+ + " operator? Considering it as failed",
+ nameOf(app),
+ noStatusTimeoutMs);
+ status = JobHandle.Status.FAILED;
+ } else {
+ status = JobHandle.Status.QUEUED;
+ }
+ } else if (QUEUED_STATES.contains(current.summary)) {
+ status = JobHandle.Status.QUEUED;
+ } else if (STARTED_STATES.contains(current.summary)) {
+ status = JobHandle.Status.STARTED;
+ } else {
+ // A state this version doesn't know, of a newer operator, before any
outcome. It can't be
+ // told whether it is final, so the job stays unfinished: started if it
has run, queued
+ // otherwise.
+ status =
+ history.stream().anyMatch(s -> STARTED_STATES.contains(s.summary))
+ ? JobHandle.Status.STARTED
+ : JobHandle.Status.QUEUED;
+ LOG.warn(
+ "Unknown state {} of SparkApplication {}, considering it as {}",
+ current.summary,
+ nameOf(app),
+ status);
+ }
+
+ return JobExecutionInfo.builder()
+ .withStatus(status)
+ .withStartedAt(startedAt)
+ .withFinishedAt(finishedAt)
+ .build();
+ }
+
+ /**
+ * Returns whether the operator has reported a state of the SparkApplication
that can be read.
+ *
+ * @param app the SparkApplication
+ * @return true if the application has a state
+ */
+ public static boolean hasState(GenericKubernetesResource app) {
+ return !stateHistory(app).isEmpty();
+ }
+
+ /**
+ * Returns whether the SparkApplication has stopped running, successfully or
not.
+ *
+ * @param app the SparkApplication
+ * @return true if the application has reached an outcome or a stopped state
+ */
+ public static boolean isStopped(GenericKubernetesResource app) {
+ List<State> history = stateHistory(app);
+ return isStopped(
+ history.isEmpty() ? null : history.get(history.size() - 1),
lastOutcome(history));
+ }
+
+ /**
+ * Returns whether the application stopped: its current state is an outcome
or a stopped state, or
+ * a state this version doesn't know that follows an outcome. A known state
after an outcome means
+ * the application runs again.
+ */
+ private static boolean isStopped(@Nullable State current, @Nullable State
outcome) {
+ if (current == null) {
+ return false;
+ }
+ boolean running =
+ QUEUED_STATES.contains(current.summary) ||
STARTED_STATES.contains(current.summary);
+ return isOutcome(current)
+ || STOPPED_STATES.contains(current.summary)
+ || (!running && outcome != null);
+ }
+
+ private static boolean isOutcome(State state) {
+ return SUCCEEDED_STATE.equals(state.summary) ||
FAILED_STATES.contains(state.summary);
+ }
+
+ private static JobHandle.Status statusOf(State outcome) {
+ return SUCCEEDED_STATE.equals(outcome.summary)
+ ? JobHandle.Status.SUCCEEDED
+ : JobHandle.Status.FAILED;
+ }
+
+ /**
+ * Returns when the job started: when the driver container did, by the
driver pod or else by the
+ * driver status the operator observed, or else when the application reached
a started state.
+ */
+ @Nullable
+ private static Instant startedAt(
+ List<State> history, @Nullable Instant driverStartedAt, @Nullable
Instant createdAt) {
+ Instant containerStartedAt =
+ driverStartedAt != null
+ ? driverStartedAt
+ : history.stream()
+ .map(s -> s.driverStartedAt)
+ .filter(Objects::nonNull)
+ .min(Comparator.naturalOrder())
+ .orElse(null);
+ // A container that failed to start can report a start time of the epoch.
A driver can't have
+ // started before its application was created.
+ if (containerStartedAt != null
+ && (createdAt == null || !containerStartedAt.isBefore(createdAt))) {
+ return containerStartedAt;
+ }
+ return history.stream()
+ .filter(s -> STARTED_STATES.contains(s.summary))
+ .findFirst()
+ .map(s -> s.time)
+ .orElse(null);
+ }
+
+ @Nullable
+ private static State lastOutcome(List<State> history) {
+ for (int i = history.size() - 1; i >= 0; i--) {
+ if (isOutcome(history.get(i))) {
+ return history.get(i);
+ }
+ }
+ return null;
+ }
+
+ /**
+ * Returns the state transition history of the SparkApplication, oldest
first. It ends with the
+ * current state, which is also recorded on its own in case the history is
trimmed.
+ */
+ @SuppressWarnings("unchecked")
+ private static List<State> stateHistory(GenericKubernetesResource app) {
+ List<State> history = new ArrayList<>();
+ Object status = app.getAdditionalProperties().get("status");
+ if (!(status instanceof Map)) {
+ return history;
+ }
+ String driverContainer = driverContainerName(app);
+
+ Object transitions = ((Map<String, Object>)
status).get("stateTransitionHistory");
+ if (transitions instanceof Map) {
+ ((Map<String, Object>) transitions)
+ .forEach(
+ (id, state) -> {
+ State parsed = parseState(id, state, driverContainer);
+ if (parsed != null) {
+ history.add(parsed);
+ }
+ });
+ history.sort(Comparator.comparingLong(s -> s.id));
+ }
+
+ State current =
+ parseState(null, ((Map<String, Object>) status).get("currentState"),
driverContainer);
+ if (current != null
+ && (history.isEmpty()
+ || !current.summary.equals(history.get(history.size() -
1).summary))) {
+ history.add(current);
+ }
+ return history;
+ }
+
+ @Nullable
+ @SuppressWarnings("unchecked")
+ private static State parseState(
+ @Nullable String id, @Nullable Object state, String driverContainer) {
+ if (!(state instanceof Map)) {
+ return null;
+ }
+ Map<String, Object> fields = (Map<String, Object>) state;
+ Object summary = fields.get("currentStateSummary");
+ if (!(summary instanceof String)) {
+ return null;
+ }
+ long order;
+ try {
+ order = id == null ? Long.MAX_VALUE : Long.parseLong(id);
+ } catch (NumberFormatException e) {
+ return null;
+ }
+
+ Map<String, Object> driverState =
+ observedDriverContainerState(fields.get("lastObservedDriverStatus"),
driverContainer);
+ return new State(
+ order,
+ (String) summary,
+ timeOf(fields.get("lastTransitionTime")),
+ driverState == null ? null : driverStartedAt(driverState),
+ driverState == null ? null : driverFinishedAt(driverState));
+ }
+
+ @Nullable
+ private static Instant driverStartedAt(Map<String, Object> driverState) {
+ Map<String, Object> terminated = mapOf(driverState.get("terminated"));
+ if (terminated != null) {
+ return timeOf(terminated.get("startedAt"));
+ }
+ Map<String, Object> running = mapOf(driverState.get("running"));
+ return running == null ? null : timeOf(running.get("startedAt"));
+ }
+
+ @Nullable
+ private static Instant driverFinishedAt(Map<String, Object> driverState) {
+ Map<String, Object> terminated = mapOf(driverState.get("terminated"));
+ return terminated == null ? null : timeOf(terminated.get("finishedAt"));
+ }
+
+ /**
+ * Returns the name of the driver container: the one a job names with a pod
template, or else the
+ * one Spark gives it.
+ */
+ private static String driverContainerName(GenericKubernetesResource app) {
+ Map<String, Object> spec =
mapOf(app.getAdditionalProperties().get("spec"));
+ Map<String, Object> sparkConf = spec == null ? null :
mapOf(spec.get("sparkConf"));
+ Object name =
+ sparkConf == null ? null :
sparkConf.get(SparkApplicationUtils.SPARK_DRIVER_CONTAINER_NAME);
+ return name instanceof String && StringUtils.isNotBlank((String) name)
+ ? (String) name
+ : SparkApplicationUtils.DRIVER_CONTAINER_NAME;
+ }
+
+ /** Returns the state of the driver container of the driver pod. */
+ @Nullable
+ private static ContainerState driverContainerState(Pod driver, String
driverContainer) {
+ List<ContainerStatus> containers =
+ driver.getStatus() == null ? null :
driver.getStatus().getContainerStatuses();
+ if (containers == null) {
+ return null;
+ }
+ for (ContainerStatus container : containers) {
+ if (driverContainer.equals(container.getName())) {
+ return container.getState();
+ }
+ }
+ // The only container is the driver, even if a pod template renamed it.
+ return containers.size() == 1 ? containers.get(0).getState() : null;
+ }
+
+ /** Returns the state of the driver container in the observed status of the
driver pod. */
+ @Nullable
+ @SuppressWarnings("unchecked")
+ private static Map<String, Object> observedDriverContainerState(
+ @Nullable Object podStatus, String driverContainer) {
+ Map<String, Object> status = mapOf(podStatus);
+ Object containers = status == null ? null :
status.get("containerStatuses");
+ if (!(containers instanceof List)) {
+ return null;
+ }
+
+ List<Object> containerStatuses = (List<Object>) containers;
+ for (Object container : containerStatuses) {
+ Map<String, Object> fields = mapOf(container);
+ if (fields != null && driverContainer.equals(fields.get("name"))) {
+ return mapOf(fields.get("state"));
+ }
+ }
+ // The only container is the driver, even if a pod template renamed it.
+ return containerStatuses.size() == 1 ? stateOf(containerStatuses.get(0)) :
null;
+ }
+
+ @Nullable
+ private static Map<String, Object> stateOf(@Nullable Object container) {
+ Map<String, Object> fields = mapOf(container);
+ return fields == null ? null : mapOf(fields.get("state"));
+ }
+
+ @Nullable
+ @SuppressWarnings("unchecked")
+ private static Map<String, Object> mapOf(@Nullable Object value) {
+ return value instanceof Map ? (Map<String, Object>) value : null;
+ }
+
+ @Nullable
+ private static Instant timeOf(@Nullable Object value) {
+ return value instanceof String ? parseTime((String) value) : null;
+ }
+
+ @Nullable
+ private static Instant parseTime(@Nullable String time) {
+ if (time == null) {
+ return null;
+ }
+ try {
+ return Instant.parse(time);
+ } catch (DateTimeParseException e) {
+ return null;
+ }
+ }
+
+ private static String nameOf(GenericKubernetesResource app) {
+ ObjectMeta metadata = app.getMetadata();
+ return metadata.getNamespace() + "/" + metadata.getName();
+ }
+}
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
new file mode 100644
index 0000000000..aa49a1c5fb
--- /dev/null
+++
b/plugins/k8s-job-executor/src/main/java/org/apache/gravitino/job/k8s/spark/SparkApplicationUtils.java
@@ -0,0 +1,245 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.gravitino.job.k8s.spark;
+
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableMap;
+import io.fabric8.kubernetes.api.model.GenericKubernetesResource;
+import io.fabric8.kubernetes.api.model.GenericKubernetesResourceBuilder;
+import io.fabric8.kubernetes.client.dsl.base.ResourceDefinitionContext;
+import java.util.ArrayList;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Locale;
+import java.util.Map;
+import java.util.TreeMap;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.gravitino.connector.job.JobContext;
+import org.apache.gravitino.job.SparkJobTemplate;
+import org.apache.gravitino.job.k8s.K8sJobExecutorConfigs;
+import org.apache.gravitino.job.k8s.K8sJobResourceUtils;
+
+/**
+ * Builds the SparkApplication of a Spark job, a custom resource of <a
+ *
href="https://github.com/apache/spark-kubernetes-operator";>apache/spark-kubernetes-operator</a>
+ * 0.9.x.
+ */
+public final class SparkApplicationUtils {
+
+ /** The SparkApplication resource definition. */
+ public static final ResourceDefinitionContext SPARK_APPLICATION =
+ new ResourceDefinitionContext.Builder()
+ .withGroup("spark.apache.org")
+ .withVersion("v1")
+ .withKind("SparkApplication")
+ .withPlural("sparkapplications")
+ .withNamespaced(true)
+ .build();
+
+ /** The label the operator sets on the pods of a SparkApplication, whose
value is its name. */
+ public static final String LABEL_SPARK_APP_NAME =
"spark.operator/spark-app-name";
+
+ /** The label Spark sets on its pods, whose value is the role of the pod. */
+ public static final String LABEL_SPARK_ROLE = "spark-role";
+
+ /** The {@link #LABEL_SPARK_ROLE} value of the driver pod. */
+ public static final String SPARK_ROLE_DRIVER = "driver";
+
+ /** The name Spark gives the driver container, unless a pod template renames
it. */
+ public static final String DRIVER_CONTAINER_NAME = "spark-kubernetes-driver";
+
+ /** The Spark configuration that names the driver container of a job's
driver pod template. */
+ public static final String SPARK_DRIVER_CONTAINER_NAME =
+ "spark.kubernetes.driver.podTemplateContainerName";
+
+ /** The Spark configuration of the namespace a job runs in. */
+ public static final String SPARK_NAMESPACE = "spark.kubernetes.namespace";
+
+ private static final String SPARK_MASTER = "spark.master";
+ private static final String SPARK_DEPLOY_MODE = "spark.submit.deployMode";
+ private static final String SPARK_IMAGE = "spark.kubernetes.container.image";
+ private static final String SPARK_DRIVER_SERVICE_ACCOUNT =
+ "spark.kubernetes.authenticate.driver.serviceAccountName";
+ 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 Pattern URI_SCHEME =
Pattern.compile("^([A-Za-z][A-Za-z0-9+.-]*):");
+
+ private static final Pattern URI_USER_INFO =
Pattern.compile("(?<=://)[^/?#@]*@");
+
+ private static final Pattern URI_QUERY_OR_FRAGMENT =
Pattern.compile("[?#].*$");
+
+ private SparkApplicationUtils() {}
+
+ /**
+ * Builds the SparkApplication of a Spark job.
+ *
+ * @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
+ * @return the SparkApplication
+ * @throws IllegalArgumentException if the job can't run on Kubernetes
+ */
+ 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()));
+ }
+
+ checkClusterReachable(template.executable());
+ template.jars().forEach(SparkApplicationUtils::checkClusterReachable);
+ template.files().forEach(SparkApplicationUtils::checkClusterReachable);
+ template.archives().forEach(SparkApplicationUtils::checkClusterReachable);
+
+ Map<String, String> sparkConf = new TreeMap<>(configs.sparkConf());
+ sparkConf.putAll(template.configs());
+ // The operator decides the master and the deploy mode.
+ sparkConf.remove(SPARK_MASTER);
+ sparkConf.remove(SPARK_DEPLOY_MODE);
+ putIfBlank(sparkConf, SPARK_IMAGE, configs.sparkImage());
+ putIfBlank(sparkConf, SPARK_DRIVER_SERVICE_ACCOUNT,
configs.sparkServiceAccount());
+ // Only a job chooses its namespace this way. A default set in the Spark
configurations of
+ // the executor would silently shadow the namespace of the executor
configurations.
+ String namespace =
StringUtils.trimToNull(template.configs().get(SPARK_NAMESPACE));
+ if (namespace == null) {
+ namespace = configs.namespace();
+ } else if (!K8sJobResourceUtils.isValidNamespace(namespace)) {
+ throw new IllegalArgumentException(
+ String.format(
+ "%s of job template %s isn't a valid namespace name: %s",
+ SPARK_NAMESPACE, template.name(), namespace));
+ }
+ sparkConf.put(SPARK_NAMESPACE, namespace);
+ appendList(sparkConf, SPARK_JARS, template.jars());
+ appendList(sparkConf, SPARK_FILES, template.files());
+ appendList(sparkConf, SPARK_ARCHIVES, template.archives());
+ // The resources may also come from the Spark configurations, of the job
or of the executor.
+ for (String key : ImmutableList.of(SPARK_JARS, SPARK_FILES,
SPARK_ARCHIVES)) {
+ String resources = sparkConf.get(key);
+ if (StringUtils.isNotBlank(resources)) {
+ checkClusterReachable(resources);
+ }
+ }
+
+ Map<String, Object> spec = new LinkedHashMap<>();
+ if (StringUtils.isNotBlank(template.className())) {
+ spec.put("mainClass", template.className());
+ }
+ // The operator takes the main resource of a Python application from
pyFiles.
+ spec.put(isPythonFile(template.executable()) ? "pyFiles" : "jars",
template.executable());
+ spec.put("driverArgs", new ArrayList<>(template.arguments()));
+ spec.put("sparkConf", sparkConf);
+ spec.put("runtimeVersions", ImmutableMap.of("sparkVersion",
configs.sparkVersion()));
+ spec.put(
+ "applicationTolerations",
+ ImmutableMap.of(
+ // A Gravitino job run is one execution. Restarts would also make
the failed states
+ // non-final, while Gravitino never changes a job's final status.
+ "restartConfig", ImmutableMap.of("restartPolicy", "Never"),
+ // Retain the driver pod after the application stops, so its
output can be read.
+ "resourceRetainPolicy", "Always",
+ "resourceRetainDurationMillis",
configs.sparkResourceRetainDurationMs(),
+ "ttlAfterStopMillis", configs.sparkTtlAfterStopMs(),
+ // The executor start timeout isn't set: the operator only checks
it against the
+ // minimum number of executors of instanceConfig, which isn't set
either.
+ "applicationTimeoutConfig",
+ ImmutableMap.of(
+ "driverStartTimeoutMillis",
configs.sparkDriverStartTimeoutMs(),
+ "driverReadyTimeoutMillis",
configs.sparkDriverReadyTimeoutMs())));
+
+ GenericKubernetesResource app =
+ new GenericKubernetesResourceBuilder()
+ .withApiVersion(SPARK_APPLICATION.getGroup() + "/" +
SPARK_APPLICATION.getVersion())
+ .withKind(SPARK_APPLICATION.getKind())
+ .withNewMetadata()
+ .withName(K8sJobResourceUtils.resourceName(configs,
context.jobId()))
+ .withNamespace(namespace)
+ .withLabels(K8sJobResourceUtils.jobLabels(context.jobId()))
+ .addToAnnotations(K8sJobResourceUtils.ANNOTATION_METALAKE,
context.metalake())
+ .endMetadata()
+ .build();
+ app.setAdditionalProperty("spec", spec);
+ return app;
+ }
+
+ /**
+ * 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.
+ */
+ private static void checkClusterReachable(String resource) {
+ String[] uris = resource.split(",");
+ for (String uri : uris) {
+ // Spark ignores an empty element of a list.
+ if (uris.length > 1 && StringUtils.isBlank(uri)) {
+ continue;
+ }
+ Matcher scheme = URI_SCHEME.matcher(uri.trim());
+ // No scheme, a Windows drive letter, or a file URI.
+ if (!scheme.find()
+ || scheme.group(1).length() == 1
+ || "file".equalsIgnoreCase(scheme.group(1))) {
+ throw new IllegalArgumentException(
+ String.format(
+ "Resource %s of the job is a path on the Gravitino server,
which the Spark job on"
+ + " Kubernetes can't reach. Use a URI reachable from the
cluster instead, such"
+ + " as local:// for a file in the image, https://, s3a://
or hdfs://",
+ withoutSecrets(uri.trim())));
+ }
+ }
+ }
+
+ /**
+ * Returns a URI without its user info, query and fragment, which may carry
credentials such as a
+ * password or a signature, to put it in a message returned to the caller.
+ */
+ private static String withoutSecrets(String uri) {
+ return URI_QUERY_OR_FRAGMENT
+ .matcher(URI_USER_INFO.matcher(uri).replaceFirst(""))
+ .replaceFirst("");
+ }
+
+ private static boolean isPythonFile(String uri) {
+ // Look at the path only, the URI may carry a query or a fragment.
+ String path = URI_QUERY_OR_FRAGMENT.matcher(uri).replaceFirst("");
+ return path.toLowerCase(Locale.ROOT).endsWith(".py");
+ }
+
+ private static void putIfBlank(Map<String, String> sparkConf, String key,
String value) {
+ if (StringUtils.isBlank(sparkConf.get(key))) {
+ sparkConf.put(key, value);
+ }
+ }
+
+ private static void appendList(Map<String, String> sparkConf, String key,
List<String> values) {
+ if (values.isEmpty()) {
+ return;
+ }
+ String joined = String.join(",", values);
+ String existing = sparkConf.get(key);
+ sparkConf.put(key, StringUtils.isBlank(existing) ? joined : existing + ","
+ joined);
+ }
+}
diff --git
a/plugins/k8s-job-executor/src/test/java/org/apache/gravitino/job/k8s/spark/TestSparkApplicationStatusUtils.java
b/plugins/k8s-job-executor/src/test/java/org/apache/gravitino/job/k8s/spark/TestSparkApplicationStatusUtils.java
new file mode 100644
index 0000000000..464bdec9cd
--- /dev/null
+++
b/plugins/k8s-job-executor/src/test/java/org/apache/gravitino/job/k8s/spark/TestSparkApplicationStatusUtils.java
@@ -0,0 +1,767 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.gravitino.job.k8s.spark;
+
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableMap;
+import io.fabric8.kubernetes.api.model.GenericKubernetesResource;
+import io.fabric8.kubernetes.api.model.GenericKubernetesResourceBuilder;
+import io.fabric8.kubernetes.api.model.Pod;
+import io.fabric8.kubernetes.api.model.PodBuilder;
+import io.fabric8.kubernetes.client.utils.Serialization;
+import java.time.Instant;
+import java.util.LinkedHashMap;
+import java.util.Map;
+import org.apache.gravitino.connector.job.JobExecutionInfo;
+import org.apache.gravitino.job.JobHandle;
+import org.apache.gravitino.job.k8s.K8sJobResourceUtils;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.CsvSource;
+
+public class TestSparkApplicationStatusUtils {
+
+ private static final Instant CREATED_AT =
Instant.parse("2026-09-30T00:00:00Z");
+
+ private static final Instant NOW = CREATED_AT.plusSeconds(60);
+
+ private static final long NO_STATUS_TIMEOUT_MS = 600_000L;
+
+ @ParameterizedTest
+ @CsvSource({
+ "Submitted, QUEUED",
+ "ScheduledToRestart, QUEUED",
+ "DriverRequested, QUEUED",
+ "DriverStarted, STARTED",
+ "DriverReady, STARTED",
+ "InitializedBelowThresholdExecutors, STARTED",
+ "RunningHealthy, STARTED",
+ "RunningWithPartialCapacity, STARTED",
+ "RunningWithBelowThresholdExecutors, STARTED",
+ "Succeeded, SUCCEEDED",
+ "Failed, FAILED",
+ "SchedulingFailure, FAILED",
+ "DriverEvicted, FAILED",
+ "DriverStartTimedOut, FAILED",
+ "ExecutorsStartTimedOut, FAILED",
+ "DriverReadyTimedOut, FAILED",
+ "ResourceReleased, FAILED",
+ "TerminatedWithoutReleaseResources, FAILED"
+ })
+ public void testCurrentState(String state, JobHandle.Status expected) {
+ GenericKubernetesResource app = app("Submitted", state);
+ Assertions.assertEquals(expected, executionInfoOf(app).status());
+ }
+
+ @Test
+ public void testUnknownState() {
+ Assertions.assertEquals(
+ JobHandle.Status.QUEUED, executionInfoOf(app("Submitted",
"SomeNewState")).status());
+ Assertions.assertEquals(
+ JobHandle.Status.STARTED,
+ executionInfoOf(app("Submitted", "DriverReady",
"SomeNewState")).status());
+ }
+
+ @Test
+ public void testNoStatus() {
+ GenericKubernetesResource app = app();
+ Assertions.assertEquals(JobHandle.Status.QUEUED,
executionInfoOf(app).status());
+ Assertions.assertEquals(
+ JobHandle.Status.FAILED,
+ SparkApplicationStatusUtils.executionInfoOf(
+ app, null, CREATED_AT.plusMillis(NO_STATUS_TIMEOUT_MS + 1),
NO_STATUS_TIMEOUT_MS)
+ .status());
+ }
+
+ @Test
+ public void testTimes() {
+ JobExecutionInfo queued = executionInfoOf(app("Submitted",
"DriverRequested"));
+ Assertions.assertNull(queued.startedAt());
+ Assertions.assertNull(queued.finishedAt());
+
+ JobExecutionInfo running =
+ executionInfoOf(app("Submitted", "DriverRequested", "DriverReady",
"RunningHealthy"));
+ Assertions.assertEquals(JobHandle.Status.STARTED, running.status());
+ Assertions.assertEquals(timeOf(2), running.startedAt());
+ Assertions.assertNull(running.finishedAt());
+
+ JobExecutionInfo succeeded =
+ executionInfoOf(
+ app(
+ "Submitted",
+ "DriverRequested",
+ "DriverReady",
+ "RunningHealthy",
+ "Succeeded",
+ "TerminatedWithoutReleaseResources",
+ "ResourceReleased"));
+ Assertions.assertEquals(
+ JobExecutionInfo.builder()
+ .withStatus(JobHandle.Status.SUCCEEDED)
+ .withStartedAt(timeOf(2))
+ .withFinishedAt(timeOf(4))
+ .build(),
+ succeeded);
+
+ JobExecutionInfo neverStarted =
+ executionInfoOf(app("Submitted", "DriverRequested",
"DriverStartTimedOut"));
+ Assertions.assertEquals(JobHandle.Status.FAILED, neverStarted.status());
+ Assertions.assertNull(neverStarted.startedAt());
+ Assertions.assertEquals(timeOf(2), neverStarted.finishedAt());
+ }
+
+ @Test
+ public void testDriverContainerTimes() {
+ Instant driverStartedAt = Instant.parse("2026-09-30T00:00:01Z");
+ Instant driverFinishedAt = Instant.parse("2026-09-30T00:00:02Z");
+
+ // The operator observed the driver running when it was ready.
+ GenericKubernetesResource running = app("Submitted", "DriverReady",
"RunningHealthy");
+ Map<String, Map<String, Object>> history = history(running);
+ history.put(
+ "1",
+ withDriverStatus(
+ history.get("1"),
+ ImmutableMap.of("running", ImmutableMap.of("startedAt",
driverStartedAt.toString()))));
+ JobExecutionInfo info = executionInfoOf(running);
+ Assertions.assertEquals(JobHandle.Status.STARTED, info.status());
+ Assertions.assertEquals(driverStartedAt, info.startedAt());
+
+ // The operator observed the driver terminated a while after it finished.
+ GenericKubernetesResource failed =
+ app("Submitted", "DriverReady", "Failed",
"TerminatedWithoutReleaseResources");
+ history = history(failed);
+ history.put(
+ "2",
+ withDriverStatus(
+ history.get("2"),
+ ImmutableMap.of(
+ "terminated",
+ ImmutableMap.of(
+ "startedAt", driverStartedAt.toString(),
+ "finishedAt", driverFinishedAt.toString(),
+ "exitCode", 1))));
+ Assertions.assertEquals(
+ JobExecutionInfo.builder()
+ .withStatus(JobHandle.Status.FAILED)
+ .withStartedAt(driverStartedAt)
+ .withFinishedAt(driverFinishedAt)
+ .build(),
+ executionInfoOf(failed));
+ }
+
+ @Test
+ public void testRenamedDriverContainer() {
+ Instant driverFinishedAt = Instant.parse("2026-09-30T00:00:02Z");
+ GenericKubernetesResource succeeded = app("Submitted", "DriverReady",
"Succeeded");
+ Map<String, Map<String, Object>> history = history(succeeded);
+ Map<String, Object> terminated =
+ ImmutableMap.of(
+ "terminated",
+ ImmutableMap.of(
+ "startedAt", "2026-09-30T00:00:01Z", "finishedAt",
driverFinishedAt.toString()));
+ history.put(
+ "2",
+ ImmutableMap.<String, Object>builder()
+ .putAll(history.get("2"))
+ .put(
+ "lastObservedDriverStatus",
+ ImmutableMap.of(
+ "containerStatuses",
+ ImmutableList.of(ImmutableMap.of("name", "main", "state",
terminated))))
+ .build());
+ Assertions.assertEquals(driverFinishedAt,
executionInfoOf(succeeded).finishedAt());
+
+ // With sidecars, the driver container can't be told apart.
+ history.put(
+ "2",
+ ImmutableMap.<String, Object>builder()
+ .putAll(history(app("Submitted", "DriverReady",
"Succeeded")).get("2"))
+ .put(
+ "lastObservedDriverStatus",
+ ImmutableMap.of(
+ "containerStatuses",
+ ImmutableList.of(
+ ImmutableMap.of("name", "main", "state", terminated),
+ ImmutableMap.of("name", "sidecar", "state",
terminated))))
+ .build());
+ Assertions.assertEquals(timeOf(2),
executionInfoOf(succeeded).finishedAt());
+ }
+
+ @Test
+ public void testDriverPodTimes() {
+ GenericKubernetesResource succeeded =
+ app("Submitted", "DriverReady", "Succeeded",
"TerminatedWithoutReleaseResources");
+ Pod driver =
+ new PodBuilder()
+ .withNewStatus()
+ .addNewContainerStatus()
+ .withName("spark-kubernetes-driver")
+ .withNewState()
+ .withNewTerminated()
+ .withStartedAt("2026-09-30T00:00:00Z")
+ .withFinishedAt("2026-09-30T00:00:01Z")
+ .endTerminated()
+ .endState()
+ .endContainerStatus()
+ .endStatus()
+ .build();
+
+ Assertions.assertEquals(
+ JobExecutionInfo.builder()
+ .withStatus(JobHandle.Status.SUCCEEDED)
+ .withStartedAt(Instant.parse("2026-09-30T00:00:00Z"))
+ .withFinishedAt(Instant.parse("2026-09-30T00:00:01Z"))
+ .build(),
+ SparkApplicationStatusUtils.executionInfoOf(succeeded, driver, NOW,
NO_STATUS_TIMEOUT_MS));
+
+ // A driver pod without status falls back to the operator's times.
+ Assertions.assertEquals(
+ timeOf(2),
+ SparkApplicationStatusUtils.executionInfoOf(
+ succeeded, new PodBuilder().build(), NOW, NO_STATUS_TIMEOUT_MS)
+ .finishedAt());
+ }
+
+ @Test
+ public void testStoppedStatesWalkBackToOutcome() {
+ Assertions.assertEquals(
+ JobHandle.Status.FAILED,
+ executionInfoOf(app("Submitted", "DriverReady", "Failed",
"ResourceReleased")).status());
+ Assertions.assertEquals(
+ JobHandle.Status.SUCCEEDED,
+ executionInfoOf(app("Submitted", "DriverReady", "Succeeded",
"ResourceReleased")).status());
+ Assertions.assertEquals(
+ JobHandle.Status.FAILED,
+ executionInfoOf(app("Submitted", "DriverReady",
"ResourceReleased")).status());
+ }
+
+ @Test
+ public void testHistoryOrderedById() {
+ GenericKubernetesResource app = app("Submitted", "DriverReady",
"Succeeded");
+ // Ids sort numerically, not as strings.
+ Map<String, Object> history = new LinkedHashMap<>();
+ for (int i = 0; i < 12; i++) {
+ history.put(String.valueOf(11 - i), state(i == 0 ? "Succeeded" :
"RunningHealthy", 11 - i));
+ }
+ history.put("0", state("Submitted", 0));
+ status(app).put("stateTransitionHistory", history);
+ status(app).put("currentState", state("Succeeded", 11));
+
+ JobExecutionInfo info = executionInfoOf(app);
+ Assertions.assertEquals(JobHandle.Status.SUCCEEDED, info.status());
+ Assertions.assertEquals(timeOf(1), info.startedAt());
+ Assertions.assertEquals(timeOf(11), info.finishedAt());
+ }
+
+ @Test
+ public void testCurrentStateMissingFromHistory() {
+ GenericKubernetesResource app = app("Submitted", "DriverReady");
+ status(app).put("currentState", state("Succeeded", 5));
+ JobExecutionInfo info = executionInfoOf(app);
+ Assertions.assertEquals(JobHandle.Status.SUCCEEDED, info.status());
+ Assertions.assertEquals(timeOf(5), info.finishedAt());
+ }
+
+ @Test
+ public void testCancelled() {
+ Instant deletedAt = NOW.minusSeconds(1);
+
+ // Cancel requested, but the deletion hasn't taken effect yet.
+ GenericKubernetesResource running = app("Submitted", "DriverReady");
+ cancel(running);
+ Assertions.assertTrue(K8sJobResourceUtils.isCancelRequested(running));
+ Assertions.assertEquals(JobHandle.Status.STARTED,
executionInfoOf(running).status());
+
+ running.getMetadata().setDeletionTimestamp(deletedAt.toString());
+ JobExecutionInfo cancelled = executionInfoOf(running);
+ Assertions.assertEquals(JobHandle.Status.CANCELLED, cancelled.status());
+ Assertions.assertEquals(timeOf(1), cancelled.startedAt());
+ Assertions.assertEquals(deletedAt, cancelled.finishedAt());
+
+ // The application stopped before the deletion.
+ GenericKubernetesResource stopped = app("Submitted", "DriverReady",
"Failed");
+ cancel(stopped);
+ JobExecutionInfo info = executionInfoOf(stopped);
+ Assertions.assertEquals(JobHandle.Status.CANCELLED, info.status());
+ Assertions.assertEquals(timeOf(2), info.finishedAt());
+ }
+
+ @Test
+ public void testDeletedBySomeoneElse() {
+ Instant deletedAt = NOW.minusSeconds(1);
+
+ GenericKubernetesResource succeeded = app("Submitted", "DriverReady",
"Succeeded");
+ succeeded.getMetadata().setDeletionTimestamp(deletedAt.toString());
+ Assertions.assertEquals(JobHandle.Status.SUCCEEDED,
executionInfoOf(succeeded).status());
+
+ GenericKubernetesResource running = app("Submitted", "DriverReady");
+ running.getMetadata().setDeletionTimestamp(deletedAt.toString());
+ JobExecutionInfo info = executionInfoOf(running);
+ Assertions.assertEquals(JobHandle.Status.FAILED, info.status());
+ Assertions.assertEquals(deletedAt, info.finishedAt());
+ }
+
+ @Test
+ public void testIsStopped() {
+ Assertions.assertFalse(SparkApplicationStatusUtils.isStopped(app()));
+
Assertions.assertFalse(SparkApplicationStatusUtils.isStopped(app("Submitted",
"DriverReady")));
+
Assertions.assertTrue(SparkApplicationStatusUtils.isStopped(app("Submitted",
"Succeeded")));
+ Assertions.assertTrue(
+ SparkApplicationStatusUtils.isStopped(app("Submitted",
"DriverStartTimedOut")));
+ Assertions.assertTrue(
+ SparkApplicationStatusUtils.isStopped(app("Submitted",
"ResourceReleased")));
+ }
+
+ @Test
+ public void testDriverStartedIsStarted() {
+ // The driver container runs from DriverStarted on, even if the pod never
gets ready.
+ GenericKubernetesResource app = app("Submitted", "DriverRequested",
"DriverStarted");
+ JobExecutionInfo info = executionInfoOf(app);
+ Assertions.assertEquals(JobHandle.Status.STARTED, info.status());
+ Assertions.assertEquals(timeOf(2), info.startedAt());
+
+ Pod running = driverPod("spark-kubernetes-driver", "2026-09-30T00:00:01Z",
null);
+ JobExecutionInfo withPod =
+ SparkApplicationStatusUtils.executionInfoOf(app, running, NOW,
NO_STATUS_TIMEOUT_MS);
+ Assertions.assertEquals(JobHandle.Status.STARTED, withPod.status());
+ Assertions.assertEquals(Instant.parse("2026-09-30T00:00:01Z"),
withPod.startedAt());
+
+ // Cancelled before DriverReady, the job still has its started time.
+ cancel(app);
+
app.getMetadata().setDeletionTimestamp(CREATED_AT.plusSeconds(10).toString());
+ JobExecutionInfo cancelled = executionInfoOf(app);
+ Assertions.assertEquals(JobHandle.Status.CANCELLED, cancelled.status());
+ Assertions.assertEquals(timeOf(2), cancelled.startedAt());
+ Assertions.assertEquals(CREATED_AT.plusSeconds(10),
cancelled.finishedAt());
+
+ // An unknown state after DriverStarted is a started one.
+ Assertions.assertEquals(
+ JobHandle.Status.STARTED,
+ executionInfoOf(app("Submitted", "DriverStarted",
"SomeNewState")).status());
+ }
+
+ @Test
+ public void testDriverTerminationTimeWithoutOutcome() {
+ // The driver terminated after the deletion was requested, and the
operator recorded no
+ // outcome. The time the driver actually terminated is the finished time.
+ Pod terminated =
+ driverPod("spark-kubernetes-driver", "2026-09-30T00:00:02Z",
"2026-09-30T00:00:25Z");
+
+ GenericKubernetesResource cancelled = app("Submitted", "DriverReady",
"RunningHealthy");
+ cancel(cancelled);
+
cancelled.getMetadata().setDeletionTimestamp(CREATED_AT.plusSeconds(20).toString());
+ JobExecutionInfo info =
+ SparkApplicationStatusUtils.executionInfoOf(
+ cancelled, terminated, NOW, NO_STATUS_TIMEOUT_MS);
+ Assertions.assertEquals(JobHandle.Status.CANCELLED, info.status());
+ Assertions.assertEquals(Instant.parse("2026-09-30T00:00:25Z"),
info.finishedAt());
+
+ // Deleted by someone else: the job failed, a terminated driver doesn't
make it succeed.
+ GenericKubernetesResource deleted = app("Submitted", "DriverReady",
"RunningHealthy");
+
deleted.getMetadata().setDeletionTimestamp(CREATED_AT.plusSeconds(20).toString());
+ info =
+ SparkApplicationStatusUtils.executionInfoOf(deleted, terminated, NOW,
NO_STATUS_TIMEOUT_MS);
+ Assertions.assertEquals(JobHandle.Status.FAILED, info.status());
+ Assertions.assertEquals(Instant.parse("2026-09-30T00:00:25Z"),
info.finishedAt());
+
+ // Without the driver pod, the deletion time is the best known one.
+ Assertions.assertEquals(CREATED_AT.plusSeconds(20),
executionInfoOf(deleted).finishedAt());
+
+ // A running job isn't finished because a driver pod terminated.
+ info =
+ SparkApplicationStatusUtils.executionInfoOf(
+ app("Submitted", "DriverReady", "RunningHealthy"),
+ terminated,
+ NOW,
+ NO_STATUS_TIMEOUT_MS);
+ Assertions.assertEquals(JobHandle.Status.STARTED, info.status());
+ Assertions.assertNull(info.finishedAt());
+ }
+
+ @Test
+ public void testUnknownStateAfterOutcome() {
+ // A newer operator may add a state after the outcome, the job must not
stay unfinished.
+ GenericKubernetesResource app = app("Submitted", "DriverReady",
"Succeeded", "SomeNewState");
+ JobExecutionInfo info = executionInfoOf(app);
+ Assertions.assertEquals(JobHandle.Status.SUCCEEDED, info.status());
+ Assertions.assertEquals(timeOf(2), info.finishedAt());
+ Assertions.assertTrue(SparkApplicationStatusUtils.isStopped(app));
+
+ // A known state after an outcome means the application runs again.
+ GenericKubernetesResource restarted =
+ app("Submitted", "DriverReady", "Failed", "ScheduledToRestart",
"RunningHealthy");
+ Assertions.assertEquals(JobHandle.Status.STARTED,
executionInfoOf(restarted).status());
+ Assertions.assertNull(executionInfoOf(restarted).finishedAt());
+ Assertions.assertFalse(SparkApplicationStatusUtils.isStopped(restarted));
+ }
+
+ @Test
+ public void testStoppedWithoutOutcome() {
+ GenericKubernetesResource app = app("Submitted", "DriverReady",
"ResourceReleased");
+ JobExecutionInfo failed = executionInfoOf(app);
+ Assertions.assertEquals(JobHandle.Status.FAILED, failed.status());
+ Assertions.assertEquals(timeOf(2), failed.finishedAt());
+
+ cancel(app);
+ JobExecutionInfo cancelled = executionInfoOf(app);
+ Assertions.assertEquals(JobHandle.Status.CANCELLED, cancelled.status());
+ Assertions.assertEquals(timeOf(2), cancelled.finishedAt());
+
+ GenericKubernetesResource stopped =
+ app("Submitted", "DriverReady", "Succeeded",
"TerminatedWithoutReleaseResources");
+ cancel(stopped);
+ Assertions.assertEquals(JobHandle.Status.CANCELLED,
executionInfoOf(stopped).status());
+ Assertions.assertEquals(timeOf(2), executionInfoOf(stopped).finishedAt());
+ }
+
+ @Test
+ public void testDeletionNotBeforeStart() {
+ // The deletion time is in whole seconds, the started time of the same
second isn't.
+ GenericKubernetesResource app = app("Submitted", "DriverReady");
+ cancel(app);
+
app.getMetadata().setDeletionTimestamp(CREATED_AT.plusSeconds(1).toString());
+
+ JobExecutionInfo info = executionInfoOf(app);
+ Assertions.assertEquals(JobHandle.Status.CANCELLED, info.status());
+ Assertions.assertEquals(timeOf(1), info.startedAt());
+ Assertions.assertEquals(timeOf(1), info.finishedAt());
+ }
+
+ @Test
+ public void testContainerStartBeforeCreationIsIgnored() {
+ // A container that failed to start reports a start time of the epoch.
+ Map<String, Object> neverStarted =
+ ImmutableMap.of(
+ "terminated",
+ ImmutableMap.of(
+ "startedAt", "1970-01-01T00:00:00Z", "finishedAt",
"2026-09-30T00:00:02Z"));
+
+ GenericKubernetesResource app = app("Submitted", "DriverRequested",
"Failed");
+ Map<String, Map<String, Object>> history = history(app);
+ history.put("2", withDriverStatus(history.get("2"), neverStarted));
+ JobExecutionInfo info = executionInfoOf(app);
+ Assertions.assertEquals(JobHandle.Status.FAILED, info.status());
+ Assertions.assertNull(info.startedAt());
+ Assertions.assertEquals(Instant.parse("2026-09-30T00:00:02Z"),
info.finishedAt());
+
+ // The operator saw it started, so its time is used instead.
+ GenericKubernetesResource started = app("Submitted", "DriverReady",
"Failed");
+ history = history(started);
+ history.put("2", withDriverStatus(history.get("2"), neverStarted));
+ Assertions.assertEquals(timeOf(1), executionInfoOf(started).startedAt());
+
+ Pod driver = driverPod("spark-kubernetes-driver", "1970-01-01T00:00:00Z",
null);
+ Assertions.assertEquals(
+ timeOf(1),
+ SparkApplicationStatusUtils.executionInfoOf(
+ app("Submitted", "DriverReady", "Succeeded"), driver, NOW,
NO_STATUS_TIMEOUT_MS)
+ .startedAt());
+ }
+
+ @Test
+ public void testDriverPodStates() {
+ GenericKubernetesResource app = app("Submitted", "DriverReady", "Failed");
+ Map<String, Map<String, Object>> history = history(app);
+ history.put(
+ "2",
+ withDriverStatus(
+ history.get("2"),
+ ImmutableMap.of(
+ "terminated",
+ ImmutableMap.of(
+ "startedAt", "2026-09-30T00:00:10Z", "finishedAt",
"2026-09-30T00:00:20Z"))));
+
+ // The driver pod wins over the driver status that the operator observed.
+ JobExecutionInfo terminated =
+ SparkApplicationStatusUtils.executionInfoOf(
+ app,
+ driverPod("spark-kubernetes-driver", "2026-09-30T00:00:11Z",
"2026-09-30T00:00:21Z"),
+ NOW,
+ NO_STATUS_TIMEOUT_MS);
+ Assertions.assertEquals(Instant.parse("2026-09-30T00:00:11Z"),
terminated.startedAt());
+ Assertions.assertEquals(Instant.parse("2026-09-30T00:00:21Z"),
terminated.finishedAt());
+
+ // A running driver has a started time only, the finished time is the
observed one.
+ JobExecutionInfo running =
+ SparkApplicationStatusUtils.executionInfoOf(
+ app,
+ driverPod("spark-kubernetes-driver", "2026-09-30T00:00:12Z", null),
+ NOW,
+ NO_STATUS_TIMEOUT_MS);
+ Assertions.assertEquals(Instant.parse("2026-09-30T00:00:12Z"),
running.startedAt());
+ Assertions.assertEquals(Instant.parse("2026-09-30T00:00:20Z"),
running.finishedAt());
+
+ // The only container is the driver, whatever its name.
+ JobExecutionInfo renamed =
+ SparkApplicationStatusUtils.executionInfoOf(
+ app,
+ driverPod("main", "2026-09-30T00:00:13Z", "2026-09-30T00:00:23Z"),
+ NOW,
+ NO_STATUS_TIMEOUT_MS);
+ Assertions.assertEquals(Instant.parse("2026-09-30T00:00:23Z"),
renamed.finishedAt());
+ }
+
+ @Test
+ public void testDriverContainerNamedByJob() {
+ // With a sidecar, the driver container is the one the job names for its
pod template.
+ GenericKubernetesResource app = app("Submitted", "DriverReady", "Failed");
+ app.setAdditionalProperty(
+ "spec",
+ ImmutableMap.of(
+ "sparkConf",
+ ImmutableMap.of(SparkApplicationUtils.SPARK_DRIVER_CONTAINER_NAME,
"main")));
+ Map<String, Object> driver =
+ ImmutableMap.of(
+ "terminated",
+ ImmutableMap.of(
+ "startedAt", "2026-09-30T00:00:10Z", "finishedAt",
"2026-09-30T00:00:20Z"));
+ Map<String, Object> sidecar =
+ ImmutableMap.of(
+ "terminated",
+ ImmutableMap.of(
+ "startedAt", "2026-09-30T00:00:05Z", "finishedAt",
"2026-09-30T00:00:30Z"));
+ Map<String, Map<String, Object>> history = history(app);
+ history.put(
+ "2",
+ ImmutableMap.<String, Object>builder()
+ .putAll(history.get("2"))
+ .put(
+ "lastObservedDriverStatus",
+ ImmutableMap.of(
+ "containerStatuses",
+ ImmutableList.of(
+ ImmutableMap.of("name", "sidecar", "state", sidecar),
+ ImmutableMap.of("name", "main", "state", driver))))
+ .build());
+
+ JobExecutionInfo info = executionInfoOf(app);
+ Assertions.assertEquals(Instant.parse("2026-09-30T00:00:10Z"),
info.startedAt());
+ Assertions.assertEquals(Instant.parse("2026-09-30T00:00:20Z"),
info.finishedAt());
+
+ Pod pod =
+ new PodBuilder(driverPod("sidecar", "2026-09-30T00:00:05Z",
"2026-09-30T00:00:30Z"))
+ .editStatus()
+ .addNewContainerStatus()
+ .withName("main")
+ .withNewState()
+ .withNewTerminated()
+ .withStartedAt("2026-09-30T00:00:11Z")
+ .withFinishedAt("2026-09-30T00:00:21Z")
+ .endTerminated()
+ .endState()
+ .endContainerStatus()
+ .endStatus()
+ .build();
+ Assertions.assertEquals(
+ Instant.parse("2026-09-30T00:00:21Z"),
+ SparkApplicationStatusUtils.executionInfoOf(app, pod, NOW,
NO_STATUS_TIMEOUT_MS)
+ .finishedAt());
+ }
+
+ @Test
+ public void testHasState() {
+ Assertions.assertFalse(SparkApplicationStatusUtils.hasState(app()));
+
Assertions.assertTrue(SparkApplicationStatusUtils.hasState(app("Submitted")));
+
+ GenericKubernetesResource unreadable = app();
+ unreadable.setAdditionalProperty("status", "not an object");
+ Assertions.assertFalse(SparkApplicationStatusUtils.hasState(unreadable));
+ unreadable.setAdditionalProperty(
+ "status", ImmutableMap.of("currentState", ImmutableMap.of("message",
"no summary")));
+ Assertions.assertFalse(SparkApplicationStatusUtils.hasState(unreadable));
+ }
+
+ @Test
+ public void testCurrentStateWithoutHistory() {
+ GenericKubernetesResource app = app();
+ app.setAdditionalProperty(
+ "status", ImmutableMap.of("currentState", state("RunningHealthy", 3)));
+
+ Assertions.assertTrue(SparkApplicationStatusUtils.hasState(app));
+ JobExecutionInfo info = executionInfoOf(app);
+ Assertions.assertEquals(JobHandle.Status.STARTED, info.status());
+ Assertions.assertEquals(timeOf(3), info.startedAt());
+ }
+
+ @Test
+ public void testMalformedStatus() {
+ GenericKubernetesResource app = app("Submitted", "DriverReady",
"Succeeded");
+ Map<String, Map<String, Object>> history = history(app);
+ // Entries that can't be read are skipped, and a time that can't be parsed
is unknown.
+ history.put("not-a-number", state("Failed", 9));
+ history.put("7", ImmutableMap.of("message", "no summary"));
+ history.put(
+ "2", ImmutableMap.of("currentStateSummary", "Succeeded",
"lastTransitionTime", "soon"));
+
+ JobExecutionInfo info = executionInfoOf(app);
+ Assertions.assertEquals(JobHandle.Status.SUCCEEDED, info.status());
+ Assertions.assertEquals(timeOf(1), info.startedAt());
+ Assertions.assertNull(info.finishedAt());
+
+ // Without a creation time, an application without status never times out.
+ GenericKubernetesResource noCreationTime = app();
+ noCreationTime.getMetadata().setCreationTimestamp(null);
+ Assertions.assertEquals(
+ JobHandle.Status.QUEUED,
+ SparkApplicationStatusUtils.executionInfoOf(
+ noCreationTime, null, NOW.plusSeconds(86_400),
NO_STATUS_TIMEOUT_MS)
+ .status());
+ }
+
+ @Test
+ public void testStatusReportedByOperator() {
+ // The status of a failed application as operator 0.9.0 reports it, read
the way the client
+ // does: the history is keyed by numbers in strings, the times have
nanoseconds, and only the
+ // Failed state carries the driver status.
+ String json =
+
"{\"apiVersion\":\"spark.apache.org/v1\",\"kind\":\"SparkApplication\","
+ +
"\"metadata\":{\"name\":\"gravitino-job-1\",\"namespace\":\"default\","
+ + "\"creationTimestamp\":\"2026-09-24T09:22:32Z\"},"
+ +
"\"status\":{\"currentState\":{\"currentStateSummary\":\"ResourceReleased\","
+ + "\"lastTransitionTime\":\"2026-09-27T01:02:22.906216807Z\"},"
+ + "\"stateTransitionHistory\":{"
+ + "\"10\":{\"currentStateSummary\":\"ResourceReleased\","
+ + "\"lastTransitionTime\":\"2026-09-27T01:02:22.906216807Z\"},"
+ + "\"0\":{\"currentStateSummary\":\"Submitted\","
+ + "\"lastTransitionTime\":\"2026-09-24T09:22:32.207220731Z\"},"
+ + "\"1\":{\"currentStateSummary\":\"DriverRequested\","
+ + "\"lastTransitionTime\":\"2026-09-24T09:22:32.436928096Z\"},"
+ + "\"2\":{\"currentStateSummary\":\"DriverStarted\","
+ + "\"lastTransitionTime\":\"2026-09-24T09:22:33.099430159Z\"},"
+ + "\"3\":{\"currentStateSummary\":\"DriverReady\","
+ + "\"lastTransitionTime\":\"2026-09-24T09:22:33.099616659Z\"},"
+ + "\"4\":{\"currentStateSummary\":\"RunningHealthy\","
+ + "\"lastTransitionTime\":\"2026-09-24T09:22:33.111225423Z\"},"
+ + "\"5\":{\"currentStateSummary\":\"Failed\","
+ + "\"lastTransitionTime\":\"2026-09-24T09:22:57.002342477Z\","
+ +
"\"lastObservedDriverStatus\":{\"phase\":\"Failed\",\"containerStatuses\":["
+ +
"{\"name\":\"spark-kubernetes-driver\",\"state\":{\"terminated\":{\"exitCode\":101,"
+ +
"\"startedAt\":\"2026-09-24T09:22:32Z\",\"finishedAt\":\"2026-09-24T09:22:33Z\"}}}]}},"
+ +
"\"6\":{\"currentStateSummary\":\"TerminatedWithoutReleaseResources\","
+ + "\"lastTransitionTime\":\"2026-09-24T09:22:57.047040866Z\"}}}}";
+ GenericKubernetesResource app = Serialization.unmarshal(json,
GenericKubernetesResource.class);
+
+ Assertions.assertTrue(SparkApplicationStatusUtils.hasState(app));
+ Assertions.assertTrue(SparkApplicationStatusUtils.isStopped(app));
+ Assertions.assertEquals(
+ JobExecutionInfo.builder()
+ .withStatus(JobHandle.Status.FAILED)
+ .withStartedAt(Instant.parse("2026-09-24T09:22:32Z"))
+ .withFinishedAt(Instant.parse("2026-09-24T09:22:33Z"))
+ .build(),
+ executionInfoOf(app));
+ }
+
+ private static JobExecutionInfo executionInfoOf(GenericKubernetesResource
app) {
+ return SparkApplicationStatusUtils.executionInfoOf(app, null, NOW,
NO_STATUS_TIMEOUT_MS);
+ }
+
+ /** Creates a SparkApplication that went through the given states, one
second apart. */
+ private static GenericKubernetesResource app(String... states) {
+ GenericKubernetesResource app =
+ new GenericKubernetesResourceBuilder()
+ .withNewMetadata()
+ .withNamespace("default")
+ .withName("gravitino-job-1")
+ .withCreationTimestamp(CREATED_AT.toString())
+ .endMetadata()
+ .build();
+ if (states.length == 0) {
+ return app;
+ }
+
+ Map<String, Object> history = new LinkedHashMap<>();
+ for (int i = 0; i < states.length; i++) {
+ history.put(String.valueOf(i), state(states[i], i));
+ }
+ Map<String, Object> status = new LinkedHashMap<>();
+ status.put("currentState", state(states[states.length - 1], states.length
- 1));
+ status.put("stateTransitionHistory", history);
+ app.setAdditionalProperty("status", status);
+ return app;
+ }
+
+ @SuppressWarnings("unchecked")
+ private static Map<String, Map<String, Object>>
history(GenericKubernetesResource app) {
+ return (Map<String, Map<String, Object>>)
status(app).get("stateTransitionHistory");
+ }
+
+ private static Map<String, Object> withDriverStatus(
+ Map<String, Object> state, Map<String, Object> driverState) {
+ return ImmutableMap.<String, Object>builder()
+ .putAll(state)
+ .put(
+ "lastObservedDriverStatus",
+ ImmutableMap.of(
+ "containerStatuses",
+ ImmutableList.of(
+ ImmutableMap.of("name", "spark-kubernetes-driver",
"state", driverState))))
+ .build();
+ }
+
+ @SuppressWarnings("unchecked")
+ private static Map<String, Object> status(GenericKubernetesResource app) {
+ return (Map<String, Object>) app.getAdditionalProperties().get("status");
+ }
+
+ private static Map<String, Object> state(String summary, int index) {
+ return ImmutableMap.of(
+ "currentStateSummary", summary, "lastTransitionTime",
timeOf(index).toString());
+ }
+
+ private static Instant timeOf(int index) {
+ return CREATED_AT.plusSeconds(index).plusNanos(123_456_789L);
+ }
+
+ private static Pod driverPod(String container, String startedAt, String
finishedAt) {
+ PodBuilder pod = new PodBuilder().withNewStatus().endStatus();
+ if (finishedAt == null) {
+ return pod.editStatus()
+ .addNewContainerStatus()
+ .withName(container)
+ .withNewState()
+ .withNewRunning()
+ .withStartedAt(startedAt)
+ .endRunning()
+ .endState()
+ .endContainerStatus()
+ .endStatus()
+ .build();
+ }
+ return pod.editStatus()
+ .addNewContainerStatus()
+ .withName(container)
+ .withNewState()
+ .withNewTerminated()
+ .withStartedAt(startedAt)
+ .withFinishedAt(finishedAt)
+ .endTerminated()
+ .endState()
+ .endContainerStatus()
+ .endStatus()
+ .build();
+ }
+
+ private static void cancel(GenericKubernetesResource app) {
+ app.getMetadata()
+
.setAnnotations(ImmutableMap.of(K8sJobResourceUtils.ANNOTATION_CANCEL_REQUESTED,
"true"));
+ }
+}
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
new file mode 100644
index 0000000000..71e3abb38b
--- /dev/null
+++
b/plugins/k8s-job-executor/src/test/java/org/apache/gravitino/job/k8s/spark/TestSparkApplicationUtils.java
@@ -0,0 +1,471 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.gravitino.job.k8s.spark;
+
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableMap;
+import io.fabric8.kubernetes.api.model.GenericKubernetesResource;
+import java.io.File;
+import java.util.HashMap;
+import java.util.Map;
+import org.apache.gravitino.connector.job.JobContext;
+import org.apache.gravitino.job.SparkJobTemplate;
+import org.apache.gravitino.job.k8s.K8sJobExecutorConfigs;
+import org.apache.gravitino.job.k8s.K8sJobResourceUtils;
+import org.apache.gravitino.job.k8s.TestK8sJobExecutorConfigs;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+public class TestSparkApplicationUtils {
+
+ private static final JobContext CONTEXT = new JobContext(42L, "ml", new
File("/tmp/job-42"));
+
+ private final K8sJobExecutorConfigs configs = newConfigs(ImmutableMap.of());
+
+ @Test
+ @SuppressWarnings("unchecked")
+ public void testBuildSparkApplication() {
+ SparkJobTemplate template =
+ SparkJobTemplate.builder()
+ .withName("user-job")
+ .withExecutable("s3a://bucket/app.jar")
+ .withClassName("org.example.App")
+ .withArguments(ImmutableList.of("--date", "2026-09-30"))
+ .withJars(ImmutableList.of("https://repo/lib.jar";))
+ .withFiles(ImmutableList.of("local:///opt/conf.properties"))
+ .withArchives(ImmutableList.of("hdfs://nn/env.tgz#env"))
+ .withConfigs(
+ ImmutableMap.of(
+ "spark.master", "local[*]",
+ "spark.submit.deployMode", "client",
+ "spark.executor.instances", "2",
+ "spark.jars", "https://repo/other.jar";))
+ .build();
+
+ GenericKubernetesResource app =
+ SparkApplicationUtils.buildSparkApplication(CONTEXT, template,
configs);
+
+ Assertions.assertEquals("spark.apache.org/v1", app.getApiVersion());
+ Assertions.assertEquals("SparkApplication", app.getKind());
+ Assertions.assertEquals("gravitino-job-42", app.getMetadata().getName());
+ Assertions.assertEquals("default", app.getMetadata().getNamespace());
+ Assertions.assertEquals(
+ ImmutableMap.of(
+ "app.kubernetes.io/managed-by", "gravitino",
"gravitino.apache.org/job-id", "42"),
+ app.getMetadata().getLabels());
+ Assertions.assertEquals(
+ "ml",
app.getMetadata().getAnnotations().get(K8sJobResourceUtils.ANNOTATION_METALAKE));
+
+ Map<String, Object> spec = (Map<String, Object>)
app.getAdditionalProperties().get("spec");
+ Assertions.assertEquals("org.example.App", spec.get("mainClass"));
+ Assertions.assertEquals("s3a://bucket/app.jar", spec.get("jars"));
+ Assertions.assertFalse(spec.containsKey("pyFiles"));
+ Assertions.assertEquals(ImmutableList.of("--date", "2026-09-30"),
spec.get("driverArgs"));
+ Assertions.assertEquals(ImmutableMap.of("sparkVersion", "3.5.9"),
spec.get("runtimeVersions"));
+
+ Map<String, String> sparkConf = (Map<String, String>)
spec.get("sparkConf");
+ Assertions.assertEquals(
+ ImmutableMap.<String, String>builder()
+ .put("spark.archives", "hdfs://nn/env.tgz#env")
+ .put("spark.driver.memory", "2g")
+ .put("spark.executor.instances", "2")
+ .put("spark.files", "local:///opt/conf.properties")
+ .put("spark.jars", "https://repo/other.jar,https://repo/lib.jar";)
+ .put("spark.kubernetes.authenticate.driver.serviceAccountName",
"spark")
+ .put("spark.kubernetes.container.image", "gravitino-spark:3.5.9")
+ .put("spark.kubernetes.namespace", "default")
+ .build(),
+ sparkConf);
+
+ Map<String, Object> tolerations = (Map<String, Object>)
spec.get("applicationTolerations");
+ Assertions.assertEquals(
+ ImmutableMap.of("restartPolicy", "Never"),
tolerations.get("restartConfig"));
+ Assertions.assertEquals("Always", tolerations.get("resourceRetainPolicy"));
+ Assertions.assertEquals(86_400_000L,
tolerations.get("resourceRetainDurationMillis"));
+ Assertions.assertEquals(604_800_000L,
tolerations.get("ttlAfterStopMillis"));
+ Assertions.assertEquals(
+ ImmutableMap.of(
+ "driverStartTimeoutMillis", 3_600_000L,
+ "driverReadyTimeoutMillis", 3_600_000L),
+ tolerations.get("applicationTimeoutConfig"));
+ }
+
+ @Test
+ @SuppressWarnings("unchecked")
+ public void testPythonApplication() {
+ SparkJobTemplate template = newTemplate("https://repo/job.py";,
ImmutableMap.of());
+
+ Map<String, Object> spec =
+ (Map<String, Object>)
+ SparkApplicationUtils.buildSparkApplication(CONTEXT, template,
configs)
+ .getAdditionalProperties()
+ .get("spec");
+
+ Assertions.assertEquals("https://repo/job.py";, spec.get("pyFiles"));
+ Assertions.assertFalse(spec.containsKey("jars"));
+ }
+
+ @Test
+ @SuppressWarnings("unchecked")
+ public void testPythonApplicationWithQuery() {
+ for (String uri : new String[] {"https://repo/job.PY?sig=abc";,
"s3a://bucket/job.py#main"}) {
+ Map<String, Object> spec =
+ (Map<String, Object>)
+ SparkApplicationUtils.buildSparkApplication(
+ CONTEXT, newTemplate(uri, ImmutableMap.of()), configs)
+ .getAdditionalProperties()
+ .get("spec");
+ Assertions.assertEquals(uri, spec.get("pyFiles"));
+ }
+
+ Map<String, Object> jar =
+ (Map<String, Object>)
+ SparkApplicationUtils.buildSparkApplication(
+ CONTEXT,
+ newTemplate("https://repo/app.jar?name=job.py";,
ImmutableMap.of()),
+ configs)
+ .getAdditionalProperties()
+ .get("spec");
+ Assertions.assertTrue(jar.containsKey("jars"));
+ }
+
+ @Test
+ @SuppressWarnings("unchecked")
+ public void testPythonApplicationWithoutMainClass() {
+ for (String className : new String[] {null, " "}) {
+ SparkJobTemplate template =
+ SparkJobTemplate.builder()
+ .withName("user-job")
+ .withExecutable("https://repo/job.py";)
+ .withClassName(className)
+ .build();
+ Map<String, Object> spec =
+ (Map<String, Object>)
+ SparkApplicationUtils.buildSparkApplication(CONTEXT, template,
configs)
+ .getAdditionalProperties()
+ .get("spec");
+ Assertions.assertEquals("https://repo/job.py";, spec.get("pyFiles"));
+ Assertions.assertFalse(spec.containsKey("mainClass"));
+ }
+ }
+
+ @Test
+ @SuppressWarnings("unchecked")
+ public void testBlankAndExistingConfigs() {
+ SparkJobTemplate template =
+ newTemplateBuilder()
+ .withFiles(ImmutableList.of("https://repo/b.conf";))
+ .withArchives(ImmutableList.of("https://repo/b.tgz";))
+ .withConfigs(
+ ImmutableMap.of(
+ // Blank values don't override the executor configurations.
+ "spark.kubernetes.container.image", " ",
+ "spark.kubernetes.authenticate.driver.serviceAccountName",
"",
+ "spark.kubernetes.namespace", " ",
+ "spark.files", "https://repo/a.conf";,
+ "spark.archives", "https://repo/a.tgz";))
+ .build();
+
+ GenericKubernetesResource app =
+ SparkApplicationUtils.buildSparkApplication(CONTEXT, template,
configs);
+
+ Assertions.assertEquals("default", app.getMetadata().getNamespace());
+ Map<String, String> sparkConf =
+ (Map<String, String>)
+ ((Map<String, Object>)
app.getAdditionalProperties().get("spec")).get("sparkConf");
+ Assertions.assertEquals(
+ "gravitino-spark:3.5.9",
sparkConf.get("spark.kubernetes.container.image"));
+ Assertions.assertEquals(
+ "spark",
sparkConf.get("spark.kubernetes.authenticate.driver.serviceAccountName"));
+ Assertions.assertEquals("default",
sparkConf.get("spark.kubernetes.namespace"));
+ Assertions.assertEquals(
+ "https://repo/a.conf,https://repo/b.conf";,
sparkConf.get("spark.files"));
+ Assertions.assertEquals(
+ "https://repo/a.tgz,https://repo/b.tgz";,
sparkConf.get("spark.archives"));
+ }
+
+ @Test
+ @SuppressWarnings("unchecked")
+ public void testDefaultSparkConfigsDoNotChooseNamespace() {
+ // Only the namespace configuration of the executor, or the job itself,
chooses the namespace.
+ K8sJobExecutorConfigs withDefault =
+ newConfigs(
+ ImmutableMap.of(
+ K8sJobExecutorConfigs.NAMESPACE,
+ "jobs",
+ "spark.conf.spark.kubernetes.namespace",
+ "other"));
+
+ GenericKubernetesResource app =
+ SparkApplicationUtils.buildSparkApplication(
+ CONTEXT, newTemplate("https://repo/app.jar";, ImmutableMap.of()),
withDefault);
+
+ Assertions.assertEquals("jobs", app.getMetadata().getNamespace());
+ Map<String, String> sparkConf =
+ (Map<String, String>)
+ ((Map<String, Object>)
app.getAdditionalProperties().get("spec")).get("sparkConf");
+ Assertions.assertEquals("jobs",
sparkConf.get("spark.kubernetes.namespace"));
+ }
+
+ @Test
+ public void testNamespaceOfJob() {
+ Assertions.assertEquals(
+ "team-a",
+ SparkApplicationUtils.buildSparkApplication(
+ CONTEXT,
+ newTemplate(
+ "https://repo/app.jar";,
+ ImmutableMap.of("spark.kubernetes.namespace", " team-a ")),
+ configs)
+ .getMetadata()
+ .getNamespace());
+
+ for (String invalid : new String[] {"Team_A", "team/a", "-team"}) {
+ SparkJobTemplate template =
+ newTemplate(
+ "https://repo/app.jar";,
ImmutableMap.of("spark.kubernetes.namespace", invalid));
+ Assertions.assertThrows(
+ IllegalArgumentException.class,
+ () -> SparkApplicationUtils.buildSparkApplication(CONTEXT, template,
configs));
+ }
+ }
+
+ @Test
+ @SuppressWarnings("unchecked")
+ public void testJobConfigsOverrideDefaults() {
+ K8sJobExecutorConfigs withDefaults =
+ newConfigs(
+ ImmutableMap.of(
+ K8sJobExecutorConfigs.NAMESPACE,
+ "jobs",
+ "spark.conf.spark.executor.instances",
+ "1",
+ "spark.conf.spark.master",
+ "k8s://ignored"));
+ SparkJobTemplate template =
+ newTemplate(
+ "https://repo/app.jar";,
+ ImmutableMap.of(
+ "spark.executor.instances", "4",
+ "spark.kubernetes.namespace", "team-a",
+ "spark.kubernetes.container.image", "custom:1.0",
+ "spark.kubernetes.authenticate.driver.serviceAccountName",
"team-a-spark"));
+
+ GenericKubernetesResource app =
+ SparkApplicationUtils.buildSparkApplication(CONTEXT, template,
withDefaults);
+
+ Assertions.assertEquals("team-a", app.getMetadata().getNamespace());
+ Map<String, String> sparkConf =
+ (Map<String, String>)
+ ((Map<String, Object>)
app.getAdditionalProperties().get("spec")).get("sparkConf");
+ Assertions.assertEquals("4", sparkConf.get("spark.executor.instances"));
+ Assertions.assertEquals("custom:1.0",
sparkConf.get("spark.kubernetes.container.image"));
+ Assertions.assertEquals(
+ "team-a-spark",
sparkConf.get("spark.kubernetes.authenticate.driver.serviceAccountName"));
+ Assertions.assertFalse(sparkConf.containsKey("spark.master"));
+ }
+
+ @Test
+ public void testDefaultNamespace() {
+ K8sJobExecutorConfigs withNamespace =
+ newConfigs(ImmutableMap.of(K8sJobExecutorConfigs.NAMESPACE, "jobs"));
+ GenericKubernetesResource app =
+ SparkApplicationUtils.buildSparkApplication(
+ CONTEXT, newTemplate("https://repo/app.jar";, ImmutableMap.of()),
withNamespace);
+ Assertions.assertEquals("jobs", app.getMetadata().getNamespace());
+ }
+
+ @Test
+ public void testRejectServerLocalResources() {
+ String[] serverPaths = {
+ "/opt/app.jar",
+ "app.jar",
+ "file:///opt/app.jar",
+ "FILE:/opt/app.jar",
+ "C:/apps/app.jar",
+ // Spark takes several resources in one entry, each of them is checked.
+ "https://repo/a.jar,/opt/app.jar";,
+ };
+ for (String uri : serverPaths) {
+ Assertions.assertThrows(
+ IllegalArgumentException.class,
+ () ->
+ SparkApplicationUtils.buildSparkApplication(
+ CONTEXT, newTemplate(uri, ImmutableMap.of()), configs),
+ uri);
+ for (SparkJobTemplate template :
+ ImmutableList.of(
+ newTemplateBuilder().withJars(ImmutableList.of(uri)).build(),
+ newTemplateBuilder().withFiles(ImmutableList.of(uri)).build(),
+
newTemplateBuilder().withArchives(ImmutableList.of(uri)).build())) {
+ Assertions.assertThrows(
+ IllegalArgumentException.class,
+ () -> SparkApplicationUtils.buildSparkApplication(CONTEXT,
template, configs),
+ uri);
+ }
+ }
+ }
+
+ @Test
+ @SuppressWarnings("unchecked")
+ public void testAcceptClusterReachableResources() {
+ // Whatever has a scheme other than file is left to Spark, even if it
isn't a strict URI.
+ String[] uris = {
+ "local:///opt/app.jar",
+ "hdfs:/apps/app.jar",
+ "s3a://bucket/my app.jar",
+ "https://repo/a.jar, https://repo/b.jar";,
+ };
+ for (String uri : uris) {
+ Map<String, Object> spec =
+ (Map<String, Object>)
+ SparkApplicationUtils.buildSparkApplication(
+ CONTEXT, newTemplate(uri, ImmutableMap.of()), configs)
+ .getAdditionalProperties()
+ .get("spec");
+ Assertions.assertEquals(uri, spec.get("jars"));
+ }
+ }
+
+ @Test
+ @SuppressWarnings("unchecked")
+ public void testResourcesInSparkConfigs() {
+ String[] keys = {"spark.jars", "spark.files", "spark.archives"};
+ String[] serverPaths = {
+ "/opt/local.jar", "file:///opt/local.jar",
"https://repo/a.jar,/opt/local.jar";
+ };
+ for (String key : keys) {
+ for (String value : serverPaths) {
+ // Set by the job.
+ SparkJobTemplate template =
+ newTemplate("https://repo/app.jar";, ImmutableMap.of(key, value));
+ Assertions.assertThrows(
+ IllegalArgumentException.class,
+ () -> SparkApplicationUtils.buildSparkApplication(CONTEXT,
template, configs),
+ key + "=" + value);
+
+ // Set by the default Spark configurations of the executor.
+ K8sJobExecutorConfigs withDefault =
newConfigs(ImmutableMap.of("spark.conf." + key, value));
+ Assertions.assertThrows(
+ IllegalArgumentException.class,
+ () ->
+ SparkApplicationUtils.buildSparkApplication(
+ CONTEXT, newTemplate("https://repo/app.jar";,
ImmutableMap.of()), withDefault),
+ key + "=" + value);
+ }
+
+ // Resources that the cluster can reach are kept, from both sources. The
job's value
+ // overrides the default one, and an empty element is ignored like Spark
does.
+ K8sJobExecutorConfigs withDefault =
+ newConfigs(ImmutableMap.of("spark.conf." + key,
"local:///opt/default.jar"));
+ Map<String, String> defaults =
+ (Map<String, String>)
+ ((Map<String, Object>)
+ SparkApplicationUtils.buildSparkApplication(
+ CONTEXT,
+ newTemplate("https://repo/app.jar";,
ImmutableMap.of()),
+ withDefault)
+ .getAdditionalProperties()
+ .get("spec"))
+ .get("sparkConf");
+ Assertions.assertEquals("local:///opt/default.jar", defaults.get(key));
+
+ String value = "https://repo/a.jar,, s3a://bucket/b.jar";
+ Map<String, String> overridden =
+ (Map<String, String>)
+ ((Map<String, Object>)
+ SparkApplicationUtils.buildSparkApplication(
+ CONTEXT,
+ newTemplate("https://repo/app.jar";,
ImmutableMap.of(key, value)),
+ withDefault)
+ .getAdditionalProperties()
+ .get("spec"))
+ .get("sparkConf");
+ Assertions.assertEquals(value, overridden.get(key));
+ }
+ }
+
+ @Test
+ public void testServerPathInSparkConfigsDoesNotLeakCredentials() {
+ SparkJobTemplate template =
+ newTemplate(
+ "https://repo/app.jar";,
+ ImmutableMap.of("spark.files",
"https://repo/a.conf,file://u:secret@h/b.conf?sig=abc";));
+ IllegalArgumentException e =
+ Assertions.assertThrows(
+ IllegalArgumentException.class,
+ () -> SparkApplicationUtils.buildSparkApplication(CONTEXT,
template, configs));
+ Assertions.assertTrue(e.getMessage().contains("file://h/b.conf "),
e.getMessage());
+ Assertions.assertFalse(e.getMessage().contains("secret"), e.getMessage());
+ Assertions.assertFalse(e.getMessage().contains("abc"), e.getMessage());
+ }
+
+ @Test
+ public void testRejectionDoesNotLeakCredentials() {
+ SparkJobTemplate template =
+
newTemplate("file://user:secret@host/opt/app.jar?X-Signature=abc#frag",
ImmutableMap.of());
+ IllegalArgumentException e =
+ Assertions.assertThrows(
+ IllegalArgumentException.class,
+ () -> SparkApplicationUtils.buildSparkApplication(CONTEXT,
template, configs));
+ Assertions.assertTrue(e.getMessage().contains("file://host/opt/app.jar "),
e.getMessage());
+ Assertions.assertFalse(e.getMessage().contains("secret"), e.getMessage());
+ Assertions.assertFalse(e.getMessage().contains("abc"), e.getMessage());
+ }
+
+ @Test
+ public void testRejectEnvironments() {
+ SparkJobTemplate template =
+ SparkJobTemplate.builder()
+ .withName("user-job")
+ .withExecutable("https://repo/app.jar";)
+ .withClassName("org.example.App")
+ .withEnvironments(ImmutableMap.of("TOKEN", "secret"))
+ .build();
+ IllegalArgumentException e =
+ Assertions.assertThrows(
+ IllegalArgumentException.class,
+ () -> SparkApplicationUtils.buildSparkApplication(CONTEXT,
template, configs));
+ Assertions.assertTrue(e.getMessage().contains("TOKEN"));
+ Assertions.assertFalse(e.getMessage().contains("secret"));
+ }
+
+ public static K8sJobExecutorConfigs newConfigs(Map<String, String> extra) {
+ Map<String, String> map = new
HashMap<>(TestK8sJobExecutorConfigs.requiredConfigs());
+ map.put("spark.conf.spark.driver.memory", "2g");
+ map.putAll(extra);
+ return new K8sJobExecutorConfigs(map);
+ }
+
+ private static SparkJobTemplate.Builder newTemplateBuilder() {
+ return SparkJobTemplate.builder()
+ .withName("user-job")
+ .withExecutable("https://repo/app.jar";)
+ .withClassName("org.example.App");
+ }
+
+ private static SparkJobTemplate newTemplate(String executable, Map<String,
String> configs) {
+ return SparkJobTemplate.builder()
+ .withName("user-job")
+ .withExecutable(executable)
+ .withClassName("org.example.App")
+ .withConfigs(configs)
+ .build();
+ }
+}
