Copilot commented on code in PR #14016:
URL: https://github.com/apache/gravitino/pull/14016#discussion_r4323397555
##########
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:
This test validates directory outcomes, but it doesn’t assert that cleanup
actually attempted both entity deletions (and continued after the first
failure). Adding explicit verifications for `entityStore.delete(failingIdent,
JOB)` and `entityStore.delete(otherIdent, JOB)` would make the test more robust
and better aligned with the stated behavior (“continues after failing to delete
job”).
##########
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:
The metalake-listing guard logic (comment + try/catch + early return) is
duplicated in both scheduled tasks. Consider extracting a small private helper
(e.g., `listInUseMetalakesSafely(String actionDescription)` or similar) to
reduce duplication and keep the two tasks consistent if logging/behavior needs
to change later.
##########
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:
The metalake-listing guard logic (comment + try/catch + early return) is
duplicated in both scheduled tasks. Consider extracting a small private helper
(e.g., `listInUseMetalakesSafely(String actionDescription)` or similar) to
reduce duplication and keep the two tasks consistent if logging/behavior needs
to change later.
##########
core/src/main/java/org/apache/gravitino/job/JobManager.java:
##########
Review Comment:
Grammar issue in the comment: “This unnecessary list all the jobs” should be
“This unnecessarily lists all the jobs” (or similar). This comment is
user-facing for maintainers and currently reads awkwardly.
--
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]
