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 361b1e6565 [#13613] feat(k8s-job-executor): Add the k8s job executor
to run Spark jobs on Kubernetes (part 3/3) (#13700)
361b1e6565 is described below
commit 361b1e6565407b1f3b3202e8ba5dfe04f0a5e568
Author: Jerry Shao <[email protected]>
AuthorDate: Fri Oct 16 14:31:08 2026 +0800
[#13613] feat(k8s-job-executor): Add the k8s job executor to run Spark jobs
on Kubernetes (part 3/3) (#13700)
### What changes were proposed in this pull request?
This is part 3 of 3 of #13613. Part 1 (#13724) and part 2 (#13725) are
merged. It adds `K8sJobExecutor`, a job executor named `k8s` that runs
SPARK job templates as `SparkApplication` resources of
[apache/spark-kubernetes-operator](https://github.com/apache/spark-kubernetes-operator)
0.9.x. It rejects SHELL templates for now. All the job state lives in
Kubernetes, so any Gravitino server can query and cancel any job.
- **Submit**: creates the SparkApplication.
- On a conflict, an existing SparkApplication is adopted only if it
belongs to the same job.
- A job that the cluster rejects, for example because its namespace
doesn't exist, is rejected with the reason. So is a job that sets
`spark.submit.deployMode=client`, as the operator only runs cluster
mode.
- If the API server doesn't answer, the executor tries, best effort, to
delete the SparkApplication in case it was created, as Gravitino records
no job for a failed submission. If that cleanup can't reach the API
server either, the SparkApplication may be left running (a sweep of such
orphans is a follow-up in #13554).
- **Status**:
- The SparkApplications of a namespace are listed once and cached for
`statusCacheTtlInMs` (`NamespacedResourceCache`). A miss falls back to a
GET.
- A SparkApplication with no `status` field after `noStatusTimeoutInMs`
is deleted before the job fails, so the operator can't run it later. One
whose status has no state this version can read stays unfinished.
- A failed job whose driver container hasn't terminated gets its driver
pod deleted, for the same reason. The job isn't failed while its driver
pod can't be looked up. A driver that exited is kept with its log, even
if a sidecar keeps the pod running.
- The started and finished times of a finished job, or of one whose
SparkApplication is being deleted such as a cancelled one, come from its
driver pod when it is still there. The driver pod is found by the
operator's labels and must be owned by the SparkApplication.
- While the API server can't be reached, the status queries fail right
away for 10 seconds after the first failure, so a round over many jobs
doesn't wait for the connection to time out once per job.
- **Cancel**: the SparkApplication gets a cancel-requested annotation,
written with a resource-version checked update so a job that stops
meanwhile isn't marked, then is deleted with `Foreground` propagation.
The job is `CANCELLED` once it is being deleted or has stopped.
- **Ownership**: the status, cancel and output paths only use a
SparkApplication that is managed by Gravitino and whose job id label
matches its name. Anything else with the name is treated as missing.
- **Cluster**: a job whose execution id names another `cluster` fails
its queries with `IllegalStateException` rather than
`NoSuchJobException`, so it isn't ended while it may still run.
`cluster` must be the same on every server and must not be renamed while
jobs run.
- **Output**: stdout is the tail of the driver pod log (`PodLogUtils`).
Kubernetes merges stderr into it, so stderr is empty.
The executor isn't registered in `JobExecutorFactory` or shipped in the
distribution yet. That is #13616.
### Why are the changes needed?
The only built-in job executor, `local`, runs jobs as child processes of
the Gravitino server, which isn't suitable for production. This is the
first step of #13554.
Fix: #13613
### Does this PR introduce _any_ user-facing change?
Yes, the configurations under `gravitino.jobExecutor.k8s.` added in part
1 take effect once the executor is registered in #13616.
- Connection: `cluster`, `kubeconfig`, `context`, `masterUrl`,
`caCertFile`, `tokenFile`, `connectionTimeoutInMs`,
`requestTimeoutInMs`, `requestRetries`
- Jobs: `namespace` (required), `namePrefix`, `statusCacheTtlInMs`,
`noStatusTimeoutInMs`
- Spark: `spark.image` and `spark.sparkVersion` (both required),
`spark.serviceAccount`, `spark.resourceRetainDurationInMs`,
`spark.ttlAfterStopInMs`, `spark.driverStartTimeoutInMs`,
`spark.driverReadyTimeoutInMs`, `spark.conf.*`
### How was this patch tested?
- Unit tests. The executor tests run against the fabric8 mock server.
- Manually on a local Kubernetes cluster (OrbStack) with operator 0.9.0,
driving the executor directly: a successful job, a failing job, a
cancelled job, a job whose driver can't be scheduled, and reading the
driver output. Re-run after the second review (Kubernetes 1.35, operator
0.9.0): a successful job, a failing job and a cancelled job, with their
times and output, and a cancel of a finished job, which keeps its
outcome.
🤖 Generated with [Claude Code](https://claude.com/claude-code)
---------
Co-authored-by: Claude Opus 5.5 <[email protected]>
---
.../apache/gravitino/job/k8s/K8sJobExecutor.java | 618 ++++++++++
.../gravitino/job/k8s/K8sJobExecutorConfigs.java | 13 +-
.../gravitino/job/k8s/K8sJobResourceUtils.java | 23 +
.../gravitino/job/k8s/NamespacedResourceCache.java | 183 +++
.../org/apache/gravitino/job/k8s/PodLogUtils.java | 121 ++
.../job/k8s/spark/SparkApplicationStatusUtils.java | 105 +-
.../job/k8s/spark/SparkApplicationUtils.java | 11 +
.../gravitino/job/k8s/TestK8sJobExecutor.java | 1195 ++++++++++++++++++++
.../gravitino/job/k8s/TestK8sJobResourceUtils.java | 30 +
.../job/k8s/TestNamespacedResourceCache.java | 248 ++++
.../apache/gravitino/job/k8s/TestPodLogUtils.java | 101 ++
.../k8s/spark/TestSparkApplicationStatusUtils.java | 95 +-
.../job/k8s/spark/TestSparkApplicationUtils.java | 31 +-
13 files changed, 2749 insertions(+), 25 deletions(-)
diff --git
a/plugins/k8s-job-executor/src/main/java/org/apache/gravitino/job/k8s/K8sJobExecutor.java
b/plugins/k8s-job-executor/src/main/java/org/apache/gravitino/job/k8s/K8sJobExecutor.java
new file mode 100644
index 0000000000..1433f860bb
--- /dev/null
+++
b/plugins/k8s-job-executor/src/main/java/org/apache/gravitino/job/k8s/K8sJobExecutor.java
@@ -0,0 +1,618 @@
+/*
+ * 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;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableSet;
+import io.fabric8.kubernetes.api.model.Container;
+import io.fabric8.kubernetes.api.model.DeletionPropagation;
+import io.fabric8.kubernetes.api.model.GenericKubernetesResource;
+import io.fabric8.kubernetes.api.model.GenericKubernetesResourceList;
+import io.fabric8.kubernetes.api.model.OwnerReference;
+import io.fabric8.kubernetes.api.model.Pod;
+import io.fabric8.kubernetes.client.KubernetesClient;
+import io.fabric8.kubernetes.client.KubernetesClientException;
+import io.fabric8.kubernetes.client.dsl.NonNamespaceOperation;
+import io.fabric8.kubernetes.client.dsl.PodResource;
+import io.fabric8.kubernetes.client.dsl.Resource;
+import java.io.IOException;
+import java.io.InputStream;
+import java.net.HttpURLConnection;
+import java.time.Clock;
+import java.util.Comparator;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.function.Predicate;
+import javax.annotation.Nullable;
+import org.apache.gravitino.connector.job.JobContext;
+import org.apache.gravitino.connector.job.JobExecutionInfo;
+import org.apache.gravitino.connector.job.JobExecutor;
+import org.apache.gravitino.exceptions.NoSuchJobException;
+import org.apache.gravitino.job.JobHandle;
+import org.apache.gravitino.job.JobTemplate;
+import org.apache.gravitino.job.SparkJobTemplate;
+import org.apache.gravitino.job.k8s.spark.SparkApplicationStatusUtils;
+import org.apache.gravitino.job.k8s.spark.SparkApplicationUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * A job executor that runs jobs on Kubernetes. Spark jobs run as
SparkApplication resources of <a
+ *
href="https://github.com/apache/spark-kubernetes-operator";>apache/spark-kubernetes-operator</a>
+ * 0.9.x, shell jobs aren't supported yet.
+ *
+ * <p>All the job state lives in Kubernetes, so every Gravitino server can
query and cancel any job.
+ * The job execution id is {@code <cluster>/<kind>/<namespace>/<name>}, see
{@link
+ * K8sJobExecutionId}.
+ */
+public class K8sJobExecutor implements JobExecutor {
+
+ private static final Logger LOG =
LoggerFactory.getLogger(K8sJobExecutor.class);
+
+ // Not defined by HttpURLConnection.
+ private static final int HTTP_UNPROCESSABLE_ENTITY = 422;
+
+ // The responses of the API server that reject a resource as it is, so
creating it again can't
+ // succeed: bad request, forbidden, not found (the namespace or the
SparkApplication CRD), and
+ // unprocessable entity.
+ private static final Set<Integer> REJECTED_REQUEST_CODES =
+ ImmutableSet.of(
+ HttpURLConnection.HTTP_BAD_REQUEST,
+ HttpURLConnection.HTTP_FORBIDDEN,
+ HttpURLConnection.HTTP_NOT_FOUND,
+ HTTP_UNPROCESSABLE_ENTITY);
+
+ // How long the status queries fail right away after the API server couldn't
be reached.
+ // Gravitino queries the jobs one after another, and each query would
otherwise wait for the
+ // connection to time out again.
+ @VisibleForTesting static final long UNREACHABLE_BACKOFF_MS = 10_000L;
+
+ // How many times a change of a SparkApplication is tried again after it
changed meanwhile.
+ private static final int UPDATE_CONFLICT_RETRIES = 3;
+
+ private final Clock clock;
+
+ // The failure of the last status query that couldn't reach the API server,
and when it failed.
+ // Reset once the backoff is over.
+ @Nullable private volatile KubernetesClientException unreachableFailure;
+
+ private volatile long unreachableAtMs;
+
+ private K8sJobExecutorConfigs configs;
+
+ private KubernetesClient client;
+
+ private NamespacedResourceCache<GenericKubernetesResource>
sparkApplicationCache;
+
+ /** Creates the job executor, which Gravitino initializes with {@link
#initialize(Map)}. */
+ public K8sJobExecutor() {
+ this(Clock.systemUTC());
+ }
+
+ @VisibleForTesting
+ K8sJobExecutor(Clock clock) {
+ this.clock = clock;
+ }
+
+ @Override
+ public void initialize(Map<String, String> configs) {
+ K8sJobExecutorConfigs parsed = new K8sJobExecutorConfigs(configs);
+ initialize(parsed, K8sClientUtils.createClient(parsed));
+ }
+
+ @VisibleForTesting
+ void initialize(K8sJobExecutorConfigs configs, KubernetesClient client) {
+ this.configs = configs;
+ this.client = client;
+ this.sparkApplicationCache =
+ new NamespacedResourceCache<>(
+ configs.statusCacheTtlMs(),
+ namespace ->
+ getSparkApplicationClient(namespace)
+ .withLabel(
+ K8sJobResourceUtils.LABEL_MANAGED_BY,
+ K8sJobResourceUtils.MANAGED_BY_GRAVITINO)
+ .list()
+ .getItems(),
+ // Getting the SparkApplication itself couldn't reach the API
server either.
+ e ->
+ e instanceof KubernetesClientException
+ && isUnreachable((KubernetesClientException) e),
+ clock::millis);
+ LOG.info(
+ "Initialized the k8s job executor for cluster {} at {}, default
namespace {}",
+ configs.cluster(),
+ client.getMasterUrl(),
+ configs.namespace());
+ }
+
+ @Override
+ public String submitJob(JobContext context, JobTemplate jobTemplate) {
+ if (!(jobTemplate instanceof SparkJobTemplate)) {
+ throw new IllegalArgumentException(
+ String.format(
+ "The k8s job executor only supports Spark jobs, but job template
%s is a %s job",
+ jobTemplate.name(), jobTemplate.jobType()));
+ }
+
+ GenericKubernetesResource app =
+ SparkApplicationUtils.buildSparkApplication(
+ context, (SparkJobTemplate) jobTemplate, configs);
+ String namespace = app.getMetadata().getNamespace();
+ String name = app.getMetadata().getName();
+ K8sJobExecutionId jobExecutionId =
+ K8sJobExecutionId.of(
+ configs.cluster(), K8sJobExecutionId.Kind.SPARK_APPLICATION,
namespace, name);
+ try {
+ getSparkApplicationClient(namespace).resource(app).create();
+ } catch (KubernetesClientException e) {
+ if (isUnreachable(e)) {
+ // The SparkApplication may have been created although no response
came back. Gravitino
+ // records no job for a failed submission, so delete it, otherwise it
would run unknown.
+ deleteUnsubmittedSparkApplication(jobExecutionId, context.jobId());
+ throw e;
+ }
+ if (REJECTED_REQUEST_CODES.contains(e.getCode())) {
+ // The cluster rejects the job as it is, for example its namespace
doesn't exist or isn't
+ // allowed, so the caller gets the reason rather than an internal
error.
+ throw new IllegalArgumentException(
+ String.format(
+ "Kubernetes rejected the SparkApplication %s/%s of job
template %s: %s",
+ namespace, name, jobTemplate.name(), e.getMessage()),
+ e);
+ }
+ if (e.getCode() != HttpURLConnection.HTTP_CONFLICT) {
+ throw e;
+ }
+ // Created by an earlier attempt to submit the same job. Never adopt a
resource of another
+ // job, for example one left by another Gravitino deployment sharing the
namespace, nor one
+ // that is being deleted.
+ GenericKubernetesResource existing =
+ getSparkApplicationClient(namespace).withName(name).get();
+ if (existing == null
+ || existing.getMetadata().getDeletionTimestamp() != null
+ || !K8sJobResourceUtils.isOfJob(existing, context.jobId())) {
+ throw new IllegalStateException(
+ String.format(
+ "SparkApplication %s/%s already exists and can't be used for
job %s. If it belongs"
+ + " to another Gravitino deployment sharing the namespace,
use a different %s"
+ + " for each of them",
+ namespace, name, context.jobId(),
K8sJobExecutorConfigs.NAME_PREFIX),
+ e);
+ }
+ }
+
+ LOG.info("Submitted job {} as {}", context.jobId(), jobExecutionId);
+ return jobExecutionId.toString();
+ }
+
+ @Override
+ public JobExecutionInfo getJobExecutionInfo(String jobId) throws
NoSuchJobException {
+ K8sJobExecutionId id = parseJobExecutionId(jobId);
+ checkApiServerReachable();
+ try {
+ return getSparkApplicationExecutionInfo(id, jobId);
+ } catch (KubernetesClientException e) {
+ if (isUnreachable(e)) {
+ unreachableAtMs = clock.millis();
+ unreachableFailure = e;
+ }
+ throw e;
+ }
+ }
+
+ @Override
+ public void cancelJob(String jobId) throws NoSuchJobException {
+ K8sJobExecutionId id = parseJobExecutionId(jobId);
+ // Mark the cancellation before deleting, so that a status query on any
server tells a
+ // cancelled job from one deleted by someone else. A job that has stopped
isn't marked, so
+ // that a finished job isn't reported as cancelled. The mark is only
written if the job hasn't
+ // changed since it was checked, so a job that stops meanwhile is checked
again.
+ GenericKubernetesResource app =
+ updateSparkApplication(
+ id,
+ current -> {
+ if (SparkApplicationStatusUtils.isStopped(current)) {
+ return false;
+ }
+ setCancelRequested(current, true);
+ return true;
+ })
+ .orElseThrow(
+ () -> new NoSuchJobException("SparkApplication of job %s
doesn't exist", jobId));
+ if (SparkApplicationStatusUtils.isStopped(app)) {
+ return;
+ }
+
+ try {
+ deleteSparkApplication(id);
+ } catch (KubernetesClientException e) {
+ // The deletion may have taken effect although its response was lost.
Then the job is being
+ // cancelled, and keeps its mark.
+ AtomicBoolean deleting = new AtomicBoolean();
+ try {
+ Optional<GenericKubernetesResource> latest =
+ updateSparkApplication(
+ id,
+ current -> {
+ deleting.set(current.getMetadata().getDeletionTimestamp() !=
null);
+ if (deleting.get()) {
+ return false;
+ }
+ // The job keeps running and the caller is told the
cancellation failed, so
+ // remove the mark, otherwise the job would be reported as
cancelled if someone
+ // else deleted it while it runs. A mark that can't be
removed is harmless
+ // otherwise: without a deletion, the job keeps the outcome
it reaches.
+ setCancelRequested(current, false);
+ return true;
+ });
+ deleting.compareAndSet(false, !latest.isPresent());
+ } catch (RuntimeException removal) {
+ e.addSuppressed(removal);
+ throw e;
+ }
+ if (!deleting.get()) {
+ throw e;
+ }
+ LOG.warn("The deletion of the SparkApplication of job {} took effect
anyway", jobId, e);
+ }
+ LOG.info("Requested to cancel job {}", jobId);
+ }
+
+ @Override
+ public List<String> getJobStdout(String jobId, int maxLines, int maxBytes) {
+ try {
+ K8sJobExecutionId id = parseJobExecutionId(jobId);
+ // The SparkApplication tells the driver pod apart from other pods with
the same labels, and
+ // names the driver container. It may already be gone, the driver pod is
retained longer.
+ GenericKubernetesResource app =
+ getSparkApplicationClient(id.namespace()).withName(id.name()).get();
+ if (app != null && !K8sJobResourceUtils.isJobResource(app)) {
+ // The name is used by a resource that isn't the job's, whose driver
log isn't the job's.
+ return ImmutableList.of();
+ }
+ Optional<Pod> driver = getDriverPod(id, app);
+ if (!driver.isPresent()) {
+ return ImmutableList.of();
+ }
+
+ // Kubernetes can only limit the bytes from the head of the log, so read
the last lines and
+ // keep their tail. So the lines bound the bytes read, and maxBytes only
the bytes kept.
+ PodResource pod =
+
client.pods().inNamespace(id.namespace()).withName(driver.get().getMetadata().getName());
+ // The container must be named if the driver pod has others, for example
a sidecar.
+ List<Container> containers =
+ driver.get().getSpec() == null ? null :
driver.get().getSpec().getContainers();
+ String driverContainer =
+ app == null
+ ? SparkApplicationUtils.DRIVER_CONTAINER_NAME
+ : SparkApplicationStatusUtils.getDriverContainerName(app);
+ boolean nameContainer = containers != null && containers.size() > 1;
+ try (InputStream log =
+ (nameContainer ? pod.inContainer(driverContainer) : pod)
+ .tailingLines(maxLines)
+ .getLogInputStream()) {
+ return PodLogUtils.readLastLines(log, maxLines, maxBytes);
+ }
+ } catch (KubernetesClientException e) {
+ if (e.getCode() == HttpURLConnection.HTTP_BAD_REQUEST) {
+ // For example, the driver container hasn't started yet.
+ LOG.debug("No driver log of job {} to read yet: {}", jobId,
e.getMessage());
+ } else {
+ LOG.warn("Failed to read the driver log of job {}", jobId, e);
+ }
+ return ImmutableList.of();
+ } catch (RuntimeException | IOException e) {
+ // The output is best effort and must never fail the caller.
+ LOG.warn("Failed to read the driver log of job {}", jobId, e);
+ return ImmutableList.of();
+ }
+ }
+
+ /**
+ * Returns an empty list: Kubernetes merges the standard error into the
standard output of a pod,
+ * so it is returned by {@link #getJobStdout(String, int, int)}.
+ */
+ @Override
+ public List<String> getJobStderr(String jobId, int maxLines, int maxBytes) {
+ return ImmutableList.of();
+ }
+
+ @Override
+ public void close() {
+ if (client != null) {
+ client.close();
+ }
+ }
+
+ /**
+ * Returns whether a request failed because the API server couldn't be
reached or didn't answer,
+ * rather than because the API server refused or failed the request.
+ *
+ * @param e the failure of the request
+ * @return true if the API server gave no response
+ */
+ @VisibleForTesting
+ static boolean isUnreachable(KubernetesClientException e) {
+ if (e.getStatus() != null) {
+ return false;
+ }
+ boolean ioFailure = false;
+ for (Throwable cause = e.getCause(); cause != null; cause =
cause.getCause()) {
+ // A response that isn't JSON, for example the login page of a proxy,
did come back.
+ if (cause instanceof JsonProcessingException) {
+ return false;
+ }
+ ioFailure |= cause instanceof IOException;
+ }
+ return ioFailure;
+ }
+
+ /**
+ * Fails right away if the API server couldn't be reached a moment ago, so
that querying the
+ * status of many jobs doesn't wait for the connection to time out once per
job.
+ */
+ private void checkApiServerReachable() {
+ KubernetesClientException failure = unreachableFailure;
+ if (failure == null) {
+ return;
+ }
+ long remainingMs = unreachableAtMs + UNREACHABLE_BACKOFF_MS -
clock.millis();
+ if (remainingMs <= 0) {
+ unreachableFailure = null;
+ return;
+ }
+ throw new KubernetesClientException(
+ String.format(
+ "The API server of cluster %s couldn't be reached a moment ago,
the status of its jobs"
+ + " isn't queried again for %s ms",
+ configs.cluster(), remainingMs),
+ failure);
+ }
+
+ private JobExecutionInfo getSparkApplicationExecutionInfo(K8sJobExecutionId
id, String jobId) {
+ Optional<GenericKubernetesResource> cached =
+ sparkApplicationCache.find(id.namespace(), id.name());
+ GenericKubernetesResource app =
+ (cached.isPresent() ? cached : getSparkApplication(id))
+ .filter(K8sJobResourceUtils::isJobResource)
+ .orElseThrow(
+ () -> new NoSuchJobException("SparkApplication of job %s
doesn't exist", jobId));
+ return getExecutionInfo(id, jobId, app);
+ }
+
+ private JobExecutionInfo getExecutionInfo(
+ K8sJobExecutionId id, String jobId, GenericKubernetesResource app) {
+ // The operator observes the driver finishing a while after it does, so a
finished job takes
+ // its times from the driver pod, which is retained after the job
finishes. So does a job whose
+ // SparkApplication is being deleted, for example a cancelled one, whose
driver may have
+ // terminated before the operator recorded any outcome.
+ Pod driver = null;
+ KubernetesClientException driverFailure = null;
+ if (SparkApplicationStatusUtils.isStopped(app)
+ || app.getMetadata().getDeletionTimestamp() != null) {
+ try {
+ driver = getDriverPod(id, app).orElse(null);
+ } catch (KubernetesClientException e) {
+ driverFailure = e;
+ }
+ }
+ JobExecutionInfo info =
+ SparkApplicationStatusUtils.executionInfoOf(
+ app, driver, clock.instant(), configs.noStatusTimeoutMs());
+ if (info.status() == JobHandle.Status.FAILED
+ && !SparkApplicationStatusUtils.hasStatus(app)
+ && app.getMetadata().getDeletionTimestamp() == null) {
+ return getExecutionInfoWithoutStatus(id, jobId, info);
+ }
+ if (info.status() == JobHandle.Status.FAILED && driverFailure != null) {
+ // Whether the driver still runs can't be told, so the job isn't failed
yet: a later query
+ // checks its driver again.
+ throw driverFailure;
+ }
+ if (driverFailure != null) {
+ LOG.warn("Failed to get the driver pod of job {}", jobId, driverFailure);
+ }
+ if (info.status() == JobHandle.Status.FAILED
+ && driver != null
+ && SparkApplicationStatusUtils.isDriverRunning(app, driver)) {
+ // The operator retains the driver pod of a stopped application, even
one that is still
+ // pending or running after the application timed out starting. Delete
it, otherwise it
+ // could still run the job after it failed. A failure to delete it fails
this status query,
+ // so that the job only fails once the pod is deleted, by a later query.
+ LOG.warn(
+ "Deleting the driver pod {} of failed job {}, as it hasn't
terminated",
+ driver.getMetadata().getName(),
+ jobId);
+
client.pods().inNamespace(id.namespace()).withName(driver.getMetadata().getName()).delete();
+ }
+ return info;
+ }
+
+ /**
+ * Parses the job execution id of a job of the connected cluster. A job of
another cluster isn't
+ * reported as missing, which would end it while it may still run, for
example after the cluster
+ * is renamed: the query fails, and the job is left as it is.
+ */
+ private K8sJobExecutionId parseJobExecutionId(String jobId) throws
NoSuchJobException {
+ K8sJobExecutionId id = K8sJobExecutionId.parse(jobId);
+ if (!configs.cluster().equals(id.cluster())) {
+ throw new IllegalStateException(
+ String.format(
+ "Job %s runs in cluster %s, but the k8s job executor is
connected to cluster %s",
+ jobId, id.cluster(), configs.cluster()));
+ }
+ return id;
+ }
+
+ /**
+ * Returns the execution info of a job whose SparkApplication got no status
from the operator in
+ * time, which is failed. The SparkApplication is deleted first, otherwise
the operator would
+ * still run it once it comes back, with the job already failed and out of
Gravitino's reach. A
+ * failure to delete it fails the status query, so that the job only fails
once it is deleted, by
+ * a later query.
+ */
+ private JobExecutionInfo getExecutionInfoWithoutStatus(
+ K8sJobExecutionId id, String jobId, JobExecutionInfo failed) {
+ // The SparkApplication may come from the cache: it may have got a status
since, or be being
+ // deleted, for example by a cancellation on another server. Then it is
mapped as usual. If it
+ // is gone meanwhile, the job is missing rather than failed, which settles
a cancelling job as
+ // cancelled.
+ GenericKubernetesResource latest =
+ getSparkApplication(id)
+ .orElseThrow(
+ () -> new NoSuchJobException("SparkApplication of job %s
doesn't exist", jobId));
+ if (SparkApplicationStatusUtils.hasStatus(latest)
+ || latest.getMetadata().getDeletionTimestamp() != null) {
+ sparkApplicationCache.invalidate(id.namespace());
+ return getExecutionInfo(id, jobId, latest);
+ }
+ deleteSparkApplication(id);
+ return failed;
+ }
+
+ /**
+ * Deletes the SparkApplication of a job whose submission failed without
telling whether it was
+ * created, if it was. Only a SparkApplication of this job is deleted, and
any failure is ignored:
+ * the submission fails anyway.
+ */
+ private void deleteUnsubmittedSparkApplication(K8sJobExecutionId id, long
jobId) {
+ try {
+ Optional<GenericKubernetesResource> app = getSparkApplication(id);
+ if (app.isPresent() && K8sJobResourceUtils.isOfJob(app.get(), jobId)) {
+ deleteSparkApplication(id);
+ LOG.info("Deleted the SparkApplication of job {}, whose submission
failed", jobId);
+ }
+ } catch (RuntimeException e) {
+ LOG.warn(
+ "Failed to clean up the SparkApplication {} of job {}, whose
submission failed",
+ id,
+ jobId,
+ e);
+ }
+ }
+
+ /**
+ * Returns the SparkApplication of a job, if it exists. A resource with its
name that isn't a job
+ * resource of Gravitino, for example one created by another system, isn't.
+ */
+ private Optional<GenericKubernetesResource>
getSparkApplication(K8sJobExecutionId id) {
+ return
Optional.ofNullable(getSparkApplicationClient(id.namespace()).withName(id.name()).get())
+ .filter(K8sJobResourceUtils::isJobResource);
+ }
+
+ /**
+ * Changes the SparkApplication of a job, only if it hasn't changed since it
was read, and reads
+ * it again to decide the change again otherwise.
+ *
+ * @param id the job execution id
+ * @param change changes the SparkApplication in place, and returns whether
it changed it
+ * @return the SparkApplication the change was decided on, or empty if it
doesn't exist
+ */
+ private Optional<GenericKubernetesResource> updateSparkApplication(
+ K8sJobExecutionId id, Predicate<GenericKubernetesResource> change) {
+ for (int attempt = 0; ; attempt++) {
+ Optional<GenericKubernetesResource> app = getSparkApplication(id);
+ if (!app.isPresent() || !change.test(app.get())) {
+ return app;
+ }
+ try {
+ // An update carries the resource version that was read, so it fails
if the
+ // SparkApplication has changed since.
+ getSparkApplicationClient(id.namespace()).resource(app.get()).update();
+ sparkApplicationCache.invalidate(id.namespace());
+ return app;
+ } catch (KubernetesClientException e) {
+ if (e.getCode() == HttpURLConnection.HTTP_NOT_FOUND) {
+ return Optional.empty();
+ }
+ if (e.getCode() != HttpURLConnection.HTTP_CONFLICT || attempt >=
UPDATE_CONFLICT_RETRIES) {
+ throw e;
+ }
+ }
+ }
+ }
+
+ /**
+ * Returns the newest driver pod of a SparkApplication. The operator names
the driver pod after
+ * the Spark app id, which it may hash-truncate, so it is looked up by the
operator's labels, and
+ * also by its owner, the SparkApplication, when that is known.
+ */
+ private Optional<Pod> getDriverPod(
+ K8sJobExecutionId id, @Nullable GenericKubernetesResource app) {
+ String appUid = app == null ? null : app.getMetadata().getUid();
+ return client
+ .pods()
+ .inNamespace(id.namespace())
+ .withLabel(SparkApplicationUtils.LABEL_SPARK_APP_NAME, id.name())
+ .withLabel(SparkApplicationUtils.LABEL_SPARK_ROLE,
SparkApplicationUtils.SPARK_ROLE_DRIVER)
+ .list()
+ .getItems()
+ .stream()
+ .filter(pod -> appUid == null || isOwnedBy(pod, appUid))
+ .max(
+ Comparator.comparing(
+ pod -> pod.getMetadata().getCreationTimestamp(),
+ Comparator.nullsFirst(Comparator.naturalOrder())));
+ }
+
+ private static boolean isOwnedBy(Pod pod, String uid) {
+ List<OwnerReference> owners = pod.getMetadata().getOwnerReferences();
+ return owners != null && owners.stream().anyMatch(owner ->
uid.equals(owner.getUid()));
+ }
+
+ private static void setCancelRequested(GenericKubernetesResource app,
boolean cancelRequested) {
+ Map<String, String> annotations = new HashMap<>();
+ if (app.getMetadata().getAnnotations() != null) {
+ annotations.putAll(app.getMetadata().getAnnotations());
+ }
+ if (cancelRequested) {
+ annotations.put(K8sJobResourceUtils.ANNOTATION_CANCEL_REQUESTED, "true");
+ } else {
+ annotations.remove(K8sJobResourceUtils.ANNOTATION_CANCEL_REQUESTED);
+ }
+ app.getMetadata().setAnnotations(annotations);
+ }
+
+ private void deleteSparkApplication(K8sJobExecutionId id) {
+ // The SparkApplication stays until its pods are deleted, so that a status
query sees it being
+ // deleted rather than gone.
+ getSparkApplicationClient(id.namespace())
+ .withName(id.name())
+ .withPropagationPolicy(DeletionPropagation.FOREGROUND)
+ .delete();
+ sparkApplicationCache.invalidate(id.namespace());
+ }
+
+ private NonNamespaceOperation<
+ GenericKubernetesResource,
+ GenericKubernetesResourceList,
+ Resource<GenericKubernetesResource>>
+ getSparkApplicationClient(String namespace) {
+ return client
+ .genericKubernetesResources(SparkApplicationUtils.SPARK_APPLICATION)
+ .inNamespace(namespace);
+ }
+}
diff --git
a/plugins/k8s-job-executor/src/main/java/org/apache/gravitino/job/k8s/K8sJobExecutorConfigs.java
b/plugins/k8s-job-executor/src/main/java/org/apache/gravitino/job/k8s/K8sJobExecutorConfigs.java
index 306322a771..b4b8eea85d 100644
---
a/plugins/k8s-job-executor/src/main/java/org/apache/gravitino/job/k8s/K8sJobExecutorConfigs.java
+++
b/plugins/k8s-job-executor/src/main/java/org/apache/gravitino/job/k8s/K8sJobExecutorConfigs.java
@@ -35,7 +35,12 @@ public class K8sJobExecutorConfigs {
/** The name of the job executor. */
public static final String K8S_JOB_EXECUTOR_NAME = "k8s";
- /** The name of the Kubernetes cluster the jobs run in, which is part of the
job execution id. */
+ /**
+ * The name of the Kubernetes cluster the jobs run in, which is part of the
job execution id. It
+ * must be the same on every Gravitino server, and must not be changed while
jobs run: the jobs
+ * submitted under another name can't be queried nor cancelled, and stay
unfinished until it is
+ * changed back.
+ */
public static final String CLUSTER = "cluster";
/** The default cluster name. */
@@ -105,7 +110,11 @@ public class K8sJobExecutorConfigs {
/** The default status cache TTL, 5 seconds. */
public static final long DEFAULT_STATUS_CACHE_TTL_MS = 5_000L;
- /** A job whose resource still has no status after this time is considered
failed. */
+ /**
+ * A job whose resource still has no status after this time is considered
failed, and its resource
+ * is deleted. A job whose resource has a status without any state this
version can read, for
+ * example of a newer operator, stays unfinished instead, until the operator
deletes the resource.
+ */
public static final String NO_STATUS_TIMEOUT_MS = "noStatusTimeoutInMs";
/** The default no status timeout, 10 minutes. */
diff --git
a/plugins/k8s-job-executor/src/main/java/org/apache/gravitino/job/k8s/K8sJobResourceUtils.java
b/plugins/k8s-job-executor/src/main/java/org/apache/gravitino/job/k8s/K8sJobResourceUtils.java
index a0ffb490e8..b9a3a3b4eb 100644
---
a/plugins/k8s-job-executor/src/main/java/org/apache/gravitino/job/k8s/K8sJobResourceUtils.java
+++
b/plugins/k8s-job-executor/src/main/java/org/apache/gravitino/job/k8s/K8sJobResourceUtils.java
@@ -20,8 +20,10 @@ package org.apache.gravitino.job.k8s;
import com.google.common.collect.ImmutableMap;
import io.fabric8.kubernetes.api.model.HasMetadata;
+import io.fabric8.kubernetes.api.model.ObjectMeta;
import java.util.Map;
import java.util.regex.Pattern;
+import org.apache.commons.lang3.StringUtils;
/** The naming and labeling of the Kubernetes resources created for the jobs.
*/
public final class K8sJobResourceUtils {
@@ -92,6 +94,27 @@ public final class K8sJobResourceUtils {
return labels != null &&
String.valueOf(jobId).equals(labels.get(LABEL_JOB_ID));
}
+ /**
+ * Returns whether the Kubernetes resource is the one Gravitino created for
the job its name was
+ * made for: it is managed by Gravitino, and its job id label is the job id
its name ends with.
+ * The name alone may be used by another system, or by another Gravitino
deployment with the same
+ * name prefix in the namespace, for another of its jobs.
+ *
+ * @param resource the resource
+ * @return true if the resource is a job resource of Gravitino matching its
name
+ */
+ public static boolean isJobResource(HasMetadata resource) {
+ ObjectMeta metadata = resource.getMetadata();
+ Map<String, String> labels = metadata == null ? null :
metadata.getLabels();
+ if (labels == null ||
!MANAGED_BY_GRAVITINO.equals(labels.get(LABEL_MANAGED_BY))) {
+ return false;
+ }
+ String jobId = labels.get(LABEL_JOB_ID);
+ return StringUtils.isNotBlank(jobId)
+ && metadata.getName() != null
+ && metadata.getName().endsWith("-" + jobId);
+ }
+
/**
* Returns whether Gravitino requested to cancel the job of the Kubernetes
resource.
*
diff --git
a/plugins/k8s-job-executor/src/main/java/org/apache/gravitino/job/k8s/NamespacedResourceCache.java
b/plugins/k8s-job-executor/src/main/java/org/apache/gravitino/job/k8s/NamespacedResourceCache.java
new file mode 100644
index 0000000000..59001e9638
--- /dev/null
+++
b/plugins/k8s-job-executor/src/main/java/org/apache/gravitino/job/k8s/NamespacedResourceCache.java
@@ -0,0 +1,183 @@
+/*
+ * 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;
+
+import com.google.common.collect.ImmutableMap;
+import io.fabric8.kubernetes.api.model.HasMetadata;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CompletionException;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.function.Function;
+import java.util.function.LongSupplier;
+import java.util.function.Predicate;
+import javax.annotation.Nullable;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Caches the resources listed in a namespace for a short time, so that
querying the status of many
+ * jobs in the same namespace costs one LIST call instead of one GET call per
job.
+ *
+ * <p>A resource missing from the cache isn't necessarily gone: it may have
been created after the
+ * namespace was listed, or the namespace can't be listed at all. The caller
must fall back to a GET
+ * call on a cache miss.
+ *
+ * @param <T> the resource type
+ */
+final class NamespacedResourceCache<T extends HasMetadata> {
+
+ private static final class Snapshot<T> {
+
+ private final long listedAtMs;
+
+ private final Map<String, T> resources;
+
+ private Snapshot(long listedAtMs, Map<String, T> resources) {
+ this.listedAtMs = listedAtMs;
+ this.resources = resources;
+ }
+ }
+
+ private static final Logger LOG =
LoggerFactory.getLogger(NamespacedResourceCache.class);
+
+ private final long ttlMs;
+
+ private final Function<String, List<T>> lister;
+
+ private final Predicate<RuntimeException> isFatal;
+
+ private final LongSupplier clock;
+
+ // The listing of each namespace, being made or made. Only one thread lists
a namespace at a
+ // time, the others wait for its listing. A listing is a network call, so it
is made outside of
+ // the map, which only holds the namespaces listed within the TTL or being
listed.
+ private final Map<String, CompletableFuture<Snapshot<T>>> snapshots = new
ConcurrentHashMap<>();
+
+ /**
+ * Creates a cache.
+ *
+ * @param ttlMs how long a namespace listing is cached, caching is disabled
if not positive
+ * @param lister lists the resources of a namespace
+ * @param isFatal tells the failures of the lister that getting a resource
by itself would run
+ * into as well, for example an API server that can't be reached. They
are thrown to the
+ * caller of {@link #find(String, String)} instead of being cached as an
empty listing.
+ * @param clock the current time in milliseconds
+ */
+ NamespacedResourceCache(
+ long ttlMs,
+ Function<String, List<T>> lister,
+ Predicate<RuntimeException> isFatal,
+ LongSupplier clock) {
+ this.ttlMs = ttlMs;
+ this.lister = lister;
+ this.isFatal = isFatal;
+ this.clock = clock;
+ }
+
+ /**
+ * Finds a resource in the cached listing of its namespace, and lists the
namespace again if the
+ * cached listing has expired. A namespace that can't be listed, for example
because listing it is
+ * forbidden, isn't listed again until the cached failure expires.
+ *
+ * @param namespace the namespace of the resource
+ * @param name the name of the resource
+ * @return the resource, or empty if it isn't in the listing or the
namespace can't be listed
+ * @throws RuntimeException the failure to list the namespace, if it is a
fatal one
+ */
+ Optional<T> find(String namespace, String name) {
+ if (ttlMs <= 0) {
+ return Optional.empty();
+ }
+
+ CompletableFuture<Snapshot<T>> snapshot = snapshots.get(namespace);
+ if (isExpired(snapshot)) {
+ // Drop the expired listings, also of the namespaces whose jobs are all
done.
+ snapshots.values().removeIf(this::isExpired);
+ CompletableFuture<Snapshot<T>> listing = new CompletableFuture<>();
+ snapshot =
+ snapshots.compute(namespace, (ns, current) -> isExpired(current) ?
listing : current);
+ if (snapshot == listing) {
+ try {
+ listing.complete(snapshotOf(namespace));
+ } catch (Throwable t) {
+ // Whatever the failure, even an error, the listing must be
completed and isn't cached,
+ // otherwise the next queries of the namespace would wait for it
forever. The waiting
+ // callers get the failure too.
+ snapshots.remove(namespace, listing);
+ listing.completeExceptionally(t);
+ throw t;
+ }
+ }
+ }
+
+ try {
+ return Optional.ofNullable(snapshot.join().resources.get(name));
+ } catch (CompletionException e) {
+ if (e.getCause() instanceof Error) {
+ throw (Error) e.getCause();
+ }
+ throw e.getCause() instanceof RuntimeException ? (RuntimeException)
e.getCause() : e;
+ }
+ }
+
+ /**
+ * Drops the cached listing of a namespace, after a resource in it is
changed. A listing of it
+ * being made isn't cached either, as it may predate the change.
+ *
+ * @param namespace the namespace
+ */
+ void invalidate(String namespace) {
+ snapshots.remove(namespace);
+ }
+
+ /** Whether a listing must be made again: there is none, it failed, or it is
too old. */
+ private boolean isExpired(@Nullable CompletableFuture<Snapshot<T>> snapshot)
{
+ if (snapshot == null || snapshot.isCompletedExceptionally()) {
+ return true;
+ }
+ // A listing being made is waited for.
+ Snapshot<T> made = snapshot.getNow(null);
+ return made != null && clock.getAsLong() - made.listedAtMs >= ttlMs;
+ }
+
+ private Snapshot<T> snapshotOf(String namespace) {
+ long listedAtMs = clock.getAsLong();
+ try {
+ ImmutableMap.Builder<String, T> resources = ImmutableMap.builder();
+ lister.apply(namespace).forEach(r ->
resources.put(r.getMetadata().getName(), r));
+ return new Snapshot<>(listedAtMs, resources.buildKeepingLast());
+ } catch (RuntimeException e) {
+ if (isFatal.test(e)) {
+ throw e;
+ }
+ // Cache the failure as an empty listing, so that the callers get the
resources one by one
+ // instead of each of them failing to list the namespace again.
+ LOG.warn(
+ "Failed to list the resources in namespace {}, getting them one by
one for the next {}"
+ + " ms: {}",
+ namespace,
+ ttlMs,
+ e.toString());
+ return new Snapshot<>(listedAtMs, ImmutableMap.of());
+ }
+ }
+}
diff --git
a/plugins/k8s-job-executor/src/main/java/org/apache/gravitino/job/k8s/PodLogUtils.java
b/plugins/k8s-job-executor/src/main/java/org/apache/gravitino/job/k8s/PodLogUtils.java
new file mode 100644
index 0000000000..578f116880
--- /dev/null
+++
b/plugins/k8s-job-executor/src/main/java/org/apache/gravitino/job/k8s/PodLogUtils.java
@@ -0,0 +1,121 @@
+/*
+ * 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;
+
+import com.google.common.base.Preconditions;
+import com.google.common.collect.ImmutableList;
+import java.io.IOException;
+import java.io.InputStream;
+import java.nio.charset.StandardCharsets;
+import java.util.Arrays;
+import java.util.List;
+
+/** Reads the output of a job from the log of its pod. */
+public final class PodLogUtils {
+
+ private static final int READ_BUFFER_SIZE = 8192;
+
+ // The largest array size that is safe to allocate on every JVM.
+ private static final int MAX_BUFFER_SIZE = Integer.MAX_VALUE - 8;
+
+ private PodLogUtils() {}
+
+ /**
+ * Reads the last lines of a log stream, within its last bytes. Only the
last bytes are kept in
+ * memory while reading. If the kept bytes start in the middle of a line,
that partial line is
+ * dropped, unless it is the only line.
+ *
+ * <p>The whole stream is read: {@code maxBytes} bounds the memory and the
lines returned, not the
+ * bytes read. The caller bounds those, for example by asking Kubernetes for
the last lines of a
+ * log only.
+ *
+ * @param log the log stream, which is read to its end
+ * @param maxLines the maximum number of lines to return
+ * @param maxBytes the maximum number of bytes to read from the tail of the
log
+ * @return the last lines of the log
+ * @throws IOException if the log can't be read
+ */
+ public static List<String> readLastLines(InputStream log, int maxLines, int
maxBytes)
+ throws IOException {
+ Preconditions.checkArgument(maxLines > 0, "maxLines must be positive");
+ Preconditions.checkArgument(maxBytes > 0, "maxBytes must be positive");
+
+ // Keep one more byte than the window, to tell whether the window starts
at a line boundary.
+ // The window is capped so that the buffer, of twice its size, fits in an
array.
+ int keep = (int) Math.min(maxBytes + 1L, MAX_BUFFER_SIZE / 2);
+ int window = keep - 1;
+ int maxCapacity = Math.max(2 * keep, READ_BUFFER_SIZE);
+ // The buffer grows with the log, as most logs are much shorter than the
window.
+ byte[] tail = new byte[READ_BUFFER_SIZE];
+ int length = 0;
+ byte[] chunk = new byte[READ_BUFFER_SIZE];
+ int read;
+ while ((read = log.read(chunk)) >= 0) {
+ if (length + read > tail.length && tail.length < maxCapacity) {
+ long capacity = Math.max(2L * tail.length, (long) length + read);
+ tail = Arrays.copyOf(tail, (int) Math.min(capacity, maxCapacity));
+ }
+ if (length + read > tail.length) {
+ // Drop the head, keeping the last bytes of the buffer and the chunk.
+ int kept = Math.max(keep - read, 0);
+ System.arraycopy(tail, length - kept, tail, 0, kept);
+ length = kept;
+ if (read > keep) {
+ System.arraycopy(chunk, read - keep, tail, 0, keep);
+ length = keep;
+ continue;
+ }
+ }
+ System.arraycopy(chunk, 0, tail, length, read);
+ length += read;
+ }
+
+ int start = 0;
+ boolean startsAtLineBoundary = true;
+ if (length > window) {
+ start = length - window;
+ startsAtLineBoundary = tail[start - 1] == '\n';
+ }
+ // Never start in the middle of a UTF-8 character. A '\n' is never part of
one.
+ while (start < length && (tail[start] & 0xC0) == 0x80) {
+ start++;
+ }
+ String content = new String(tail, start, length - start,
StandardCharsets.UTF_8);
+ if (!startsAtLineBoundary) {
+ int firstNewline = content.indexOf('\n');
+ if (firstNewline >= 0 && firstNewline < content.length() - 1) {
+ content = content.substring(firstNewline + 1);
+ }
+ }
+ if (content.isEmpty()) {
+ return ImmutableList.of();
+ }
+
+ List<String> lines = Arrays.asList(content.split("\\r?\\n", -1));
+ // A trailing line terminator doesn't start another line.
+ if (lines.get(lines.size() - 1).isEmpty()) {
+ lines = lines.subList(0, lines.size() - 1);
+ }
+ // The window only holds the line terminator of a line that starts before
it.
+ if (!startsAtLineBoundary && lines.size() == 1 && lines.get(0).isEmpty()) {
+ return ImmutableList.of();
+ }
+ return ImmutableList.copyOf(lines.subList(Math.max(0, lines.size() -
maxLines), lines.size()));
+ }
+}
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
index 90c17a664e..f78a498137 100644
---
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
@@ -115,12 +115,21 @@ public final class SparkApplicationStatusUtils {
*
* <ul>
* <li>A SparkApplication that Gravitino requested to cancel is cancelled
once it is being
- * deleted or has stopped.
+ * deleted, unless it had stopped before, as the operator recorded it:
by an outcome or a
+ * stopped state recorded before the deletion, or whose time is
unknown. A driver that
+ * terminated before the deletion doesn't count, as the operator may
never record its
+ * outcome once the application is deleted. The cancel mark alone
doesn't cancel it: a
+ * cancellation that failed after marking it, for example because the
server stopped or a
+ * response was lost, leaves the mark on a job that keeps running,
which then keeps the
+ * outcome it reaches.
* <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.
+ * namespace isn't watched by the operator. One with a status that has
no state this version
+ * can read stays queued, even if it was started before, and nothing
ends it until the
+ * operator deletes it after its TTL. It means the operator is newer
than this version
+ * supports, which must be fixed rather than waited out.
* <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.
@@ -133,8 +142,8 @@ public final class SparkApplicationStatusUtils {
* 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.
+ * container terminated if that is known, else when it entered its stopped
state if that is known,
+ * or else when it was deleted.
*
* @param app the SparkApplication
* @param driver the driver pod, or null if unknown
@@ -151,7 +160,7 @@ public final class SparkApplicationStatusUtils {
Instant deletedAt = parseTime(app.getMetadata().getDeletionTimestamp());
ContainerState driverState =
- driver == null ? null : driverContainerState(driver,
driverContainerName(app));
+ driver == null ? null : driverContainerState(driver,
getDriverContainerName(app));
Instant driverStartedAt = null;
Instant driverFinishedAt = null;
if (driverState != null && driverState.getTerminated() != null) {
@@ -163,24 +172,33 @@ public final class SparkApplicationStatusUtils {
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.
+ // when it was deleted if it hadn't stopped by then, 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 deletedOrStartedAt =
+ deletedAt != null && startedAt != null && deletedAt.isBefore(startedAt)
+ ? startedAt
+ : deletedAt;
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 if (deletedAt != null && !isStopped(current, outcome)) {
+ stoppedAt = deletedOrStartedAt;
} else {
stoppedAt = current == null ? null : current.time;
}
+ if (stoppedAt == null) {
+ // The time of the state it stopped in is unknown, the deletion is the
best known one.
+ stoppedAt = deletedOrStartedAt;
+ }
JobHandle.Status status;
Instant finishedAt = null;
if (K8sJobResourceUtils.isCancelRequested(app)
- && (deletedAt != null || isStopped(current, outcome))) {
+ && deletedAt != null
+ && !stoppedBefore(history, deletedAt)) {
status = JobHandle.Status.CANCELLED;
finishedAt = stoppedAt;
} else if (deletedAt != null || isStopped(current, outcome)) {
@@ -188,7 +206,15 @@ public final class SparkApplicationStatusUtils {
status = outcome == null ? JobHandle.Status.FAILED : statusOf(outcome);
finishedAt = stoppedAt;
} else if (current == null) {
- if (createdAt != null &&
createdAt.plusMillis(noStatusTimeoutMs).isBefore(now)) {
+ if (hasStatus(app)) {
+ // The operator wrote a status this version can't read, for example a
newer operator. The
+ // application is known to the operator, so the job stays unfinished
rather than failing.
+ status = JobHandle.Status.QUEUED;
+ LOG.warn(
+ "Can't read the status of SparkApplication {}, considering it as
{}",
+ nameOf(app),
+ status);
+ } else 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",
@@ -224,6 +250,36 @@ public final class SparkApplicationStatusUtils {
.build();
}
+ /**
+ * Returns whether the driver of a SparkApplication is still pending or
running. It is decided by
+ * the driver container, as a sidecar can keep the pod running after the
driver exited, and by the
+ * pod only if the driver container can't be told.
+ *
+ * @param app the SparkApplication
+ * @param driver the driver pod
+ * @return true if the driver hasn't terminated
+ */
+ public static boolean isDriverRunning(GenericKubernetesResource app, Pod
driver) {
+ ContainerState state = driverContainerState(driver,
getDriverContainerName(app));
+ if (state != null) {
+ return state.getTerminated() == null;
+ }
+ String phase = driver.getStatus() == null ? null :
driver.getStatus().getPhase();
+ return !"Succeeded".equals(phase) && !"Failed".equals(phase);
+ }
+
+ /**
+ * Returns whether the operator has written a status of the
SparkApplication, whether or not this
+ * version can read a state from it.
+ *
+ * @param app the SparkApplication
+ * @return true if the application has a non-empty status
+ */
+ public static boolean hasStatus(GenericKubernetesResource app) {
+ Object status = app.getAdditionalProperties().get("status");
+ return status instanceof Map && !((Map<?, ?>) status).isEmpty();
+ }
+
/**
* Returns whether the operator has reported a state of the SparkApplication
that can be read.
*
@@ -262,6 +318,20 @@ public final class SparkApplicationStatusUtils {
|| (!running && outcome != null);
}
+ /**
+ * Returns whether the application had reached an outcome or a stopped state
before it was
+ * deleted. A deletion time is in whole seconds, so a state of the same
second counts as after. A
+ * state whose time is unknown counts as before, like {@link
#isStopped(State, State)}, which
+ * doesn't need the times either.
+ */
+ private static boolean stoppedBefore(List<State> history, Instant deletedAt)
{
+ return history.stream()
+ .anyMatch(
+ s ->
+ (isOutcome(s) || STOPPED_STATES.contains(s.summary))
+ && (s.time == null || s.time.isBefore(deletedAt)));
+ }
+
private static boolean isOutcome(State state) {
return SUCCEEDED_STATE.equals(state.summary) ||
FAILED_STATES.contains(state.summary);
}
@@ -321,7 +391,7 @@ public final class SparkApplicationStatusUtils {
if (!(status instanceof Map)) {
return history;
}
- String driverContainer = driverContainerName(app);
+ String driverContainer = getDriverContainerName(app);
Object transitions = ((Map<String, Object>)
status).get("stateTransitionHistory");
if (transitions instanceof Map) {
@@ -392,10 +462,13 @@ public final class SparkApplicationStatusUtils {
}
/**
- * Returns the name of the driver container: the one a job names with a pod
template, or else the
- * one Spark gives it.
+ * Returns the name of the driver container of a SparkApplication: the one
the job names for its
+ * driver pod template, or else the one Spark gives it.
+ *
+ * @param app the SparkApplication
+ * @return the name of the driver container
*/
- private static String driverContainerName(GenericKubernetesResource app) {
+ public static String getDriverContainerName(GenericKubernetesResource app) {
Map<String, Object> spec =
mapOf(app.getAdditionalProperties().get("spec"));
Map<String, Object> sparkConf = spec == null ? null :
mapOf(spec.get("sparkConf"));
Object name =
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 aa49a1c5fb..73ea8d6c9c 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
@@ -75,6 +75,7 @@ public final class SparkApplicationUtils {
private static final String SPARK_MASTER = "spark.master";
private static final String SPARK_DEPLOY_MODE = "spark.submit.deployMode";
+ private static final String CLUSTER_DEPLOY_MODE = "cluster";
private static final String SPARK_IMAGE = "spark.kubernetes.container.image";
private static final String SPARK_DRIVER_SERVICE_ACCOUNT =
"spark.kubernetes.authenticate.driver.serviceAccountName";
@@ -117,6 +118,16 @@ public final class SparkApplicationUtils {
Map<String, String> sparkConf = new TreeMap<>(configs.sparkConf());
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.
+ String deployMode =
StringUtils.trimToNull(sparkConf.get(SPARK_DEPLOY_MODE));
+ if (deployMode != null &&
!CLUSTER_DEPLOY_MODE.equalsIgnoreCase(deployMode)) {
+ throw new IllegalArgumentException(
+ String.format(
+ "%s is set to %s by job template %s or by the k8s job executor
configurations, but"
+ + " the k8s job executor only runs Spark jobs in %s mode",
+ SPARK_DEPLOY_MODE, deployMode, template.name(),
CLUSTER_DEPLOY_MODE));
+ }
// The operator decides the master and the deploy mode.
sparkConf.remove(SPARK_MASTER);
sparkConf.remove(SPARK_DEPLOY_MODE);
diff --git
a/plugins/k8s-job-executor/src/test/java/org/apache/gravitino/job/k8s/TestK8sJobExecutor.java
b/plugins/k8s-job-executor/src/test/java/org/apache/gravitino/job/k8s/TestK8sJobExecutor.java
new file mode 100644
index 0000000000..dd5bd4125c
--- /dev/null
+++
b/plugins/k8s-job-executor/src/test/java/org/apache/gravitino/job/k8s/TestK8sJobExecutor.java
@@ -0,0 +1,1195 @@
+/*
+ * 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;
+
+import com.fasterxml.jackson.core.JsonParseException;
+import com.fasterxml.jackson.core.JsonParser;
+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.OwnerReference;
+import io.fabric8.kubernetes.api.model.OwnerReferenceBuilder;
+import io.fabric8.kubernetes.api.model.PodBuilder;
+import io.fabric8.kubernetes.api.model.StatusBuilder;
+import io.fabric8.kubernetes.client.Config;
+import io.fabric8.kubernetes.client.ConfigBuilder;
+import io.fabric8.kubernetes.client.KubernetesClient;
+import io.fabric8.kubernetes.client.KubernetesClientBuilder;
+import io.fabric8.kubernetes.client.KubernetesClientException;
+import io.fabric8.kubernetes.client.dsl.base.CustomResourceDefinitionContext;
+import io.fabric8.kubernetes.client.server.mock.KubernetesMixedDispatcher;
+import io.fabric8.kubernetes.client.server.mock.KubernetesMockServer;
+import io.fabric8.mockwebserver.Context;
+import io.fabric8.mockwebserver.MockWebServer;
+import io.fabric8.mockwebserver.ServerRequest;
+import io.fabric8.mockwebserver.ServerResponse;
+import io.fabric8.mockwebserver.http.RecordedRequest;
+import java.io.File;
+import java.io.IOException;
+import java.net.ConnectException;
+import java.time.Clock;
+import java.time.Instant;
+import java.time.ZoneId;
+import java.time.ZoneOffset;
+import java.time.temporal.ChronoUnit;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Queue;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicReference;
+import org.apache.gravitino.connector.job.JobContext;
+import org.apache.gravitino.connector.job.JobExecutionInfo;
+import org.apache.gravitino.exceptions.NoSuchJobException;
+import org.apache.gravitino.job.JobHandle;
+import org.apache.gravitino.job.ShellJobTemplate;
+import org.apache.gravitino.job.SparkJobTemplate;
+import org.apache.gravitino.job.k8s.spark.SparkApplicationUtils;
+import org.apache.gravitino.job.k8s.spark.TestSparkApplicationUtils;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+public class TestK8sJobExecutor {
+
+ private static final String APPS_PATH =
+ "/apis/spark.apache.org/v1/namespaces/default/sparkapplications";
+
+ private KubernetesMockServer server;
+
+ private KubernetesClient client;
+
+ private K8sJobExecutor executor;
+
+ private final AtomicReference<Instant> now = new
AtomicReference<>(Instant.now());
+
+ @BeforeEach
+ public void setUp() {
+ Map<ServerRequest, Queue<ServerResponse>> responses = new HashMap<>();
+ server =
+ new KubernetesMockServer(
+ new Context(),
+ new MockWebServer(),
+ responses,
+ new KubernetesMixedDispatcher(responses),
+ false);
+ server.init();
+ server.expectCustomResource(
+ new CustomResourceDefinitionContext.Builder()
+ .withGroup(SparkApplicationUtils.SPARK_APPLICATION.getGroup())
+ .withVersion(SparkApplicationUtils.SPARK_APPLICATION.getVersion())
+ .withKind(SparkApplicationUtils.SPARK_APPLICATION.getKind())
+ .withPlural(SparkApplicationUtils.SPARK_APPLICATION.getPlural())
+ .withScope("Namespaced")
+ .build());
+ client = server.createClient();
+
+ executor =
+ new K8sJobExecutor(
+ new Clock() {
+ @Override
+ public ZoneId getZone() {
+ return ZoneOffset.UTC;
+ }
+
+ @Override
+ public Clock withZone(ZoneId zone) {
+ return this;
+ }
+
+ @Override
+ public Instant instant() {
+ return now.get();
+ }
+ });
+ executor.initialize(
+ TestSparkApplicationUtils.newConfigs(
+ ImmutableMap.of(K8sJobExecutorConfigs.STATUS_CACHE_TTL_MS, "0")),
+ client);
+ }
+
+ @AfterEach
+ public void tearDown() {
+ executor.close();
+ server.destroy();
+ }
+
+ @Test
+ public void testSubmitAndQuery() {
+ String jobExecutionId = executor.submitJob(context(1L), sparkTemplate());
+
+ Assertions.assertEquals("default/sparkapp/default/gravitino-job-1",
jobExecutionId);
+ GenericKubernetesResource app = sparkApplication("gravitino-job-1");
+ Assertions.assertNotNull(app);
+ Assertions.assertEquals(
+ "1",
app.getMetadata().getLabels().get(K8sJobResourceUtils.LABEL_JOB_ID));
+ Assertions.assertEquals(
+ JobHandle.Status.QUEUED,
executor.getJobExecutionInfo(jobExecutionId).status());
+
+ setStates("gravitino-job-1", "Submitted", "DriverReady", "RunningHealthy");
+ JobExecutionInfo running = executor.getJobExecutionInfo(jobExecutionId);
+ Assertions.assertEquals(JobHandle.Status.STARTED, running.status());
+ Assertions.assertNotNull(running.startedAt());
+
+ setStates("gravitino-job-1", "Submitted", "DriverReady", "Succeeded",
"ResourceReleased");
+ Assertions.assertEquals(
+ JobHandle.Status.SUCCEEDED,
executor.getJobExecutionInfo(jobExecutionId).status());
+
+ // The times of a finished job come from its driver pod. The driver can't
start before its
+ // SparkApplication was created, which the mock server does at the current
time.
+ Instant driverStartedAt =
Instant.now().plusSeconds(60).truncatedTo(ChronoUnit.SECONDS);
+ Instant driverFinishedAt = driverStartedAt.plusSeconds(1);
+ client
+ .pods()
+ .inNamespace("default")
+ .resource(
+ new PodBuilder()
+ .withNewMetadata()
+ .addToOwnerReferences(ownerReference("gravitino-job-1"))
+ .withName("driver")
+ .addToLabels(SparkApplicationUtils.LABEL_SPARK_APP_NAME,
"gravitino-job-1")
+ .addToLabels(SparkApplicationUtils.LABEL_SPARK_ROLE, "driver")
+ .endMetadata()
+ .withNewStatus()
+ .addNewContainerStatus()
+ .withName("spark-kubernetes-driver")
+ .withNewState()
+ .withNewTerminated()
+ .withStartedAt(driverStartedAt.toString())
+ .withFinishedAt(driverFinishedAt.toString())
+ .endTerminated()
+ .endState()
+ .endContainerStatus()
+ .endStatus()
+ .build())
+ .create();
+ JobExecutionInfo succeeded = executor.getJobExecutionInfo(jobExecutionId);
+ Assertions.assertEquals(driverStartedAt, succeeded.startedAt());
+ Assertions.assertEquals(driverFinishedAt, succeeded.finishedAt());
+ }
+
+ @Test
+ public void testStatusCacheFallsBackToGet() {
+ executor.initialize(
+ TestSparkApplicationUtils.newConfigs(
+ ImmutableMap.of(K8sJobExecutorConfigs.STATUS_CACHE_TTL_MS,
"60000")),
+ client);
+ String job1 = executor.submitJob(context(1L), sparkTemplate());
+ Assertions.assertEquals(JobHandle.Status.QUEUED,
executor.getJobExecutionInfo(job1).status());
+
+ // Created after the namespace was listed, so it isn't in the cached
listing.
+ String job2 = executor.submitJob(context(2L), sparkTemplate());
+ Assertions.assertEquals(JobHandle.Status.QUEUED,
executor.getJobExecutionInfo(job2).status());
+
+ // The cached listing is used within its TTL, so the new state isn't seen
yet.
+ setStates("gravitino-job-1", "Submitted", "DriverReady");
+ Assertions.assertEquals(JobHandle.Status.QUEUED,
executor.getJobExecutionInfo(job1).status());
+ }
+
+ @Test
+ public void testNamespaceOfJob() {
+ SparkJobTemplate template =
+ SparkJobTemplate.builder()
+ .withName("user-job")
+ .withExecutable("local:///opt/app.jar")
+ .withClassName("org.example.App")
+
.withConfigs(ImmutableMap.of(SparkApplicationUtils.SPARK_NAMESPACE, "team-a"))
+ .build();
+
+ String jobExecutionId = executor.submitJob(context(1L), template);
+
+ Assertions.assertEquals("default/sparkapp/team-a/gravitino-job-1",
jobExecutionId);
+ Assertions.assertNull(sparkApplication("gravitino-job-1"));
+ Assertions.assertEquals(
+ JobHandle.Status.QUEUED,
executor.getJobExecutionInfo(jobExecutionId).status());
+ executor.cancelJob(jobExecutionId);
+ Assertions.assertThrows(
+ NoSuchJobException.class, () ->
executor.getJobExecutionInfo(jobExecutionId));
+ }
+
+ @Test
+ public void testRejectedByKubernetes() {
+ for (int code : new int[] {400, 403, 404, 422}) {
+ server
+ .expect()
+ .post()
+ .withPath(APPS_PATH)
+ .andReturn(
+ code, new StatusBuilder().withCode(code).withMessage("rejected "
+ code).build())
+ .once();
+
+ IllegalArgumentException e =
+ Assertions.assertThrows(
+ IllegalArgumentException.class,
+ () -> executor.submitJob(context(1L), sparkTemplate()));
+ Assertions.assertTrue(e.getMessage().contains("rejected " + code),
e.getMessage());
+ }
+
+ // Other failures aren't the job's fault. The client retries server
errors, so use another.
+ server
+ .expect()
+ .post()
+ .withPath(APPS_PATH)
+ .andReturn(410, new StatusBuilder().withCode(410).build())
+ .once();
+ Assertions.assertThrows(
+ KubernetesClientException.class, () -> executor.submitJob(context(1L),
sparkTemplate()));
+ }
+
+ @Test
+ public void testFailJobWithoutStatus() {
+ String jobExecutionId = executor.submitJob(context(1L), sparkTemplate());
+ now.set(now.get().plusSeconds(601));
+
+ // The SparkApplication can't be deleted, so the job isn't failed yet.
+ server
+ .expect()
+ .delete()
+ .withPath(APPS_PATH + "/gravitino-job-1")
+ .andReturn(403, new StatusBuilder().withCode(403).build())
+ .once();
+ Assertions.assertThrows(
+ KubernetesClientException.class, () ->
executor.getJobExecutionInfo(jobExecutionId));
+ Assertions.assertNotNull(sparkApplication("gravitino-job-1"));
+
+ // Deleted, so that the operator doesn't run it after the job failed.
+ Assertions.assertEquals(
+ JobHandle.Status.FAILED,
executor.getJobExecutionInfo(jobExecutionId).status());
+ Assertions.assertNull(sparkApplication("gravitino-job-1"));
+ }
+
+ @Test
+ public void testDeleteAliveDriverOfFailedJob() {
+ String jobExecutionId = executor.submitJob(context(1L), sparkTemplate());
+ setStates(
+ "gravitino-job-1",
+ "Submitted",
+ "DriverRequested",
+ "DriverStartTimedOut",
+ "TerminatedWithoutReleaseResources");
+ // The pending driver pod is retained, and would run once it can be
scheduled.
+ createDriverPod("driver", "spark-kubernetes-driver");
+
+ Assertions.assertEquals(
+ JobHandle.Status.FAILED,
executor.getJobExecutionInfo(jobExecutionId).status());
+
Assertions.assertNull(client.pods().inNamespace("default").withName("driver").get());
+ Assertions.assertNotNull(sparkApplication("gravitino-job-1"));
+ }
+
+ @Test
+ public void testKeepTerminatedDriverOfFailedJob() {
+ String jobExecutionId = executor.submitJob(context(1L), sparkTemplate());
+ setStates("gravitino-job-1", "Submitted", "DriverReady", "Failed");
+ client
+ .pods()
+ .inNamespace("default")
+ .resource(
+ new PodBuilder()
+ .withNewMetadata()
+ .addToOwnerReferences(ownerReference("gravitino-job-1"))
+ .withName("driver")
+ .addToLabels(SparkApplicationUtils.LABEL_SPARK_APP_NAME,
"gravitino-job-1")
+ .addToLabels(SparkApplicationUtils.LABEL_SPARK_ROLE, "driver")
+ .endMetadata()
+ .withNewStatus()
+ .withPhase("Failed")
+ .endStatus()
+ .build())
+ .create();
+
+ Assertions.assertEquals(
+ JobHandle.Status.FAILED,
executor.getJobExecutionInfo(jobExecutionId).status());
+ // Kept, so that its output can still be read.
+
Assertions.assertNotNull(client.pods().inNamespace("default").withName("driver").get());
+ }
+
+ @Test
+ public void testIsUnreachable() {
+ Assertions.assertTrue(
+ K8sJobExecutor.isUnreachable(
+ new KubernetesClientException("failed", new IOException(new
ConnectException()))));
+ Assertions.assertTrue(
+ K8sJobExecutor.isUnreachable(
+ new KubernetesClientException(
+ "failed", new IllegalStateException(new IOException("timed
out")))));
+ // The API server answered, or the failure has nothing to do with the
connection.
+ Assertions.assertFalse(
+ K8sJobExecutor.isUnreachable(
+ new KubernetesClientException(new
StatusBuilder().withCode(503).build())));
+ Assertions.assertFalse(K8sJobExecutor.isUnreachable(new
KubernetesClientException("failed")));
+ // A response came back, only not a JSON one, for example the login page
of a proxy.
+ Assertions.assertFalse(
+ K8sJobExecutor.isUnreachable(
+ new KubernetesClientException(
+ "failed", new JsonParseException((JsonParser) null,
"<html>"))));
+ Assertions.assertFalse(
+ K8sJobExecutor.isUnreachable(
+ new KubernetesClientException("failed", new
IllegalStateException())));
+ }
+
+ @Test
+ public void testFailFastWhileApiServerUnreachable() {
+ String job1 = "default/sparkapp/default/gravitino-job-1";
+ String job2 = "default/sparkapp/other/gravitino-job-2";
+ // With and without the namespace listing: nothing listens on the port.
+ for (String statusCacheTtlMs : new String[] {"60000", "0"}) {
+ try (KubernetesClient unreachable =
+ new KubernetesClientBuilder()
+ .withConfig(
+ new ConfigBuilder(Config.empty())
+ .withMasterUrl("http://127.0.0.1:1";)
+ .withRequestRetryBackoffLimit(0)
+ .build())
+ .build()) {
+ executor.initialize(
+ TestSparkApplicationUtils.newConfigs(
+ ImmutableMap.of(K8sJobExecutorConfigs.STATUS_CACHE_TTL_MS,
statusCacheTtlMs)),
+ unreachable);
+
+ KubernetesClientException first =
+ Assertions.assertThrows(
+ KubernetesClientException.class, () ->
executor.getJobExecutionInfo(job1));
+ Assertions.assertTrue(K8sJobExecutor.isUnreachable(first),
first.toString());
+
+ // The other jobs fail right away, whatever their namespace, without
another request.
+ for (String jobExecutionId : new String[] {job2, job1}) {
+ KubernetesClientException next =
+ Assertions.assertThrows(
+ KubernetesClientException.class,
+ () -> executor.getJobExecutionInfo(jobExecutionId));
+ Assertions.assertTrue(next.getMessage().contains("isn't queried
again"), next.toString());
+ Assertions.assertSame(first, next.getCause());
+ }
+
+ // Once the backoff is over, the API server is tried again.
+ now.set(now.get().plusMillis(K8sJobExecutor.UNREACHABLE_BACKOFF_MS));
+ KubernetesClientException again =
+ Assertions.assertThrows(
+ KubernetesClientException.class, () ->
executor.getJobExecutionInfo(job2));
+ Assertions.assertFalse(
+ again.getMessage().contains("isn't queried again"),
again.toString());
+ Assertions.assertNotSame(first, again);
+ now.set(now.get().plusMillis(K8sJobExecutor.UNREACHABLE_BACKOFF_MS));
+ }
+ }
+ }
+
+ @Test
+ public void testApiServerErrorDoesNotFailFast() {
+ String jobExecutionId = executor.submitJob(context(1L), sparkTemplate());
+ server
+ .expect()
+ .get()
+ .withPath(APPS_PATH + "/gravitino-job-1")
+ .andReturn(403, new StatusBuilder().withCode(403).build())
+ .once();
+
+ // The API server answered, so the next query goes to it again.
+ Assertions.assertThrows(
+ KubernetesClientException.class, () ->
executor.getJobExecutionInfo(jobExecutionId));
+ Assertions.assertEquals(
+ JobHandle.Status.QUEUED,
executor.getJobExecutionInfo(jobExecutionId).status());
+ }
+
+ @Test
+ public void testDriverLookupFailureKeepsFailedJob() {
+ String jobExecutionId = executor.submitJob(context(1L), sparkTemplate());
+ setStates(
+ "gravitino-job-1",
+ "Submitted",
+ "DriverRequested",
+ "DriverStartTimedOut",
+ "TerminatedWithoutReleaseResources");
+ createDriverPod("driver", "spark-kubernetes-driver");
+ // Listing the pods fails once, it can't be told whether the driver still
runs.
+ server
+ .expect()
+ .get()
+ .withPath(
+ "/api/v1/namespaces/default/pods?labelSelector="
+ +
"spark.operator%2Fspark-app-name%3Dgravitino-job-1%2Cspark-role%3Ddriver")
+ .andReturn(403, new StatusBuilder().withCode(403).build())
+ .once();
+
+ // The job isn't failed while its pending driver could still run.
+ Assertions.assertThrows(
+ KubernetesClientException.class, () ->
executor.getJobExecutionInfo(jobExecutionId));
+
Assertions.assertNotNull(client.pods().inNamespace("default").withName("driver").get());
+
+ Assertions.assertEquals(
+ JobHandle.Status.FAILED,
executor.getJobExecutionInfo(jobExecutionId).status());
+
Assertions.assertNull(client.pods().inNamespace("default").withName("driver").get());
+ }
+
+ @Test
+ public void testKeepDriverWithSidecarOfFailedJob() {
+ String jobExecutionId = executor.submitJob(context(1L), sparkTemplate());
+ setStates("gravitino-job-1", "Submitted", "DriverReady", "Failed");
+ // The driver exited, a sidecar keeps the pod running.
+ client
+ .pods()
+ .inNamespace("default")
+ .resource(
+ new PodBuilder()
+ .withNewMetadata()
+ .addToOwnerReferences(ownerReference("gravitino-job-1"))
+ .withName("driver")
+ .addToLabels(SparkApplicationUtils.LABEL_SPARK_APP_NAME,
"gravitino-job-1")
+ .addToLabels(SparkApplicationUtils.LABEL_SPARK_ROLE, "driver")
+ .endMetadata()
+ .withNewStatus()
+ .withPhase("Running")
+ .addNewContainerStatus()
+ .withName("spark-kubernetes-driver")
+ .withNewState()
+ .withNewTerminated()
+ .withExitCode(1)
+ .endTerminated()
+ .endState()
+ .endContainerStatus()
+ .addNewContainerStatus()
+ .withName("istio-proxy")
+ .withNewState()
+ .withNewRunning()
+ .endRunning()
+ .endState()
+ .endContainerStatus()
+ .endStatus()
+ .build())
+ .create();
+
+ Assertions.assertEquals(
+ JobHandle.Status.FAILED,
executor.getJobExecutionInfo(jobExecutionId).status());
+ // Kept, so that its output can still be read.
+
Assertions.assertNotNull(client.pods().inNamespace("default").withName("driver").get());
+ }
+
+ @Test
+ public void testDriverPodOfAnotherSparkApplicationIsIgnored() {
+ String jobExecutionId = executor.submitJob(context(1L), sparkTemplate());
+ setStates("gravitino-job-1", "Submitted", "DriverRequested",
"DriverStartTimedOut");
+ // A pending pod with the same labels, owned by another SparkApplication.
+ client
+ .pods()
+ .inNamespace("default")
+ .resource(
+ new PodBuilder()
+ .withNewMetadata()
+ .addToOwnerReferences(
+ new OwnerReferenceBuilder()
+ .withApiVersion("spark.apache.org/v1")
+ .withKind("SparkApplication")
+ .withName("gravitino-job-1")
+ .withUid("another-uid")
+ .build())
+ .withName("other-driver")
+ .addToLabels(SparkApplicationUtils.LABEL_SPARK_APP_NAME,
"gravitino-job-1")
+ .addToLabels(SparkApplicationUtils.LABEL_SPARK_ROLE, "driver")
+ .endMetadata()
+ .withNewStatus()
+ .withPhase("Pending")
+ .endStatus()
+ .build())
+ .create();
+
+ Assertions.assertEquals(
+ JobHandle.Status.FAILED,
executor.getJobExecutionInfo(jobExecutionId).status());
+
Assertions.assertNotNull(client.pods().inNamespace("default").withName("other-driver").get());
+ }
+
+ @Test
+ public void testJobWithoutStatusCancelledMeanwhile() {
+ executor.initialize(
+ TestSparkApplicationUtils.newConfigs(
+ ImmutableMap.of(K8sJobExecutorConfigs.STATUS_CACHE_TTL_MS,
"3600000")),
+ client);
+ String jobExecutionId = executor.submitJob(context(1L), sparkTemplate());
+ Assertions.assertEquals(
+ JobHandle.Status.QUEUED,
executor.getJobExecutionInfo(jobExecutionId).status());
+
+ // Another server cancels it, while this one still has it without status
in its cache.
+ GenericKubernetesResource app = sparkApplication("gravitino-job-1");
+ app.getMetadata()
+
.setAnnotations(ImmutableMap.of(K8sJobResourceUtils.ANNOTATION_CANCEL_REQUESTED,
"true"));
+ app.getMetadata().setFinalizers(ImmutableList.of("foregroundDeletion"));
+ app.getMetadata().setDeletionTimestamp(Instant.now().toString());
+ client
+ .genericKubernetesResources(SparkApplicationUtils.SPARK_APPLICATION)
+ .inNamespace("default")
+ .resource(app)
+ .update();
+ now.set(now.get().plusSeconds(601));
+
+ Assertions.assertEquals(
+ JobHandle.Status.CANCELLED,
executor.getJobExecutionInfo(jobExecutionId).status());
+ }
+
+ @Test
+ public void testAdoptNoResourceBeingDeleted() {
+ // A SparkApplication of the same job, but being deleted.
+ GenericKubernetesResource deleting =
+ SparkApplicationUtils.buildSparkApplication(
+ context(1L), sparkTemplate(),
TestSparkApplicationUtils.newConfigs(ImmutableMap.of()));
+
deleting.getMetadata().setFinalizers(ImmutableList.of("foregroundDeletion"));
+ client
+ .genericKubernetesResources(SparkApplicationUtils.SPARK_APPLICATION)
+ .inNamespace("default")
+ .resource(deleting)
+ .create();
+ GenericKubernetesResource created = sparkApplication("gravitino-job-1");
+ created.getMetadata().setDeletionTimestamp(Instant.now().toString());
+ client
+ .genericKubernetesResources(SparkApplicationUtils.SPARK_APPLICATION)
+ .inNamespace("default")
+ .resource(created)
+ .update();
+
+ Assertions.assertThrows(
+ IllegalStateException.class, () -> executor.submitJob(context(1L),
sparkTemplate()));
+ }
+
+ @Test
+ public void testSubmitWithoutResponse() {
+ // Whether the SparkApplication was created can't be told, the submission
fails anyway.
+ try (KubernetesClient unreachable =
+ new KubernetesClientBuilder()
+ .withConfig(
+ new ConfigBuilder(Config.empty())
+ .withMasterUrl("http://127.0.0.1:1";)
+ .withRequestRetryBackoffLimit(0)
+ .build())
+ .build()) {
+
executor.initialize(TestSparkApplicationUtils.newConfigs(ImmutableMap.of()),
unreachable);
+ KubernetesClientException e =
+ Assertions.assertThrows(
+ KubernetesClientException.class,
+ () -> executor.submitJob(context(1L), sparkTemplate()));
+ Assertions.assertTrue(K8sJobExecutor.isUnreachable(e), e.toString());
+ }
+ }
+
+ @Test
+ public void testStdoutOfRenamedDriverContainer() throws InterruptedException
{
+ String jobExecutionId = executor.submitJob(context(1L), sparkTemplate());
+ GenericKubernetesResource app = sparkApplication("gravitino-job-1");
+ Map<String, Object> spec =
+ new HashMap<>((Map<String, Object>)
app.getAdditionalProperties().get("spec"));
+ Map<String, Object> sparkConf = new HashMap<>((Map<String, Object>)
spec.get("sparkConf"));
+ sparkConf.put(SparkApplicationUtils.SPARK_DRIVER_CONTAINER_NAME, "main");
+ spec.put("sparkConf", sparkConf);
+ app.setAdditionalProperty("spec", spec);
+ client
+ .genericKubernetesResources(SparkApplicationUtils.SPARK_APPLICATION)
+ .inNamespace("default")
+ .resource(app)
+ .update();
+ createDriverPod("driver", "main", "sidecar");
+
+ executor.getJobStdout(jobExecutionId, 10, 1024);
+
+ RecordedRequest log = null;
+ RecordedRequest request;
+ while ((request = server.takeRequest(10, TimeUnit.MILLISECONDS)) != null) {
+ if (request.getPath().contains("/pods/driver/log")) {
+ log = request;
+ }
+ }
+ Assertions.assertNotNull(log);
+ Assertions.assertTrue(log.getPath().contains("container=main"),
log.getPath());
+ }
+
+ @Test
+ public void testStdoutBeforeInitialize() {
+ Assertions.assertEquals(
+ ImmutableList.of(),
+ new
K8sJobExecutor().getJobStdout("default/sparkapp/default/gravitino-job-1", 10,
1024));
+ }
+
+ @Test
+ public void testListFailureFallsBackToGet() throws InterruptedException {
+ executor.initialize(
+ TestSparkApplicationUtils.newConfigs(
+ ImmutableMap.of(K8sJobExecutorConfigs.STATUS_CACHE_TTL_MS,
"60000")),
+ client);
+ String jobExecutionId = executor.submitJob(context(1L), sparkTemplate());
+ server
+ .expect()
+ .get()
+ .withPath(APPS_PATH +
"?labelSelector=app.kubernetes.io%2Fmanaged-by%3Dgravitino")
+ .andReturn(403, new StatusBuilder().withCode(403).build())
+ .always();
+
+ Assertions.assertEquals(
+ JobHandle.Status.QUEUED,
executor.getJobExecutionInfo(jobExecutionId).status());
+ // Each query gets the SparkApplication itself, so it sees the new state
right away.
+ setStates("gravitino-job-1", "Submitted", "DriverReady");
+ Assertions.assertEquals(
+ JobHandle.Status.STARTED,
executor.getJobExecutionInfo(jobExecutionId).status());
+
+ // The namespace is listed once, then the failure is cached.
+ int lists = 0;
+ RecordedRequest request;
+ while ((request = server.takeRequest(10, TimeUnit.MILLISECONDS)) != null) {
+ if ("GET".equals(request.getMethod()) &&
request.getPath().contains("labelSelector")) {
+ lists++;
+ }
+ }
+ Assertions.assertEquals(1, lists);
+ }
+
+ @Test
+ public void testResubmitSameJob() {
+ String first = executor.submitJob(context(1L), sparkTemplate());
+ String second = executor.submitJob(context(1L), sparkTemplate());
+ Assertions.assertEquals(first, second);
+ }
+
+ @Test
+ public void testRejectResourceOfAnotherJob() {
+ GenericKubernetesResource other =
+ SparkApplicationUtils.buildSparkApplication(
+ context(2L), sparkTemplate(),
TestSparkApplicationUtils.newConfigs(ImmutableMap.of()));
+ other.getMetadata().setName("gravitino-job-1");
+ client
+ .genericKubernetesResources(SparkApplicationUtils.SPARK_APPLICATION)
+ .inNamespace("default")
+ .resource(other)
+ .create();
+
+ Assertions.assertThrows(
+ IllegalStateException.class, () -> executor.submitJob(context(1L),
sparkTemplate()));
+ }
+
+ @Test
+ public void testRejectShellJob() {
+ ShellJobTemplate template =
+
ShellJobTemplate.builder().withName("shell-job").withExecutable("echo").build();
+ Assertions.assertThrows(
+ IllegalArgumentException.class, () -> executor.submitJob(context(1L),
template));
+ }
+
+ @Test
+ public void testNoSuchJob() {
+ Assertions.assertThrows(
+ NoSuchJobException.class,
+ () ->
executor.getJobExecutionInfo("default/sparkapp/default/gravitino-job-9"));
+ Assertions.assertThrows(
+ NoSuchJobException.class,
+ () -> executor.cancelJob("default/sparkapp/default/gravitino-job-9"));
+ Assertions.assertThrows(
+ NoSuchJobException.class, () ->
executor.getJobExecutionInfo("gravitino-job-9"));
+
+ String jobExecutionId = executor.submitJob(context(1L), sparkTemplate());
+ Assertions.assertThrows(
+ NoSuchJobException.class,
+ () ->
executor.getJobExecutionInfo(jobExecutionId.replace("/sparkapp/", "/job/")));
+ }
+
+ @Test
+ public void testJobOfAnotherCluster() {
+ // For example, submitted before the cluster was renamed. The job isn't
missing, which would
+ // end it while it may still run, so its queries fail and leave it as it
is.
+ String jobExecutionId =
+ executor.submitJob(context(1L),
sparkTemplate()).replaceFirst("^default/", "other/");
+ Assertions.assertThrows(
+ IllegalStateException.class, () ->
executor.getJobExecutionInfo(jobExecutionId));
+ Assertions.assertThrows(IllegalStateException.class, () ->
executor.cancelJob(jobExecutionId));
+ Assertions.assertFalse(
+
K8sJobResourceUtils.isCancelRequested(sparkApplication("gravitino-job-1")));
+ Assertions.assertEquals(ImmutableList.of(),
executor.getJobStdout(jobExecutionId, 10, 1024));
+ }
+
+ @Test
+ public void testIgnoreResourceOfAnotherJob() {
+ // The SparkApplication of the job is gone, and its name is used by
another job's, for example
+ // of another Gravitino deployment with the same name prefix in the
namespace.
+ String jobExecutionId = executor.submitJob(context(1L), sparkTemplate());
+ client
+ .genericKubernetesResources(SparkApplicationUtils.SPARK_APPLICATION)
+ .inNamespace("default")
+ .withName("gravitino-job-1")
+ .delete();
+ GenericKubernetesResource other =
+ SparkApplicationUtils.buildSparkApplication(
+ context(2L), sparkTemplate(),
TestSparkApplicationUtils.newConfigs(ImmutableMap.of()));
+ other.getMetadata().setName("gravitino-job-1");
+ client
+ .genericKubernetesResources(SparkApplicationUtils.SPARK_APPLICATION)
+ .inNamespace("default")
+ .resource(other)
+ .create();
+ setStates("gravitino-job-1", "Submitted", "DriverReady");
+ createDriverPod("driver", "spark-kubernetes-driver");
+ server
+ .expect()
+ .get()
+
.withPath("/api/v1/namespaces/default/pods/driver/log?pretty=false&tailLines=10")
+ .andReturn(200, "other job\n")
+ .always();
+
+ Assertions.assertThrows(
+ NoSuchJobException.class, () ->
executor.getJobExecutionInfo(jobExecutionId));
+ Assertions.assertThrows(NoSuchJobException.class, () ->
executor.cancelJob(jobExecutionId));
+ Assertions.assertEquals(ImmutableList.of(),
executor.getJobStdout(jobExecutionId, 10, 1024));
+ GenericKubernetesResource kept = sparkApplication("gravitino-job-1");
+ Assertions.assertNotNull(kept);
+ Assertions.assertFalse(K8sJobResourceUtils.isCancelRequested(kept));
+
Assertions.assertNotNull(client.pods().inNamespace("default").withName("driver").get());
+
+ // Nor one that isn't managed by Gravitino.
+ kept.getMetadata().setLabels(ImmutableMap.of());
+ client
+ .genericKubernetesResources(SparkApplicationUtils.SPARK_APPLICATION)
+ .inNamespace("default")
+ .resource(kept)
+ .update();
+ Assertions.assertThrows(
+ NoSuchJobException.class, () ->
executor.getJobExecutionInfo(jobExecutionId));
+ Assertions.assertEquals(ImmutableList.of(),
executor.getJobStdout(jobExecutionId, 10, 1024));
+ }
+
+ @Test
+ public void testKeepJobWithUnreadableStatus() {
+ String jobExecutionId = executor.submitJob(context(1L), sparkTemplate());
+ // A status this version can't read, for example of a newer operator.
+ GenericKubernetesResource app = sparkApplication("gravitino-job-1");
+ app.setAdditionalProperty("status", ImmutableMap.of("phase", "Running"));
+ client
+ .genericKubernetesResources(SparkApplicationUtils.SPARK_APPLICATION)
+ .inNamespace("default")
+ .resource(app)
+ .update();
+ now.set(now.get().plusSeconds(601));
+
+ Assertions.assertEquals(
+ JobHandle.Status.QUEUED,
executor.getJobExecutionInfo(jobExecutionId).status());
+ Assertions.assertNotNull(sparkApplication("gravitino-job-1"));
+ }
+
+ @Test
+ public void testJobWithoutStatusGoneMeanwhile() {
+ executor.initialize(
+ TestSparkApplicationUtils.newConfigs(
+ ImmutableMap.of(K8sJobExecutorConfigs.STATUS_CACHE_TTL_MS,
"3600000")),
+ client);
+ String jobExecutionId = executor.submitJob(context(1L), sparkTemplate());
+ Assertions.assertEquals(
+ JobHandle.Status.QUEUED,
executor.getJobExecutionInfo(jobExecutionId).status());
+
+ // Another server cancels it and it is gone, while this one still has it
in its cache. It is
+ // reported missing rather than failed, which settles a cancelling job as
cancelled.
+ client
+ .genericKubernetesResources(SparkApplicationUtils.SPARK_APPLICATION)
+ .inNamespace("default")
+ .withName("gravitino-job-1")
+ .delete();
+ now.set(now.get().plusSeconds(601));
+
+ Assertions.assertThrows(
+ NoSuchJobException.class, () ->
executor.getJobExecutionInfo(jobExecutionId));
+ }
+
+ @Test
+ public void testCancelWhoseDeletionResponseIsLost() {
+ String jobExecutionId = executor.submitJob(context(1L), sparkTemplate());
+ setStates("gravitino-job-1", "Submitted", "DriverReady");
+ GenericKubernetesResource running = sparkApplication("gravitino-job-1");
+ // The deletion takes effect, but its response is lost. From then on the
SparkApplication is
+ // marked and being deleted.
+ GenericKubernetesResource deleting = sparkApplication("gravitino-job-1");
+ deleting
+ .getMetadata()
+
.setAnnotations(ImmutableMap.of(K8sJobResourceUtils.ANNOTATION_CANCEL_REQUESTED,
"true"));
+ deleting.getMetadata().setDeletionTimestamp(Instant.now().toString());
+ AtomicBoolean deleted = new AtomicBoolean();
+ server
+ .expect()
+ .get()
+ .withPath(APPS_PATH + "/gravitino-job-1")
+ .andReply(200, request -> deleted.get() ? deleting : running)
+ .always();
+ server
+ .expect()
+ .delete()
+ .withPath(APPS_PATH + "/gravitino-job-1")
+ .andReply(
+ 403,
+ request -> {
+ deleted.set(true);
+ return new StatusBuilder().withCode(403).build();
+ })
+ .once();
+
+ // The job is being cancelled, so the cancellation succeeds and the job
keeps its mark.
+ executor.cancelJob(jobExecutionId);
+
+ Assertions.assertTrue(deleted.get());
+ Assertions.assertTrue(
+
K8sJobResourceUtils.isCancelRequested(sparkApplication("gravitino-job-1")));
+ Assertions.assertEquals(
+ JobHandle.Status.CANCELLED,
executor.getJobExecutionInfo(jobExecutionId).status());
+ }
+
+ @Test
+ public void testCancel() throws InterruptedException {
+ String jobExecutionId = executor.submitJob(context(1L), sparkTemplate());
+ setStates("gravitino-job-1", "Submitted", "DriverReady");
+ // Keep the SparkApplication, as it stays until its pods are deleted.
+ server
+ .expect()
+ .delete()
+ .withPath(APPS_PATH + "/gravitino-job-1")
+ .andReturn(200, sparkApplication("gravitino-job-1"))
+ .always();
+
+ executor.cancelJob(jobExecutionId);
+
+ GenericKubernetesResource app = sparkApplication("gravitino-job-1");
+ Assertions.assertTrue(K8sJobResourceUtils.isCancelRequested(app));
+ RecordedRequest delete = lastRequest("DELETE");
+
Assertions.assertTrue(delete.getBody().readUtf8().contains("\"Foreground\""));
+
+ // The job is only cancelled once the deletion takes effect, and a status
query never deletes.
+ Assertions.assertEquals(
+ JobHandle.Status.STARTED,
executor.getJobExecutionInfo(jobExecutionId).status());
+ Assertions.assertNull(lastRequest("DELETE"));
+ }
+
+ @Test
+ public void testCancelledJobFinishedWithDriver() {
+ String jobExecutionId = executor.submitJob(context(1L), sparkTemplate());
+ setStates("gravitino-job-1", "Submitted", "DriverReady");
+ // Cancelled while running: being deleted, with no outcome recorded by the
operator.
+ GenericKubernetesResource app = sparkApplication("gravitino-job-1");
+ app.getMetadata()
+
.setAnnotations(ImmutableMap.of(K8sJobResourceUtils.ANNOTATION_CANCEL_REQUESTED,
"true"));
+ // A finalizer keeps it while it is being deleted, as the Foreground
deletion does.
+ app.getMetadata().setFinalizers(ImmutableList.of("foregroundDeletion"));
+ app.getMetadata().setDeletionTimestamp(Instant.now().toString());
+ client
+ .genericKubernetesResources(SparkApplicationUtils.SPARK_APPLICATION)
+ .inNamespace("default")
+ .resource(app)
+ .update();
+ Assertions.assertNotNull(
+
sparkApplication("gravitino-job-1").getMetadata().getDeletionTimestamp());
+
+ // The driver terminated after the deletion was requested.
+ Instant driverStartedAt =
Instant.now().plusSeconds(60).truncatedTo(ChronoUnit.SECONDS);
+ Instant driverFinishedAt = driverStartedAt.plusSeconds(30);
+ client
+ .pods()
+ .inNamespace("default")
+ .resource(
+ new PodBuilder()
+ .withNewMetadata()
+ .addToOwnerReferences(ownerReference("gravitino-job-1"))
+ .withName("driver")
+ .addToLabels(SparkApplicationUtils.LABEL_SPARK_APP_NAME,
"gravitino-job-1")
+ .addToLabels(SparkApplicationUtils.LABEL_SPARK_ROLE, "driver")
+ .endMetadata()
+ .withNewStatus()
+ .withPhase("Failed")
+ .addNewContainerStatus()
+ .withName("spark-kubernetes-driver")
+ .withNewState()
+ .withNewTerminated()
+ .withStartedAt(driverStartedAt.toString())
+ .withFinishedAt(driverFinishedAt.toString())
+ .endTerminated()
+ .endState()
+ .endContainerStatus()
+ .endStatus()
+ .build())
+ .create();
+
+ JobExecutionInfo cancelled = executor.getJobExecutionInfo(jobExecutionId);
+ Assertions.assertEquals(JobHandle.Status.CANCELLED, cancelled.status());
+ Assertions.assertEquals(driverStartedAt, cancelled.startedAt());
+ Assertions.assertEquals(driverFinishedAt, cancelled.finishedAt());
+ }
+
+ @Test
+ public void testCancelAgainAfterFailedDeletion() {
+ String jobExecutionId = executor.submitJob(context(1L), sparkTemplate());
+ setStates("gravitino-job-1", "Submitted", "DriverReady");
+ server
+ .expect()
+ .delete()
+ .withPath(APPS_PATH + "/gravitino-job-1")
+ .andReturn(403, new StatusBuilder().withCode(403).build())
+ .once();
+
+ // The cancellation fails, and the job keeps running, without the cancel
mark, so that it
+ // isn't reported as cancelled once it finishes by itself.
+ Assertions.assertThrows(
+ KubernetesClientException.class, () ->
executor.cancelJob(jobExecutionId));
+ Assertions.assertFalse(
+
K8sJobResourceUtils.isCancelRequested(sparkApplication("gravitino-job-1")));
+ Assertions.assertEquals(
+ JobHandle.Status.STARTED,
executor.getJobExecutionInfo(jobExecutionId).status());
+
+ // Cancelling it again deletes it.
+ executor.cancelJob(jobExecutionId);
+ Assertions.assertNull(sparkApplication("gravitino-job-1"));
+ Assertions.assertThrows(
+ NoSuchJobException.class, () ->
executor.getJobExecutionInfo(jobExecutionId));
+ }
+
+ @Test
+ public void testCancelWhoseDeletionFailsAfterTheJobIsGone() {
+ String jobExecutionId = executor.submitJob(context(1L), sparkTemplate());
+ setStates("gravitino-job-1", "Submitted", "DriverReady");
+ // The deletion takes effect and the SparkApplication is gone, but the
response is lost: it is
+ // read once to be marked, and is gone when read after the failed deletion.
+ server
+ .expect()
+ .get()
+ .withPath(APPS_PATH + "/gravitino-job-1")
+ .andReturn(200, sparkApplication("gravitino-job-1"))
+ .once();
+ server
+ .expect()
+ .delete()
+ .withPath(APPS_PATH + "/gravitino-job-1")
+ .andReturn(403, new StatusBuilder().withCode(403).build())
+ .once();
+ server
+ .expect()
+ .get()
+ .withPath(APPS_PATH + "/gravitino-job-1")
+ .andReturn(404, new StatusBuilder().withCode(404).build())
+ .always();
+
+ // The job is gone, so the cancellation succeeds.
+ executor.cancelJob(jobExecutionId);
+ }
+
+ @Test
+ public void testCancelConflictRetriesRunOut() {
+ String jobExecutionId = executor.submitJob(context(1L), sparkTemplate());
+ setStates("gravitino-job-1", "Submitted", "DriverReady");
+ server
+ .expect()
+ .put()
+ .withPath(APPS_PATH + "/gravitino-job-1")
+ .andReturn(409, new StatusBuilder().withCode(409).build())
+ .times(4);
+
+ KubernetesClientException e =
+ Assertions.assertThrows(
+ KubernetesClientException.class, () ->
executor.cancelJob(jobExecutionId));
+ Assertions.assertEquals(409, e.getCode());
+ // Not marked nor deleted, the job keeps running.
+ GenericKubernetesResource app = sparkApplication("gravitino-job-1");
+ Assertions.assertFalse(K8sJobResourceUtils.isCancelRequested(app));
+ Assertions.assertNull(app.getMetadata().getDeletionTimestamp());
+ Assertions.assertEquals(
+ JobHandle.Status.STARTED,
executor.getJobExecutionInfo(jobExecutionId).status());
+
+ // Once the conflicts are over, it can be cancelled.
+ executor.cancelJob(jobExecutionId);
+ Assertions.assertNull(sparkApplication("gravitino-job-1"));
+ }
+
+ @Test
+ public void testCancelJobThatJustStopped() {
+ String jobExecutionId = executor.submitJob(context(1L), sparkTemplate());
+ // The job succeeds between the read of the cancellation and its mark.
+ GenericKubernetesResource running = sparkApplication("gravitino-job-1");
+ setStates("gravitino-job-1", "Submitted", "DriverReady", "Succeeded");
+ server.expect().get().withPath(APPS_PATH +
"/gravitino-job-1").andReturn(200, running).once();
+
+ executor.cancelJob(jobExecutionId);
+
+ GenericKubernetesResource app = sparkApplication("gravitino-job-1");
+ Assertions.assertFalse(K8sJobResourceUtils.isCancelRequested(app));
+ Assertions.assertEquals(
+ JobHandle.Status.SUCCEEDED,
executor.getJobExecutionInfo(jobExecutionId).status());
+ }
+
+ @Test
+ public void testCancelStoppedJob() {
+ String jobExecutionId = executor.submitJob(context(1L), sparkTemplate());
+ setStates("gravitino-job-1", "Submitted", "DriverReady", "Succeeded");
+
+ executor.cancelJob(jobExecutionId);
+
+ GenericKubernetesResource app = sparkApplication("gravitino-job-1");
+ Assertions.assertFalse(K8sJobResourceUtils.isCancelRequested(app));
+ Assertions.assertEquals(
+ JobHandle.Status.SUCCEEDED,
executor.getJobExecutionInfo(jobExecutionId).status());
+ }
+
+ @Test
+ public void testStdout() {
+ String jobExecutionId = executor.submitJob(context(1L), sparkTemplate());
+ Assertions.assertEquals(ImmutableList.of(),
executor.getJobStdout(jobExecutionId, 10, 1024));
+
+ for (String name : new String[] {"driver", "executor"}) {
+ client
+ .pods()
+ .inNamespace("default")
+ .resource(
+ new PodBuilder()
+ .withNewMetadata()
+ .addToOwnerReferences(ownerReference("gravitino-job-1"))
+ .withName(name)
+ .addToLabels(SparkApplicationUtils.LABEL_SPARK_APP_NAME,
"gravitino-job-1")
+ .addToLabels(
+ SparkApplicationUtils.LABEL_SPARK_ROLE,
+ name.equals("executor") ? "executor" : "driver")
+ .endMetadata()
+ .build())
+ .create();
+ }
+ server
+ .expect()
+ .get()
+
.withPath("/api/v1/namespaces/default/pods/driver/log?pretty=false&tailLines=10")
+ .andReturn(200, "line 1\nline 2\n")
+ .always();
+
+ Assertions.assertEquals(
+ ImmutableList.of("line 1", "line 2"),
executor.getJobStdout(jobExecutionId, 10, 1024));
+ Assertions.assertEquals(ImmutableList.of(),
executor.getJobStderr(jobExecutionId, 10, 1024));
+ Assertions.assertEquals(ImmutableList.of(),
executor.getJobStdout("invalid", 10, 1024));
+ }
+
+ @Test
+ public void testStdoutFailure() {
+ String jobExecutionId = executor.submitJob(context(1L), sparkTemplate());
+ createDriverPod("driver", "spark-kubernetes-driver");
+ server
+ .expect()
+ .get()
+
.withPath("/api/v1/namespaces/default/pods/driver/log?pretty=false&tailLines=10")
+ .andReturn(403, "forbidden")
+ .always();
+
+ Assertions.assertEquals(ImmutableList.of(),
executor.getJobStdout(jobExecutionId, 10, 1024));
+ }
+
+ @Test
+ public void testStdoutOfDriverWithSidecar() throws InterruptedException {
+ String jobExecutionId = executor.submitJob(context(1L), sparkTemplate());
+ createDriverPod("driver", "spark-kubernetes-driver", "sidecar");
+
+ executor.getJobStdout(jobExecutionId, 10, 1024);
+
+ RecordedRequest log = null;
+ RecordedRequest request;
+ while ((request = server.takeRequest(10, TimeUnit.MILLISECONDS)) != null) {
+ if (request.getPath().contains("/pods/driver/log")) {
+ log = request;
+ }
+ }
+ Assertions.assertNotNull(log);
+ Assertions.assertTrue(
+ log.getPath().contains("container=spark-kubernetes-driver"),
log.getPath());
+ }
+
+ private void createDriverPod(String name, String... containers) {
+ PodBuilder pod =
+ new PodBuilder()
+ .withNewMetadata()
+ .addToOwnerReferences(ownerReference("gravitino-job-1"))
+ .withName(name)
+ .addToLabels(SparkApplicationUtils.LABEL_SPARK_APP_NAME,
"gravitino-job-1")
+ .addToLabels(SparkApplicationUtils.LABEL_SPARK_ROLE, "driver")
+ .endMetadata()
+ .withNewSpec()
+ .endSpec();
+ for (String container : containers) {
+
pod.editSpec().addNewContainer().withName(container).endContainer().endSpec();
+ }
+ client.pods().inNamespace("default").resource(pod.build()).create();
+ }
+
+ /** The owner reference the operator sets on the driver pod of a
SparkApplication. */
+ private OwnerReference ownerReference(String appName) {
+ GenericKubernetesResource app = sparkApplication(appName);
+ return new OwnerReferenceBuilder()
+ .withApiVersion("spark.apache.org/v1")
+ .withKind("SparkApplication")
+ .withName(appName)
+ .withUid(app.getMetadata().getUid())
+ .build();
+ }
+
+ private GenericKubernetesResource sparkApplication(String name) {
+ return client
+ .genericKubernetesResources(SparkApplicationUtils.SPARK_APPLICATION)
+ .inNamespace("default")
+ .withName(name)
+ .get();
+ }
+
+ private void setStates(String name, String... states) {
+ Map<String, Object> history = new HashMap<>();
+ Instant time = Instant.parse("2026-09-30T00:00:00Z");
+ for (int i = 0; i < states.length; i++) {
+ history.put(
+ String.valueOf(i),
+ ImmutableMap.of(
+ "currentStateSummary",
+ states[i],
+ "lastTransitionTime",
+ time.plusSeconds(i).toString()));
+ }
+ GenericKubernetesResource app = sparkApplication(name);
+ app.setAdditionalProperty(
+ "status",
+ ImmutableMap.of(
+ "currentState",
+ history.get(String.valueOf(states.length - 1)),
+ "stateTransitionHistory",
+ history));
+ client
+ .genericKubernetesResources(SparkApplicationUtils.SPARK_APPLICATION)
+ .inNamespace("default")
+ .resource(app)
+ .update();
+ }
+
+ private RecordedRequest lastRequest(String method) throws
InterruptedException {
+ RecordedRequest last = null;
+ RecordedRequest request;
+ while ((request = server.takeRequest(10, TimeUnit.MILLISECONDS)) != null) {
+ if (method.equals(request.getMethod())) {
+ last = request;
+ }
+ }
+ return last;
+ }
+
+ private static JobContext context(long jobId) {
+ return new JobContext(jobId, "ml", new File("/tmp/job-" + jobId));
+ }
+
+ private static SparkJobTemplate sparkTemplate() {
+ return SparkJobTemplate.builder()
+ .withName("user-job")
+ .withExecutable("local:///opt/spark/examples/jars/spark-examples.jar")
+ .withClassName("org.apache.spark.examples.SparkPi")
+ .withArguments(ImmutableList.of("10"))
+ .build();
+ }
+}
diff --git
a/plugins/k8s-job-executor/src/test/java/org/apache/gravitino/job/k8s/TestK8sJobResourceUtils.java
b/plugins/k8s-job-executor/src/test/java/org/apache/gravitino/job/k8s/TestK8sJobResourceUtils.java
index 76c7661738..74104fc084 100644
---
a/plugins/k8s-job-executor/src/test/java/org/apache/gravitino/job/k8s/TestK8sJobResourceUtils.java
+++
b/plugins/k8s-job-executor/src/test/java/org/apache/gravitino/job/k8s/TestK8sJobResourceUtils.java
@@ -54,6 +54,27 @@ public class TestK8sJobResourceUtils {
Assertions.assertFalse(K8sJobResourceUtils.isOfJob(new ConfigMap(), 42L));
}
+ @Test
+ public void testIsJobResource() {
+
Assertions.assertTrue(K8sJobResourceUtils.isJobResource(resource("gravitino-job-42",
42L)));
+
Assertions.assertTrue(K8sJobResourceUtils.isJobResource(resource("other-prefix-42",
42L)));
+ // Another job's resource with this name, for example of another
deployment.
+
Assertions.assertFalse(K8sJobResourceUtils.isJobResource(resource("gravitino-job-42",
43L)));
+
Assertions.assertFalse(K8sJobResourceUtils.isJobResource(resource("gravitino-job-142",
42L)));
+ // Not managed by Gravitino.
+ ConfigMap unmanaged = resource("gravitino-job-42", 42L);
+
unmanaged.getMetadata().setLabels(ImmutableMap.of(K8sJobResourceUtils.LABEL_JOB_ID,
"42"));
+ Assertions.assertFalse(K8sJobResourceUtils.isJobResource(unmanaged));
+ Assertions.assertFalse(
+ K8sJobResourceUtils.isJobResource(
+ new ConfigMapBuilder()
+ .withNewMetadata()
+ .withName("gravitino-job-42")
+ .endMetadata()
+ .build()));
+ Assertions.assertFalse(K8sJobResourceUtils.isJobResource(new ConfigMap()));
+ }
+
@Test
public void testIsCancelRequested() {
Assertions.assertFalse(
@@ -75,6 +96,15 @@ public class TestK8sJobResourceUtils {
}
}
+ private static ConfigMap resource(String name, long jobId) {
+ return new ConfigMapBuilder()
+ .withNewMetadata()
+ .withName(name)
+ .withLabels(K8sJobResourceUtils.jobLabels(jobId))
+ .endMetadata()
+ .build();
+ }
+
private static ConfigMap annotated(String cancelRequested) {
return new ConfigMapBuilder()
.withNewMetadata()
diff --git
a/plugins/k8s-job-executor/src/test/java/org/apache/gravitino/job/k8s/TestNamespacedResourceCache.java
b/plugins/k8s-job-executor/src/test/java/org/apache/gravitino/job/k8s/TestNamespacedResourceCache.java
new file mode 100644
index 0000000000..aa358c0ee4
--- /dev/null
+++
b/plugins/k8s-job-executor/src/test/java/org/apache/gravitino/job/k8s/TestNamespacedResourceCache.java
@@ -0,0 +1,248 @@
+/*
+ * 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;
+
+import com.google.common.collect.ImmutableList;
+import io.fabric8.kubernetes.api.model.ConfigMap;
+import io.fabric8.kubernetes.api.model.ConfigMapBuilder;
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Optional;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.concurrent.atomic.AtomicReference;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+public class TestNamespacedResourceCache {
+
+ private final AtomicLong now = new AtomicLong(1_000L);
+
+ private final List<String> listed = new ArrayList<>();
+
+ @Test
+ public void testCachedWithinTtl() {
+ NamespacedResourceCache<ConfigMap> cache =
+ new NamespacedResourceCache<>(
+ 100L,
+ namespace -> {
+ listed.add(namespace);
+ return ImmutableList.of(resource(namespace, "a"),
resource(namespace, "b"));
+ },
+ e -> false,
+ now::get);
+
+ Assertions.assertEquals("a", cache.find("ns1",
"a").get().getMetadata().getName());
+ Assertions.assertEquals("b", cache.find("ns1",
"b").get().getMetadata().getName());
+ Assertions.assertFalse(cache.find("ns1", "c").isPresent());
+ Assertions.assertTrue(cache.find("ns2", "a").isPresent());
+ Assertions.assertEquals(ImmutableList.of("ns1", "ns2"), listed);
+
+ now.addAndGet(100L);
+ Assertions.assertTrue(cache.find("ns1", "a").isPresent());
+ Assertions.assertEquals(ImmutableList.of("ns1", "ns2", "ns1"), listed);
+
+ cache.invalidate("ns1");
+ Assertions.assertTrue(cache.find("ns1", "a").isPresent());
+ Assertions.assertEquals(ImmutableList.of("ns1", "ns2", "ns1", "ns1"),
listed);
+ }
+
+ @Test
+ public void testDisabled() {
+ NamespacedResourceCache<ConfigMap> cache =
+ new NamespacedResourceCache<>(
+ 0L,
+ namespace -> {
+ listed.add(namespace);
+ return ImmutableList.of(resource(namespace, "a"));
+ },
+ e -> false,
+ now::get);
+
+ Assertions.assertFalse(cache.find("ns1", "a").isPresent());
+ Assertions.assertTrue(listed.isEmpty());
+ }
+
+ @Test
+ public void testListFailureIsCached() {
+ NamespacedResourceCache<ConfigMap> cache =
+ new NamespacedResourceCache<>(
+ 100L,
+ namespace -> {
+ listed.add(namespace);
+ throw new IllegalStateException("forbidden");
+ },
+ e -> false,
+ now::get);
+
+ Assertions.assertFalse(cache.find("ns1", "a").isPresent());
+ // The namespace isn't listed again until the failure expires.
+ Assertions.assertFalse(cache.find("ns1", "b").isPresent());
+ Assertions.assertEquals(ImmutableList.of("ns1"), listed);
+
+ now.addAndGet(100L);
+ Assertions.assertFalse(cache.find("ns1", "a").isPresent());
+ Assertions.assertEquals(ImmutableList.of("ns1", "ns1"), listed);
+ }
+
+ @Test
+ public void testInvalidatedWhileListing() {
+ AtomicReference<NamespacedResourceCache<ConfigMap>> cache = new
AtomicReference<>();
+ cache.set(
+ new NamespacedResourceCache<>(
+ 100L,
+ namespace -> {
+ listed.add(namespace);
+ // A resource changes while the namespace is being listed.
+ cache.get().invalidate(namespace);
+ return ImmutableList.of(resource(namespace, "a"));
+ },
+ e -> false,
+ now::get));
+
+ Assertions.assertTrue(cache.get().find("ns1", "a").isPresent());
+ // The listing may predate the change, so it isn't cached.
+ Assertions.assertTrue(cache.get().find("ns1", "a").isPresent());
+ Assertions.assertEquals(ImmutableList.of("ns1", "ns1"), listed);
+ }
+
+ @Test
+ public void testFatalListFailureIsThrown() {
+ NamespacedResourceCache<ConfigMap> cache =
+ new NamespacedResourceCache<>(
+ 100L,
+ namespace -> {
+ listed.add(namespace);
+ throw new IllegalStateException("unreachable");
+ },
+ e -> e instanceof IllegalStateException,
+ now::get);
+
+ Assertions.assertThrows(IllegalStateException.class, () ->
cache.find("ns1", "a"));
+ // Not cached as an empty listing, so the next query fails the same way.
+ Assertions.assertThrows(IllegalStateException.class, () ->
cache.find("ns1", "a"));
+ Assertions.assertEquals(ImmutableList.of("ns1", "ns1"), listed);
+ }
+
+ @Test
+ public void testInvalidationOfAnotherNamespace() {
+ AtomicReference<NamespacedResourceCache<ConfigMap>> cache = new
AtomicReference<>();
+ cache.set(
+ new NamespacedResourceCache<>(
+ 100L,
+ namespace -> {
+ listed.add(namespace);
+ // Another namespace changes while this one is being listed.
+ cache.get().invalidate("other");
+ return ImmutableList.of(resource(namespace, "a"));
+ },
+ e -> false,
+ now::get));
+
+ Assertions.assertTrue(cache.get().find("ns1", "a").isPresent());
+ // The listing of ns1 is still cached.
+ Assertions.assertTrue(cache.get().find("ns1", "a").isPresent());
+ Assertions.assertEquals(ImmutableList.of("ns1"), listed);
+ }
+
+ @Test
+ public void testConcurrentFindsListOnce() throws Exception {
+ CountDownLatch listing = new CountDownLatch(1);
+ CountDownLatch release = new CountDownLatch(1);
+ AtomicInteger lists = new AtomicInteger();
+ NamespacedResourceCache<ConfigMap> cache =
+ new NamespacedResourceCache<>(
+ 100L,
+ namespace -> {
+ lists.incrementAndGet();
+ // Only the listing of ns1 takes a while.
+ if ("ns1".equals(namespace)) {
+ listing.countDown();
+ try {
+ release.await(10, TimeUnit.SECONDS);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+ }
+ return ImmutableList.of(resource(namespace, "a"));
+ },
+ e -> false,
+ now::get);
+
+ AtomicReference<Optional<ConfigMap>> first = new AtomicReference<>();
+ AtomicReference<Optional<ConfigMap>> second = new AtomicReference<>();
+ Thread firstFind = new Thread(() -> first.set(cache.find("ns1", "a")));
+ firstFind.start();
+ Assertions.assertTrue(listing.await(10, TimeUnit.SECONDS));
+
+ // The second find waits for the listing being made, instead of listing
again. Release the
+ // listing only once it is waiting, so that it can't find a listing
already made.
+ Thread secondFind = new Thread(() -> second.set(cache.find("ns1", "a")));
+ secondFind.start();
+ long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(10);
+ while (secondFind.getState() != Thread.State.WAITING && System.nanoTime()
< deadline) {
+ Thread.sleep(10);
+ }
+ Assertions.assertEquals(Thread.State.WAITING, secondFind.getState());
+ // Another namespace isn't held up by it.
+ Assertions.assertFalse(cache.find("ns2", "b").isPresent());
+ release.countDown();
+
+ firstFind.join(10_000);
+ secondFind.join(10_000);
+ Assertions.assertTrue(first.get().isPresent());
+ Assertions.assertTrue(second.get().isPresent());
+ // ns1 once, and ns2.
+ Assertions.assertEquals(2, lists.get());
+ }
+
+ @Test
+ public void testErrorWhileListing() {
+ AtomicInteger lists = new AtomicInteger();
+ NamespacedResourceCache<ConfigMap> cache =
+ new NamespacedResourceCache<>(
+ 100L,
+ namespace -> {
+ if (lists.incrementAndGet() == 1) {
+ throw new NoClassDefFoundError("boom");
+ }
+ return ImmutableList.of(resource(namespace, "a"));
+ },
+ e -> false,
+ now::get);
+
+ Assertions.assertThrows(NoClassDefFoundError.class, () ->
cache.find("ns1", "a"));
+ // The failed listing isn't left behind, the next query lists again
instead of waiting.
+ Assertions.assertTimeoutPreemptively(
+ Duration.ofSeconds(10), () -> Assertions.assertTrue(cache.find("ns1",
"a").isPresent()));
+ Assertions.assertEquals(2, lists.get());
+ }
+
+ private static ConfigMap resource(String namespace, String name) {
+ return new ConfigMapBuilder()
+ .withNewMetadata()
+ .withNamespace(namespace)
+ .withName(name)
+ .endMetadata()
+ .build();
+ }
+}
diff --git
a/plugins/k8s-job-executor/src/test/java/org/apache/gravitino/job/k8s/TestPodLogUtils.java
b/plugins/k8s-job-executor/src/test/java/org/apache/gravitino/job/k8s/TestPodLogUtils.java
new file mode 100644
index 0000000000..bdee1d4f98
--- /dev/null
+++
b/plugins/k8s-job-executor/src/test/java/org/apache/gravitino/job/k8s/TestPodLogUtils.java
@@ -0,0 +1,101 @@
+/*
+ * 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;
+
+import com.google.common.collect.ImmutableList;
+import java.io.ByteArrayInputStream;
+import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+import java.util.List;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+public class TestPodLogUtils {
+
+ @Test
+ public void testEmpty() throws IOException {
+ Assertions.assertEquals(ImmutableList.of(), read("", 10, 100));
+ }
+
+ @Test
+ public void testWithinLimits() throws IOException {
+ Assertions.assertEquals(ImmutableList.of("a", "b"), read("a\nb\n", 10,
100));
+ Assertions.assertEquals(ImmutableList.of("a", "", "b"), read("a\r\n\r\nb",
10, 100));
+ }
+
+ @Test
+ public void testMaxLines() throws IOException {
+ Assertions.assertEquals(ImmutableList.of("b", "c"), read("a\nb\nc\n", 2,
100));
+ }
+
+ @Test
+ public void testMaxBytesDropsPartialLine() throws IOException {
+ // The last 6 bytes are "bb\ncc\n", "bb" is a partial line.
+ Assertions.assertEquals(ImmutableList.of("cc"), read("aaa\nbbbb\ncc\n",
10, 6));
+ // The last 8 bytes are "bbbb\ncc\n", which start right after a newline.
+ Assertions.assertEquals(ImmutableList.of("bbbb", "cc"),
read("aaa\nbbbb\ncc\n", 10, 8));
+ }
+
+ @Test
+ public void testMaxBytesKeepsSingleLongLine() throws IOException {
+ Assertions.assertEquals(ImmutableList.of("6789"), read("0123456789\n", 10,
5));
+ }
+
+ @Test
+ public void testMaxBytesSkipsPartialCharacter() throws IOException {
+ // "é" is 2 bytes in UTF-8, the window starts at its second byte.
+ Assertions.assertEquals(ImmutableList.of("xy"), read("éxy", 10, 3));
+ }
+
+ @Test
+ public void testWindowOnlyHoldsLineTerminator() throws IOException {
+ Assertions.assertEquals(ImmutableList.of(), read("abc\n", 10, 1));
+ Assertions.assertEquals(ImmutableList.of(), read("abc\r\n", 10, 1));
+ Assertions.assertEquals(ImmutableList.of(), read("abc\r\n", 10, 2));
+ // An empty last line is a line of its own.
+ Assertions.assertEquals(ImmutableList.of(""), read("abc\n\n", 10, 1));
+ }
+
+ @Test
+ public void testHugeMaxBytes() throws IOException {
+ for (int maxBytes : new int[] {(1 << 30) - 1, 1 << 30, Integer.MAX_VALUE})
{
+ Assertions.assertEquals(ImmutableList.of("a", "b"), read("a\nb\n", 10,
maxBytes));
+ }
+ }
+
+ @Test
+ public void testLongLog() throws IOException {
+ String log =
+ IntStream.range(0, 100_000).mapToObj(i -> "line-" +
i).collect(Collectors.joining("\n"));
+ Assertions.assertEquals(
+ ImmutableList.of("line-99998", "line-99999"),
+ read(log, 1000, "line-99998\nline-99999".length()));
+ Assertions.assertEquals(ImmutableList.of("line-99999"), read(log, 1,
100_000));
+ List<String> lines = read(log, 1000, 1 << 20);
+ Assertions.assertEquals(1000, lines.size());
+ Assertions.assertEquals("line-99000", lines.get(0));
+ }
+
+ private static List<String> read(String log, int maxLines, int maxBytes)
throws IOException {
+ return PodLogUtils.readLastLines(
+ new ByteArrayInputStream(log.getBytes(StandardCharsets.UTF_8)),
maxLines, maxBytes);
+ }
+}
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
index 464bdec9cd..499a94ed5d 100644
---
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
@@ -90,6 +90,30 @@ public class TestSparkApplicationStatusUtils {
.status());
}
+ @Test
+ public void testUnreadableStatus() {
+ // A status without any state this version can read, for example of a
newer operator, keeps
+ // the job unfinished after the no-status timeout.
+ GenericKubernetesResource app = app();
+ app.setAdditionalProperty("status", ImmutableMap.of("phase", "Running"));
+ Assertions.assertTrue(SparkApplicationStatusUtils.hasStatus(app));
+ Assertions.assertFalse(SparkApplicationStatusUtils.hasState(app));
+ Assertions.assertEquals(
+ JobHandle.Status.QUEUED,
+ SparkApplicationStatusUtils.executionInfoOf(
+ app, null, CREATED_AT.plusMillis(NO_STATUS_TIMEOUT_MS + 1),
NO_STATUS_TIMEOUT_MS)
+ .status());
+
+ // An empty status is no status.
+ app.setAdditionalProperty("status", ImmutableMap.of());
+ Assertions.assertFalse(SparkApplicationStatusUtils.hasStatus(app));
+ 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"));
@@ -296,12 +320,40 @@ public class TestSparkApplicationStatusUtils {
Assertions.assertEquals(timeOf(1), cancelled.startedAt());
Assertions.assertEquals(deletedAt, cancelled.finishedAt());
- // The application stopped before the deletion.
+ // The application stopped by itself after it was marked, but before the
deletion.
GenericKubernetesResource stopped = app("Submitted", "DriverReady",
"Failed");
cancel(stopped);
+ stopped.getMetadata().setDeletionTimestamp(timeOf(3).toString());
JobExecutionInfo info = executionInfoOf(stopped);
- Assertions.assertEquals(JobHandle.Status.CANCELLED, info.status());
+ Assertions.assertEquals(JobHandle.Status.FAILED, info.status());
Assertions.assertEquals(timeOf(2), info.finishedAt());
+
+ // An outcome recorded in the second of the deletion counts as after it.
+ GenericKubernetesResource sameSecond = app("Submitted", "DriverReady",
"Failed");
+ cancel(sameSecond);
+
sameSecond.getMetadata().setDeletionTimestamp(CREATED_AT.plusSeconds(2).toString());
+ Assertions.assertEquals(JobHandle.Status.CANCELLED,
executionInfoOf(sameSecond).status());
+ }
+
+ @Test
+ public void testCancelMarkWithoutDeletion() {
+ // A cancellation that failed after marking the job, for example because
its response was
+ // lost, leaves the mark. The job keeps running and keeps the outcome it
reaches.
+ GenericKubernetesResource succeeded = app("Submitted", "DriverReady",
"Succeeded");
+ cancel(succeeded);
+ Assertions.assertEquals(JobHandle.Status.SUCCEEDED,
executionInfoOf(succeeded).status());
+
+ GenericKubernetesResource failed = app("Submitted", "DriverReady",
"Failed");
+ cancel(failed);
+ Assertions.assertEquals(JobHandle.Status.FAILED,
executionInfoOf(failed).status());
+
+ GenericKubernetesResource running = app("Submitted", "DriverReady",
"RunningHealthy");
+ cancel(running);
+ Assertions.assertEquals(JobHandle.Status.STARTED,
executionInfoOf(running).status());
+
+ // Deleted by its TTL after it succeeded, it still succeeded.
+ succeeded.getMetadata().setDeletionTimestamp(NOW.toString());
+ Assertions.assertEquals(JobHandle.Status.SUCCEEDED,
executionInfoOf(succeeded).status());
}
@Test
@@ -420,18 +472,41 @@ public class TestSparkApplicationStatusUtils {
Assertions.assertEquals(JobHandle.Status.FAILED, failed.status());
Assertions.assertEquals(timeOf(2), failed.finishedAt());
+ // Stopped before it was cancelled, it isn't cancelled.
cancel(app);
- JobExecutionInfo cancelled = executionInfoOf(app);
- Assertions.assertEquals(JobHandle.Status.CANCELLED, cancelled.status());
- Assertions.assertEquals(timeOf(2), cancelled.finishedAt());
+ app.getMetadata().setDeletionTimestamp(NOW.toString());
+ Assertions.assertEquals(JobHandle.Status.FAILED,
executionInfoOf(app).status());
+ Assertions.assertEquals(timeOf(2), executionInfoOf(app).finishedAt());
GenericKubernetesResource stopped =
app("Submitted", "DriverReady", "Succeeded",
"TerminatedWithoutReleaseResources");
cancel(stopped);
- Assertions.assertEquals(JobHandle.Status.CANCELLED,
executionInfoOf(stopped).status());
+ stopped.getMetadata().setDeletionTimestamp(NOW.toString());
+ Assertions.assertEquals(JobHandle.Status.SUCCEEDED,
executionInfoOf(stopped).status());
Assertions.assertEquals(timeOf(2), executionInfoOf(stopped).finishedAt());
}
+ @Test
+ public void testStoppedStateWithoutTime() {
+ GenericKubernetesResource app = app("Submitted", "DriverReady",
"ResourceReleased");
+ history(app)
+ .put(
+ "2",
+ ImmutableMap.of(
+ "currentStateSummary", "ResourceReleased",
"lastTransitionTime", "soon"));
+ status(app).put("currentState", history(app).get("2"));
+ app.getMetadata().setDeletionTimestamp(NOW.toString());
+
+ // Deleted after it stopped at an unknown time, it finished when it was
deleted at the latest.
+ JobExecutionInfo info = executionInfoOf(app);
+ Assertions.assertEquals(JobHandle.Status.FAILED, info.status());
+ Assertions.assertEquals(NOW, info.finishedAt());
+
+ // Nor is it cancelled.
+ cancel(app);
+ Assertions.assertEquals(JobHandle.Status.FAILED,
executionInfoOf(app).status());
+ }
+
@Test
public void testDeletionNotBeforeStart() {
// The deletion time is in whole seconds, the started time of the same
second isn't.
@@ -616,6 +691,14 @@ public class TestSparkApplicationStatusUtils {
Assertions.assertEquals(timeOf(1), info.startedAt());
Assertions.assertNull(info.finishedAt());
+ // Marked and deleted, it had still stopped before, as its outcome time is
unknown. It finished
+ // when it was deleted at the latest.
+ cancel(app);
+ app.getMetadata().setDeletionTimestamp(NOW.toString());
+ info = executionInfoOf(app);
+ Assertions.assertEquals(JobHandle.Status.SUCCEEDED, info.status());
+ Assertions.assertEquals(NOW, info.finishedAt());
+
// Without a creation time, an application without status never times out.
GenericKubernetesResource noCreationTime = app();
noCreationTime.getMetadata().setCreationTimestamp(null);
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 71e3abb38b..e9fb7b8a09 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
@@ -38,6 +38,35 @@ public class TestSparkApplicationUtils {
private final K8sJobExecutorConfigs configs = newConfigs(ImmutableMap.of());
+ @Test
+ public void testRejectClientDeployMode() {
+ SparkJobTemplate template =
+ SparkJobTemplate.builder()
+ .withName("user-job")
+ .withExecutable("s3a://bucket/app.jar")
+ .withClassName("org.example.App")
+ .withConfigs(ImmutableMap.of("spark.submit.deployMode", " Client
"))
+ .build();
+ IllegalArgumentException e =
+ Assertions.assertThrows(
+ IllegalArgumentException.class,
+ () -> SparkApplicationUtils.buildSparkApplication(CONTEXT,
template, configs));
+ Assertions.assertTrue(e.getMessage().contains("Client"), e.getMessage());
+
+ // Nor set by the executor configurations.
+ K8sJobExecutorConfigs clientMode =
+ newConfigs(ImmutableMap.of("spark.conf.spark.submit.deployMode",
"client"));
+ SparkJobTemplate noMode =
+ SparkJobTemplate.builder()
+ .withName("user-job")
+ .withExecutable("s3a://bucket/app.jar")
+ .withClassName("org.example.App")
+ .build();
+ Assertions.assertThrows(
+ IllegalArgumentException.class,
+ () -> SparkApplicationUtils.buildSparkApplication(CONTEXT, noMode,
clientMode));
+ }
+
@Test
@SuppressWarnings("unchecked")
public void testBuildSparkApplication() {
@@ -53,7 +82,7 @@ public class TestSparkApplicationUtils {
.withConfigs(
ImmutableMap.of(
"spark.master", "local[*]",
- "spark.submit.deployMode", "client",
+ "spark.submit.deployMode", "cluster",
"spark.executor.instances", "2",
"spark.jars", "https://repo/other.jar";))
.build();
