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 4872551328 [#14058] improvement(k8s-job-executor): Harden the k8s job
executor and quiet the job status poll (#14230)
4872551328 is described below
commit 48725513281f572a6647d4de44d1c626adf9f133
Author: Jerry Shao <[email protected]>
AuthorDate: Tue Oct 27 19:41:45 2026 +0800
[#14058] improvement(k8s-job-executor): Harden the k8s job executor and
quiet the job status poll (#14230)
### What changes were proposed in this pull request?
Follow-ups of the `k8s` job executor (#13613, #13616, #13617).
**Harden the k8s job executor**
- A driver pod whose phase is `Succeeded` or `Failed` has terminated,
whatever its container status says, for example after its node was lost.
Only otherwise the driver container decides, as before.
- A create that fails with a 5xx gets the same best-effort cleanup as
one without a response: the SparkApplication may have been created. If
the cleanup fails, a WARN tells how to delete it by hand. Gravitino
doesn't sweep the SparkApplications of jobs it doesn't know (see
#13554).
- The SparkApplications of a namespace are listed page by page (500 per
request). A listing that times out waiting for the response no longer
counts as an unreachable API server: the jobs are got one by one instead
of failing every status query for 10 s.
- Where a read decides the deletion (the no-status cleanup and the
failed-submit cleanup), the SparkApplication is deleted with a
`resourceVersion` precondition and the `Foreground` propagation, and
read again on a 409 (up to 3 retries). An application that got a status
meanwhile follows the normal status path. fabric8 7.2.0 types
`lockResourceVersion()` as `ReplaceDeletable`, without
`withPropagationPolicy`, so the resource is used as the
`ExtensibleResource` it is; both end up in the `DeleteOptions` of the
request.
**Job status poll**
- `JobManager` logged an ERROR with a full stack trace for every active
job on every poll while the job executor failed, for example while the
API server was unreachable. It now logs one WARN per poll and metalake,
with the number of failed jobs, the first five of them and the first
failure, and each failed job at DEBUG.
**Output size**
- `JobExecutor#getJobStdout`/`getJobStderr`,
`gravitino.job.outputMaxBytes` and the docs said `maxBytes` bounds both
the read cost and the response size. It bounds the response size; an
executor may not bound the bytes it reads, as the k8s job executor reads
the last lines of the driver log.
### Why are the changes needed?
They are the follow-ups of the k8s job executor collected in #13554: a
terminated driver pod could be deleted or kept by mistake, a
SparkApplication could run unknown after a create that failed with a
5xx, a namespace with many SparkApplications could stop all status
updates, a deletion could act on a SparkApplication that changed after
it was read, and the status poll flooded the log while the API server
was unreachable.
Fix: #14058
### Does this PR introduce _any_ user-facing change?
- The status poll failures are logged once per poll and metalake at WARN
instead of once per job at ERROR.
- The description of `gravitino.job.outputMaxBytes` changed. No new
configuration.
### How was this patch tested?
- Unit tests with the fabric8 mock server for the submit cleanup after a
5xx (also of a resource of another job, and after a conflict), the logs
of a deleted SparkApplication, the paged listing, a slow listing, the
conditional deletion of a job without status (request body, a status
written meanwhile, retries running out), `isDriverRunning`, the cache
eviction, and the summarized poll log in `TestJobManager`. The key tests
fail when their fix is reverted.
- Manually against a local Kubernetes cluster (OrbStack) with
spark-kubernetes-operator 0.9.0: Spark Pi succeeded with its driver
output, a running job was cancelled, a SparkApplication that never got a
status (in a namespace the operator doesn't watch) was deleted and its
job failed, and the API server rejected a `Foreground` deletion with a
stale `resourceVersion` precondition with a 409 and accepted it with the
current one.
🤖 Generated with [Claude Code](https://claude.com/claude-code)
---------
Co-authored-by: Claude Opus 5.5 <[email protected]>
---
.../main/java/org/apache/gravitino/Configs.java | 12 +-
.../gravitino/connector/job/JobExecutor.java | 16 +-
.../java/org/apache/gravitino/job/JobManager.java | 76 +++--
.../gravitino/job/JobOperationDispatcher.java | 2 +-
.../org/apache/gravitino/job/TestJobManager.java | 91 ++++++
docs/gravitino-server-config.md | 2 +-
docs/manage-jobs-in-gravitino.md | 2 +-
docs/open-api/jobs.yaml | 2 +-
.../apache/gravitino/job/k8s/K8sJobExecutor.java | 150 +++++++--
.../gravitino/job/k8s/NamespacedResourceCache.java | 34 ++-
.../job/k8s/spark/SparkApplicationStatusUtils.java | 20 +-
.../gravitino/job/k8s/TestK8sJobExecutor.java | 340 ++++++++++++++++++++-
.../job/k8s/TestNamespacedResourceCache.java | 64 +++-
.../k8s/spark/TestSparkApplicationStatusUtils.java | 31 ++
14 files changed, 754 insertions(+), 88 deletions(-)
diff --git a/core/src/main/java/org/apache/gravitino/Configs.java
b/core/src/main/java/org/apache/gravitino/Configs.java
index f85ac413d2..fe19e24f3a 100644
--- a/core/src/main/java/org/apache/gravitino/Configs.java
+++ b/core/src/main/java/org/apache/gravitino/Configs.java
@@ -663,12 +663,12 @@ public class Configs {
public static final ConfigEntry<Integer> JOB_OUTPUT_MAX_BYTES =
new ConfigBuilder("gravitino.job.outputMaxBytes")
.doc(
- "The maximum number of bytes read from the tail of a job's
captured stdout/stderr "
- + "when retrieving its output. Bounds both the read cost and
the response size "
- + "regardless of how the content is shaped (e.g. a single
very long line). "
- + "This is resolved by JobManager and passed as an argument
to "
- + "JobExecutor#getJobStdout and JobExecutor#getJobStderr, so
all executors "
- + "honor the same cap.")
+ "The maximum number of bytes returned from the tail of a job's
captured "
+ + "stdout/stderr when retrieving its output. Bounds the
response size regardless "
+ + "of how the content is shaped (e.g. a single very long
line). Whether it also "
+ + "bounds the bytes read depends on the job executor. This
is resolved by "
+ + "JobManager and passed as an argument to
JobExecutor#getJobStdout and "
+ + "JobExecutor#getJobStderr, so all executors honor the same
cap.")
.version(ConfigConstants.VERSION_2_0_0)
.intConf()
.checkValue(value -> value > 0,
ConfigConstants.POSITIVE_NUMBER_ERROR_MSG)
diff --git
a/core/src/main/java/org/apache/gravitino/connector/job/JobExecutor.java
b/core/src/main/java/org/apache/gravitino/connector/job/JobExecutor.java
index 32fd9d6783..c15389e738 100644
--- a/core/src/main/java/org/apache/gravitino/connector/job/JobExecutor.java
+++ b/core/src/main/java/org/apache/gravitino/connector/job/JobExecutor.java
@@ -189,9 +189,11 @@ public interface JobExecutor extends Closeable {
* @param jobId The unique identifier of the job.
* @param maxLines The maximum number of (most recent) lines to return,
resolved by the caller
* from the {@code gravitino.job.outputMaxLines} configuration.
- * @param maxBytes The maximum number of (most recent) bytes to read from
the underlying output,
- * resolved by the caller from the {@code gravitino.job.outputMaxBytes}
configuration. Bounds
- * both the read cost and the response size regardless of how the
content is shaped.
+ * @param maxBytes The maximum number of (most recent) bytes to return,
resolved by the caller
+ * from the {@code gravitino.job.outputMaxBytes} configuration. Bounds
the response size
+ * regardless of how the content is shaped. An implementation should
also bound the bytes it
+ * reads from the underlying output when it can, but may not, for
example when the output can
+ * only be read by lines.
* @return the stdout lines of the job, or an empty list if not available.
*/
default List<String> getJobStdout(String jobId, int maxLines, int maxBytes) {
@@ -212,9 +214,11 @@ public interface JobExecutor extends Closeable {
* @param jobId The unique identifier of the job.
* @param maxLines The maximum number of (most recent) lines to return,
resolved by the caller
* from the {@code gravitino.job.outputMaxLines} configuration.
- * @param maxBytes The maximum number of (most recent) bytes to read from
the underlying output,
- * resolved by the caller from the {@code gravitino.job.outputMaxBytes}
configuration. Bounds
- * both the read cost and the response size regardless of how the
content is shaped.
+ * @param maxBytes The maximum number of (most recent) bytes to return,
resolved by the caller
+ * from the {@code gravitino.job.outputMaxBytes} configuration. Bounds
the response size
+ * regardless of how the content is shaped. An implementation should
also bound the bytes it
+ * reads from the underlying output when it can, but may not, for
example when the output can
+ * only be read by lines.
* @return the stderr lines of the job, or an empty list if not available.
*/
default List<String> getJobStderr(String jobId, int maxLines, int maxBytes) {
diff --git a/core/src/main/java/org/apache/gravitino/job/JobManager.java
b/core/src/main/java/org/apache/gravitino/job/JobManager.java
index 86887db750..2ad5a3b1a3 100644
--- a/core/src/main/java/org/apache/gravitino/job/JobManager.java
+++ b/core/src/main/java/org/apache/gravitino/job/JobManager.java
@@ -121,6 +121,9 @@ public class JobManager implements JobOperationDispatcher {
// stop.
private static final int CANCELLATION_TIMEOUT_IN_PULL_INTERVALS = 3;
+ // How many of the jobs whose status pull failed are named in the log of a
poll.
+ private static final int MAX_LOGGED_FAILED_JOBS = 5;
+
private final EntityStore entityStore;
private final File stagingDir;
@@ -488,7 +491,7 @@ public class JobManager implements JobOperationDispatcher {
maxBytes == null || maxBytes > 0, "maxBytes must be positive if
specified");
// A caller-specified maxLines/maxBytes can only narrow the globally
configured cap, never
- // widen it - the global configuration remains a hard upper bound on read
cost/response size.
+ // widen it - the global configuration remains a hard upper bound on the
response size.
int effectiveMaxLines = clampToGlobalMax(maxLines, jobOutputMaxLines);
int effectiveMaxBytes = clampToGlobalMax(maxBytes, jobOutputMaxBytes);
@@ -1046,30 +1049,47 @@ public class JobManager implements
JobOperationDispatcher {
|| job.status() == JobHandle.Status.CANCELLING)
.toList();
+ // A job whose status can't be pulled or updated is kept unchanged, and
retried in the next
+ // poll. While the job executor fails, for example while its backend is
unreachable, every job
+ // fails the same way on every poll, so the failures are logged once per
poll.
long now = System.currentTimeMillis();
- activeJobs.forEach(
- job -> {
- // A failure on one job must never stop the status pull of the
others.
- try {
- // Only the job executor instance owning the job can query its
status. The jobs
- // owned by other servers are skipped, except a node local job
whose cancellation its
- // owner has not answered in time: the owner has likely exited,
e.g. the server
- // restarted, so nobody else would ever settle it. The other jobs
left behind by a
- // server that has exited are settled by cleanUpStagingDirs() once
they expire.
- if (jobExecutor.ownsJob(job.jobExecutionId())) {
- pullAndUpdateOwnedJobStatus(metalake, job);
- } else if (jobExecutor.isJobStateNodeLocal() &&
isCancellationUnanswered(job, now)) {
- settleUnansweredCancellation(metalake, job, now);
- }
- } catch (Throwable t) {
- LOG.error(
- "Failed to update the status of job {} with execution id {}
under metalake {}",
- job.name(),
- job.jobExecutionId(),
- metalake,
- t);
- }
- });
+ List<String> failedJobs = new ArrayList<>();
+ Throwable firstFailure = null;
+ for (JobEntity job : activeJobs) {
+ // A failure on one job must never stop the status pull of the others.
+ try {
+ // Only the job executor instance owning the job can query its status.
The jobs owned by
+ // other servers are skipped, except a node local job whose
cancellation its owner has not
+ // answered in time: the owner has likely exited, e.g. the server
restarted, so nobody else
+ // would ever settle it. The other jobs left behind by a server that
has exited are settled
+ // by cleanUpStagingDirs() once they expire.
+ if (jobExecutor.ownsJob(job.jobExecutionId())) {
+ pullAndUpdateOwnedJobStatus(metalake, job);
+ } else if (jobExecutor.isJobStateNodeLocal() &&
isCancellationUnanswered(job, now)) {
+ settleUnansweredCancellation(metalake, job, now);
+ }
+ } catch (Throwable t) {
+ LOG.debug(
+ "Failed to pull or update the status of job {} with execution id
{} under metalake {}",
+ job.name(),
+ job.jobExecutionId(),
+ metalake,
+ t);
+ failedJobs.add(job.name());
+ firstFailure = firstFailure == null ? t : firstFailure;
+ }
+ }
+
+ if (firstFailure != null) {
+ LOG.warn(
+ "Failed to pull or update the status of {} of {} active jobs under
metalake {}, for"
+ + " example {}, retrying in the next poll. The first failure:",
+ failedJobs.size(),
+ activeJobs.size(),
+ metalake,
+ failedJobs.subList(0, Math.min(failedJobs.size(),
MAX_LOGGED_FAILED_JOBS)),
+ firstFailure);
+ }
}
private void pullAndUpdateOwnedJobStatus(String metalake, JobEntity job) {
@@ -1104,14 +1124,6 @@ public class JobManager implements
JobOperationDispatcher {
job.jobExecutionId(),
metalake,
observed.status());
- } catch (Exception e) {
- // Keep the job unchanged, and retry it in the next poll.
- LOG.error(
- "Failed to pull or cancel job {} by execution id {}",
- job.name(),
- job.jobExecutionId(),
- e);
- return;
}
// Besides the status, the job executor may report the timestamps of the
job later than its
diff --git
a/core/src/main/java/org/apache/gravitino/job/JobOperationDispatcher.java
b/core/src/main/java/org/apache/gravitino/job/JobOperationDispatcher.java
index 7c2c340fef..a12b8612da 100644
--- a/core/src/main/java/org/apache/gravitino/job/JobOperationDispatcher.java
+++ b/core/src/main/java/org/apache/gravitino/job/JobOperationDispatcher.java
@@ -138,7 +138,7 @@ public interface JobOperationDispatcher extends Closeable {
* @param includeOutput whether to also fetch and attach the job's
stdout/stderr output
* @param maxLines the maximum number of (most recent) output lines to
return, or {@code null} to
* use the global default
- * @param maxBytes the maximum number of (most recent) output bytes to read,
or {@code null} to
+ * @param maxBytes the maximum number of (most recent) output bytes to
return, or {@code null} to
* use the global default
* @return the job entity associated with the specified ID
* @throws NoSuchJobException if no job with the specified ID exists
diff --git a/core/src/test/java/org/apache/gravitino/job/TestJobManager.java
b/core/src/test/java/org/apache/gravitino/job/TestJobManager.java
index a2a05f8fbb..e908f9f2a2 100644
--- a/core/src/test/java/org/apache/gravitino/job/TestJobManager.java
+++ b/core/src/test/java/org/apache/gravitino/job/TestJobManager.java
@@ -41,12 +41,14 @@ import java.nio.file.Files;
import java.time.Duration;
import java.time.Instant;
import java.time.temporal.ChronoUnit;
+import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.Random;
import java.util.UUID;
+import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReference;
@@ -93,6 +95,12 @@ import org.apache.gravitino.storage.IdGenerator;
import org.apache.gravitino.storage.RandomIdGenerator;
import org.apache.gravitino.utils.NameIdentifierUtil;
import org.apache.gravitino.utils.NamespaceUtil;
+import org.apache.logging.log4j.Level;
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.core.LogEvent;
+import org.apache.logging.log4j.core.LoggerContext;
+import org.apache.logging.log4j.core.appender.AbstractAppender;
+import org.apache.logging.log4j.core.layout.PatternLayout;
import org.awaitility.Awaitility;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.Assertions;
@@ -104,6 +112,19 @@ import org.mockito.Mockito;
public class TestJobManager {
+ private static class LogCollector extends AbstractAppender {
+ private final List<LogEvent> events = new CopyOnWriteArrayList<>();
+
+ private LogCollector() {
+ super("jobManagerLogCollector", null,
PatternLayout.createDefaultLayout(), true, null);
+ }
+
+ @Override
+ public void append(LogEvent event) {
+ events.add(event.toImmutable());
+ }
+ }
+
private JobManager jobManager;
private EntityStore entityStore;
@@ -1966,6 +1987,54 @@ public class TestJobManager {
Assertions.assertTrue(cancelled.finishedAt() > 0);
}
+ @Test
+ public void testPullJobStatusLogsFailuresOncePerPoll() throws IOException {
+ List<JobEntity> jobs = new ArrayList<>();
+ for (int i = 0; i < 7; i++) {
+ jobs.add(newJobEntity("execution-" + i, JobHandle.Status.STARTED,
Instant.now(), null));
+ }
+ mockListActiveJobs(jobs.toArray(new JobEntity[0]));
+ // The backend of the job executor is unreachable, the job executor fails
every job but one.
+ RuntimeException unreachable = new RuntimeException("API server
unreachable");
+ when(jobExecutor.getJobExecutionInfo(any())).thenThrow(unreachable);
+ doReturn(JobExecutionInfo.of(JobHandle.Status.STARTED))
+ .when(jobExecutor)
+ .getJobExecutionInfo(jobs.get(6).jobExecutionId());
+ stubEntityStoreUpdateToApply(jobs.get(6));
+
+ List<LogEvent> events = captureLogs(() ->
jobManager.pullAndUpdateJobStatus());
+
+ // Each job is logged at DEBUG, and the poll at WARN once, with the first
failure.
+ Assertions.assertEquals(
+ 6,
+ events.stream()
+ .filter(
+ event ->
+ Level.DEBUG.equals(event.getLevel())
+ && event.getThrown() == unreachable
+ && event
+ .getMessage()
+ .getFormattedMessage()
+ .startsWith("Failed to pull or update the status
of job "))
+ .count());
+ List<LogEvent> warnings =
+ events.stream().filter(event ->
event.getLevel().isMoreSpecificThan(Level.WARN)).toList();
+ Assertions.assertEquals(1, warnings.size());
+ LogEvent warning = warnings.get(0);
+ Assertions.assertEquals(Level.WARN, warning.getLevel());
+ Assertions.assertSame(unreachable, warning.getThrown());
+ String message = warning.getMessage().getFormattedMessage();
+ Assertions.assertTrue(
+ message.contains("6 of 7 active jobs under metalake " + metalake),
message);
+ Assertions.assertTrue(
+ message.contains(
+ jobs.subList(0, 5).stream()
+ .map(JobEntity::name)
+ .collect(Collectors.toList())
+ .toString()),
+ message);
+ }
+
@Test
public void testPullJobStatusKeepsCancellingWhenOwnerFailsToCancel() throws
IOException {
JobEntity job =
@@ -3376,6 +3445,28 @@ public class TestJobManager {
.build();
}
+ private static List<LogEvent> captureLogs(Runnable action) {
+ LoggerContext context =
+ (LoggerContext)
LogManager.getContext(JobManager.class.getClassLoader(), false);
+ var logger = context.getLogger(JobManager.class.getName());
+ LogCollector collector = new LogCollector();
+ collector.start();
+ // Attach to this logger without replacing its configuration or its
existing appenders.
+ synchronized (logger) {
+ Level previous = logger.getLevel();
+ logger.addAppender(collector);
+ logger.setLevel(Level.DEBUG);
+ try {
+ action.run();
+ } finally {
+ logger.removeAppender(collector);
+ logger.setLevel(previous);
+ collector.stop();
+ }
+ }
+ return List.copyOf(collector.events);
+ }
+
private void mockListActiveJobs(JobEntity... jobs) {
mockedMetalake
.when(() -> MetalakeManager.listInUseMetalakes(entityStore))
diff --git a/docs/gravitino-server-config.md b/docs/gravitino-server-config.md
index ce60086496..54224446dd 100644
--- a/docs/gravitino-server-config.md
+++ b/docs/gravitino-server-config.md
@@ -657,7 +657,7 @@ server, are documented with those services. See
| `gravitino.job.stagingDirKeepTimeInMs` | How long in milliseconds a finished
job's staging files are kept. Use at least 10 minutes outside testing.
| `604800000` (7 days) |
| `gravitino.job.statusPullIntervalInMs` | Interval in milliseconds between
job status polls. Use at least 1 minute outside testing.
| `300000` (5 minutes) |
| `gravitino.job.outputMaxLines` | Maximum number of lines returned
when fetching a job's stdout/stderr output.
| `1000` |
-| `gravitino.job.outputMaxBytes` | Maximum number of bytes read from
the tail of a job's stdout/stderr when fetching its output.
| `262144` (256KB) |
+| `gravitino.job.outputMaxBytes` | Maximum number of bytes returned
from the tail of a job's stdout/stderr when fetching its output.
| `262144` (256KB) |
#### Table maintenance cooldowns
diff --git a/docs/manage-jobs-in-gravitino.md b/docs/manage-jobs-in-gravitino.md
index b09e5f23ca..5d2465a42c 100644
--- a/docs/manage-jobs-in-gravitino.md
+++ b/docs/manage-jobs-in-gravitino.md
@@ -403,7 +403,7 @@ default configurations:
| `gravitino.job.stagingDirKeepTimeInMs` | The time in milliseconds to keep
the staging directory after the job is completed
| `604800000` (7 days)
| No |
| `gravitino.job.statusPullIntervalInMs` | The interval in milliseconds to
pull the job status from the job executor
| `300000` (5 minutes)
| No |
| `gravitino.job.outputMaxLines` | The maximum number of lines
returned when fetching a job's stdout/stderr output
| `1000`
| No |
-| `gravitino.job.outputMaxBytes` | The maximum number of bytes read
from the tail of a job's stdout/stderr output
| `262144` (256KB)
| No |
+| `gravitino.job.outputMaxBytes` | The maximum number of bytes
returned from the tail of a job's stdout/stderr output
| `262144` (256KB)
| No |
#### Configurations for Local Job Executor
diff --git a/docs/open-api/jobs.yaml b/docs/open-api/jobs.yaml
index bd194d64c7..00ed2cd991 100644
--- a/docs/open-api/jobs.yaml
+++ b/docs/open-api/jobs.yaml
@@ -395,7 +395,7 @@ components:
name: outputMaxBytes
in: query
description: >
- The maximum number of (most recent) output bytes to read, when
includeOutput is true.
+ The maximum number of (most recent) output bytes to return, when
includeOutput is true.
Ignored otherwise. Can only narrow the globally configured
gravitino.job.outputMaxBytes
cap, never widen it - a value larger than the global cap is clamped
down to it. Omit to
use the global default.
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
index 5e6583f32d..3911f94d6a 100644
---
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
@@ -26,6 +26,7 @@ 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.ListOptionsBuilder;
import io.fabric8.kubernetes.api.model.OwnerReference;
import io.fabric8.kubernetes.api.model.Pod;
import io.fabric8.kubernetes.client.KubernetesClient;
@@ -33,10 +34,14 @@ 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 io.fabric8.kubernetes.client.extension.ExtensibleResource;
import java.io.IOException;
import java.io.InputStream;
import java.net.HttpURLConnection;
+import java.net.http.HttpConnectTimeoutException;
+import java.net.http.HttpTimeoutException;
import java.time.Clock;
+import java.util.ArrayList;
import java.util.Comparator;
import java.util.HashMap;
import java.util.List;
@@ -46,6 +51,7 @@ import java.util.Set;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.function.Predicate;
import javax.annotation.Nullable;
+import org.apache.commons.lang3.StringUtils;
import org.apache.gravitino.connector.job.JobContext;
import org.apache.gravitino.connector.job.JobExecutionInfo;
import org.apache.gravitino.connector.job.JobExecutor;
@@ -92,6 +98,10 @@ public class K8sJobExecutor implements JobExecutor {
// How many times a change of a SparkApplication is tried again after it
changed meanwhile.
private static final int UPDATE_CONFLICT_RETRIES = 3;
+ // How many SparkApplications a listing gets per request, so that a
namespace with many of them
+ // is listed in several shorter requests rather than one that may time out.
+ @VisibleForTesting static final long LIST_PAGE_SIZE = 500L;
+
private final Clock clock;
// The failure of the last status query that couldn't reach the API server,
and when it failed.
@@ -129,17 +139,13 @@ public class K8sJobExecutor implements JobExecutor {
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.
+ this::listSparkApplications,
+ // Getting the SparkApplication itself couldn't reach the API
server either. A listing
+ // that the API server is too slow to answer isn't, it gets the
jobs one by one instead.
e ->
e instanceof KubernetesClientException
- && isUnreachable((KubernetesClientException) e),
+ && isUnreachable((KubernetesClientException) e)
+ && !isResponseTimedOut((KubernetesClientException) e),
clock::millis);
LOG.info(
"Initialized the k8s job executor for cluster {} at {}, default
namespace {}, built-in"
@@ -170,9 +176,10 @@ public class K8sJobExecutor implements JobExecutor {
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.
+ if (isUnreachable(e) || e.getCode() >=
HttpURLConnection.HTTP_INTERNAL_ERROR) {
+ // The SparkApplication may have been created although no response or
an error came back.
+ // Gravitino records no job for a failed submission, so delete it,
otherwise it would run
+ // unknown.
deleteUnsubmittedSparkApplication(jobExecutionId, context.jobId());
throw e;
}
@@ -372,6 +379,23 @@ public class K8sJobExecutor implements JobExecutor {
return ioFailure;
}
+ /**
+ * Returns whether a request failed because the API server didn't answer it
in time, although it
+ * could be connected to.
+ *
+ * @param e the failure of the request
+ * @return true if the response timed out
+ */
+ @VisibleForTesting
+ static boolean isResponseTimedOut(KubernetesClientException e) {
+ for (Throwable cause = e.getCause(); cause != null; cause =
cause.getCause()) {
+ if (cause instanceof HttpTimeoutException) {
+ return !(cause instanceof HttpConnectTimeoutException);
+ }
+ }
+ return false;
+ }
+
/**
* 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.
@@ -479,11 +503,15 @@ public class K8sJobExecutor implements JobExecutor {
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.
+ // deleted, for example by a cancellation on another server. Then it is
mapped as usual, also
+ // if it gets one between the read and the deletion. If it is gone
meanwhile, the job is
+ // missing rather than failed, which settles a cancelling job as cancelled.
GenericKubernetesResource latest =
- getSparkApplication(id)
+ deleteSparkApplicationIf(
+ id,
+ app ->
+ !SparkApplicationStatusUtils.hasStatus(app)
+ && app.getMetadata().getDeletionTimestamp() == null)
.orElseThrow(
() -> new NoSuchJobException("SparkApplication of job %s
doesn't exist", jobId));
if (SparkApplicationStatusUtils.hasStatus(latest)
@@ -491,7 +519,6 @@ public class K8sJobExecutor implements JobExecutor {
sparkApplicationCache.invalidate(id.namespace());
return getExecutionInfo(id, jobId, latest);
}
- deleteSparkApplication(id);
return failed;
}
@@ -502,16 +529,24 @@ public class K8sJobExecutor implements JobExecutor {
*/
private void deleteUnsubmittedSparkApplication(K8sJobExecutionId id, long
jobId) {
try {
- Optional<GenericKubernetesResource> app = getSparkApplication(id);
+ Optional<GenericKubernetesResource> app =
+ deleteSparkApplicationIf(id, current ->
K8sJobResourceUtils.isOfJob(current, jobId));
if (app.isPresent() && K8sJobResourceUtils.isOfJob(app.get(), jobId)) {
- deleteSparkApplication(id);
LOG.info("Deleted the SparkApplication of job {}, whose submission
failed", jobId);
}
} catch (RuntimeException e) {
+ // Gravitino doesn't sweep the resources of unknown jobs, so tell how to
delete it by hand.
LOG.warn(
- "Failed to clean up the SparkApplication {} of job {}, whose
submission failed",
- id,
+ "Failed to clean up the SparkApplication {}/{} of job {}, whose
submission failed. It may"
+ + " run although Gravitino has no record of it. If it exists
with the label {}={},"
+ + " delete it with: kubectl delete sparkapplication -n {} {}",
+ id.namespace(),
+ id.name(),
+ jobId,
+ K8sJobResourceUtils.LABEL_JOB_ID,
jobId,
+ id.namespace(),
+ id.name(),
e);
}
}
@@ -557,6 +592,59 @@ public class K8sJobExecutor implements JobExecutor {
}
}
+ /**
+ * Deletes the SparkApplication of a job if it should be, only if it hasn't
changed since it was
+ * read, and reads it again to decide again otherwise.
+ *
+ * @param id the job execution id
+ * @param shouldDelete tells whether to delete the SparkApplication
+ * @return the SparkApplication the deletion was decided on, or empty if it
doesn't exist, also if
+ * it was gone by the time it was deleted
+ */
+ private Optional<GenericKubernetesResource> deleteSparkApplicationIf(
+ K8sJobExecutionId id, Predicate<GenericKubernetesResource> shouldDelete)
{
+ for (int attempt = 0; ; attempt++) {
+ Optional<GenericKubernetesResource> app = getSparkApplication(id);
+ if (!app.isPresent() || !shouldDelete.test(app.get())) {
+ return app;
+ }
+ try {
+ // fabric8 reports that nothing was deleted, rather than failing, for
a SparkApplication
+ // that is gone, for example deleted by a cancellation on another
server meanwhile.
+ boolean deleted = !lockSparkApplication(app.get()).delete().isEmpty();
+ sparkApplicationCache.invalidate(id.namespace());
+ return deleted ? app : Optional.empty();
+ } catch (KubernetesClientException e) {
+ if (e.getCode() != HttpURLConnection.HTTP_CONFLICT || attempt >=
UPDATE_CONFLICT_RETRIES) {
+ throw e;
+ }
+ }
+ }
+ }
+
+ /**
+ * Returns a SparkApplication as a resource locked to the version it was
read at. Deleting it
+ * fails with the code 409 if it has changed since, and deletes its pods
first, as {@link
+ * #deleteSparkApplication(K8sJobExecutionId)} does.
+ *
+ * @param app the SparkApplication as it was read
+ * @return the locked resource
+ */
+ @SuppressWarnings("unchecked")
+ private ExtensibleResource<GenericKubernetesResource> lockSparkApplication(
+ GenericKubernetesResource app) {
+ // fabric8 types a resource locked to a version as one that can't set the
propagation policy,
+ // but its implementation can: it sends both the precondition and the
policy in the options of
+ // the deletion.
+ ExtensibleResource<GenericKubernetesResource> resource =
+ (ExtensibleResource<GenericKubernetesResource>)
+ getSparkApplicationClient(app.getMetadata().getNamespace())
+ .withName(app.getMetadata().getName());
+ return resource
+ .lockResourceVersion(app.getMetadata().getResourceVersion())
+ .withPropagationPolicy(DeletionPropagation.FOREGROUND);
+ }
+
/**
* 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
@@ -608,6 +696,26 @@ public class K8sJobExecutor implements JobExecutor {
sparkApplicationCache.invalidate(id.namespace());
}
+ /** Lists the SparkApplications managed by Gravitino in a namespace, page by
page. */
+ private List<GenericKubernetesResource> listSparkApplications(String
namespace) {
+ List<GenericKubernetesResource> apps = new ArrayList<>();
+ String continueToken = null;
+ do {
+ GenericKubernetesResourceList page =
+ getSparkApplicationClient(namespace)
+ .withLabel(
+ K8sJobResourceUtils.LABEL_MANAGED_BY,
K8sJobResourceUtils.MANAGED_BY_GRAVITINO)
+ .list(
+ new ListOptionsBuilder()
+ .withLimit(LIST_PAGE_SIZE)
+ .withContinue(continueToken)
+ .build());
+ apps.addAll(page.getItems());
+ continueToken = page.getMetadata() == null ? null :
page.getMetadata().getContinue();
+ } while (StringUtils.isNotBlank(continueToken));
+ return apps;
+ }
+
private NonNamespaceOperation<
GenericKubernetesResource,
GenericKubernetesResourceList,
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
index 59001e9638..db0356f770 100644
---
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
@@ -18,11 +18,14 @@
*/
package org.apache.gravitino.job.k8s;
+import com.google.common.annotations.VisibleForTesting;
import com.google.common.collect.ImmutableMap;
+import com.google.common.collect.ImmutableSet;
import io.fabric8.kubernetes.api.model.HasMetadata;
import java.util.List;
import java.util.Map;
import java.util.Optional;
+import java.util.Set;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionException;
import java.util.concurrent.ConcurrentHashMap;
@@ -41,24 +44,31 @@ import org.slf4j.LoggerFactory;
* 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.
*
+ * <p>A namespace that can't be listed isn't listed again for {@link
#FAILED_LIST_TTL_MS}, longer
+ * than a listing is cached: a failed listing may have taken long, for example
one that timed out
+ * after its retries, and getting the resources one by one meanwhile gives the
same results.
+ *
* @param <T> the resource type
*/
final class NamespacedResourceCache<T extends HasMetadata> {
private static final class Snapshot<T> {
- private final long listedAtMs;
+ private final long expiresAtMs;
private final Map<String, T> resources;
- private Snapshot(long listedAtMs, Map<String, T> resources) {
- this.listedAtMs = listedAtMs;
+ private Snapshot(long expiresAtMs, Map<String, T> resources) {
+ this.expiresAtMs = expiresAtMs;
this.resources = resources;
}
}
private static final Logger LOG =
LoggerFactory.getLogger(NamespacedResourceCache.class);
+ /** How long a namespace that can't be listed isn't listed again, at least
the TTL. */
+ @VisibleForTesting static final long FAILED_LIST_TTL_MS = 10 * 60 * 1000L;
+
private final long ttlMs;
private final Function<String, List<T>> lister;
@@ -149,6 +159,12 @@ final class NamespacedResourceCache<T extends HasMetadata>
{
snapshots.remove(namespace);
}
+ /** Returns the namespaces whose listing is cached or being made. */
+ @VisibleForTesting
+ Set<String> getCachedNamespaces() {
+ return ImmutableSet.copyOf(snapshots.keySet());
+ }
+
/** 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()) {
@@ -156,7 +172,7 @@ final class NamespacedResourceCache<T extends HasMetadata> {
}
// A listing being made is waited for.
Snapshot<T> made = snapshot.getNow(null);
- return made != null && clock.getAsLong() - made.listedAtMs >= ttlMs;
+ return made != null && clock.getAsLong() >= made.expiresAtMs;
}
private Snapshot<T> snapshotOf(String namespace) {
@@ -164,20 +180,22 @@ final class NamespacedResourceCache<T extends
HasMetadata> {
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());
+ return new Snapshot<>(listedAtMs + ttlMs, 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.
+ // instead of each of them failing to list the namespace again. It is
cached from when the
+ // listing failed: a listing that timed out may have taken longer than
the TTL.
+ long failedTtlMs = Math.max(ttlMs, FAILED_LIST_TTL_MS);
LOG.warn(
"Failed to list the resources in namespace {}, getting them one by
one for the next {}"
+ " ms: {}",
namespace,
- ttlMs,
+ failedTtlMs,
e.toString());
- return new Snapshot<>(listedAtMs, ImmutableMap.of());
+ return new Snapshot<>(clock.getAsLong() + failedTtlMs,
ImmutableMap.of());
}
}
}
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 f78a498137..8bf81e2409 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
@@ -108,6 +108,9 @@ public final class SparkApplicationStatusUtils {
private static final Set<String> STOPPED_STATES =
ImmutableSet.of("ResourceReleased", "TerminatedWithoutReleaseResources");
+ // The terminal phases of a pod. fabric8 has no public constants for the pod
phases.
+ private static final Set<String> POD_TERMINATED_PHASES =
ImmutableSet.of("Succeeded", "Failed");
+
private SparkApplicationStatusUtils() {}
/**
@@ -251,21 +254,22 @@ public final class SparkApplicationStatusUtils {
}
/**
- * 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.
+ * Returns whether the driver of a SparkApplication is still pending or
running. A pod that has
+ * terminated runs nothing, even if its container status still says running,
for example after its
+ * node was lost. Otherwise it is decided by the driver container, as a
sidecar can keep the pod
+ * running after the driver exited, and the driver runs if its 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);
+ if (POD_TERMINATED_PHASES.contains(phase)) {
+ return false;
+ }
+ ContainerState state = driverContainerState(driver,
getDriverContainerName(app));
+ return state == null || state.getTerminated() == null;
}
/**
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
index a5fdc9c92b..01f479941c 100644
---
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
@@ -23,6 +23,8 @@ 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.GenericKubernetesResourceList;
+import io.fabric8.kubernetes.api.model.ListMetaBuilder;
import io.fabric8.kubernetes.api.model.OwnerReference;
import io.fabric8.kubernetes.api.model.OwnerReferenceBuilder;
import io.fabric8.kubernetes.api.model.PodBuilder;
@@ -43,6 +45,8 @@ import io.fabric8.mockwebserver.http.RecordedRequest;
import java.io.File;
import java.io.IOException;
import java.net.ConnectException;
+import java.net.http.HttpConnectTimeoutException;
+import java.net.http.HttpTimeoutException;
import java.time.Clock;
import java.time.Instant;
import java.time.ZoneId;
@@ -285,6 +289,87 @@ public class TestK8sJobExecutor {
Assertions.assertNull(sparkApplication("gravitino-job-1"));
}
+ @Test
+ public void testJobWithoutStatusDeletedIfUnchanged() throws
InterruptedException {
+ String jobExecutionId = executor.submitJob(context(1L), sparkTemplate());
+ String resourceVersion =
sparkApplication("gravitino-job-1").getMetadata().getResourceVersion();
+ now.set(now.get().plusSeconds(601));
+
+ Assertions.assertEquals(
+ JobHandle.Status.FAILED,
executor.getJobExecutionInfo(jobExecutionId).status());
+
+ String delete = lastRequest("DELETE").getBody().readUtf8();
+ Assertions.assertTrue(
+ delete.contains("\"resourceVersion\":\"" + resourceVersion + "\""),
delete);
+ Assertions.assertTrue(delete.contains("\"Foreground\""), delete);
+ }
+
+ @Test
+ public void testJobWithoutStatusGetsStatusBeforeDeletion() {
+ String jobExecutionId = executor.submitJob(context(1L), sparkTemplate());
+ GenericKubernetesResource withoutStatus =
sparkApplication("gravitino-job-1");
+ now.set(now.get().plusSeconds(601));
+ // The operator writes the status after the SparkApplication was read
without one, and before
+ // it is deleted, which then conflicts. Read again, it is mapped as usual.
+ setStates("gravitino-job-1", "Submitted", "DriverReady");
+ server
+ .expect()
+ .get()
+ .withPath(APPS_PATH + "/gravitino-job-1")
+ .andReturn(200, withoutStatus)
+ .times(2);
+ server
+ .expect()
+ .delete()
+ .withPath(APPS_PATH + "/gravitino-job-1")
+ .andReturn(409, new StatusBuilder().withCode(409).build())
+ .once();
+
+ Assertions.assertEquals(
+ JobHandle.Status.STARTED,
executor.getJobExecutionInfo(jobExecutionId).status());
+ Assertions.assertNotNull(sparkApplication("gravitino-job-1"));
+ }
+
+ @Test
+ public void testJobWithoutStatusGoneBeforeDeletion() {
+ String jobExecutionId = executor.submitJob(context(1L), sparkTemplate());
+ now.set(now.get().plusSeconds(601));
+ // Deleted by a cancellation on another server between the read and the
deletion.
+ server
+ .expect()
+ .delete()
+ .withPath(APPS_PATH + "/gravitino-job-1")
+ .andReturn(404, new StatusBuilder().withCode(404).build())
+ .once();
+
+ // Missing rather than failed, which settles a cancelling job as cancelled.
+ Assertions.assertThrows(
+ NoSuchJobException.class, () ->
executor.getJobExecutionInfo(jobExecutionId));
+ }
+
+ @Test
+ public void testJobWithoutStatusConflictRetriesRunOut() {
+ String jobExecutionId = executor.submitJob(context(1L), sparkTemplate());
+ now.set(now.get().plusSeconds(601));
+ server
+ .expect()
+ .delete()
+ .withPath(APPS_PATH + "/gravitino-job-1")
+ .andReturn(409, new StatusBuilder().withCode(409).build())
+ .times(4);
+
+ // The job isn't failed while it can't be deleted.
+ KubernetesClientException e =
+ Assertions.assertThrows(
+ KubernetesClientException.class, () ->
executor.getJobExecutionInfo(jobExecutionId));
+ Assertions.assertEquals(409, e.getCode());
+ Assertions.assertNotNull(sparkApplication("gravitino-job-1"));
+
+ 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());
@@ -595,6 +680,242 @@ public class TestK8sJobExecutor {
}
}
+ @Test
+ public void testSubmitWithServerError() throws InterruptedException {
+ // The SparkApplication was created, but an error came back instead of the
response.
+ GenericKubernetesResource created =
+ SparkApplicationUtils.buildSparkApplication(
+ context(1L), sparkTemplate(),
TestSparkApplicationUtils.newConfigs(ImmutableMap.of()));
+ client
+ .genericKubernetesResources(SparkApplicationUtils.SPARK_APPLICATION)
+ .inNamespace("default")
+ .resource(created)
+ .create();
+ String resourceVersion =
sparkApplication("gravitino-job-1").getMetadata().getResourceVersion();
+ server
+ .expect()
+ .post()
+ .withPath(APPS_PATH)
+ .andReturn(500, new StatusBuilder().withCode(500).withMessage("etcd
timeout").build())
+ .always();
+
+ try (KubernetesClient noRetries =
clientWithoutRetries(client.getConfiguration())) {
+
executor.initialize(TestSparkApplicationUtils.newConfigs(ImmutableMap.of()),
noRetries);
+ KubernetesClientException e =
+ Assertions.assertThrows(
+ KubernetesClientException.class,
+ () -> executor.submitJob(context(1L), sparkTemplate()));
+ Assertions.assertEquals(500, e.getCode());
+ }
+
+ // Gravitino records no job, so the SparkApplication is deleted, if it is
still the one read.
+ Assertions.assertNull(sparkApplication("gravitino-job-1"));
+ String delete = lastRequest("DELETE").getBody().readUtf8();
+ Assertions.assertTrue(
+ delete.contains("\"resourceVersion\":\"" + resourceVersion + "\""),
delete);
+ Assertions.assertTrue(delete.contains("\"Foreground\""), delete);
+ }
+
+ @Test
+ public void testSubmitWithServerErrorAndFailedCleanup() {
+ server
+ .expect()
+ .post()
+ .withPath(APPS_PATH)
+ .andReturn(500, new StatusBuilder().withCode(500).withMessage("etcd
timeout").build())
+ .always();
+ server
+ .expect()
+ .get()
+ .withPath(APPS_PATH + "/gravitino-job-1")
+ .andReturn(403, new StatusBuilder().withCode(403).build())
+ .always();
+
+ try (KubernetesClient noRetries =
clientWithoutRetries(client.getConfiguration())) {
+
executor.initialize(TestSparkApplicationUtils.newConfigs(ImmutableMap.of()),
noRetries);
+ // The failure of the cleanup doesn't hide the failure of the submission.
+ KubernetesClientException e =
+ Assertions.assertThrows(
+ KubernetesClientException.class,
+ () -> executor.submitJob(context(1L), sparkTemplate()));
+ Assertions.assertEquals(500, e.getCode());
+ }
+ }
+
+ @Test
+ public void testSubmitCleanupKeepsResourceOfAnotherJob() {
+ // A SparkApplication with the name of the job, but of another one.
+ 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();
+ server
+ .expect()
+ .post()
+ .withPath(APPS_PATH)
+ .andReturn(503, new StatusBuilder().withCode(503).build())
+ .always();
+
+ try (KubernetesClient noRetries =
clientWithoutRetries(client.getConfiguration())) {
+
executor.initialize(TestSparkApplicationUtils.newConfigs(ImmutableMap.of()),
noRetries);
+ Assertions.assertThrows(
+ KubernetesClientException.class, () ->
executor.submitJob(context(1L), sparkTemplate()));
+ }
+ Assertions.assertNotNull(sparkApplication("gravitino-job-1"));
+ }
+
+ @Test
+ public void testSubmitCleanupRereadsChangedResource() {
+ GenericKubernetesResource created =
+ SparkApplicationUtils.buildSparkApplication(
+ context(1L), sparkTemplate(),
TestSparkApplicationUtils.newConfigs(ImmutableMap.of()));
+ client
+ .genericKubernetesResources(SparkApplicationUtils.SPARK_APPLICATION)
+ .inNamespace("default")
+ .resource(created)
+ .create();
+ server
+ .expect()
+ .post()
+ .withPath(APPS_PATH)
+ .andReturn(500, new StatusBuilder().withCode(500).build())
+ .always();
+ // It changes between the read and the deletion, so it is read again and
deleted then.
+ server
+ .expect()
+ .delete()
+ .withPath(APPS_PATH + "/gravitino-job-1")
+ .andReturn(409, new StatusBuilder().withCode(409).build())
+ .once();
+
+ try (KubernetesClient noRetries =
clientWithoutRetries(client.getConfiguration())) {
+
executor.initialize(TestSparkApplicationUtils.newConfigs(ImmutableMap.of()),
noRetries);
+ Assertions.assertThrows(
+ KubernetesClientException.class, () ->
executor.submitJob(context(1L), sparkTemplate()));
+ }
+ Assertions.assertNull(sparkApplication("gravitino-job-1"));
+ }
+
+ @Test
+ public void testStdoutOfDeletedSparkApplication() throws
InterruptedException {
+ String jobExecutionId = executor.submitJob(context(1L), sparkTemplate());
+ createDriverPod("driver", "spark-kubernetes-driver", "sidecar");
+ // The SparkApplication is gone, its driver pod is retained longer.
+ client
+ .genericKubernetesResources(SparkApplicationUtils.SPARK_APPLICATION)
+ .inNamespace("default")
+ .withName("gravitino-job-1")
+ .delete();
+ server
+ .expect()
+ .get()
+ .withPath(
+ "/api/v1/namespaces/default/pods/driver/log?pretty=false"
+ + "&container=spark-kubernetes-driver&tailLines=10")
+ .andReturn(200, "line 1\nline 2\n")
+ .always();
+
+ // The driver container has the default name, as the SparkApplication
can't tell another one.
+ Assertions.assertEquals(
+ ImmutableList.of("line 1", "line 2"),
executor.getJobStdout(jobExecutionId, 10, 1024));
+ }
+
+ @Test
+ public void testListPageByPage() {
+ executor.initialize(
+ TestSparkApplicationUtils.newConfigs(
+ ImmutableMap.of(K8sJobExecutorConfigs.STATUS_CACHE_TTL_MS,
"60000")),
+ client);
+ String job1 = executor.submitJob(context(1L), sparkTemplate());
+ String job2 = executor.submitJob(context(2L), sparkTemplate());
+ GenericKubernetesResource app1 = sparkApplication("gravitino-job-1");
+ GenericKubernetesResource app2 = sparkApplication("gravitino-job-2");
+ setStates("gravitino-job-1", "Submitted", "DriverReady");
+ setStates("gravitino-job-2", "Submitted", "DriverReady");
+ String list = APPS_PATH +
"?labelSelector=app.kubernetes.io%2Fmanaged-by%3Dgravitino";
+ server
+ .expect()
+ .get()
+ .withPath(list + "&limit=" + K8sJobExecutor.LIST_PAGE_SIZE)
+ .andReturn(200, listOf("next", app1))
+ .once();
+ server
+ .expect()
+ .get()
+ .withPath(
+ APPS_PATH
+ +
"?continue=next&labelSelector=app.kubernetes.io%2Fmanaged-by%3Dgravitino&limit="
+ + K8sJobExecutor.LIST_PAGE_SIZE)
+ .andReturn(200, listOf(null, app2))
+ .once();
+
+ // Both come from the listing, which predates their new state.
+ Assertions.assertEquals(JobHandle.Status.QUEUED,
executor.getJobExecutionInfo(job1).status());
+ Assertions.assertEquals(JobHandle.Status.QUEUED,
executor.getJobExecutionInfo(job2).status());
+ }
+
+ @Test
+ public void testSlowListingFallsBackToGet() {
+ String jobExecutionId = executor.submitJob(context(1L), sparkTemplate());
+ server
+ .expect()
+ .get()
+ .delay(2, TimeUnit.SECONDS)
+ .withPath(
+ APPS_PATH
+ +
"?labelSelector=app.kubernetes.io%2Fmanaged-by%3Dgravitino&limit="
+ + K8sJobExecutor.LIST_PAGE_SIZE)
+ .andReturn(200, listOf(null))
+ .always();
+
+ try (KubernetesClient slow =
+ // Retried like the client of the executor, after which the timeout is
still the cause.
+ new KubernetesClientBuilder()
+ .withConfig(
+ new ConfigBuilder(client.getConfiguration())
+ .withRequestTimeout(200)
+ .withRequestRetryBackoffLimit(2)
+ .withRequestRetryBackoffInterval(10)
+ .build())
+ .build()) {
+ executor.initialize(
+ TestSparkApplicationUtils.newConfigs(
+ ImmutableMap.of(K8sJobExecutorConfigs.STATUS_CACHE_TTL_MS,
"60000")),
+ slow);
+
+ // The API server answers, only too slowly to list the namespace: the
job is got by itself,
+ // and the next queries aren't failed fast.
+ Assertions.assertEquals(
+ JobHandle.Status.QUEUED,
executor.getJobExecutionInfo(jobExecutionId).status());
+ setStates("gravitino-job-1", "Submitted", "DriverReady");
+ Assertions.assertEquals(
+ JobHandle.Status.STARTED,
executor.getJobExecutionInfo(jobExecutionId).status());
+ }
+ }
+
+ @Test
+ public void testIsResponseTimedOut() {
+ Assertions.assertTrue(
+ K8sJobExecutor.isResponseTimedOut(
+ new KubernetesClientException(
+ "failed", new IOException(new HttpTimeoutException("request
timed out")))));
+ Assertions.assertFalse(
+ K8sJobExecutor.isResponseTimedOut(
+ new KubernetesClientException(
+ "failed", new HttpConnectTimeoutException("connect timed
out"))));
+ Assertions.assertFalse(
+ K8sJobExecutor.isResponseTimedOut(
+ new KubernetesClientException("failed", new IOException(new
ConnectException()))));
+ Assertions.assertFalse(
+ K8sJobExecutor.isResponseTimedOut(
+ new KubernetesClientException(new
StatusBuilder().withCode(504).build())));
+ }
+
@Test
public void testStdoutOfRenamedDriverContainer() throws InterruptedException
{
String jobExecutionId = executor.submitJob(context(1L), sparkTemplate());
@@ -642,7 +963,10 @@ public class TestK8sJobExecutor {
server
.expect()
.get()
- .withPath(APPS_PATH +
"?labelSelector=app.kubernetes.io%2Fmanaged-by%3Dgravitino")
+ .withPath(
+ APPS_PATH
+ +
"?labelSelector=app.kubernetes.io%2Fmanaged-by%3Dgravitino&limit="
+ + K8sJobExecutor.LIST_PAGE_SIZE)
.andReturn(403, new StatusBuilder().withCode(403).build())
.always();
@@ -1190,6 +1514,20 @@ public class TestK8sJobExecutor {
return last;
}
+ private static GenericKubernetesResourceList listOf(
+ String continueToken, GenericKubernetesResource... items) {
+ GenericKubernetesResourceList list = new GenericKubernetesResourceList();
+ list.setMetadata(new
ListMetaBuilder().withContinue(continueToken).build());
+ list.setItems(ImmutableList.copyOf(items));
+ return list;
+ }
+
+ private static KubernetesClient clientWithoutRetries(Config config) {
+ return new KubernetesClientBuilder()
+ .withConfig(new
ConfigBuilder(config).withRequestRetryBackoffLimit(0).build())
+ .build();
+ }
+
private static JobContext context(long jobId) {
return new JobContext(jobId, "ml", new File("/tmp/job-" + jobId));
}
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
index aa358c0ee4..8587524210 100644
---
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
@@ -19,6 +19,7 @@
package org.apache.gravitino.job.k8s;
import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableSet;
import io.fabric8.kubernetes.api.model.ConfigMap;
import io.fabric8.kubernetes.api.model.ConfigMapBuilder;
import java.time.Duration;
@@ -95,11 +96,39 @@ public class TestNamespacedResourceCache {
now::get);
Assertions.assertFalse(cache.find("ns1", "a").isPresent());
- // The namespace isn't listed again until the failure expires.
+ // The namespace isn't listed again until the failure expires, which takes
longer than the TTL
+ // of a listing.
+ Assertions.assertFalse(cache.find("ns1", "b").isPresent());
+ now.addAndGet(100L);
Assertions.assertFalse(cache.find("ns1", "b").isPresent());
Assertions.assertEquals(ImmutableList.of("ns1"), listed);
- now.addAndGet(100L);
+ now.addAndGet(NamespacedResourceCache.FAILED_LIST_TTL_MS - 100L);
+ Assertions.assertFalse(cache.find("ns1", "a").isPresent());
+ Assertions.assertEquals(ImmutableList.of("ns1", "ns1"), listed);
+ }
+
+ @Test
+ public void testSlowListFailureIsCachedFromWhenItFailed() {
+ NamespacedResourceCache<ConfigMap> cache =
+ new NamespacedResourceCache<>(
+ 100L,
+ namespace -> {
+ listed.add(namespace);
+ // The listing times out after longer than a failure is cached.
+ now.addAndGet(NamespacedResourceCache.FAILED_LIST_TTL_MS + 50L);
+ throw new IllegalStateException("timed out");
+ },
+ e -> false,
+ now::get);
+
+ Assertions.assertFalse(cache.find("ns1", "a").isPresent());
+ // The failure isn't expired as soon as it is cached, so the next query
doesn't wait for
+ // another listing.
+ Assertions.assertFalse(cache.find("ns1", "b").isPresent());
+ Assertions.assertEquals(ImmutableList.of("ns1"), listed);
+
+ now.addAndGet(NamespacedResourceCache.FAILED_LIST_TTL_MS);
Assertions.assertFalse(cache.find("ns1", "a").isPresent());
Assertions.assertEquals(ImmutableList.of("ns1", "ns1"), listed);
}
@@ -143,6 +172,37 @@ public class TestNamespacedResourceCache {
Assertions.assertEquals(ImmutableList.of("ns1", "ns1"), listed);
}
+ @Test
+ public void testExpiredListingsAreEvicted() {
+ NamespacedResourceCache<ConfigMap> cache =
+ new NamespacedResourceCache<>(
+ 100L,
+ namespace -> {
+ listed.add(namespace);
+ return ImmutableList.of(resource(namespace, "a"));
+ },
+ e -> false,
+ now::get);
+ Assertions.assertTrue(cache.find("ns1", "a").isPresent());
+ now.addAndGet(50L);
+ Assertions.assertTrue(cache.find("ns2", "a").isPresent());
+ Assertions.assertEquals(ImmutableSet.of("ns1", "ns2"),
cache.getCachedNamespaces());
+
+ // Listing another namespace drops the expired listings, also of the
namespaces no longer
+ // queried, and keeps the others.
+ now.addAndGet(50L);
+ Assertions.assertTrue(cache.find("ns3", "a").isPresent());
+ Assertions.assertEquals(ImmutableSet.of("ns2", "ns3"),
cache.getCachedNamespaces());
+ // A cached listing is found without listing again, so nothing is evicted.
+ Assertions.assertTrue(cache.find("ns2", "a").isPresent());
+ Assertions.assertEquals(ImmutableSet.of("ns2", "ns3"),
cache.getCachedNamespaces());
+
+ now.addAndGet(100L);
+ Assertions.assertTrue(cache.find("ns2", "a").isPresent());
+ Assertions.assertEquals(ImmutableSet.of("ns2"),
cache.getCachedNamespaces());
+ Assertions.assertEquals(ImmutableList.of("ns1", "ns2", "ns3", "ns2"),
listed);
+ }
+
@Test
public void testInvalidationOfAnotherNamespace() {
AtomicReference<NamespacedResourceCache<ConfigMap>> cache = new
AtomicReference<>();
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 499a94ed5d..5404cc33eb 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
@@ -651,6 +651,33 @@ public class TestSparkApplicationStatusUtils {
.finishedAt());
}
+ @Test
+ public void testIsDriverRunning() {
+ GenericKubernetesResource app = app("Submitted", "DriverRequested",
"DriverStartTimedOut");
+ Pod running = driverPod("spark-kubernetes-driver", "2026-09-30T00:00:10Z",
null);
+ Pod terminated =
+ driverPod("spark-kubernetes-driver", "2026-09-30T00:00:10Z",
"2026-09-30T00:00:20Z");
+
+ Assertions.assertTrue(SparkApplicationStatusUtils.isDriverRunning(app,
running));
+ Assertions.assertFalse(SparkApplicationStatusUtils.isDriverRunning(app,
terminated));
+ // Nothing tells the driver has terminated: a pending pod, or one without
a status.
+ Assertions.assertTrue(SparkApplicationStatusUtils.isDriverRunning(app,
podInPhase("Pending")));
+ Assertions.assertTrue(SparkApplicationStatusUtils.isDriverRunning(app, new
Pod()));
+
+ // A terminated pod runs nothing, even if its container status still says
running, for example
+ // after its node was lost.
+ for (String phase : new String[] {"Succeeded", "Failed"}) {
+ running.getStatus().setPhase(phase);
+ Assertions.assertFalse(SparkApplicationStatusUtils.isDriverRunning(app,
running), phase);
+ Assertions.assertFalse(
+ SparkApplicationStatusUtils.isDriverRunning(app, podInPhase(phase)),
phase);
+ }
+
+ // A sidecar keeps the pod running after the driver exited.
+ terminated.getStatus().setPhase("Running");
+ Assertions.assertFalse(SparkApplicationStatusUtils.isDriverRunning(app,
terminated));
+ }
+
@Test
public void testHasState() {
Assertions.assertFalse(SparkApplicationStatusUtils.hasState(app()));
@@ -843,6 +870,10 @@ public class TestSparkApplicationStatusUtils {
.build();
}
+ private static Pod podInPhase(String phase) {
+ return new
PodBuilder().withNewStatus().withPhase(phase).endStatus().build();
+ }
+
private static void cancel(GenericKubernetesResource app) {
app.getMetadata()
.setAnnotations(ImmutableMap.of(K8sJobResourceUtils.ANNOTATION_CANCEL_REQUESTED,
"true"));
