lasdf1234 commented on code in PR #14016:
URL: https://github.com/apache/gravitino/pull/14016#discussion_r4332717405
##########
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)) {
+ // 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) {
Review Comment:
**[Medium] Inconsistent catch width vs status-poll path**
`expireStaleActiveJob` is wrapped in `catch (RuntimeException)`, while the
status-poll per-job loop and `forEachInUseMetalake` use `catch (Throwable)`.
If an `Error` (e.g. plugin `LinkageError` / `NoClassDefFoundError`) escapes
here mid-loop, it bypasses this catch and lands in `forEachInUseMetalake`'s
metalake-level handler. That aborts the rest of `cleanUpStagingDirs(metalake)`
for this run — including the `expiredJobs.forEach(...)` delete batch that may
already have collected finished jobs before this stale one.
Suggest aligning to `catch (Throwable)` here (and ideally on the delete loop
below) so one bad expire/delete only skips that job, same as status polling.
##########
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:
**[Low / suggestion] Catching `Throwable` at the task boundary**
Catching `Error` here is what keeps `scheduleAtFixedRate` alive after plugin
linkage failures — good for that goal. Consider still rethrowing
`VirtualMachineError` (OOM, StackOverflow) after logging, so the JVM/scheduler
is not kept on life support in a known-bad state:
```java
} catch (Throwable t) {
LOG.error(...);
if (t instanceof VirtualMachineError) {
throw t;
}
return;
}
```
Same pattern could apply to the per-metalake / per-job `Throwable` catches
if you want consistency.
##########
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)) {
+ // 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 | RuntimeException e) {
Review Comment:
**[Low] Same inconsistency on the delete loop**
`catch (IOException | RuntimeException)` still lets an `Error` escape to the
metalake-level handler and skip remaining expired jobs in this batch.
If the intent of this PR is "nothing escapes a job-scoped unit of work",
widen to `Throwable` for symmetry with `pullAndUpdateJobStatus`'s per-job guard.
##########
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) {
+ LOG.error("Failed to list the in-use metalakes to {}", 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 to {} under metalake {}: {}", task, metalake,
e.getMessage());
Review Comment:
**[Nit] Log wording**
`Skipped to {} under metalake {}` reads awkwardly with `task` values like
`"pull the job status"` → *"Skipped to pull the job status under metalake …"*.
Something like `Skipped {} for metalake {}: {}` or `Skipping metalake {}
while trying to {}: {}` reads more naturally. Same for the listing failure
message (`Failed to list the in-use metalakes to {}`).
--
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]
