This is an automated email from the ASF dual-hosted git repository.
jerryshao pushed a commit to branch branch-1.3
in repository https://gitbox.apache.org/repos/asf/gravitino.git
The following commit(s) were added to refs/heads/branch-1.3 by this push:
new 6f6050d776 [Cherry-pick to branch-1.3] [#13146] fix(core): Track local
jobs only by their owning executor in multi-node deployments (#13147) (#13164)
6f6050d776 is described below
commit 6f6050d77659c31294f31f4cd5c0e8649201eb81
Author: github-actions[bot]
<41898282+github-actions[bot]@users.noreply.github.com>
AuthorDate: Tue Sep 15 13:55:19 2026 +0800
[Cherry-pick to branch-1.3] [#13146] fix(core): Track local jobs only by
their owning executor in multi-node deployments (#13147) (#13164)
**Cherry-pick Information:**
- Original commit: 8d2c01bd111676a4b03e54b00b3d2472756e5a3d
- Target branch: `branch-1.3`
- Status: ✅ **Conflicts resolved**
Conflicts in `JobManager.java` and `TestJobManager.java` were resolved
by keeping the changes from the original commit. Code depending on
features that are not in `branch-1.3` was removed:
`OptimisticLockException`, `JobEntity#startedAt`/`runtimeJobTemplate`
and `JobHandle#finishedAt`, as well as an unrelated OCC cleanup test
pulled in by the conflict block.
`TestJobManagerMultiNode` now restores the `GravitinoEnv` config and
lock manager after it runs, since `MetricsSource` on `branch-1.3` still
reads the global config and the metrics tests running later failed
otherwise.
Validation on `branch-1.3`: `./gradlew :core:test -PskipITs` (1579
tests) and `JobIT` pass.
---------
Co-authored-by: Jerry Shao <[email protected]>
Co-authored-by: Claude Opus 5 <[email protected]>
Co-authored-by: Jerry Shao <[email protected]>
---
.../gravitino/client/integration/test/JobIT.java | 99 +++++-
.../gravitino/connector/job/JobExecutor.java | 32 ++
.../java/org/apache/gravitino/job/JobManager.java | 307 +++++++++++------
.../gravitino/job/local/LocalJobExecutor.java | 35 +-
.../org/apache/gravitino/job/TestJobManager.java | 367 +++++++++++++++++++++
.../gravitino/job/TestJobManagerMultiNode.java | 306 +++++++++++++++++
.../gravitino/job/local/TestLocalJobExecutor.java | 83 +++++
docs/manage-jobs-in-gravitino.md | 38 ++-
8 files changed, 1163 insertions(+), 104 deletions(-)
diff --git
a/clients/client-java/src/test/java/org/apache/gravitino/client/integration/test/JobIT.java
b/clients/client-java/src/test/java/org/apache/gravitino/client/integration/test/JobIT.java
index 8681e5cd25..cea050cf55 100644
---
a/clients/client-java/src/test/java/org/apache/gravitino/client/integration/test/JobIT.java
+++
b/clients/client-java/src/test/java/org/apache/gravitino/client/integration/test/JobIT.java
@@ -21,29 +21,39 @@ package org.apache.gravitino.client.integration.test;
import com.google.common.collect.ImmutableMap;
import com.google.common.collect.Lists;
import java.io.File;
+import java.io.IOException;
import java.nio.file.Files;
+import java.time.Duration;
+import java.time.Instant;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.Set;
+import java.util.UUID;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;
import org.apache.commons.io.FileUtils;
+import org.apache.gravitino.GravitinoEnv;
import org.apache.gravitino.client.GravitinoMetalake;
import org.apache.gravitino.exceptions.JobTemplateAlreadyExistsException;
import org.apache.gravitino.exceptions.NoSuchJobException;
import org.apache.gravitino.exceptions.NoSuchJobTemplateException;
import org.apache.gravitino.integration.test.util.BaseIT;
import org.apache.gravitino.integration.test.util.GravitinoITUtils;
+import org.apache.gravitino.integration.test.util.ITUtils;
import org.apache.gravitino.job.JobHandle;
import org.apache.gravitino.job.JobTemplate;
import org.apache.gravitino.job.JobTemplateChange;
import org.apache.gravitino.job.ShellJobTemplate;
import org.apache.gravitino.job.SparkJobTemplate;
+import org.apache.gravitino.meta.AuditInfo;
+import org.apache.gravitino.meta.JobEntity;
+import org.apache.gravitino.utils.NamespaceUtil;
import org.awaitility.Awaitility;
import org.junit.jupiter.api.AfterAll;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Assumptions;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
@@ -52,6 +62,16 @@ public class JobIT extends BaseIT {
private static final String METALAKE_NAME =
GravitinoITUtils.genRandomName("job_it_metalake");
+ private static final long STATUS_PULL_INTERVAL_IN_MS = 3000L;
+
+ // Finished jobs, and active jobs not updated, for this long are cleaned up.
The cleanup runs
+ // every tenth of it.
+ private static final long JOB_KEEP_TIME_IN_MS = 60_000L;
+
+ // A job execution id that no job executor instance of this server owns, as
if the job was
+ // submitted by the local job executor of another Gravitino server sharing
the metadata store.
+ private static final String OTHER_SERVER_EXECUTION_ID_PREFIX =
"local-job-otherserver-";
+
private File testStagingDir;
private File testSparkHome;
private String testEntryScriptPath;
@@ -83,7 +103,9 @@ public class JobIT extends BaseIT {
"gravitino.job.stagingDir",
testStagingDir.getAbsolutePath(),
"gravitino.job.statusPullIntervalInMs",
- "3000",
+ String.valueOf(STATUS_PULL_INTERVAL_IN_MS),
+ "gravitino.job.stagingDirKeepTimeInMs",
+ String.valueOf(JOB_KEEP_TIME_IN_MS),
"gravitino.jobExecutor.local.sparkHome",
testSparkHome.getAbsolutePath());
registerCustomConfigs(configs);
@@ -458,6 +480,81 @@ public class JobIT extends BaseIT {
NoSuchJobException.class, () ->
metalake.cancelJob("non_existent_job_id"));
}
+ @Test
+ public void testJobOwnedByAnotherServerIsNotFailed() throws Exception {
+ Assumptions.assumeTrue(
+ ITUtils.EMBEDDED_TEST_MODE.equals(testMode),
+ "Simulating another server needs direct access to the server's
metadata store");
+ JobTemplate template = builder.withName("test_other_server_job").build();
+ metalake.registerJobTemplate(template);
+
+ // Another server's job can't be found in this server's job executor, but
this server must not
+ // mark it as FAILED when pulling job statuses.
+ JobEntity job = insertOtherServerJob(template.name(),
JobHandle.Status.STARTED, Instant.now());
+ Awaitility.await()
+ .during(STATUS_PULL_INTERVAL_IN_MS * 2, TimeUnit.MILLISECONDS)
+ .atMost(STATUS_PULL_INTERVAL_IN_MS * 2 + 1000, TimeUnit.MILLISECONDS)
+ .until(() -> metalake.getJob(job.name()).jobStatus() ==
JobHandle.Status.STARTED);
+ }
+
+ @Test
+ public void testCancelJobOwnedByAnotherServer() throws Exception {
+ Assumptions.assumeTrue(
+ ITUtils.EMBEDDED_TEST_MODE.equals(testMode),
+ "Simulating another server needs direct access to the server's
metadata store");
+ JobTemplate template =
builder.withName("test_other_server_cancel").build();
+ metalake.registerJobTemplate(template);
+ JobEntity job = insertOtherServerJob(template.name(),
JobHandle.Status.STARTED, Instant.now());
+
+ // This server can't cancel another server's job, so it marks the job as
CANCELLING for its
+ // owner to cancel, instead of failing the request.
+ JobHandle cancellingJob = metalake.cancelJob(job.name());
+ Assertions.assertEquals(JobHandle.Status.CANCELLING,
cancellingJob.jobStatus());
+ Awaitility.await()
+ .during(STATUS_PULL_INTERVAL_IN_MS * 2, TimeUnit.MILLISECONDS)
+ .atMost(STATUS_PULL_INTERVAL_IN_MS * 2 + 1000, TimeUnit.MILLISECONDS)
+ .until(() -> metalake.getJob(job.name()).jobStatus() ==
JobHandle.Status.CANCELLING);
+ }
+
+ @Test
+ public void testStaleActiveJobIsMarkedFailed() throws Exception {
+ Assumptions.assumeTrue(
+ ITUtils.EMBEDDED_TEST_MODE.equals(testMode),
+ "Simulating another server needs direct access to the server's
metadata store");
+ JobTemplate template = builder.withName("test_stale_job").build();
+ metalake.registerJobTemplate(template);
+
+ // The job was left behind by a server that exited long ago, so nobody
updates it anymore.
+ JobEntity job =
+ insertOtherServerJob(
+ template.name(), JobHandle.Status.STARTED,
Instant.now().minus(Duration.ofDays(30)));
+ Assertions.assertEquals(JobHandle.Status.STARTED,
metalake.getJob(job.name()).jobStatus());
+
+ // The cleanup marks the job as FAILED, as it has not been updated for
longer than the keep
+ // time. The failed job is kept for another keep time before being removed.
+ Awaitility.await()
+ .atMost(1, TimeUnit.MINUTES)
+ .until(() -> metalake.getJob(job.name()).jobStatus() ==
JobHandle.Status.FAILED);
+ }
+
+ private JobEntity insertOtherServerJob(
+ String templateName, JobHandle.Status status, Instant createTime) throws
IOException {
+ long jobId = GravitinoEnv.getInstance().idGenerator().nextId();
+ JobEntity job =
+ JobEntity.builder()
+ .withId(jobId)
+ .withJobExecutionId(OTHER_SERVER_EXECUTION_ID_PREFIX +
UUID.randomUUID())
+ .withJobTemplateName(templateName)
+ .withStatus(status)
+ .withNamespace(NamespaceUtil.ofJob(METALAKE_NAME))
+ .withAuditInfo(
+
AuditInfo.builder().withCreator("test").withCreateTime(createTime).build())
+ .withFinishedAt(0L)
+ .build();
+ GravitinoEnv.getInstance().entityStore().put(job, false /* overwrite */);
+ return job;
+ }
+
private String generateTestEntryScript() {
String content =
"#!/bin/bash\n"
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 40eacac9de..e14fb6dce1 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
@@ -80,4 +80,36 @@ public interface JobExecutor extends Closeable {
* @throws NoSuchJobException If the job with the given identifier does not
exist.
*/
void cancelJob(String jobId) throws NoSuchJobException;
+
+ /**
+ * Whether the job with the given identifier is owned by this job executor
instance, which means
+ * this instance is able to query and cancel it.
+ *
+ * <p>In a multi-node deployment every Gravitino server has its own job
executor instance, and all
+ * of them see the same jobs from the shared metadata store. Gravitino only
queries the status of,
+ * or cancels, a job through the executor instance that owns it. The default
implementation
+ * returns {@code true}, which fits job executors backed by an external job
runner that any
+ * Gravitino server can reach.
+ *
+ * @param jobId The unique identifier of the job.
+ * @return {@code true} if this executor instance owns the job, {@code
false} otherwise.
+ */
+ default boolean ownsJob(String jobId) {
+ return true;
+ }
+
+ /**
+ * Whether the job state is only kept by the executor instance that owns the
job, for example the
+ * processes launched on the local node. If so, only the owning instance can
cancel the job, so a
+ * cancellation requested on another Gravitino server is carried out by the
owner when it pulls
+ * the job status.
+ *
+ * <p>The default implementation returns {@code false}.
+ *
+ * @return {@code true} if the job state is local to the owning executor
instance, {@code false}
+ * otherwise.
+ */
+ default boolean isJobStateNodeLocal() {
+ return false;
+ }
}
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 501eb88a9c..bd6e8ecc8e 100644
--- a/core/src/main/java/org/apache/gravitino/job/JobManager.java
+++ b/core/src/main/java/org/apache/gravitino/job/JobManager.java
@@ -28,6 +28,7 @@ import java.io.IOException;
import java.net.URI;
import java.nio.file.Files;
import java.time.Instant;
+import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.Map;
@@ -35,6 +36,8 @@ import java.util.Optional;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.function.Function;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
import java.util.stream.Collectors;
@@ -498,12 +501,32 @@ public class JobManager implements JobOperationDispatcher
{
return jobEntity;
}
- // Cancel the job using the job executor
- try {
- jobExecutor.cancelJob(jobEntity.jobExecutionId());
- } catch (Exception e) {
- throw new RuntimeException(
- String.format("Failed to cancel job with ID %s under metalake %s",
jobId, metalake), e);
+ // Cancel the job using the job executor if this server owns the job.
Otherwise, the job runs
+ // on another server, so only mark it as CANCELLING below, and the owning
server cancels it
+ // when it pulls the job status next time.
+ if (jobExecutor.ownsJob(jobEntity.jobExecutionId())) {
+ try {
+ jobExecutor.cancelJob(jobEntity.jobExecutionId());
+ } catch (NoSuchJobException e) {
+ // The job is lost by the job executor, mark it as CANCELLING and the
status pull will
+ // settle it as CANCELLED.
+ LOG.warn(
+ "Job {} with execution id {} under metalake {} is not found in the
job executor, "
+ + "marking it as CANCELLING",
+ jobId,
+ jobEntity.jobExecutionId(),
+ metalake);
+ } catch (Exception e) {
+ throw new RuntimeException(
+ String.format("Failed to cancel job with ID %s under metalake %s",
jobId, metalake), e);
+ }
+ } else {
+ LOG.info(
+ "Job {} with execution id {} under metalake {} is owned by another
job executor "
+ + "instance, marking it as CANCELLING for the owner to cancel
it",
+ jobId,
+ jobEntity.jobExecutionId(),
+ metalake);
}
// Update the job status to CANCELING
@@ -591,85 +614,11 @@ public class JobManager implements JobOperationDispatcher
{
activeJobs.forEach(
job -> {
- JobHandle.Status newStatus = job.status();
- try {
- newStatus = jobExecutor.getJobStatus(job.jobExecutionId());
- } catch (NoSuchJobException e) {
- // If the job is not found in the external job executor, we
assume the job is
- // FAILED if it is not in CANCELLING status, otherwise we assume
it is CANCELLED.
- if (job.status() == JobHandle.Status.CANCELLING) {
- newStatus = JobHandle.Status.CANCELLED;
- } else {
- newStatus = JobHandle.Status.FAILED;
- }
- LOG.warn(
- "Job {} with execution id {} under metalake {} is not found
in the "
- + "external job executor, marking it as {}. This could
be due to the job "
- + "being deleted by the external job executor. Please
check the external job "
- + "executor to know more details.",
- job.name(),
- job.jobExecutionId(),
- metalake,
- newStatus);
- } catch (Exception e) {
- LOG.error(
- "Failed to get job status for job {} by execution id {}",
- job.name(),
- job.jobExecutionId(),
- e);
- }
-
- if (newStatus != job.status()) {
- // Update the job entity with new status. entityStore.update()
re-fetches the
- // latest entity itself right before applying the updater, so
the transition below
- // is derived from latestJobEntity - the state as of right
before the write - rather
- // than the possibly-stale `job` snapshot taken by listJobs()
above. A concurrent
- // writer (e.g. cancelJob(), or another poll run) may have
already moved the job to
- // a terminal state, into CANCELLING, or recorded a real
finishedAt in the gap
- // between that snapshot and this point; the updater must not
regress any of that
- // using the stale snapshot's view of the world.
- JobHandle.Status finalNewStatus = newStatus;
- JobEntity updated;
- try {
- updated =
- TreeLockUtils.doWithTreeLock(
- NameIdentifierUtil.ofJob(metalake, job.name()),
- LockType.WRITE,
- () -> {
- try {
- return entityStore.update(
- NameIdentifierUtil.ofJob(metalake, job.name()),
- JobEntity.class,
- Entity.EntityType.JOB,
- latestJobEntity ->
- toUpdatedStatusJobEntity(latestJobEntity,
finalNewStatus));
- } catch (IOException e) {
- throw new RuntimeException(
- String.format(
- "Failed to update job entity %s to status
%s",
- job.name(), finalNewStatus),
- e);
- }
- });
- } catch (NoSuchEntityException e) {
- // The job could have been deleted concurrently (e.g. by
legacy-timeline cleanup)
- // in the gap between the listJobs() snapshot above and this
update. Skip it rather
- // than letting the exception escape this scheduled task,
which would silently
- // cancel all future status-pull runs
(ScheduledExecutorService semantics).
- LOG.warn(
- "Job {} under metalake {} no longer exists, skipping
status update to {}. "
- + "This could be due to the job being deleted
concurrently.",
- job.name(),
- metalake,
- finalNewStatus);
- return;
- }
-
- LOG.info(
- "Updated the job {} with execution id {} status to {}",
- job.name(),
- job.jobExecutionId(),
- updated.status());
+ // Only the job executor instance owning the job can query its
status. The jobs
+ // owned by other servers are skipped, and the jobs left behind by
a server that has
+ // exited are settled by cleanUpStagingDirs() once they expire.
+ if (jobExecutor.ownsJob(job.jobExecutionId())) {
+ pullAndUpdateOwnedJobStatus(metalake, job);
}
});
}
@@ -725,21 +674,27 @@ public class JobManager implements JobOperationDispatcher
{
List<String> metalakes = MetalakeManager.listInUseMetalakes(entityStore);
for (String metalake : metalakes) {
- List<JobEntity> finishedJobs =
- listJobs(metalake, Optional.empty()).stream()
- .filter(
- job ->
- job.status() == JobHandle.Status.CANCELLED
- || job.status() == JobHandle.Status.SUCCEEDED
- || job.status() == JobHandle.Status.FAILED)
- .filter(
- job ->
- job.finishedAt() > 0
- && job.finishedAt() + jobStagingDirKeepTimeInMs
- < System.currentTimeMillis())
- .toList();
+ long now = System.currentTimeMillis();
+ List<JobEntity> expiredJobs = new ArrayList<>();
+ for (JobEntity job : listJobs(metalake, Optional.empty())) {
+ if (isFinishedStatus(job.status())) {
+ if (job.finishedAt() > 0 && job.finishedAt() +
jobStagingDirKeepTimeInMs < now) {
+ expiredJobs.add(job);
+ }
+ } else if (jobExecutor.isJobStateNodeLocal() && isStaleActiveJob(job,
now)) {
+ // The state of a node local job is lost when the Gravitino server
running it exits, so
+ // an active job that has not been updated for the whole retention
time is considered
+ // left behind. Mark it as finished, and it is cleaned up once it
expires as a finished
+ // job. Jobs of other job executors can be tracked by any server, so
they never expire.
+ try {
+ expireStaleActiveJob(metalake, job, now);
+ } catch (RuntimeException e) {
+ LOG.error("Failed to expire job {} under metalake {}", job.name(),
metalake, e);
+ }
+ }
+ }
- finishedJobs.forEach(
+ expiredJobs.forEach(
job -> {
try {
entityStore.delete(
@@ -1041,4 +996,158 @@ public class JobManager implements
JobOperationDispatcher {
private <T> T updatedValue(T currentValue, Optional<T> newValue) {
return newValue.orElse(currentValue);
}
+
+ private void pullAndUpdateOwnedJobStatus(String metalake, JobEntity job) {
+ JobHandle.Status newStatus = job.status();
+ try {
+ newStatus = jobExecutor.getJobStatus(job.jobExecutionId());
+ // The job was marked as CANCELLING by another server, which can't
cancel it itself, so
+ // cancel it here as the owner. This only applies to node local job
state, other job
+ // executors are cancelled directly by the server handling the request,
and they may keep
+ // reporting the job as running while cancelling it asynchronously.
+ if (jobExecutor.isJobStateNodeLocal()
+ && job.status() == JobHandle.Status.CANCELLING
+ && (newStatus == JobHandle.Status.QUEUED || newStatus ==
JobHandle.Status.STARTED)) {
+ newStatus = cancelOwnedJob(metalake, job);
+ }
+ } catch (NoSuchJobException e) {
+ // If the job is not found in the external job executor, we assume the
job is
+ // FAILED if it is not in CANCELLING status, otherwise we assume it is
CANCELLED.
+ if (job.status() == JobHandle.Status.CANCELLING) {
+ newStatus = JobHandle.Status.CANCELLED;
+ } else {
+ newStatus = JobHandle.Status.FAILED;
+ }
+ LOG.warn(
+ "Job {} with execution id {} under metalake {} is not found in the "
+ + "external job executor, marking it as {}. This could be due to
the job "
+ + "being deleted by the external job executor. Please check the
external job "
+ + "executor to know more details.",
+ job.name(),
+ job.jobExecutionId(),
+ metalake,
+ newStatus);
+ } catch (Exception e) {
+ // Keep the job unchanged, and retry it in the next poll.
+ newStatus = job.status();
+ LOG.error(
+ "Failed to pull or cancel job {} by execution id {}",
+ job.name(),
+ job.jobExecutionId(),
+ e);
+ }
+
+ if (newStatus != job.status()) {
+ // Update the job entity with new status. entityStore.update()
re-fetches the
+ // latest entity itself right before applying the updater, so the
transition below
+ // is derived from latestJobEntity - the state as of right before the
write - rather
+ // than the possibly-stale `job` snapshot taken by listJobs() above. A
concurrent
+ // writer (e.g. cancelJob(), or another poll run) may have already moved
the job to
+ // a terminal state, into CANCELLING, or recorded a real finishedAt in
the
+ // gap between that snapshot and this point; the updater must not
regress any of
+ // that using the stale snapshot's view of the world.
+ JobHandle.Status finalNewStatus = newStatus;
+ updateJobEntity(
+ metalake,
+ job,
+ latestJobEntity -> toUpdatedStatusJobEntity(latestJobEntity,
finalNewStatus))
+ .ifPresent(
+ updated ->
+ LOG.info(
+ "Updated the job {} with execution id {} status to {}",
+ job.name(),
+ job.jobExecutionId(),
+ updated.status()));
+ }
+ }
+
+ private JobHandle.Status cancelOwnedJob(String metalake, JobEntity job) {
+ LOG.info(
+ "Cancelling job {} with execution id {} under metalake {} as it is
marked as CANCELLING",
+ job.name(),
+ job.jobExecutionId(),
+ metalake);
+ jobExecutor.cancelJob(job.jobExecutionId());
+ return jobExecutor.getJobStatus(job.jobExecutionId());
+ }
+
+ private Optional<JobEntity> updateJobEntity(
+ String metalake, JobEntity job, Function<JobEntity, JobEntity> updater) {
+ try {
+ return Optional.of(
+ TreeLockUtils.doWithTreeLock(
+ NameIdentifierUtil.ofJob(metalake, job.name()),
+ LockType.WRITE,
+ () -> {
+ try {
+ return entityStore.update(
+ NameIdentifierUtil.ofJob(metalake, job.name()),
+ JobEntity.class,
+ Entity.EntityType.JOB,
+ updater);
+ } catch (IOException e) {
+ throw new RuntimeException(
+ String.format("Failed to update job entity %s",
job.name()), e);
+ }
+ }));
+ } catch (NoSuchEntityException e) {
+ // The job could have been deleted concurrently (e.g. by legacy-timeline
cleanup)
+ // in the gap between the listJobs() snapshot above and this update.
Skip it rather
+ // than letting the exception escape this scheduled task, which would
silently
+ // cancel all future status-pull runs (ScheduledExecutorService
semantics).
+ LOG.warn(
+ "Job {} under metalake {} no longer exists, skipping the update.
This could be due "
+ + "to the job being deleted concurrently.",
+ job.name(),
+ metalake);
+ return Optional.empty();
+ }
+ }
+
+ private void expireStaleActiveJob(String metalake, JobEntity job, long now) {
+ AtomicBoolean expired = new AtomicBoolean(false);
+ updateJobEntity(
+ metalake,
+ job,
+ latestJobEntity -> {
+ // The job may have been updated since the listJobs() snapshot.
+ if (!isStaleActiveJob(latestJobEntity, now)) {
+ return latestJobEntity;
+ }
+ expired.set(true);
+ // The expired job gets the current time as its finished time,
so it's kept for
+ // another retention time like any other finished job before
being cleaned up.
+ return toUpdatedStatusJobEntity(
+ latestJobEntity,
+ latestJobEntity.status() == JobHandle.Status.CANCELLING
+ ? JobHandle.Status.CANCELLED
+ : JobHandle.Status.FAILED);
+ })
+ .filter(expiredJob -> expired.get())
+ .ifPresent(
+ expiredJob ->
+ LOG.warn(
+ "Job {} with execution id {} under metalake {} has not
been updated for more "
+ + "than {} ms, marking it as {}. This could be due to
the Gravitino server "
+ + "running the job having exited.",
+ job.name(),
+ job.jobExecutionId(),
+ metalake,
+ jobStagingDirKeepTimeInMs,
+ expiredJob.status()));
+ }
+
+ private boolean isStaleActiveJob(JobEntity job, long now) {
+ return !isFinishedStatus(job.status())
+ && lastUpdatedTimeInMs(job) + jobStagingDirKeepTimeInMs < now;
+ }
+
+ private static long lastUpdatedTimeInMs(JobEntity job) {
+ AuditInfo auditInfo = job.auditInfo();
+ Instant lastUpdatedTime =
+ auditInfo.lastModifiedTime() != null
+ ? auditInfo.lastModifiedTime()
+ : auditInfo.createTime();
+ return lastUpdatedTime == null ? 0L : lastUpdatedTime.toEpochMilli();
+ }
}
diff --git
a/core/src/main/java/org/apache/gravitino/job/local/LocalJobExecutor.java
b/core/src/main/java/org/apache/gravitino/job/local/LocalJobExecutor.java
index 295a0a8483..e441e7b207 100644
--- a/core/src/main/java/org/apache/gravitino/job/local/LocalJobExecutor.java
+++ b/core/src/main/java/org/apache/gravitino/job/local/LocalJobExecutor.java
@@ -37,6 +37,7 @@ import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ThreadLocalRandom;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import org.apache.commons.lang3.tuple.Pair;
@@ -56,6 +57,12 @@ public class LocalJobExecutor implements JobExecutor {
private static final long UNEXPIRED_TIME_IN_MS = -1L;
+ // A random id of this executor instance, it is embedded in every job id
submitted by this
+ // instance, so that each Gravitino server only tracks the jobs it runs
itself.
+ private String executorId;
+
+ private String ownedJobIdPrefix;
+
private Map<String, String> configs;
private BlockingQueue<Pair<String, JobTemplate>> waitingQueue;
@@ -80,6 +87,9 @@ public class LocalJobExecutor implements JobExecutor {
@Override
public void initialize(Map<String, String> configs) {
this.configs = configs;
+ this.executorId = String.format("%08x",
ThreadLocalRandom.current().nextInt());
+ this.ownedJobIdPrefix = LOCAL_JOB_PREFIX + executorId + "-";
+ LOG.info("Initializing local job executor with executor id {}",
executorId);
int waitingQueueSize =
configs.containsKey(WAITING_QUEUE_SIZE)
@@ -109,9 +119,11 @@ public class LocalJobExecutor implements JobExecutor {
Preconditions.checkArgument(
maxRunningJobs > 0, "Max running jobs must be greater than 0, but got:
%s", maxRunningJobs);
- this.jobExecutorService =
+ // With an unbounded queue, the pool never grows beyond its core size, so
the core size must be
+ // the max running jobs. Idle core threads are still allowed to time out.
+ ThreadPoolExecutor threadPoolExecutor =
new ThreadPoolExecutor(
- 0,
+ maxRunningJobs,
maxRunningJobs,
60L,
TimeUnit.SECONDS,
@@ -122,6 +134,8 @@ public class LocalJobExecutor implements JobExecutor {
thread.setDaemon(true);
return thread;
});
+ threadPoolExecutor.allowCoreThreadTimeOut(true);
+ this.jobExecutorService = threadPoolExecutor;
this.jobStatus = Maps.newHashMap();
@@ -171,7 +185,7 @@ public class LocalJobExecutor implements JobExecutor {
SparkProcessBuilder.resolveSparkSubmit(configs);
}
- String newJobId = LOCAL_JOB_PREFIX + UUID.randomUUID();
+ String newJobId = ownedJobIdPrefix + UUID.randomUUID();
Pair<String, JobTemplate> jobPair = Pair.of(newJobId, jobTemplate);
synchronized (lock) {
@@ -240,6 +254,16 @@ public class LocalJobExecutor implements JobExecutor {
}
}
+ @Override
+ public boolean ownsJob(String jobId) {
+ return jobId != null && jobId.startsWith(ownedJobIdPrefix);
+ }
+
+ @Override
+ public boolean isJobStateNodeLocal() {
+ return true;
+ }
+
@Override
public void close() throws IOException {
// Mark the executor as finished to stop processing jobs
@@ -340,6 +364,11 @@ public class LocalJobExecutor implements JobExecutor {
}
}
+ @VisibleForTesting
+ String executorId() {
+ return executorId;
+ }
+
@VisibleForTesting
void cleanupJobStatus() {
long currentTime = System.currentTimeMillis();
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 6ffb8c2688..1d73004281 100644
--- a/core/src/test/java/org/apache/gravitino/job/TestJobManager.java
+++ b/core/src/test/java/org/apache/gravitino/job/TestJobManager.java
@@ -41,6 +41,7 @@ import java.nio.charset.StandardCharsets;
import java.nio.file.Files;
import java.nio.file.Path;
import java.time.Instant;
+import java.time.temporal.ChronoUnit;
import java.util.Collections;
import java.util.List;
import java.util.Optional;
@@ -49,6 +50,7 @@ import java.util.UUID;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.function.Function;
+import javax.annotation.Nullable;
import org.apache.commons.io.FileUtils;
import org.apache.commons.lang3.ArrayUtils;
import org.apache.commons.lang3.reflect.FieldUtils;
@@ -68,6 +70,7 @@ import org.apache.gravitino.exceptions.NoSuchEntityException;
import org.apache.gravitino.exceptions.NoSuchJobException;
import org.apache.gravitino.exceptions.NoSuchJobTemplateException;
import org.apache.gravitino.exceptions.NoSuchMetalakeException;
+import org.apache.gravitino.job.local.LocalJobExecutor;
import org.apache.gravitino.lock.LockManager;
import org.apache.gravitino.meta.AuditInfo;
import org.apache.gravitino.meta.BaseMetalake;
@@ -120,6 +123,8 @@ public class TestJobManager {
entityStore = Mockito.mock(EntityStore.class);
jobExecutor = Mockito.mock(JobExecutor.class);
+ // Mocks don't call the default interface methods, so stub the defaults
explicitly.
+ when(jobExecutor.ownsJob(any())).thenReturn(true);
idGenerator = new RandomIdGenerator();
JobManager jm = new JobManager(config, entityStore, idGenerator,
jobExecutor);
jobManager = Mockito.spy(jm);
@@ -895,6 +900,302 @@ public class TestJobManager {
Assertions.assertEquals(0L, result.finishedAt());
}
+ @Test
+ public void testPullJobStatusSkipsJobOwnedByAnotherExecutor() throws
IOException {
+ // The jobs are run by the job executor on another server, so this server
must neither query
+ // their status (the local executor can't find them) nor touch the job
entities, no matter how
+ // long they have not been updated.
+ Instant longAgo = Instant.now().minus(30, ChronoUnit.DAYS);
+ JobEntity freshJob =
+ newJobEntity("local-job-other-1", JobHandle.Status.QUEUED,
Instant.now(), null);
+ JobEntity staleJob =
+ newJobEntity("local-job-other-2", JobHandle.Status.STARTED, longAgo,
longAgo);
+ mockListActiveJobs(freshJob, staleJob);
+ when(jobExecutor.ownsJob(any())).thenReturn(false);
+ when(jobExecutor.isJobStateNodeLocal()).thenReturn(true);
+
+ Assertions.assertDoesNotThrow(() -> jobManager.pullAndUpdateJobStatus());
+
+ verify(jobExecutor, never()).getJobStatus(any());
+ verify(entityStore, never())
+ .update(any(), eq(JobEntity.class), eq(Entity.EntityType.JOB), any());
+ }
+
+ @Test
+ public void
testPullJobStatusCancelsOwnedJobMarkedCancellingByAnotherServer() throws
IOException {
+ // A STARTED job marked as CANCELLING by another server is cancelled by
its owner, and stays
+ // CANCELLING until the process exits, so there's nothing to update.
+ JobEntity startedJob =
+ newJobEntity("local-job-mine-1", JobHandle.Status.CANCELLING,
Instant.now(), null);
+ mockListActiveJobs(startedJob);
+ when(jobExecutor.isJobStateNodeLocal()).thenReturn(true);
+ when(jobExecutor.getJobStatus(startedJob.jobExecutionId()))
+ .thenReturn(JobHandle.Status.STARTED, JobHandle.Status.CANCELLING);
+ stubEntityStoreUpdateToApply(startedJob);
+
+ Assertions.assertDoesNotThrow(() -> jobManager.pullAndUpdateJobStatus());
+
+ verify(jobExecutor, times(1)).cancelJob(startedJob.jobExecutionId());
+ verify(entityStore, never())
+ .update(any(), eq(JobEntity.class), eq(Entity.EntityType.JOB), any());
+
+ // A QUEUED job marked as CANCELLING by another server is cancelled right
away.
+ Mockito.clearInvocations(entityStore, jobExecutor);
+ JobEntity queuedJob =
+ newJobEntity("local-job-mine-2", JobHandle.Status.CANCELLING,
Instant.now(), null);
+ mockListActiveJobs(queuedJob);
+ when(jobExecutor.getJobStatus(queuedJob.jobExecutionId()))
+ .thenReturn(JobHandle.Status.QUEUED, JobHandle.Status.CANCELLED);
+ stubEntityStoreUpdateToApply(queuedJob);
+
+ Assertions.assertDoesNotThrow(() -> jobManager.pullAndUpdateJobStatus());
+
+ verify(jobExecutor, times(1)).cancelJob(queuedJob.jobExecutionId());
+ JobEntity cancelled = captureUpdatedJobEntity(queuedJob);
+ Assertions.assertEquals(JobHandle.Status.CANCELLED, cancelled.status());
+ Assertions.assertTrue(cancelled.finishedAt() > 0);
+ }
+
+ @Test
+ public void testPullJobStatusKeepsCancellingWhenOwnerFailsToCancel() throws
IOException {
+ JobEntity job =
+ newJobEntity("local-job-mine-1", JobHandle.Status.CANCELLING,
Instant.now(), null);
+ mockListActiveJobs(job);
+ when(jobExecutor.isJobStateNodeLocal()).thenReturn(true);
+
when(jobExecutor.getJobStatus(job.jobExecutionId())).thenReturn(JobHandle.Status.STARTED);
+ doThrow(new RuntimeException("cancel
failed")).when(jobExecutor).cancelJob(any());
+ stubEntityStoreUpdateToApply(job);
+
+ Assertions.assertDoesNotThrow(() -> jobManager.pullAndUpdateJobStatus());
+
+ // Nothing changes, and the next poll retries the cancellation.
+ verify(jobExecutor, times(1)).cancelJob(job.jobExecutionId());
+ verify(entityStore, never())
+ .update(any(), eq(JobEntity.class), eq(Entity.EntityType.JOB), any());
+ }
+
+ @Test
+ public void testPullJobStatusDoesNotRecancelJobOfNonNodeLocalExecutor()
throws IOException {
+ // An external job executor is cancelled directly by the server handling
the cancel request,
+ // and may keep reporting the job as running while cancelling it
asynchronously. The status
+ // pull must not cancel it again on every poll.
+ JobEntity job =
+ newJobEntity("external-job-1", JobHandle.Status.CANCELLING,
Instant.now(), null);
+ mockListActiveJobs(job);
+
when(jobExecutor.getJobStatus(job.jobExecutionId())).thenReturn(JobHandle.Status.STARTED);
+ stubEntityStoreUpdateToApply(job);
+
+ Assertions.assertDoesNotThrow(() -> jobManager.pullAndUpdateJobStatus());
+
+ verify(jobExecutor, never()).cancelJob(any());
+ // The observed STARTED status never regresses the CANCELLING job.
+ Assertions.assertEquals(JobHandle.Status.CANCELLING,
captureUpdatedJobEntity(job).status());
+ }
+
+ @Test
+ public void testPullJobStatusDoesNotCancelFinishedJobMarkedCancelling()
throws IOException {
+ // The job finished before its owner noticed the cancellation request.
+ JobEntity job =
+ newJobEntity("local-job-mine-1", JobHandle.Status.CANCELLING,
Instant.now(), null);
+ mockListActiveJobs(job);
+ when(jobExecutor.isJobStateNodeLocal()).thenReturn(true);
+
when(jobExecutor.getJobStatus(job.jobExecutionId())).thenReturn(JobHandle.Status.SUCCEEDED);
+ stubEntityStoreUpdateToApply(job);
+
+ Assertions.assertDoesNotThrow(() -> jobManager.pullAndUpdateJobStatus());
+
+ verify(jobExecutor, never()).cancelJob(any());
+ Assertions.assertEquals(JobHandle.Status.SUCCEEDED,
captureUpdatedJobEntity(job).status());
+ }
+
+ @Test
+ public void testCancelJobOwnedByAnotherExecutor() throws IOException {
+ mockedMetalake
+ .when(() -> MetalakeManager.checkMetalake(metalakeIdent, entityStore))
+ .thenAnswer(a -> null);
+
+ JobEntity job =
+ newJobEntity("local-job-other-1", JobHandle.Status.STARTED,
Instant.now(), null);
+ doReturn(job).when(jobManager).getJob(metalake, job.name());
+ when(jobExecutor.ownsJob(job.jobExecutionId())).thenReturn(false);
+ stubEntityStoreUpdateToApply(job);
+
+ // The job runs on another server, so only mark it as CANCELLING for its
owner to cancel.
+ JobEntity cancellingJob = jobManager.cancelJob(metalake, job.name());
+ Assertions.assertEquals(JobHandle.Status.CANCELLING,
cancellingJob.status());
+ verify(jobExecutor, never()).cancelJob(any());
+ }
+
+ @Test
+ public void testCancelJobNotFoundInOwnedExecutor() throws IOException {
+ mockedMetalake
+ .when(() -> MetalakeManager.checkMetalake(metalakeIdent, entityStore))
+ .thenAnswer(a -> null);
+
+ JobEntity job = newJobEntity("local-job-mine-1", JobHandle.Status.STARTED,
Instant.now(), null);
+ doReturn(job).when(jobManager).getJob(metalake, job.name());
+ doThrow(new
NoSuchJobException("lost")).when(jobExecutor).cancelJob(job.jobExecutionId());
+ stubEntityStoreUpdateToApply(job);
+
+ // The status pull later settles a job lost by the executor as CANCELLED.
+ JobEntity cancellingJob = jobManager.cancelJob(metalake, job.name());
+ Assertions.assertEquals(JobHandle.Status.CANCELLING,
cancellingJob.status());
+ }
+
+ @Test
+ public void testPullJobStatusAcrossServersWithLocalJobExecutors() throws
Exception {
+ // Reproduces the multi-node deployment: two servers share the same
metadata store, and each
+ // has its own local job executor. The server that didn't run the job must
not mark it FAILED.
+ LocalJobExecutor ownerExecutor = new LocalJobExecutor();
+ LocalJobExecutor otherExecutor = new LocalJobExecutor();
+ ownerExecutor.initialize(Collections.emptyMap());
+ otherExecutor.initialize(Collections.emptyMap());
+ JobManager ownerManager =
+ Mockito.spy(new JobManager(config, entityStore, idGenerator,
ownerExecutor));
+ JobManager otherManager =
+ Mockito.spy(new JobManager(config, entityStore, idGenerator,
otherExecutor));
+ File jobStagingDir =
Files.createTempDirectory("gravitino-test-multi-node-job").toFile();
+
+ try {
+ JobTemplate jobTemplate =
+ JobManager.createRuntimeJobTemplate(
+ newShellJobTemplateEntity("shell_job", "echo"),
+ Collections.emptyMap(),
+ jobStagingDir);
+ String executionId = ownerExecutor.submitJob(jobTemplate);
+ Awaitility.await()
+ .atMost(1, TimeUnit.MINUTES)
+ .until(() -> ownerExecutor.getJobStatus(executionId) ==
JobHandle.Status.SUCCEEDED);
+
+ JobEntity job =
+ newJobEntity(
+ idGenerator.nextId(), executionId, JobHandle.Status.QUEUED,
Instant.now(), null);
+ mockedMetalake
+ .when(() -> MetalakeManager.listInUseMetalakes(entityStore))
+ .thenReturn(ImmutableList.of(metalake));
+ doReturn(ImmutableList.of(job)).when(otherManager).listJobs(metalake,
Optional.empty());
+ doReturn(ImmutableList.of(job)).when(ownerManager).listJobs(metalake,
Optional.empty());
+ stubEntityStoreUpdateToApply(job);
+
+ otherManager.pullAndUpdateJobStatus();
+ verify(entityStore, never())
+ .update(any(), eq(JobEntity.class), eq(Entity.EntityType.JOB),
any());
+
+ ownerManager.pullAndUpdateJobStatus();
+ Assertions.assertEquals(JobHandle.Status.SUCCEEDED,
captureUpdatedJobEntity(job).status());
+ } finally {
+ ownerManager.close();
+ otherManager.close();
+ FileUtils.deleteDirectory(jobStagingDir);
+ }
+ }
+
+ @Test
+ public void testCleanUpStagingDirsExpiresStaleActiveJobs() throws
IOException {
+ // Active jobs that have not been updated for the whole retention time are
left behind, e.g. by
+ // a server that exited while running them. They are marked as finished,
and kept for another
+ // retention time before being cleaned up.
+ Instant longAgo = Instant.now().minus(30, ChronoUnit.DAYS);
+ JobEntity queuedJob = newJobEntity("local-job-gone-1",
JobHandle.Status.QUEUED, longAgo, null);
+ JobEntity startedJob =
+ newJobEntity("local-job-gone-2", JobHandle.Status.STARTED, longAgo,
longAgo);
+ JobEntity cancellingJob =
+ newJobEntity("local-job-gone-3", JobHandle.Status.CANCELLING, longAgo,
longAgo);
+ JobEntity activeJob =
+ newJobEntity("local-job-mine-1", JobHandle.Status.STARTED, longAgo,
Instant.now());
+ mockListActiveJobs(queuedJob, startedJob, cancellingJob, activeJob);
+ when(jobExecutor.isJobStateNodeLocal()).thenReturn(true);
+ for (JobEntity job : ImmutableList.of(queuedJob, startedJob,
cancellingJob, activeJob)) {
+ stubEntityStoreUpdateToApply(job, job);
+ }
+ File jobStagingDir = new File(testStagingDir, metalake + "/shell_job/" +
startedJob.name());
+ Assertions.assertTrue(jobStagingDir.mkdirs());
+
+ long beforeCleanUp = System.currentTimeMillis();
+ Assertions.assertDoesNotThrow(() -> jobManager.cleanUpStagingDirs());
+
+ Assertions.assertEquals(
+ JobHandle.Status.FAILED, captureUpdatedJobEntity(queuedJob,
queuedJob).status());
+ JobEntity expiredStartedJob = captureUpdatedJobEntity(startedJob,
startedJob);
+ Assertions.assertEquals(JobHandle.Status.FAILED,
expiredStartedJob.status());
+ // The expire time is used as the finished time, so the job is kept for
another retention time.
+ Assertions.assertTrue(expiredStartedJob.finishedAt() >= beforeCleanUp);
+ Assertions.assertEquals(
+ JobHandle.Status.CANCELLED, captureUpdatedJobEntity(cancellingJob,
cancellingJob).status());
+ verify(entityStore, never()).delete(any(), any());
+ Assertions.assertTrue(jobStagingDir.exists());
+
+ // A job updated recently is still active, so it's not expired.
+ verify(entityStore, never())
+ .update(
+ eq(NameIdentifierUtil.ofJob(metalake, activeJob.name())),
+ eq(JobEntity.class),
+ eq(Entity.EntityType.JOB),
+ any());
+ }
+
+ @Test
+ public void testCleanUpStagingDirsDoesNotExpireJobOfNonNodeLocalExecutor()
throws IOException {
+ // Any server can track the jobs of an external job executor, so they are
never considered left
+ // behind, no matter how long their status has not changed.
+ Instant longAgo = Instant.now().minus(30, ChronoUnit.DAYS);
+ JobEntity job = newJobEntity("external-job-1", JobHandle.Status.STARTED,
longAgo, longAgo);
+ mockListActiveJobs(job);
+
+ Assertions.assertDoesNotThrow(() -> jobManager.cleanUpStagingDirs());
+
+ verify(entityStore, never())
+ .update(any(), eq(JobEntity.class), eq(Entity.EntityType.JOB), any());
+ verify(entityStore, never()).delete(any(), any());
+ }
+
+ @Test
+ public void testCleanUpStagingDirsDoesNotExpireJobUpdatedConcurrently()
throws IOException {
+ // listJobs() observes a stale job, but it is updated before
entityStore.update() re-fetches
+ // it, so it's still active and must be kept.
+ Instant longAgo = Instant.now().minus(30, ChronoUnit.DAYS);
+ JobEntity staleSnapshot =
+ newJobEntity("local-job-other-1", JobHandle.Status.STARTED, longAgo,
longAgo);
+ JobEntity latestUpdated =
+ newJobEntity(
+ staleSnapshot.id(),
+ staleSnapshot.jobExecutionId(),
+ JobHandle.Status.STARTED,
+ longAgo,
+ Instant.now());
+ mockListActiveJobs(staleSnapshot);
+ when(jobExecutor.isJobStateNodeLocal()).thenReturn(true);
+ stubEntityStoreUpdateToApply(latestUpdated);
+
+ Assertions.assertDoesNotThrow(() -> jobManager.cleanUpStagingDirs());
+
+ Assertions.assertSame(latestUpdated,
captureUpdatedJobEntity(latestUpdated));
+ verify(entityStore, never()).delete(any(), any());
+ }
+
+ @Test
+ public void testCleanUpStagingDirsContinuesAfterFailingToExpireJob() throws
IOException {
+ Instant longAgo = Instant.now().minus(30, ChronoUnit.DAYS);
+ JobEntity failingJob =
+ newJobEntity("local-job-gone-1", JobHandle.Status.STARTED, longAgo,
longAgo);
+ JobEntity otherJob =
+ newJobEntity("local-job-gone-2", JobHandle.Status.STARTED, longAgo,
longAgo);
+ mockListActiveJobs(failingJob, otherJob);
+ when(jobExecutor.isJobStateNodeLocal()).thenReturn(true);
+ when(entityStore.update(
+ eq(NameIdentifierUtil.ofJob(metalake, failingJob.name())),
+ eq(JobEntity.class),
+ eq(Entity.EntityType.JOB),
+ any()))
+ .thenThrow(new IOException("store error"));
+ stubEntityStoreUpdateToApply(otherJob, otherJob);
+
+ Assertions.assertDoesNotThrow(() -> jobManager.cleanUpStagingDirs());
+
+ Assertions.assertEquals(
+ JobHandle.Status.FAILED, captureUpdatedJobEntity(otherJob,
otherJob).status());
+ }
+
@Test
public void testCleanUpStagingDirs() throws IOException,
InterruptedException {
JobEntity job = newJobEntity("shell_job", JobHandle.Status.STARTED);
@@ -1202,6 +1503,44 @@ public class TestJobManager {
.build();
}
+ private void mockListActiveJobs(JobEntity... jobs) {
+ mockedMetalake
+ .when(() -> MetalakeManager.listInUseMetalakes(entityStore))
+ .thenReturn(ImmutableList.of(metalake));
+ doReturn(ImmutableList.copyOf(jobs)).when(jobManager).listJobs(metalake,
Optional.empty());
+ }
+
+ private JobEntity newJobEntity(
+ String executionId,
+ JobHandle.Status status,
+ Instant createTime,
+ @Nullable Instant lastModifiedTime) {
+ return newJobEntity(idGenerator.nextId(), executionId, status, createTime,
lastModifiedTime);
+ }
+
+ private JobEntity newJobEntity(
+ long id,
+ String executionId,
+ JobHandle.Status status,
+ Instant createTime,
+ @Nullable Instant lastModifiedTime) {
+ return JobEntity.builder()
+ .withId(id)
+ .withJobExecutionId(executionId)
+ .withNamespace(NamespaceUtil.ofJob(metalake))
+ .withJobTemplateName("shell_job")
+ .withStatus(status)
+ .withFinishedAt(0L)
+ .withAuditInfo(
+ AuditInfo.builder()
+ .withCreator("test")
+ .withCreateTime(createTime)
+ .withLastModifier(lastModifiedTime == null ? null : "modifier")
+ .withLastModifiedTime(lastModifiedTime)
+ .build())
+ .build();
+ }
+
// cancelJob/pullAndUpdateJobStatus now go through entityStore.update(),
which re-fetches the
// latest entity and applies an updater function internally. Since
entityStore is a full mock,
// this stubs that re-fetch to hand back the given entity, mirroring what
the real
@@ -1224,6 +1563,34 @@ public class TestJobManager {
return captor.getValue().apply(latestJobEntity);
}
+ @SuppressWarnings("unchecked")
+ private void stubEntityStoreUpdateToApply(JobEntity job, JobEntity
latestJobEntity)
+ throws IOException {
+ when(entityStore.update(
+ eq(NameIdentifierUtil.ofJob(metalake, job.name())),
+ eq(JobEntity.class),
+ eq(Entity.EntityType.JOB),
+ any()))
+ .thenAnswer(
+ invocation -> {
+ Function<JobEntity, JobEntity> updater =
invocation.getArgument(3);
+ return updater.apply(latestJobEntity);
+ });
+ }
+
+ @SuppressWarnings("unchecked")
+ private JobEntity captureUpdatedJobEntity(JobEntity job, JobEntity
latestJobEntity)
+ throws IOException {
+ ArgumentCaptor<Function<JobEntity, JobEntity>> captor =
ArgumentCaptor.forClass(Function.class);
+ verify(entityStore, times(1))
+ .update(
+ eq(NameIdentifierUtil.ofJob(metalake, job.name())),
+ eq(JobEntity.class),
+ eq(Entity.EntityType.JOB),
+ captor.capture());
+ return captor.getValue().apply(latestJobEntity);
+ }
+
private HttpServer createLoopbackHttpServer(String response) throws
IOException {
HttpServer server = HttpServer.create(new InetSocketAddress("127.0.0.1",
0), 0);
server.createContext(
diff --git
a/core/src/test/java/org/apache/gravitino/job/TestJobManagerMultiNode.java
b/core/src/test/java/org/apache/gravitino/job/TestJobManagerMultiNode.java
new file mode 100644
index 0000000000..abed136a55
--- /dev/null
+++ b/core/src/test/java/org/apache/gravitino/job/TestJobManagerMultiNode.java
@@ -0,0 +1,306 @@
+/*
+ * 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;
+
+import com.google.common.collect.ImmutableMap;
+import com.google.common.collect.Lists;
+import java.io.File;
+import java.io.IOException;
+import java.nio.file.Files;
+import java.time.Instant;
+import java.util.Collections;
+import java.util.concurrent.TimeUnit;
+import org.apache.commons.io.FileUtils;
+import org.apache.commons.lang3.reflect.FieldUtils;
+import org.apache.gravitino.Config;
+import org.apache.gravitino.Configs;
+import org.apache.gravitino.Entity;
+import org.apache.gravitino.EntityStore;
+import org.apache.gravitino.GravitinoEnv;
+import org.apache.gravitino.cache.NoOpsCache;
+import org.apache.gravitino.connector.job.JobExecutor;
+import org.apache.gravitino.exceptions.NoSuchJobException;
+import org.apache.gravitino.job.local.LocalJobExecutor;
+import org.apache.gravitino.lock.LockManager;
+import org.apache.gravitino.meta.AuditInfo;
+import org.apache.gravitino.meta.JobEntity;
+import org.apache.gravitino.meta.JobTemplateEntity;
+import org.apache.gravitino.storage.RandomIdGenerator;
+import org.apache.gravitino.storage.relational.RelationalEntityStore;
+import org.apache.gravitino.storage.relational.TestJDBCBackend;
+import org.apache.gravitino.utils.NameIdentifierUtil;
+import org.apache.gravitino.utils.NamespaceUtil;
+import org.awaitility.Awaitility;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.TestTemplate;
+
+/**
+ * Tests the job system in a multi-node deployment: two Gravitino servers,
each with its own {@link
+ * JobManager} and {@link LocalJobExecutor}, share the same relational
metadata store. The status
+ * pull and the cleanup are triggered manually, so that the test controls
which node runs them when.
+ */
+public class TestJobManagerMultiNode extends TestJDBCBackend {
+
+ private static final String METALAKE = "metalake_job_multi_node";
+
+ private static final String TEMPLATE = "sleep_job";
+
+ // Active jobs not updated for this long are expired, and finished jobs are
cleaned up after it.
+ // The tests move the job timestamps back instead of waiting for it to
elapse.
+ private static final long JOB_KEEP_TIME_IN_MS = TimeUnit.HOURS.toMillis(1);
+
+ private File testDir;
+
+ private LocalJobExecutor executorA;
+
+ private LocalJobExecutor executorB;
+
+ private JobManager nodeA;
+
+ private JobManager nodeB;
+
+ private EntityStore entityStore;
+
+ private Object originalConfig;
+
+ private Object originalLockManager;
+
+ @BeforeAll
+ public void saveGravitinoEnv() throws IllegalAccessException {
+ // The backend extension and this test replace the global config and lock
manager, restore them
+ // afterwards so that other tests running in the same JVM are not affected.
+ originalConfig = FieldUtils.readField(GravitinoEnv.getInstance(),
"config", true);
+ originalLockManager = FieldUtils.readField(GravitinoEnv.getInstance(),
"lockManager", true);
+ }
+
+ @AfterAll
+ public void restoreGravitinoEnv() throws IllegalAccessException {
+ FieldUtils.writeField(GravitinoEnv.getInstance(), "config",
originalConfig, true);
+ FieldUtils.writeField(GravitinoEnv.getInstance(), "lockManager",
originalLockManager, true);
+ }
+
+ @BeforeEach
+ public void setUpNodes() throws Exception {
+ testDir =
Files.createTempDirectory("gravitino-test-job-multi-node").toFile();
+
+ Config config = new Config(false) {};
+ config.set(Configs.JOB_STAGING_DIR, new File(testDir,
"staging").getAbsolutePath());
+ config.set(Configs.JOB_STAGING_DIR_KEEP_TIME_IN_MS, JOB_KEEP_TIME_IN_MS);
+ FieldUtils.writeField(GravitinoEnv.getInstance(), "lockManager", new
LockManager(config), true);
+
+ // Both nodes share the same metadata store, backed by the relational
backend under test.
+ RelationalEntityStore relationalEntityStore = new RelationalEntityStore();
+ FieldUtils.writeField(relationalEntityStore, "backend", backend, true);
+ FieldUtils.writeField(relationalEntityStore, "cache", new
NoOpsCache(config), true);
+ entityStore = relationalEntityStore;
+
+ createAndInsertMakeLake(METALAKE);
+ backend.insert(newSleepJobTemplateEntity(), false);
+
+ executorA = new LocalJobExecutor();
+ executorA.initialize(Collections.emptyMap());
+ executorB = new LocalJobExecutor();
+ executorB.initialize(Collections.emptyMap());
+ nodeA = newJobManager(config, entityStore, executorA);
+ nodeB = newJobManager(config, entityStore, executorB);
+ }
+
+ @AfterEach
+ public void tearDownNodes() throws IOException {
+ // Closing a node also kills the job processes it runs.
+ if (nodeA != null) {
+ nodeA.close();
+ }
+ if (nodeB != null) {
+ nodeB.close();
+ }
+ FileUtils.deleteDirectory(testDir);
+ }
+
+ @TestTemplate
+ public void testJobIsOnlyTrackedByItsOwnerNode() throws IOException {
+ JobEntity job = nodeA.runJob(METALAKE, TEMPLATE,
ImmutableMap.of("seconds", "0"));
+ Assertions.assertTrue(executorA.ownsJob(job.jobExecutionId()));
+ Assertions.assertFalse(executorB.ownsJob(job.jobExecutionId()));
+ Awaitility.await()
+ .atMost(1, TimeUnit.MINUTES)
+ .until(() -> executorA.getJobStatus(job.jobExecutionId()) ==
JobHandle.Status.SUCCEEDED);
+
+ // Before the fix, node B couldn't find the job in its own executor and
marked it as FAILED.
+ nodeB.pullAndUpdateJobStatus();
+ JobEntity afterNodeBPull = getJob(job.name());
+ Assertions.assertEquals(JobHandle.Status.QUEUED, afterNodeBPull.status());
+ Assertions.assertEquals(job.auditInfo(), afterNodeBPull.auditInfo());
+
+ nodeA.pullAndUpdateJobStatus();
+ JobEntity afterNodeAPull = getJob(job.name());
+ Assertions.assertEquals(JobHandle.Status.SUCCEEDED,
afterNodeAPull.status());
+ Assertions.assertTrue(afterNodeAPull.finishedAt() > 0);
+ }
+
+ @TestTemplate
+ public void testCancelJobFromAnotherNode() throws IOException {
+ JobEntity job = runLongJobOnNodeA();
+
+ // Node B can't cancel the job itself, so it only marks the job as
CANCELLING.
+ JobEntity cancelling = nodeB.cancelJob(METALAKE, job.name());
+ Assertions.assertEquals(JobHandle.Status.CANCELLING, cancelling.status());
+ Assertions.assertEquals(JobHandle.Status.STARTED,
executorA.getJobStatus(job.jobExecutionId()));
+
+ // Node A cancels the job when it pulls the job status next time.
+ nodeA.pullAndUpdateJobStatus();
+ Awaitility.await()
+ .atMost(1, TimeUnit.MINUTES)
+ .until(() -> executorA.getJobStatus(job.jobExecutionId()) ==
JobHandle.Status.CANCELLED);
+ Assertions.assertEquals(JobHandle.Status.CANCELLING,
getJob(job.name()).status());
+
+ nodeA.pullAndUpdateJobStatus();
+ JobEntity cancelled = getJob(job.name());
+ Assertions.assertEquals(JobHandle.Status.CANCELLED, cancelled.status());
+ Assertions.assertTrue(cancelled.finishedAt() > 0);
+ }
+
+ @TestTemplate
+ public void testJobsLeftByExitedNodeAreExpiredByAnotherNode() throws
Exception {
+ JobEntity startedJob = runLongJobOnNodeA();
+ // Cancel this job from node B, node A exits before cancelling it.
+ JobEntity cancellingJob = nodeA.runJob(METALAKE, TEMPLATE,
ImmutableMap.of("seconds", "600"));
+ nodeB.cancelJob(METALAKE, cancellingJob.name());
+
+ // Node A exits, and nobody updates its jobs anymore.
+ nodeA.close();
+ nodeA = null;
+
+ // Neither the status pull nor the cleanup of node B touches the jobs
before they expire.
+ nodeB.pullAndUpdateJobStatus();
+ nodeB.cleanUpStagingDirs();
+ Assertions.assertEquals(JobHandle.Status.STARTED,
getJob(startedJob.name()).status());
+ Assertions.assertEquals(JobHandle.Status.CANCELLING,
getJob(cancellingJob.name()).status());
+
+ // Once the jobs have not been updated for the keep time, node B marks
them as finished.
+ moveJobTimestampsBack(startedJob.name());
+ moveJobTimestampsBack(cancellingJob.name());
+ nodeB.cleanUpStagingDirs();
+ JobEntity failedJob = getJob(startedJob.name());
+ JobEntity cancelledJob = getJob(cancellingJob.name());
+ Assertions.assertEquals(JobHandle.Status.FAILED, failedJob.status());
+ Assertions.assertEquals(JobHandle.Status.CANCELLED, cancelledJob.status());
+
+ // The finished jobs are kept for another keep time, and then cleaned up.
+ nodeB.cleanUpStagingDirs();
+ Assertions.assertTrue(jobExists(startedJob.name()));
+ Assertions.assertTrue(jobExists(cancellingJob.name()));
+
+ moveJobTimestampsBack(startedJob.name());
+ moveJobTimestampsBack(cancellingJob.name());
+ nodeB.cleanUpStagingDirs();
+ Assertions.assertFalse(jobExists(startedJob.name()));
+ Assertions.assertFalse(jobExists(cancellingJob.name()));
+ }
+
+ private JobEntity runLongJobOnNodeA() throws IOException {
+ JobEntity job = nodeA.runJob(METALAKE, TEMPLATE,
ImmutableMap.of("seconds", "600"));
+ Awaitility.await()
+ .atMost(1, TimeUnit.MINUTES)
+ .until(() -> executorA.getJobStatus(job.jobExecutionId()) ==
JobHandle.Status.STARTED);
+ nodeA.pullAndUpdateJobStatus();
+ Assertions.assertEquals(JobHandle.Status.STARTED,
getJob(job.name()).status());
+ return job;
+ }
+
+ private JobEntity getJob(String jobName) {
+ // Read through node B, as it works the same from any node sharing the
metadata store.
+ return nodeB.getJob(METALAKE, jobName);
+ }
+
+ // Simulates that the keep time has elapsed since the job was last updated,
or finished.
+ private void moveJobTimestampsBack(String jobName) throws IOException {
+ long offsetInMs = JOB_KEEP_TIME_IN_MS + TimeUnit.MINUTES.toMillis(1);
+ entityStore.update(
+ NameIdentifierUtil.ofJob(METALAKE, jobName),
+ JobEntity.class,
+ Entity.EntityType.JOB,
+ job ->
+ JobEntity.builder()
+ .withId(job.id())
+ .withJobExecutionId(job.jobExecutionId())
+ .withJobTemplateName(job.jobTemplateName())
+ .withStatus(job.status())
+ .withNamespace(job.namespace())
+ .withAuditInfo(
+ AuditInfo.builder()
+ .withCreator(job.auditInfo().creator())
+
.withCreateTime(job.auditInfo().createTime().minusMillis(offsetInMs))
+ .withLastModifier(job.auditInfo().lastModifier())
+ .withLastModifiedTime(
+ job.auditInfo().lastModifiedTime() == null
+ ? null
+ :
job.auditInfo().lastModifiedTime().minusMillis(offsetInMs))
+ .build())
+ .withFinishedAt(job.finishedAt() > 0 ? job.finishedAt() -
offsetInMs : 0L)
+ .build());
+ }
+
+ private boolean jobExists(String jobName) {
+ try {
+ getJob(jobName);
+ return true;
+ } catch (NoSuchJobException e) {
+ return false;
+ }
+ }
+
+ private JobTemplateEntity newSleepJobTemplateEntity() throws IOException {
+ File script = new File(testDir, "sleep-job.sh");
+ // Exec the sleep, so that killing the job process also stops the sleep.
+ Files.writeString(script.toPath(), "#!/bin/bash\nexec sleep \"$1\"\n");
+ Assertions.assertTrue(script.setExecutable(true));
+
+ return JobTemplateEntity.builder()
+ .withId(RandomIdGenerator.INSTANCE.nextId())
+ .withName(TEMPLATE)
+ .withNamespace(NamespaceUtil.ofJobTemplate(METALAKE))
+ .withTemplateContent(
+ JobTemplateEntity.TemplateContent.builder()
+ .withJobType(JobTemplate.JobType.SHELL)
+ .withExecutable(script.getAbsolutePath())
+ .withArguments(Lists.newArrayList("{{seconds}}"))
+ .withEnvironments(Collections.emptyMap())
+ .withCustomFields(Collections.emptyMap())
+ .withScripts(Collections.emptyList())
+ .build())
+ .withAuditInfo(
+
AuditInfo.builder().withCreator("test").withCreateTime(Instant.now()).build())
+ .build();
+ }
+
+ private static JobManager newJobManager(
+ Config config, EntityStore entityStore, JobExecutor jobExecutor) {
+ JobManager jobManager =
+ new JobManager(config, entityStore, RandomIdGenerator.INSTANCE,
jobExecutor);
+ // Stop the background schedulers, the test pulls the job status manually.
+ jobManager.statusPullExecutor.shutdownNow();
+ jobManager.cleanUpExecutor.shutdownNow();
+ return jobManager;
+ }
+}
diff --git
a/core/src/test/java/org/apache/gravitino/job/local/TestLocalJobExecutor.java
b/core/src/test/java/org/apache/gravitino/job/local/TestLocalJobExecutor.java
index 45b49e01e4..ec5f4d58d8 100644
---
a/core/src/test/java/org/apache/gravitino/job/local/TestLocalJobExecutor.java
+++
b/core/src/test/java/org/apache/gravitino/job/local/TestLocalJobExecutor.java
@@ -27,12 +27,15 @@ import java.net.URL;
import java.nio.file.Files;
import java.util.Collections;
import java.util.Map;
+import java.util.UUID;
import java.util.concurrent.TimeUnit;
import org.apache.commons.io.FileUtils;
import org.apache.gravitino.connector.job.JobExecutor;
+import org.apache.gravitino.exceptions.NoSuchJobException;
import org.apache.gravitino.job.JobHandle;
import org.apache.gravitino.job.JobManager;
import org.apache.gravitino.job.JobTemplate;
+import org.apache.gravitino.job.ShellJobTemplate;
import org.apache.gravitino.job.SparkJobTemplate;
import org.apache.gravitino.meta.AuditInfo;
import org.apache.gravitino.meta.JobTemplateEntity;
@@ -133,6 +136,67 @@ public class TestLocalJobExecutor {
Assertions.assertEquals(JobHandle.Status.SUCCEEDED,
jobExecutor.getJobStatus(jobId));
}
+ @Test
+ public void testJobOwnership() throws IOException {
+ LocalJobExecutor executor = (LocalJobExecutor) jobExecutor;
+ Assertions.assertTrue(executor.isJobStateNodeLocal());
+ Assertions.assertTrue(executor.executorId().matches("[0-9a-f]{8}"));
+
+ JobTemplate template =
+ JobManager.createRuntimeJobTemplate(
+ jobTemplateEntity,
+ ImmutableMap.of("arg1", "value1", "arg2", "success", "var",
"value3"),
+ workingDir);
+ String jobId = executor.submitJob(template);
+ Assertions.assertTrue(
+ jobId.matches("local-job-" + executor.executorId() +
"-[0-9a-f-]{36}"), jobId);
+ Assertions.assertTrue(executor.ownsJob(jobId));
+
+ // Jobs submitted before the executor id was introduced aren't owned by
any executor.
+ Assertions.assertFalse(executor.ownsJob("local-job-" + UUID.randomUUID()));
+ Assertions.assertFalse(executor.ownsJob(null));
+
+ LocalJobExecutor anotherExecutor = new LocalJobExecutor();
+ try {
+ anotherExecutor.initialize(Collections.emptyMap());
+ Assertions.assertNotEquals(executor.executorId(),
anotherExecutor.executorId());
+ Assertions.assertFalse(anotherExecutor.ownsJob(jobId));
+ Assertions.assertThrows(NoSuchJobException.class, () ->
anotherExecutor.getJobStatus(jobId));
+ } finally {
+ anotherExecutor.close();
+ }
+
+ Awaitility.await()
+ .atMost(3, TimeUnit.MINUTES)
+ .until(() -> executor.getJobStatus(jobId) ==
JobHandle.Status.SUCCEEDED);
+ }
+
+ @Test
+ public void testRunJobsConcurrentlyUpToMaxRunningJobs() throws IOException {
+ LocalJobExecutor executor = new LocalJobExecutor();
+ try {
+
executor.initialize(ImmutableMap.of(LocalJobExecutorConfigs.MAX_RUNNING_JOBS,
"2"));
+ String jobId1 = executor.submitJob(newSleepJobTemplate("sleep-1"));
+ String jobId2 = executor.submitJob(newSleepJobTemplate("sleep-2"));
+ String jobId3 = executor.submitJob(newSleepJobTemplate("sleep-3"));
+
+ // Up to maxRunningJobs jobs run at the same time, the others wait in
the queue.
+ Awaitility.await()
+ .atMost(1, TimeUnit.MINUTES)
+ .until(
+ () ->
+ executor.getJobStatus(jobId1) == JobHandle.Status.STARTED
+ && executor.getJobStatus(jobId2) ==
JobHandle.Status.STARTED);
+ Awaitility.await()
+ .during(1, TimeUnit.SECONDS)
+ .atMost(2, TimeUnit.SECONDS)
+ .until(() -> executor.getJobStatus(jobId3) ==
JobHandle.Status.QUEUED);
+ } finally {
+ // Closing the executor kills the running jobs.
+ executor.close();
+ }
+ }
+
@Test
public void testSubmitJobFailure() throws IOException {
Map<String, String> jobConf =
@@ -289,4 +353,23 @@ public class TestLocalJobExecutor {
long actualValue = (long) field.get(exec);
Assertions.assertEquals(11L, actualValue);
}
+
+ private JobTemplate newSleepJobTemplate(String name) throws IOException {
+ // The job runs in the directory of its executable, so give each job its
own directory.
+ File jobDir = new File(workingDir, name);
+ Assertions.assertTrue(jobDir.mkdirs());
+ File script = new File(jobDir, "sleep.sh");
+ // Exec the sleep, so that killing the job process also stops the sleep.
+ Files.writeString(script.toPath(), "#!/bin/bash\nexec sleep 600\n");
+ Assertions.assertTrue(script.setExecutable(true));
+
+ return ShellJobTemplate.builder()
+ .withName(name)
+ .withExecutable(script.getAbsolutePath())
+ .withArguments(Collections.emptyList())
+ .withEnvironments(Collections.emptyMap())
+ .withCustomFields(Collections.emptyMap())
+ .withScripts(Collections.emptyList())
+ .build();
+ }
}
diff --git a/docs/manage-jobs-in-gravitino.md b/docs/manage-jobs-in-gravitino.md
index c3e93de708..e210530b2c 100644
--- a/docs/manage-jobs-in-gravitino.md
+++ b/docs/manage-jobs-in-gravitino.md
@@ -236,7 +236,6 @@ 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 |
-
#### Configurations for Local Job Executor
The local job executor is used for testing and development purposes, it runs
the job in the local process.
@@ -249,6 +248,43 @@ The following are the default configurations for the local
job executor:
| `gravitino.jobExecutor.local.jobStatusKeepTimeInMs` | The time in
milliseconds to keep the job status in the local job executor
| `3600000` (1 hour)
| No |
| `gravitino.jobExecutor.local.sparkHome` | The home directory of
Spark, Gravitino checks this configuration firstly and then `SPARK_HOME` env.
Either of them should be set to run Spark job | `None`
| No |
+The local job executor runs up to `gravitino.jobExecutor.local.maxRunningJobs`
jobs at the same
+time, each in its own process on the Gravitino server host, and queues the
others. Make sure the
+host has enough resources for that many jobs, or lower this value, especially
when running Spark
+jobs.
+
+When multiple Gravitino servers share the same metadata store, each server's
local job executor
+only tracks the jobs it runs itself:
+
+- A job can only be run and tracked by the server that received the run
request. Other servers
+ skip it when pulling job statuses.
+- Cancelling a job on a server that doesn't run it marks the job as
`CANCELLING`, and the server
+ running the job cancels it the next time it pulls job statuses. This can
take up to
+ `gravitino.job.statusPullIntervalInMs`.
+- If a server exits while running jobs, nobody can track these jobs anymore.
When such a job has
+ not been updated for `gravitino.job.stagingDirKeepTimeInMs`, it is marked as
`FAILED`, or as
+ `CANCELLED` if it was being cancelled. Like other finished jobs, it is then
kept for another
+ `gravitino.job.stagingDirKeepTimeInMs` before being cleaned up together with
its staging
+ directory.
+
+:::caution
+The local job executor can't tell a job left behind by an exited server from a
job that is still
+running without changing its status. A job of the local job executor that is
still queued, started
+or cancelling after `gravitino.job.stagingDirKeepTimeInMs` is marked as
`FAILED` (or `CANCELLED`),
+even if the job is still running, and keeps this status even if it later
finishes. Set this time
+longer than any job can run, or stay queued, without changing its status.
+:::
+
+:::caution
+The local job executor gets a new identity every time the Gravitino server
starts, so a restarted
+server doesn't recognize the jobs it ran before the restart. This also applies
to a single-server
+deployment. The processes of these jobs are usually gone with the previous
server process, but the
+jobs are only marked as `FAILED` once they expire as described above, which
can take up to about
+1.1 times `gravitino.job.stagingDirKeepTimeInMs` (about 7.7 days by default),
as the cleanup runs
+every tenth of that time. Until then, they are still reported as queued,
started or cancelling.
+Cancelling such a job only marks it as `CANCELLING`, which also restarts the
expiration.
+:::
+
## Future Work
The job system still needs more work:
