jerryshao commented on code in PR #14016:
URL: https://github.com/apache/gravitino/pull/14016#discussion_r4332803994
##########
core/src/main/java/org/apache/gravitino/job/JobManager.java:
##########
@@ -1363,6 +1349,51 @@ private Optional<JobEntity> updateJobEntity(
}
}
+ 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 if (jobExecutor.isJobStateNodeLocal() && isStaleActiveJob(job,
now)) {
Review Comment:
Moved the `isJobStateNodeLocal()` / `isStaleActiveJob` check inside the
per-job try, as suggested. Added
`testCleanUpStagingDirsContinuesAfterFailingToCheckJobExecutor`: with the list
`[expired, active]` and `isJobStateNodeLocal()` throwing `AbstractMethodError`,
the expired job is still deleted. In 401d914.
_🤖 Addressed by [Claude Code](https://claude.com/claude-code)_
##########
core/src/main/java/org/apache/gravitino/job/JobManager.java:
##########
@@ -1059,6 +986,65 @@ private <T> T updatedValue(T currentValue, Optional<T>
newValue) {
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) {
Review Comment:
Left as is. `InterruptedException` is checked, and nothing inside these
guards declares it, so it can't reach them short of a sneaky throw. Catching an
exception also doesn't clear the interrupt flag; only a method that throws
`InterruptedException` does, and that would be the place to restore it. A
`shutdownNow()` interrupt during a run therefore stays set for whatever checks
it next.
_🤖 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]
