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 8d2c01bd11 [#13146] fix(core): Track local jobs only by their owning
executor in multi-node deployments (#13147)
8d2c01bd11 is described below
commit 8d2c01bd111676a4b03e54b00b3d2472756e5a3d
Author: Jerry Shao <[email protected]>
AuthorDate: Tue Sep 15 12:02:08 2026 +0800
[#13146] fix(core): Track local jobs only by their owning executor in
multi-node deployments (#13147)
### What changes were proposed in this pull request?
Make each server only track the local jobs it runs itself, so that
servers sharing the same metadata store no longer overwrite each other's
job statuses.
- `LocalJobExecutor`
- Generates a random 8-hex-digit executor id on initialization and
embeds it in the job execution ids: `local-job-<executorId>-<uuid>`.
- Implements the new `JobExecutor` methods `ownsJob(jobId)` and
`isJobStateNodeLocal()`.
- Fixes the thread pool to run up to `maxRunningJobs` jobs concurrently
(core size was 0 with an unbounded queue, so only one job ran at a
time).
- `JobExecutor`: adds the default methods `ownsJob` (default `true`) and
`isJobStateNodeLocal` (default `false`), so external job executors keep
their current behavior.
- `JobManager`
- The status pull only queries the jobs owned by this server's executor,
and skips the others.
- `cancelJob` on a job owned by another server only marks it as
`CANCELLING`. The owning server cancels it on its next status pull. This
only applies to node-local job state.
- For node-local job executors, `cleanUpStagingDirs` marks active jobs
not updated for `gravitino.job.stagingDirKeepTimeInMs` as `FAILED` (or
`CANCELLED` if cancelling), e.g. jobs left behind by a server that
exited. They are then kept for another keep time and cleaned up like
other finished jobs.
- Docs: describe the multi-node behavior of the local job executor and
its limitations.
### Why are the changes needed?
`LocalJobExecutor` keeps job status only in the memory of the server
that submitted the job, but every server pulls the status of all active
jobs from the shared store. A server that doesn't own a job gets
`NoSuchJobException` from its own executor and marks the job as
`FAILED`, even if the job succeeded on the other server. Cancelling a
job through a non-owning server also failed with a 500 error.
Fix: #13146
### Does this PR introduce _any_ user-facing change?
- Job execution ids of the local job executor change from
`local-job-<uuid>` to `local-job-<executorId>-<uuid>`.
- Cancelling a job through a server that doesn't run it now returns
`CANCELLING` instead of failing. The cancellation takes effect on the
owning server's next status pull (up to
`gravitino.job.statusPullIntervalInMs`).
- For the local job executor, `gravitino.job.stagingDirKeepTimeInMs` now
also expires active jobs whose status has not changed for that long.
Jobs of other job executors are not affected.
- `gravitino.jobExecutor.local.maxRunningJobs` now takes effect.
- New default methods on the `JobExecutor` developer API: `ownsJob` and
`isJobStateNodeLocal`.
Known limitations, documented in `manage-jobs-in-gravitino.md`:
- A local job running, or staying queued, longer than
`gravitino.job.stagingDirKeepTimeInMs` without a status change is marked
as `FAILED`, even if it is still running. A normal local job isn't
expected to run that long; a lease-based liveness check would be the
long-term solution.
- The executor id changes on every restart, so a restarted server
doesn't recognize the jobs it ran before. These jobs are only marked as
`FAILED` once they expire, up to about 7.7 days with the default config.
Active jobs created before upgrading, with the old id format, are
handled the same way.
- During a rolling upgrade, servers still running the old version may
mark jobs run by upgraded servers as `FAILED`.
### How was this patch tested?
- Unit tests
- `TestLocalJobExecutor`: job ownership by executor id, and running jobs
concurrently up to `maxRunningJobs`.
- `TestJobManager`: skipping jobs owned by other executors, cross-node
cancellation, not re-cancelling jobs of non-node-local executors, and
expiring stale active jobs in the cleanup (including concurrent updates
and failures).
- New `TestJobManagerMultiNode`: two `JobManager`s with real
`LocalJobExecutor`s sharing a real relational store (H2, plus
MySQL/PostgreSQL with `dockerTest=true`) cover owner-only tracking,
cancelling through the other node, and expiring jobs left by an exited
node.
- Integration test `JobIT` (embedded mode) simulates another server's
job in the metadata store: it is not marked as `FAILED`, cancelling it
returns `CANCELLING`, and a stale one is marked as `FAILED`.
🤖 Generated with [Claude Code](https://claude.com/claude-code)
---------
Co-authored-by: Claude Opus 5 <[email protected]>
---
.../gravitino/client/integration/test/JobIT.java | 101 +++++-
.../gravitino/connector/job/JobExecutor.java | 32 ++
.../java/org/apache/gravitino/job/JobManager.java | 319 ++++++++++++------
.../gravitino/job/local/LocalJobExecutor.java | 35 +-
.../org/apache/gravitino/job/TestJobManager.java | 369 +++++++++++++++++++++
.../gravitino/job/TestJobManagerMultiNode.java | 288 ++++++++++++++++
.../gravitino/job/local/TestLocalJobExecutor.java | 83 +++++
docs/manage-jobs-in-gravitino.md | 38 ++-
8 files changed, 1157 insertions(+), 108 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 a093fca5a8..380affdcde 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);
@@ -529,6 +551,83 @@ 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);
+ Assertions.assertNotNull(metalake.getJob(job.name()).finishedAt());
+ }
+
+ 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())
+ .withStartedAt(status == JobHandle.Status.QUEUED ? 0L :
createTime.toEpochMilli())
+ .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 44e287a6a4..a86e7d4d91 100644
--- a/core/src/main/java/org/apache/gravitino/job/JobManager.java
+++ b/core/src/main/java/org/apache/gravitino/job/JobManager.java
@@ -29,6 +29,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;
@@ -36,6 +37,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;
@@ -542,12 +545,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
@@ -638,93 +661,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
startedAt/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 (OptimisticLockException e) {
- // A later poll re-reads both executor state and metadata.
Never stop the scheduled
- // task or replay external submission/cancellation because a
metadata CAS lost.
- LOG.info(
- "Job {} under metalake {} changed concurrently; deferring
status update",
- job.name(),
- metalake);
- return;
- } 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);
}
});
}
@@ -802,17 +743,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 -> isFinishedStatus(job.status()))
- .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(
@@ -1121,4 +1072,166 @@ 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
startedAt/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 (OptimisticLockException e) {
+ // A later poll re-reads both executor state and metadata. Never stop
the scheduled
+ // task or replay external submission/cancellation because a metadata
CAS lost.
+ LOG.info(
+ "Job {} under metalake {} changed concurrently; deferring status
update",
+ job.name(),
+ metalake);
+ return Optional.empty();
+ } 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 1ca2f43e6a..b3605dd576 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;
@@ -72,6 +74,7 @@ import
org.apache.gravitino.exceptions.NoSuchJobTemplateException;
import org.apache.gravitino.exceptions.NoSuchMetalakeException;
import org.apache.gravitino.exceptions.NonEmptyEntityException;
import org.apache.gravitino.exceptions.OptimisticLockException;
+import org.apache.gravitino.job.local.LocalJobExecutor;
import org.apache.gravitino.json.JsonUtils;
import org.apache.gravitino.lock.LockManager;
import org.apache.gravitino.meta.AuditInfo;
@@ -125,6 +128,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);
@@ -1290,6 +1295,303 @@ 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());
+ Assertions.assertEquals(startedJob.startedAt(),
expiredStartedJob.startedAt());
+ // 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());
+ }
+
/** Conflicts preserve files without stopping this cleanup batch or its next
scheduled run. */
@Test
public void testCleanUpStagingDirsContinuesAfterOccConflict() throws
IOException {
@@ -1692,6 +1994,45 @@ 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)
+ .withStartedAt(0L)
+ .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
@@ -1714,6 +2055,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..09bf25a3ca
--- /dev/null
+++ b/core/src/test/java/org/apache/gravitino/job/TestJobManagerMultiNode.java
@@ -0,0 +1,288 @@
+/*
+ * 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.AfterEach;
+import org.junit.jupiter.api.Assertions;
+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;
+
+ @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())
+ .withStartedAt(job.startedAt())
+ .withFinishedAt(job.finishedAt() > 0 ? job.finishedAt() -
offsetInMs : 0L)
+ .withRuntimeJobTemplate(job.runtimeJobTemplate())
+ .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:
