jerryshao commented on code in PR #14016:
URL: https://github.com/apache/gravitino/pull/14016#discussion_r4332727052
##########
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:
The expire guard now catches `Throwable`, so an `Error` skips only that job.
Added `testCleanUpStagingDirsContinuesAfterErrorExpiringJob`, which checks that
the expired jobs found before it are still deleted. In d595660.
_🤖 Addressed by [Claude Code](https://claude.com/claude-code)_
##########
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:
The delete loop now catches `Throwable` after `OptimisticLockException`.
Added `testCleanUpStagingDirsContinuesAfterErrorDeletingJob`. In d595660.
_🤖 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) {
+ 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:
Task names are now nouns ("job status pull", "job staging directory
cleanup"), and the logs read "Failed to list the in-use metalakes for the {}",
"Skipped the {} for metalake {}: {}" and "The {} failed for metalake {}". In
d595660.
_🤖 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. Rethrowing from a `scheduleAtFixedRate` task doesn't stop the
JVM: the error is kept in the future nobody reads and only cancels that task
for good. So a transient OOM or a deep-recursion `StackOverflowError` on one
metalake would stop job status updates permanently, which is the outcome this
PR removes. The error is logged before the next run, and a JVM that is really
broken is better handled by `-XX:+ExitOnOutOfMemoryError` /
`-XX:+CrashOnOutOfMemoryError` than by quietly cancelling one background task.
_🤖 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]
