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 02954c331b [#14014] fix(core): keep job status polling and staging
cleanup running after a failure (#14016)
02954c331b is described below
commit 02954c331b74d7a05f029276980ae744fb85db41
Author: Jerry Shao <[email protected]>
AuthorDate: Wed Oct 21 14:47:46 2026 +0800
[#14014] fix(core): keep job status polling and staging cleanup running
after a failure (#14016)
### What changes were proposed in this pull request?
Guard both scheduled tasks in `JobManager`, `pullAndUpdateJobStatus` and
`cleanUpStagingDirs`:
- A failure in `MetalakeManager.listInUseMetalakes` is logged and ends
the current run only.
- Each metalake is processed in its own try, so a metalake that was
disabled or dropped after being listed, or a store error while listing
its jobs, only skips that metalake.
- In the status poll, the `ownsJob` check is inside the per-job try.
- The cleanup's per-job delete loop also catches `RuntimeException`, and
its log names the metalake.
### Why are the changes needed?
Both tasks are scheduled with `scheduleAtFixedRate`. Any exception
escaping a run cancels all later runs, and the exception stays in a
future nobody reads, so it is never logged. One transient store error,
or disabling a metalake at the wrong moment, stopped all job status
updates or staging cleanup until the server restarted.
Fix: #14014
### Does this PR introduce _any_ user-facing change?
No.
### How was this patch tested?
Added unit tests to `TestJobManager` for both tasks: they recover after
a failure to list metalakes, keep processing the other metalakes when
one fails, and keep deleting other expired jobs when one delete fails
unexpectedly. All five fail without this change.
🤖 Generated with [Claude Code](https://claude.com/claude-code)
---------
Co-authored-by: Claude Opus 5.5 <[email protected]>
---
.../java/org/apache/gravitino/job/JobManager.java | 187 ++++++++++-------
.../org/apache/gravitino/job/TestJobManager.java | 222 +++++++++++++++++++++
2 files changed, 332 insertions(+), 77 deletions(-)
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 20feaf1ca5..c69d76332d 100644
--- a/core/src/main/java/org/apache/gravitino/job/JobManager.java
+++ b/core/src/main/java/org/apache/gravitino/job/JobManager.java
@@ -42,6 +42,7 @@ 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.Consumer;
import java.util.function.Function;
import java.util.regex.Pattern;
import javax.annotation.Nullable;
@@ -61,6 +62,7 @@ import org.apache.gravitino.dto.job.JobTemplateDTO;
import org.apache.gravitino.dto.util.DTOConverters;
import org.apache.gravitino.exceptions.InUseException;
import org.apache.gravitino.exceptions.JobTemplateAlreadyExistsException;
+import org.apache.gravitino.exceptions.MetalakeNotInUseException;
import org.apache.gravitino.exceptions.NoSuchEntityException;
import org.apache.gravitino.exceptions.NoSuchJobException;
import org.apache.gravitino.exceptions.NoSuchJobTemplateException;
@@ -757,87 +759,12 @@ public class JobManager implements JobOperationDispatcher
{
@VisibleForTesting
void pullAndUpdateJobStatus() {
- List<String> metalakes = MetalakeManager.listInUseMetalakes(entityStore);
- for (String metalake : metalakes) {
- // This unnecessary list all the jobs, we need to improve the code to
only list the active
- // jobs.
- List<JobEntity> activeJobs =
- listJobs(metalake, Optional.empty()).stream()
- .filter(
- job ->
- job.status() == JobHandle.Status.QUEUED
- || job.status() == JobHandle.Status.STARTED
- || job.status() == JobHandle.Status.CANCELLING)
- .toList();
-
- activeJobs.forEach(
- job -> {
- // 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;
- }
- // An exception escaping this scheduled task would cancel all its
later runs, so a
- // failure on one job must never stop the status pull of the
others.
- try {
- pullAndUpdateOwnedJobStatus(metalake, job);
- } catch (RuntimeException e) {
- LOG.error(
- "Failed to update the status of job {} with execution id {}
under metalake {}",
- job.name(),
- job.jobExecutionId(),
- metalake,
- e);
- }
- });
- }
+ forEachInUseMetalake("job status pull", this::pullAndUpdateJobStatus);
}
@VisibleForTesting
void cleanUpStagingDirs() {
- List<String> metalakes = MetalakeManager.listInUseMetalakes(entityStore);
-
- for (String metalake : metalakes) {
- 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);
- }
- }
- }
-
- expiredJobs.forEach(
- job -> {
- try {
- entityStore.delete(
- NameIdentifierUtil.ofJob(metalake, job.name()),
Entity.EntityType.JOB);
- deleteJobStagingDir(job);
- } catch (OptimisticLockException e) {
- // Keep files when deletion conflicts or times out on a lock.
The next cleanup run
- // re-reads the job and checks retention eligibility again;
other jobs in this batch
- // can still be processed.
- LOG.info(
- "Job {} under metalake {} could not be deleted because of a
concurrent operation; deferring cleanup",
- job.name(),
- metalake);
- } catch (IOException e) {
- LOG.error("Failed to delete job and staging directory for job
{}", job.name(), e);
- }
- });
- }
+ forEachInUseMetalake("job staging directory cleanup",
this::cleanUpStagingDirs);
}
@VisibleForTesting
@@ -1067,6 +994,65 @@ public class JobManager implements JobOperationDispatcher
{
return newValue.orElse(currentValue);
}
+ // Runs a scheduled task on every in-use metalake. Anything escaping a
scheduled task, including
+ // an Error such as a LinkageError from a job executor plugin, would cancel
all its later runs,
+ // so a failure is logged and only skips the metalake it affects.
+ private void forEachInUseMetalake(String task, Consumer<String> action) {
+ List<String> metalakes;
+ try {
+ metalakes = MetalakeManager.listInUseMetalakes(entityStore);
+ } catch (Throwable t) {
+ LOG.error("Failed to list the in-use metalakes for the {}", task, t);
+ return;
+ }
+
+ for (String metalake : metalakes) {
+ try {
+ action.accept(metalake);
+ } catch (MetalakeNotInUseException | NoSuchMetalakeException e) {
+ // The metalake was disabled or dropped after it was listed as in use,
which is an expected
+ // race, so it is logged without the stack trace.
+ LOG.warn("Skipped the {} for metalake {}: {}", task, metalake,
e.getMessage());
+ } catch (Throwable t) {
+ LOG.error("The {} failed for metalake {}", task, metalake, t);
+ }
+ }
+ }
+
+ private void pullAndUpdateJobStatus(String metalake) {
+ // This unnecessarily lists all the jobs, we need to improve the code to
only list the
+ // active jobs.
+ List<JobEntity> activeJobs =
+ listJobs(metalake, Optional.empty()).stream()
+ .filter(
+ job ->
+ job.status() == JobHandle.Status.QUEUED
+ || job.status() == JobHandle.Status.STARTED
+ || job.status() == JobHandle.Status.CANCELLING)
+ .toList();
+
+ 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;
+ }
+ pullAndUpdateOwnedJobStatus(metalake, job);
+ } catch (Throwable t) {
+ LOG.error(
+ "Failed to update the status of job {} with execution id {}
under metalake {}",
+ job.name(),
+ job.jobExecutionId(),
+ metalake,
+ t);
+ }
+ });
+ }
+
private void pullAndUpdateOwnedJobStatus(String metalake, JobEntity job) {
JobExecutionInfo observed;
try {
@@ -1371,6 +1357,53 @@ public class JobManager implements
JobOperationDispatcher {
}
}
+ private void cleanUpStagingDirs(String metalake) {
+ 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 {
+ // 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 {
+ if (jobExecutor.isJobStateNodeLocal() && isStaleActiveJob(job, now))
{
+ expireStaleActiveJob(metalake, job, now);
+ }
+ } catch (Throwable t) {
+ LOG.error("Failed to expire job {} under metalake {}", job.name(),
metalake, t);
+ }
+ }
+ }
+
+ expiredJobs.forEach(
+ job -> {
+ try {
+ entityStore.delete(
+ NameIdentifierUtil.ofJob(metalake, job.name()),
Entity.EntityType.JOB);
+ deleteJobStagingDir(job);
+ } catch (OptimisticLockException e) {
+ // Keep files when deletion conflicts or times out on a lock. The
next cleanup run
+ // re-reads the job and checks retention eligibility again; other
jobs in this batch
+ // can still be processed.
+ LOG.info(
+ "Job {} under metalake {} could not be deleted because of a
concurrent operation; deferring cleanup",
+ job.name(),
+ metalake);
+ } catch (Throwable t) {
+ LOG.error(
+ "Failed to delete job and staging directory for job {} under
metalake {}",
+ job.name(),
+ metalake,
+ t);
+ }
+ });
+ }
+
private void expireStaleActiveJob(String metalake, JobEntity job, long now) {
AtomicBoolean expired = new AtomicBoolean(false);
updateJobEntity(
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 1b2d7f79fa..0699490a4a 100644
--- a/core/src/test/java/org/apache/gravitino/job/TestJobManager.java
+++ b/core/src/test/java/org/apache/gravitino/job/TestJobManager.java
@@ -72,6 +72,7 @@ import org.apache.gravitino.dto.job.ShellJobTemplateDTO;
import org.apache.gravitino.exceptions.InUseException;
import org.apache.gravitino.exceptions.JobTemplateAlreadyExistsException;
import org.apache.gravitino.exceptions.MetalakeInUseException;
+import org.apache.gravitino.exceptions.MetalakeNotInUseException;
import org.apache.gravitino.exceptions.NoSuchEntityException;
import org.apache.gravitino.exceptions.NoSuchJobException;
import org.apache.gravitino.exceptions.NoSuchJobTemplateException;
@@ -2005,6 +2006,227 @@ public class TestJobManager {
verify(entityStore, times(1)).delete(otherIdent, Entity.EntityType.JOB);
}
+ /** A failure to list the metalakes is logged and does not stop later status
polls. */
+ @Test
+ public void testPullJobStatusContinuesAfterFailingToListMetalakes() throws
IOException {
+ JobEntity job = newJobEntity("shell_job", JobHandle.Status.QUEUED);
+ mockedMetalake
+ .when(() -> MetalakeManager.listInUseMetalakes(entityStore))
+ .thenThrow(new RuntimeException("Failed to list in-use metalakes"))
+ .thenReturn(ImmutableList.of(metalake));
+ doReturn(ImmutableList.of(job)).when(jobManager).listJobs(metalake,
Optional.empty());
+ when(jobExecutor.getJobExecutionInfo(job.jobExecutionId()))
+ .thenReturn(JobExecutionInfo.of(JobHandle.Status.SUCCEEDED));
+ stubEntityStoreUpdateToApply(job);
+
+ Assertions.assertDoesNotThrow(() -> jobManager.pullAndUpdateJobStatus());
+ verify(jobExecutor, never()).getJobExecutionInfo(any());
+
+ Assertions.assertDoesNotThrow(() -> jobManager.pullAndUpdateJobStatus());
+ Assertions.assertEquals(JobHandle.Status.SUCCEEDED,
captureUpdatedJobEntity(job).status());
+ }
+
+ /** A metalake disabled after being listed as in use only skips the status
poll of itself. */
+ @Test
+ public void testPullJobStatusContinuesAfterFailingOnOneMetalake() throws
IOException {
+ String disabledMetalake = "disabled_metalake";
+ JobEntity job = newJobEntity("shell_job", JobHandle.Status.QUEUED);
+ mockedMetalake
+ .when(() -> MetalakeManager.listInUseMetalakes(entityStore))
+ .thenReturn(ImmutableList.of(disabledMetalake, metalake));
+ doThrow(new MetalakeNotInUseException("Metalake %s is not in use",
disabledMetalake))
+ .when(jobManager)
+ .listJobs(disabledMetalake, Optional.empty());
+ doReturn(ImmutableList.of(job)).when(jobManager).listJobs(metalake,
Optional.empty());
+ when(jobExecutor.getJobExecutionInfo(job.jobExecutionId()))
+ .thenReturn(JobExecutionInfo.of(JobHandle.Status.SUCCEEDED));
+ stubEntityStoreUpdateToApply(job, job);
+
+ Assertions.assertDoesNotThrow(() -> jobManager.pullAndUpdateJobStatus());
+
+ Assertions.assertEquals(JobHandle.Status.SUCCEEDED,
captureUpdatedJobEntity(job, job).status());
+ }
+
+ /** An Error on one metalake, such as a LinkageError from a plugin, only
skips that metalake. */
+ @Test
+ public void testPullJobStatusContinuesAfterErrorOnOneMetalake() throws
IOException {
+ String brokenMetalake = "broken_metalake";
+ JobEntity job = newJobEntity("shell_job", JobHandle.Status.QUEUED);
+ mockedMetalake
+ .when(() -> MetalakeManager.listInUseMetalakes(entityStore))
+ .thenReturn(ImmutableList.of(brokenMetalake, metalake));
+ doThrow(new NoClassDefFoundError("org/example/MissingClass"))
+ .when(jobManager)
+ .listJobs(brokenMetalake, Optional.empty());
+ doReturn(ImmutableList.of(job)).when(jobManager).listJobs(metalake,
Optional.empty());
+ when(jobExecutor.getJobExecutionInfo(job.jobExecutionId()))
+ .thenReturn(JobExecutionInfo.of(JobHandle.Status.SUCCEEDED));
+ stubEntityStoreUpdateToApply(job, job);
+
+ Assertions.assertDoesNotThrow(() -> jobManager.pullAndUpdateJobStatus());
+
+ Assertions.assertEquals(JobHandle.Status.SUCCEEDED,
captureUpdatedJobEntity(job, job).status());
+ }
+
+ /** A failure checking the owner of one job does not stop the status pull of
the others. */
+ @Test
+ public void testPullJobStatusContinuesAfterFailingToCheckJobOwner() throws
IOException {
+ JobEntity failingJob = newJobEntity("shell_job", JobHandle.Status.QUEUED);
+ JobEntity otherJob = newJobEntity("shell_job", JobHandle.Status.QUEUED);
+ mockListActiveJobs(failingJob, otherJob);
+ when(jobExecutor.ownsJob(failingJob.jobExecutionId()))
+ .thenThrow(new AbstractMethodError("ownsJob"));
+ when(jobExecutor.getJobExecutionInfo(otherJob.jobExecutionId()))
+ .thenReturn(JobExecutionInfo.of(JobHandle.Status.SUCCEEDED));
+ stubEntityStoreUpdateToApply(otherJob, otherJob);
+
+ Assertions.assertDoesNotThrow(() -> jobManager.pullAndUpdateJobStatus());
+
+ verify(jobExecutor,
never()).getJobExecutionInfo(failingJob.jobExecutionId());
+ Assertions.assertEquals(
+ JobHandle.Status.SUCCEEDED, captureUpdatedJobEntity(otherJob,
otherJob).status());
+ }
+
+ /** An Error expiring a stale job still lets the expired jobs found before
it be deleted. */
+ @Test
+ public void testCleanUpStagingDirsContinuesAfterErrorExpiringJob() throws
IOException {
+ Instant longAgo = Instant.now().minus(30, ChronoUnit.DAYS);
+ JobEntity expired = expiredJob();
+ JobEntity stale = newJobEntity("local-job-gone-1",
JobHandle.Status.STARTED, longAgo, longAgo);
+ mockListActiveJobs(expired, stale);
+ when(jobExecutor.isJobStateNodeLocal()).thenReturn(true);
+ when(entityStore.update(
+ eq(NameIdentifierUtil.ofJob(metalake, stale.name())),
+ eq(JobEntity.class),
+ eq(Entity.EntityType.JOB),
+ any()))
+ .thenThrow(new NoClassDefFoundError("org/example/MissingClass"));
+ NameIdentifier expiredIdent = NameIdentifierUtil.ofJob(metalake,
expired.name());
+ when(entityStore.delete(expiredIdent,
Entity.EntityType.JOB)).thenReturn(true);
+ File expiredDir = jobManager.jobStagingDir(expired.id());
+ Assertions.assertTrue(expiredDir.mkdirs());
+
+ Assertions.assertDoesNotThrow(() -> jobManager.cleanUpStagingDirs());
+
+ Assertions.assertFalse(expiredDir.exists());
+ verify(entityStore, times(1)).delete(expiredIdent, Entity.EntityType.JOB);
+ }
+
+ /** A failure checking the job executor still lets the expired jobs found
before be deleted. */
+ @Test
+ public void testCleanUpStagingDirsContinuesAfterFailingToCheckJobExecutor()
throws IOException {
+ Instant longAgo = Instant.now().minus(30, ChronoUnit.DAYS);
+ JobEntity expired = expiredJob();
+ JobEntity active = newJobEntity("local-job-gone-1",
JobHandle.Status.STARTED, longAgo, longAgo);
+ mockListActiveJobs(expired, active);
+ when(jobExecutor.isJobStateNodeLocal())
+ .thenThrow(new AbstractMethodError("isJobStateNodeLocal"));
+ NameIdentifier expiredIdent = NameIdentifierUtil.ofJob(metalake,
expired.name());
+ when(entityStore.delete(expiredIdent,
Entity.EntityType.JOB)).thenReturn(true);
+ File expiredDir = jobManager.jobStagingDir(expired.id());
+ Assertions.assertTrue(expiredDir.mkdirs());
+
+ Assertions.assertDoesNotThrow(() -> jobManager.cleanUpStagingDirs());
+
+ Assertions.assertFalse(expiredDir.exists());
+ verify(entityStore, times(1)).delete(expiredIdent, Entity.EntityType.JOB);
+ }
+
+ /** An Error deleting one expired job does not stop the cleanup of the
others. */
+ @Test
+ public void testCleanUpStagingDirsContinuesAfterErrorDeletingJob() throws
IOException {
+ JobEntity failing = expiredJob();
+ JobEntity other = expiredJob();
+ mockListActiveJobs(failing, other);
+ NameIdentifier failingIdent = NameIdentifierUtil.ofJob(metalake,
failing.name());
+ NameIdentifier otherIdent = NameIdentifierUtil.ofJob(metalake,
other.name());
+ when(entityStore.delete(failingIdent, Entity.EntityType.JOB))
+ .thenThrow(new NoClassDefFoundError("org/example/MissingClass"));
+ when(entityStore.delete(otherIdent,
Entity.EntityType.JOB)).thenReturn(true);
+ File otherDir = jobManager.jobStagingDir(other.id());
+ Assertions.assertTrue(otherDir.mkdirs());
+
+ Assertions.assertDoesNotThrow(() -> jobManager.cleanUpStagingDirs());
+
+ Assertions.assertFalse(otherDir.exists());
+ verify(entityStore, times(1)).delete(otherIdent, Entity.EntityType.JOB);
+ }
+
+ /** A failure to list the metalakes is logged and does not stop later
cleanup runs. */
+ @Test
+ public void testCleanUpStagingDirsContinuesAfterFailingToListMetalakes()
throws IOException {
+ JobEntity job = expiredJob();
+ mockedMetalake
+ .when(() -> MetalakeManager.listInUseMetalakes(entityStore))
+ .thenThrow(new RuntimeException("Failed to list in-use metalakes"))
+ .thenReturn(ImmutableList.of(metalake));
+ doReturn(ImmutableList.of(job)).when(jobManager).listJobs(metalake,
Optional.empty());
+ NameIdentifier jobIdent = NameIdentifierUtil.ofJob(metalake, job.name());
+ when(entityStore.delete(jobIdent, Entity.EntityType.JOB)).thenReturn(true);
+ File jobDir = jobManager.jobStagingDir(job.id());
+ Assertions.assertTrue(jobDir.mkdirs());
+
+ Assertions.assertDoesNotThrow(() -> jobManager.cleanUpStagingDirs());
+ Assertions.assertTrue(jobDir.exists());
+ verify(entityStore, never()).delete(any(), any());
+
+ Assertions.assertDoesNotThrow(() -> jobManager.cleanUpStagingDirs());
+ Assertions.assertFalse(jobDir.exists());
+ verify(entityStore, times(1)).delete(jobIdent, Entity.EntityType.JOB);
+ }
+
+ /** A store failure while listing the jobs of one metalake only skips the
cleanup of itself. */
+ @Test
+ public void testCleanUpStagingDirsContinuesAfterFailingOnOneMetalake()
throws IOException {
+ String failingMetalake = "failing_metalake";
+ JobEntity job = expiredJob();
+ mockedMetalake
+ .when(() -> MetalakeManager.listInUseMetalakes(entityStore))
+ .thenReturn(ImmutableList.of(failingMetalake, metalake));
+ doThrow(new RuntimeException(new IOException("store error")))
+ .when(jobManager)
+ .listJobs(failingMetalake, Optional.empty());
+ doReturn(ImmutableList.of(job)).when(jobManager).listJobs(metalake,
Optional.empty());
+ NameIdentifier jobIdent = NameIdentifierUtil.ofJob(metalake, job.name());
+ when(entityStore.delete(jobIdent, Entity.EntityType.JOB)).thenReturn(true);
+ File jobDir = jobManager.jobStagingDir(job.id());
+ Assertions.assertTrue(jobDir.mkdirs());
+
+ Assertions.assertDoesNotThrow(() -> jobManager.cleanUpStagingDirs());
+
+ Assertions.assertFalse(jobDir.exists());
+ verify(entityStore, times(1)).delete(jobIdent, Entity.EntityType.JOB);
+ }
+
+ /** An unexpected failure deleting one expired job does not stop the cleanup
of the others. */
+ @Test
+ public void testCleanUpStagingDirsContinuesAfterFailingToDeleteJob() throws
IOException {
+ JobEntity failing = expiredJob();
+ JobEntity other = expiredJob();
+ doReturn(ImmutableList.of(failing, other))
+ .when(jobManager)
+ .listJobs(metalake, Optional.empty());
+ mockedMetalake
+ .when(() -> MetalakeManager.listInUseMetalakes(entityStore))
+ .thenReturn(ImmutableList.of(metalake));
+ NameIdentifier failingIdent = NameIdentifierUtil.ofJob(metalake,
failing.name());
+ NameIdentifier otherIdent = NameIdentifierUtil.ofJob(metalake,
other.name());
+ when(entityStore.delete(failingIdent, Entity.EntityType.JOB))
+ .thenThrow(new RuntimeException("store error"));
+ when(entityStore.delete(otherIdent,
Entity.EntityType.JOB)).thenReturn(true);
+ File failingDir = jobManager.jobStagingDir(failing.id());
+ File otherDir = jobManager.jobStagingDir(other.id());
+ Assertions.assertTrue(failingDir.mkdirs());
+ Assertions.assertTrue(otherDir.mkdirs());
+
+ Assertions.assertDoesNotThrow(() -> jobManager.cleanUpStagingDirs());
+
+ Assertions.assertTrue(failingDir.exists());
+ Assertions.assertFalse(otherDir.exists());
+ verify(entityStore, times(1)).delete(failingIdent, Entity.EntityType.JOB);
+ verify(entityStore, times(1)).delete(otherIdent, Entity.EntityType.JOB);
+ }
+
@Test
public void testCleanUpStagingDirsDeletesLegacyStagingDirOfRenamedTemplate()
throws IOException {
// The job ran from template "shell_job" before an upgrade, in the legacy
layout, and the
