jerryshao commented on code in PR #14016:
URL: https://github.com/apache/gravitino/pull/14016#discussion_r4323761282
##########
core/src/test/java/org/apache/gravitino/job/TestJobManager.java:
##########
@@ -1977,6 +1978,120 @@ public void
testCleanUpStagingDirsContinuesAfterOccConflict() throws IOException
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());
+ }
+
+ /** 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());
+ }
Review Comment:
Added verifications that cleanup attempts both deletes, in 337a7df.
_🤖 Addressed by [Claude Code](https://claude.com/claude-code)_
##########
core/src/main/java/org/apache/gravitino/job/JobManager.java:
##########
@@ -749,87 +749,123 @@ public void close() throws IOException {
@VisibleForTesting
void pullAndUpdateJobStatus() {
- List<String> metalakes = MetalakeManager.listInUseMetalakes(entityStore);
+ // An exception escaping this scheduled task would cancel all its later
runs, so a failure is
+ // logged and only skips the metalake or the job it affects.
+ List<String> metalakes;
+ try {
+ metalakes = MetalakeManager.listInUseMetalakes(entityStore);
+ } catch (RuntimeException e) {
+ LOG.error("Failed to list the in-use metalakes to pull the job status",
e);
+ return;
+ }
Review Comment:
Extracted the listing and per-metalake guard into `forEachInUseMetalake`,
shared by both tasks, in 337a7df.
_🤖 Addressed by [Claude Code](https://claude.com/claude-code)_
##########
core/src/main/java/org/apache/gravitino/job/JobManager.java:
##########
@@ -749,87 +749,123 @@ public void close() throws IOException {
@VisibleForTesting
void pullAndUpdateJobStatus() {
- List<String> metalakes = MetalakeManager.listInUseMetalakes(entityStore);
+ // An exception escaping this scheduled task would cancel all its later
runs, so a failure is
+ // logged and only skips the metalake or the job it affects.
+ List<String> metalakes;
+ try {
+ metalakes = MetalakeManager.listInUseMetalakes(entityStore);
+ } catch (RuntimeException e) {
+ LOG.error("Failed to list the in-use metalakes to pull the job status",
e);
+ return;
+ }
+
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 -> {
+ try {
+ pullAndUpdateJobStatus(metalake);
+ } catch (RuntimeException e) {
+ LOG.error("Failed to pull the job status under metalake {}", metalake,
e);
+ }
+ }
+ }
+
+ private void pullAndUpdateJobStatus(String metalake) {
+ // 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();
Review Comment:
Same as above: both tasks now use `forEachInUseMetalake`, in 337a7df.
_🤖 Addressed by [Claude Code](https://claude.com/claude-code)_
##########
core/src/main/java/org/apache/gravitino/job/JobManager.java:
##########
Review Comment:
Fixed the wording to "This unnecessarily lists all the jobs", in 337a7df.
_🤖 Addressed by [Claude Code](https://claude.com/claude-code)_
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
