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 23c489d761 [#14046] fix(core): Settle a local job whose cancellation
its owner never answers (#14047)
23c489d761 is described below
commit 23c489d7612860da40358b0b3c819b07acf0910a
Author: Jerry Shao <[email protected]>
AuthorDate: Thu Oct 22 10:01:32 2026 +0800
[#14046] fix(core): Settle a local job whose cancellation its owner never
answers (#14047)
### What changes were proposed in this pull request?
When a node-local job owned by another executor is still `CANCELLING`
three status pull intervals after the cancel request, the server pulling
job statuses marks it as `CANCELLED`.
- **`JobManager`**:
- In `pullAndUpdateJobStatus`, a job this server doesn't own is no
longer always skipped. If the executor is node-local, the job is
`CANCELLING`, and its last update is older than 3 ×
`gravitino.job.statusPullIntervalInMs`, it is marked as `CANCELLED` and
a warning is logged.
- Like `expireStaleActiveJob`, the condition is checked again on the
entity that the update re-reads. A job the owner settled in the meantime
is left as it is.
- **Docs**:
- Cancelling a job left behind by an exited or restarted server now
clears it, and its job template can then be deleted.
- A new caution states the assumptions: every server uses the same pull
interval, and the servers' clocks are in sync.
### Why are the changes needed?
The local job executor gets a new identity on every server start, so a
restarted server no longer owns the jobs it ran before. Cancelling such
a job only marks it as `CANCELLING` for an owner that is gone. The job
then stays `CANCELLING`, and its job template can't be deleted, until it
expires after `gravitino.job.stagingDirKeepTimeInMs` (about 7.7 days by
default).
A live owner answers a cancellation within about two pull intervals: it
cancels the job at its next pull, and records it as `CANCELLED` at the
pull after the process exits. So a job still `CANCELLING` after three
intervals has most likely been left behind, and cancelling across
servers keeps working. When several servers settle the same job at once,
OCC lets only one write through. The others are deferred to the next
pull, which then sees the job as finished.
Fix: #14046
### Does this PR introduce _any_ user-facing change?
Yes. With the local job executor, a job that another server owns and
that has been `CANCELLING` for more than 3 ×
`gravitino.job.statusPullIntervalInMs` is now marked as `CANCELLED`. If
its owner is still alive but doesn't answer in time, the job's process
may keep running. That can happen when the servers use different pull
intervals or their clocks are off, when the owner can't pull for that
long, or when the process takes longer than one interval to exit. The
docs describe these cases.
There are no API or configuration changes.
### How was this patch tested?
- **`TestJobManager`**:
- Settles a stale unanswered cancellation, including a job whose last
modified time is unset.
- Leaves a recent cancellation alone, and a job owned by another server
that isn't `CANCELLING`.
- Leaves jobs of non-node-local executors alone.
- Keeps a job that its owner settled concurrently.
- Tolerates an OCC conflict, and the next pull leaves the job as it is.
- **`TestJobManagerMultiNode`**:
- Node A exits and node B cancels its job. The job stays `CANCELLING`
and blocks deleting the template until the timeout, then node B marks it
as `CANCELLED` and the template can be deleted.
- A pull on node B before the timeout doesn't affect a cancellation that
the live node A then carries out.
- **`JobIT`** (embedded): a job of another server that is cancelled
becomes `CANCELLED`, and its template can then be deleted.
🤖 Generated with [Claude Code](https://claude.com/claude-code)
---------
Co-authored-by: Claude Opus 5.5 <[email protected]>
---
.../gravitino/client/integration/test/JobIT.java | 9 +
.../java/org/apache/gravitino/job/JobManager.java | 62 +++++-
.../org/apache/gravitino/job/TestJobManager.java | 240 +++++++++++++++++++++
.../gravitino/job/TestJobManagerMultiNode.java | 44 +++-
docs/manage-jobs-in-gravitino.md | 31 ++-
5 files changed, 378 insertions(+), 8 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 1442354417..363812907d 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
@@ -633,6 +633,15 @@ public class JobIT extends BaseIT {
.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);
+
+ // The other server never answers the cancellation, as it doesn't exist.
After three pull
+ // intervals this server considers it gone and marks the job as CANCELLED,
so the job template
+ // can be deleted.
+ Awaitility.await()
+ .atMost(STATUS_PULL_INTERVAL_IN_MS * 5, TimeUnit.MILLISECONDS)
+ .until(() -> metalake.getJob(job.name()).jobStatus() ==
JobHandle.Status.CANCELLED);
+ Assertions.assertNotNull(metalake.getJob(job.name()).finishedAt());
+ Assertions.assertTrue(metalake.deleteJobTemplate(template.name()));
}
@Test
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 c69d76332d..86887db750 100644
--- a/core/src/main/java/org/apache/gravitino/job/JobManager.java
+++ b/core/src/main/java/org/apache/gravitino/job/JobManager.java
@@ -110,6 +110,17 @@ public class JobManager implements JobOperationDispatcher {
private static final long JOB_STATUS_PULL_MIN_INTERVAL_IN_MS = 60 * 1000L;
// 1 minute
+ // A live owner of a node local job cancels it at its next status pull, and
records it as
+ // CANCELLED at the pull after the process exits, so it answers a
cancellation within about two
+ // pull intervals plus the time the process takes to exit. A job still
CANCELLING after this many
+ // pull intervals is considered left behind by an owner that has exited.
This assumes all servers
+ // use the same pull interval and have their clocks in sync, and that a
cancelled process exits
+ // within one pull interval, i.e.
gravitino.jobExecutor.local.cancelForceKillDelayInMs is shorter
+ // than it, which holds for the recommended intervals of 1 minute or more.
Otherwise a job may be
+ // marked as CANCELLED while its process is still exiting, after its owner
has already asked it to
+ // stop.
+ private static final int CANCELLATION_TIMEOUT_IN_PULL_INTERVALS = 3;
+
private final EntityStore entityStore;
private final File stagingDir;
@@ -124,6 +135,8 @@ public class JobManager implements JobOperationDispatcher {
private final int jobOutputMaxBytes;
+ private final long cancellationTimeoutInMs;
+
@VisibleForTesting final ScheduledExecutorService cleanUpExecutor;
@VisibleForTesting final ScheduledExecutorService statusPullExecutor;
@@ -200,6 +213,8 @@ public class JobManager implements JobOperationDispatcher {
jobStatusPullIntervalInMs,
JOB_STATUS_PULL_MIN_INTERVAL_IN_MS);
}
+ this.cancellationTimeoutInMs =
+ jobStatusPullIntervalInMs * CANCELLATION_TIMEOUT_IN_PULL_INTERVALS;
this.statusPullExecutor =
Executors.newSingleThreadScheduledExecutor(
runnable -> {
@@ -1031,17 +1046,21 @@ public class JobManager implements
JobOperationDispatcher {
|| job.status() == JobHandle.Status.CANCELLING)
.toList();
+ long now = System.currentTimeMillis();
activeJobs.forEach(
job -> {
// A failure on one job must never stop the status pull of the
others.
try {
// Only the job executor instance owning the job can query its
status. The jobs
- // owned by other servers are skipped, and the jobs left behind by
a server that has
- // exited are settled by cleanUpStagingDirs() once they expire.
- if (!jobExecutor.ownsJob(job.jobExecutionId())) {
- return;
+ // owned by other servers are skipped, except a node local job
whose cancellation its
+ // owner has not answered in time: the owner has likely exited,
e.g. the server
+ // restarted, so nobody else would ever settle it. The other jobs
left behind by a
+ // server that has exited are settled by cleanUpStagingDirs() once
they expire.
+ if (jobExecutor.ownsJob(job.jobExecutionId())) {
+ pullAndUpdateOwnedJobStatus(metalake, job);
+ } else if (jobExecutor.isJobStateNodeLocal() &&
isCancellationUnanswered(job, now)) {
+ settleUnansweredCancellation(metalake, job, now);
}
- pullAndUpdateOwnedJobStatus(metalake, job);
} catch (Throwable t) {
LOG.error(
"Failed to update the status of job {} with execution id {}
under metalake {}",
@@ -1443,6 +1462,39 @@ public class JobManager implements
JobOperationDispatcher {
&& lastUpdatedTimeInMs(job) + jobStagingDirKeepTimeInMs < now;
}
+ private void settleUnansweredCancellation(String metalake, JobEntity job,
long now) {
+ AtomicBoolean settled = new AtomicBoolean(false);
+ updateJobEntity(
+ metalake,
+ job,
+ latestJobEntity -> {
+ // The owner may have answered the cancellation since the
listJobs() snapshot.
+ if (!isCancellationUnanswered(latestJobEntity, now)) {
+ return latestJobEntity;
+ }
+ settled.set(true);
+ return toUpdatedStatusJobEntity(
+ latestJobEntity,
JobExecutionInfo.of(JobHandle.Status.CANCELLED));
+ })
+ .filter(settledJob -> settled.get())
+ .ifPresent(
+ settledJob ->
+ LOG.warn(
+ "Job {} with execution id {} under metalake {} has been
cancelling for more "
+ + "than {} ms without its owner answering, marking it
as {}. This could be "
+ + "due to the Gravitino server running the job having
exited.",
+ job.name(),
+ job.jobExecutionId(),
+ metalake,
+ cancellationTimeoutInMs,
+ settledJob.status()));
+ }
+
+ private boolean isCancellationUnanswered(JobEntity job, long now) {
+ return job.status() == JobHandle.Status.CANCELLING
+ && now - lastUpdatedTimeInMs(job) > cancellationTimeoutInMs;
+ }
+
private static long lastUpdatedTimeInMs(JobEntity job) {
AuditInfo auditInfo = job.auditInfo();
Instant lastUpdatedTime =
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 0699490a4a..a2a05f8fbb 100644
--- a/core/src/test/java/org/apache/gravitino/job/TestJobManager.java
+++ b/core/src/test/java/org/apache/gravitino/job/TestJobManager.java
@@ -1687,6 +1687,246 @@ public class TestJobManager {
.update(any(), eq(JobEntity.class), eq(Entity.EntityType.JOB), any());
}
+ @Test
+ public void
testPullJobStatusSettlesUnansweredCancellationOfJobOwnedByAnotherExecutor()
+ throws IOException {
+ // The job was cancelled an hour ago, far longer than a live owner takes
to answer, so its
+ // owner has likely exited, e.g. the server running it restarted.
+ Instant queuedAt = Instant.now().minus(2, ChronoUnit.HOURS);
+ Instant cancelledAt = Instant.now().minus(1, ChronoUnit.HOURS);
+ JobEntity job =
+ newJobEntity("local-job-gone-1", JobHandle.Status.CANCELLING,
queuedAt, cancelledAt);
+ mockListActiveJobs(job);
+ when(jobExecutor.ownsJob(any())).thenReturn(false);
+ when(jobExecutor.isJobStateNodeLocal()).thenReturn(true);
+ stubEntityStoreUpdateToApply(job, job);
+
+ long beforePull = System.currentTimeMillis();
+ Assertions.assertDoesNotThrow(() -> jobManager.pullAndUpdateJobStatus());
+
+ JobEntity settled = captureUpdatedJobEntity(job, job);
+ Assertions.assertEquals(JobHandle.Status.CANCELLED, settled.status());
+ Assertions.assertTrue(settled.finishedAt() >= beforePull);
+ Assertions.assertEquals(job.auditInfo().createTime(),
settled.auditInfo().createTime());
+ // The job is not run by this server, so its executor is never asked about
it.
+ verify(jobExecutor, never()).getJobExecutionInfo(any());
+ verify(jobExecutor, never()).cancelJob(any());
+ }
+
+ @Test
+ public void testPullJobStatusSettlesCancellationOnlyAfterTimeout() throws
IOException {
+ // The timeout is three pull intervals. A job cancelled just within it is
left to its owner, one
+ // cancelled just beyond it is settled.
+ long timeoutInMs = 3 * config.get(Configs.JOB_STATUS_PULL_INTERVAL_IN_MS);
+ Instant longAgo = Instant.now().minus(30, ChronoUnit.DAYS);
+ JobEntity withinTimeout =
+ newJobEntity(
+ "local-job-other-1",
+ JobHandle.Status.CANCELLING,
+ longAgo,
+ Instant.now().minusMillis(timeoutInMs).plusSeconds(10));
+ JobEntity beyondTimeout =
+ newJobEntity(
+ "local-job-other-2",
+ JobHandle.Status.CANCELLING,
+ longAgo,
+ Instant.now().minusMillis(timeoutInMs).minusSeconds(10));
+ mockListActiveJobs(withinTimeout, beyondTimeout);
+ when(jobExecutor.ownsJob(any())).thenReturn(false);
+ when(jobExecutor.isJobStateNodeLocal()).thenReturn(true);
+ stubEntityStoreUpdateToApply(beyondTimeout, beyondTimeout);
+
+ Assertions.assertDoesNotThrow(() -> jobManager.pullAndUpdateJobStatus());
+
+ verify(entityStore, never())
+ .update(
+ eq(NameIdentifierUtil.ofJob(metalake, withinTimeout.name())),
+ eq(JobEntity.class),
+ eq(Entity.EntityType.JOB),
+ any());
+ Assertions.assertEquals(
+ JobHandle.Status.CANCELLED, captureUpdatedJobEntity(beyondTimeout,
beyondTimeout).status());
+ }
+
+ @Test
+ public void testPullJobStatusContinuesAfterFailingToSettleCancellation()
throws IOException {
+ // Failing to settle one job must not stop settling the others.
+ Instant longAgo = Instant.now().minus(1, ChronoUnit.HOURS);
+ JobEntity failingJob =
+ newJobEntity("local-job-gone-1", JobHandle.Status.CANCELLING, longAgo,
longAgo);
+ JobEntity otherJob =
+ newJobEntity("local-job-gone-2", JobHandle.Status.CANCELLING, longAgo,
longAgo);
+ mockListActiveJobs(failingJob, otherJob);
+ when(jobExecutor.ownsJob(any())).thenReturn(false);
+ 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.pullAndUpdateJobStatus());
+
+ Assertions.assertEquals(
+ JobHandle.Status.CANCELLED, captureUpdatedJobEntity(otherJob,
otherJob).status());
+ }
+
+ @Test
+ public void testPullJobStatusContinuesAfterFailingToCheckNodeLocalJobState()
throws IOException {
+ // The job executor fails to tell whether its job state is node local for
the first job, which
+ // must not stop settling the next one.
+ Instant longAgo = Instant.now().minus(1, ChronoUnit.HOURS);
+ JobEntity failingJob =
+ newJobEntity("local-job-gone-1", JobHandle.Status.CANCELLING, longAgo,
longAgo);
+ JobEntity otherJob =
+ newJobEntity("local-job-gone-2", JobHandle.Status.CANCELLING, longAgo,
longAgo);
+ mockListActiveJobs(failingJob, otherJob);
+ when(jobExecutor.ownsJob(any())).thenReturn(false);
+ when(jobExecutor.isJobStateNodeLocal())
+ .thenThrow(new AbstractMethodError("isJobStateNodeLocal"))
+ .thenReturn(true);
+ stubEntityStoreUpdateToApply(otherJob, otherJob);
+
+ Assertions.assertDoesNotThrow(() -> jobManager.pullAndUpdateJobStatus());
+
+ verify(entityStore, never())
+ .update(
+ eq(NameIdentifierUtil.ofJob(metalake, failingJob.name())),
+ eq(JobEntity.class),
+ eq(Entity.EntityType.JOB),
+ any());
+ Assertions.assertEquals(
+ JobHandle.Status.CANCELLED, captureUpdatedJobEntity(otherJob,
otherJob).status());
+ }
+
+ @Test
+ public void testPullJobStatusWaitsForOwnerToAnswerCancellation() throws
IOException {
+ // The job was cancelled just now, so its owner may still answer it. A job
owned by another
+ // server that is not cancelling is skipped however long it has not been
updated.
+ Instant longAgo = Instant.now().minus(30, ChronoUnit.DAYS);
+ JobEntity recentlyCancelled =
+ newJobEntity("local-job-other-1", JobHandle.Status.CANCELLING,
longAgo, Instant.now());
+ JobEntity notCancelled =
+ newJobEntity("local-job-other-2", JobHandle.Status.STARTED, longAgo,
longAgo);
+ mockListActiveJobs(recentlyCancelled, notCancelled);
+ when(jobExecutor.ownsJob(any())).thenReturn(false);
+ when(jobExecutor.isJobStateNodeLocal()).thenReturn(true);
+
+ Assertions.assertDoesNotThrow(() -> jobManager.pullAndUpdateJobStatus());
+
+ verify(entityStore, never())
+ .update(any(), eq(JobEntity.class), eq(Entity.EntityType.JOB), any());
+ }
+
+ @Test
+ public void
testPullJobStatusDoesNotSettleCancellationOfNonNodeLocalExecutor()
+ throws IOException {
+ // Any server can track the jobs of an external job executor, so a
cancelling job is left to
+ // the job executor however long it takes.
+ Instant longAgo = Instant.now().minus(30, ChronoUnit.DAYS);
+ JobEntity job = newJobEntity("external-job-1",
JobHandle.Status.CANCELLING, longAgo, longAgo);
+ mockListActiveJobs(job);
+ when(jobExecutor.ownsJob(any())).thenReturn(false);
+
+ Assertions.assertDoesNotThrow(() -> jobManager.pullAndUpdateJobStatus());
+
+ verify(entityStore, never())
+ .update(any(), eq(JobEntity.class), eq(Entity.EntityType.JOB), any());
+ }
+
+ @Test
+ public void testPullJobStatusDoesNotSettleCancellationAnsweredConcurrently()
throws IOException {
+ // listJobs() observes an unanswered cancellation, but the owner records
the job as CANCELLED
+ // before entityStore.update() re-fetches it, so the recorded job is kept
as it is.
+ Instant longAgo = Instant.now().minus(1, ChronoUnit.HOURS);
+ JobEntity staleSnapshot =
+ newJobEntity("local-job-other-1", JobHandle.Status.CANCELLING,
longAgo, longAgo);
+ JobEntity answered =
+ newJobEntity(
+ staleSnapshot.id(),
+ staleSnapshot.jobExecutionId(),
+ JobHandle.Status.CANCELLED,
+ longAgo,
+ Instant.now());
+ mockListActiveJobs(staleSnapshot);
+ when(jobExecutor.ownsJob(any())).thenReturn(false);
+ when(jobExecutor.isJobStateNodeLocal()).thenReturn(true);
+ stubEntityStoreUpdateToApply(answered);
+
+ Assertions.assertDoesNotThrow(() -> jobManager.pullAndUpdateJobStatus());
+
+ Assertions.assertSame(answered, captureUpdatedJobEntity(answered));
+ }
+
+ @Test
+ public void
testPullJobStatusSkipsCancellationSettledConcurrentlyByAnotherServer()
+ throws IOException {
+ // Another server settles the first job between this server's read and
write, so the write
+ // conflicts. The other jobs are still settled, and the next poll sees the
first job as finished
+ // and leaves it as it is.
+ Instant longAgo = Instant.now().minus(1, ChronoUnit.HOURS);
+ JobEntity conflictedJob =
+ newJobEntity("local-job-gone-1", JobHandle.Status.CANCELLING, longAgo,
longAgo);
+ JobEntity otherJob =
+ newJobEntity("local-job-gone-2", JobHandle.Status.CANCELLING, longAgo,
longAgo);
+ mockListActiveJobs(conflictedJob, otherJob);
+ when(jobExecutor.ownsJob(any())).thenReturn(false);
+ when(jobExecutor.isJobStateNodeLocal()).thenReturn(true);
+ when(entityStore.update(
+ eq(NameIdentifierUtil.ofJob(metalake, conflictedJob.name())),
+ eq(JobEntity.class),
+ eq(Entity.EntityType.JOB),
+ any()))
+ .thenThrow(new OptimisticLockException("job changed"));
+ stubEntityStoreUpdateToApply(otherJob, otherJob);
+
+ Assertions.assertDoesNotThrow(() -> jobManager.pullAndUpdateJobStatus());
+
+ Assertions.assertEquals(
+ JobHandle.Status.CANCELLED, captureUpdatedJobEntity(otherJob,
otherJob).status());
+
+ JobEntity settledByOther =
+ newJobEntity(
+ conflictedJob.id(),
+ conflictedJob.jobExecutionId(),
+ JobHandle.Status.CANCELLED,
+ longAgo,
+ Instant.now());
+ mockListActiveJobs(settledByOther);
+
+ Assertions.assertDoesNotThrow(() -> jobManager.pullAndUpdateJobStatus());
+
+ verify(entityStore, times(1))
+ .update(
+ eq(NameIdentifierUtil.ofJob(metalake, conflictedJob.name())),
+ eq(JobEntity.class),
+ eq(Entity.EntityType.JOB),
+ any());
+ }
+
+ @Test
+ public void
testPullJobStatusSettlesUnansweredCancellationWithoutLastModifiedTime()
+ throws IOException {
+ // The job has no last modified time, so how long it has been cancelling
is taken from when it
+ // was created.
+ JobEntity job =
+ newJobEntity(
+ "local-job-gone-1",
+ JobHandle.Status.CANCELLING,
+ Instant.now().minus(1, ChronoUnit.HOURS),
+ null);
+ mockListActiveJobs(job);
+ when(jobExecutor.ownsJob(any())).thenReturn(false);
+ when(jobExecutor.isJobStateNodeLocal()).thenReturn(true);
+ stubEntityStoreUpdateToApply(job, job);
+
+ Assertions.assertDoesNotThrow(() -> jobManager.pullAndUpdateJobStatus());
+
+ Assertions.assertEquals(JobHandle.Status.CANCELLED,
captureUpdatedJobEntity(job, job).status());
+ }
+
@Test
public void
testPullJobStatusCancelsOwnedJobMarkedCancellingByAnotherServer() throws
IOException {
// A STARTED job marked as CANCELLING by another server is cancelled by
its owner, and stays
diff --git
a/core/src/test/java/org/apache/gravitino/job/TestJobManagerMultiNode.java
b/core/src/test/java/org/apache/gravitino/job/TestJobManagerMultiNode.java
index 94c84fd324..b696363999 100644
--- a/core/src/test/java/org/apache/gravitino/job/TestJobManagerMultiNode.java
+++ b/core/src/test/java/org/apache/gravitino/job/TestJobManagerMultiNode.java
@@ -42,6 +42,7 @@ import org.apache.gravitino.GravitinoEnv;
import org.apache.gravitino.cache.NoOpsCache;
import org.apache.gravitino.connector.job.JobExecutionInfo;
import org.apache.gravitino.connector.job.JobExecutor;
+import org.apache.gravitino.exceptions.InUseException;
import org.apache.gravitino.exceptions.NoSuchJobException;
import org.apache.gravitino.job.local.LocalJobExecutor;
import org.apache.gravitino.job.local.LocalJobExecutorConfigs;
@@ -81,6 +82,9 @@ public class TestJobManagerMultiNode extends TestJDBCBackend {
// 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);
+ // A cancellation of a local job is answered by its owner within a few pull
intervals.
+ private static final long JOB_STATUS_PULL_INTERVAL_IN_MS =
TimeUnit.MINUTES.toMillis(1);
+
private File testDir;
private LocalJobExecutor executorA;
@@ -181,6 +185,12 @@ public class TestJobManagerMultiNode extends
TestJDBCBackend {
Assertions.assertEquals(JobHandle.Status.CANCELLING, cancelling.status());
Assertions.assertEquals(JobHandle.Status.STARTED,
executorA.getJobStatus(job.jobExecutionId()));
+ // Node B leaves the cancellation to node A when it pulls job statuses, as
node A still has
+ // time to answer it.
+ nodeB.pullAndUpdateJobStatus();
+ Assertions.assertEquals(JobHandle.Status.CANCELLING,
getJob(job.name()).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()
@@ -235,6 +245,34 @@ public class TestJobManagerMultiNode extends
TestJDBCBackend {
Assertions.assertFalse(jobExists(cancellingJob.name()));
}
+ @TestTemplate
+ public void testCancelJobLeftByExitedNodeFromAnotherNode() throws Exception {
+ JobEntity job = runLongJobOnNodeA();
+
+ // Node A exits, e.g. its server restarts, and the job process is gone
with it.
+ nodeA.close();
+ nodeA = null;
+
+ // Node B can't cancel the job itself, so it marks the job as CANCELLING
and gives the owner
+ // time to answer. Until then, the job still blocks deleting its template.
+ Assertions.assertEquals(
+ JobHandle.Status.CANCELLING, nodeB.cancelJob(METALAKE,
job.name()).status());
+ nodeB.pullAndUpdateJobStatus();
+ Assertions.assertEquals(JobHandle.Status.CANCELLING,
getJob(job.name()).status());
+ Assertions.assertThrows(
+ InUseException.class, () -> nodeB.deleteJobTemplate(METALAKE,
TEMPLATE));
+
+ // Nobody answers the cancellation in time, so node B marks the job as
CANCELLED, and the
+ // template can be deleted.
+ moveJobTimestampsBack(
+ job.name(), 3 * JOB_STATUS_PULL_INTERVAL_IN_MS +
TimeUnit.MINUTES.toMillis(1));
+ nodeB.pullAndUpdateJobStatus();
+ JobEntity cancelled = getJob(job.name());
+ Assertions.assertEquals(JobHandle.Status.CANCELLED, cancelled.status());
+ Assertions.assertTrue(cancelled.finishedAt() > 0);
+ Assertions.assertTrue(nodeB.deleteJobTemplate(METALAKE, TEMPLATE));
+ }
+
@TestTemplate
public void testGetJobOutputFromAnotherNode() throws IOException {
JobEntity job = runEchoJobOnNodeA("a");
@@ -429,7 +467,10 @@ public class TestJobManagerMultiNode extends
TestJDBCBackend {
// 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);
+ moveJobTimestampsBack(jobName, JOB_KEEP_TIME_IN_MS +
TimeUnit.MINUTES.toMillis(1));
+ }
+
+ private void moveJobTimestampsBack(String jobName, long offsetInMs) throws
IOException {
entityStore.update(
NameIdentifierUtil.ofJob(METALAKE, jobName),
JobEntity.class,
@@ -505,6 +546,7 @@ public class TestJobManagerMultiNode extends
TestJDBCBackend {
Config config = new Config(false) {};
config.set(Configs.JOB_STAGING_DIR, stagingDir.getAbsolutePath());
config.set(Configs.JOB_STAGING_DIR_KEEP_TIME_IN_MS, JOB_KEEP_TIME_IN_MS);
+ config.set(Configs.JOB_STATUS_PULL_INTERVAL_IN_MS,
JOB_STATUS_PULL_INTERVAL_IN_MS);
return config;
}
diff --git a/docs/manage-jobs-in-gravitino.md b/docs/manage-jobs-in-gravitino.md
index 451819781e..b09e5f23ca 100644
--- a/docs/manage-jobs-in-gravitino.md
+++ b/docs/manage-jobs-in-gravitino.md
@@ -441,6 +441,10 @@ only tracks the jobs it runs itself:
`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.
+- To clear such a job sooner, cancel it. If the job is still `CANCELLING` more
than three times
+ `gravitino.job.statusPullIntervalInMs` after the cancel request, the server
running it is
+ considered gone, and another server marks the job as `CANCELLED` when it
pulls job statuses. Its
+ job template can then be deleted.
:::caution
The local job executor can't tell a job left behind by an exited server from a
job that is still
@@ -456,8 +460,31 @@ server doesn't recognize the jobs it ran before the
restart. This also applies t
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.
+every tenth of that time. Until then, they are still reported as queued,
started or cancelling, and
+they keep their job templates from being deleted.
+
+To clear these jobs sooner, cancel them. A cancelled job is marked as
`CANCELLED` within about four
+times `gravitino.job.statusPullIntervalInMs`, as described above. Lowering
+`gravitino.job.stagingDirKeepTimeInMs` also shortens the time, but finished
jobs are then kept for
+a shorter time, and a job that runs longer than that without changing its
status is marked as
+`FAILED`.
+:::
+
+:::caution
+Telling a gone server from a live one by how long a job stays `CANCELLING`
requires all servers to
+use the same `gravitino.job.statusPullIntervalInMs` and to have their clocks
in sync. A job is also
+marked as `CANCELLED` while its server is still running it if that server
doesn't answer the
+cancellation in time, for example because:
+
+- it uses a longer `gravitino.job.statusPullIntervalInMs` than the other
servers, or its clock is
+ off,
+- it fails to pull job statuses for that long, or
+- the job process takes longer than one pull interval to exit, e.g. as
+ `gravitino.jobExecutor.local.cancelForceKillDelayInMs` is longer than the
pull interval.
+
+If the server hasn't asked the job process to stop yet, it never does once the
job is `CANCELLED`,
+so the process keeps running until it finishes, and deleting its job template
also deletes the
+staging directory the process uses.
:::
## Future Work
