lasdf1234 commented on code in PR #14016:
URL: https://github.com/apache/gravitino/pull/14016#discussion_r4332791943
##########
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:
**[Medium] `isJobStateNodeLocal()` is still outside the per-job try**
`expireStaleActiveJob` is now guarded with `catch (Throwable)`, but the
`else if` condition calls `jobExecutor.isJobStateNodeLocal()` *before* entering
that try.
If that call throws an `Error` / `AbstractMethodError` (same plugin-linkage
class of failure this PR is hardening against) on a non-finished job:
1. the exception escapes the `for` loop,
2. lands in `forEachInUseMetalake`'s metalake-level handler,
3. and aborts the rest of this metalake's cleanup — including the
`expiredJobs.forEach(...)` delete batch for finished jobs already collected
earlier in the loop.
Status polling already keeps `ownsJob` inside its per-job try; cleanup
should match. Suggest wrapping the whole stale-active branch (or the entire
per-job loop body) in the try:
```java
} else {
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);
}
}
```
A test with list order `[expired, active]` where `isJobStateNodeLocal()`
throws would lock this in (today's expire test only covers the call inside the
try).
##########
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:
**[Nit] InterruptedException clears the interrupt flag**
Broad `catch (Throwable)` (here and the other new guards) will swallow
`InterruptedException` and clear the thread's interrupt status. On
`shutdownNow()` that can delay cooperative stop of a long cleanup/poll tick.
Low practical impact with single-thread daemon executors, but the usual
pattern is:
```java
} catch (Throwable t) {
if (t instanceof InterruptedException) {
Thread.currentThread().interrupt();
}
LOG.error(...);
return;
}
```
Optional; not blocking if you prefer to keep the guards minimal.
--
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]
