jerryshao commented on code in PR #14016:
URL: https://github.com/apache/gravitino/pull/14016#discussion_r4332523307
##########
core/src/main/java/org/apache/gravitino/job/JobManager.java:
##########
@@ -1059,6 +985,60 @@ private <T> T updatedValue(T currentValue, Optional<T>
newValue) {
return newValue.orElse(currentValue);
}
+ // Runs a scheduled task on every in-use metalake. An exception escaping a
scheduled task 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 (RuntimeException e) {
+ LOG.error("Failed to list the in-use metalakes to {}", task, e);
+ return;
+ }
+
+ for (String metalake : metalakes) {
+ try {
+ action.accept(metalake);
+ } catch (RuntimeException e) {
Review Comment:
Both guards in `forEachInUseMetalake` now catch `Throwable`, and so does the
per-job guard in the status poll, so a `LinkageError` from `cancelJob` skips
only that job instead of the rest of its metalake. Added
`testPullJobStatusContinuesAfterErrorOnOneMetalake` (`NoClassDefFoundError`).
In 81f54d3.
_🤖 Addressed by [Claude Code](https://claude.com/claude-code)_
##########
core/src/main/java/org/apache/gravitino/job/JobManager.java:
##########
@@ -1363,6 +1343,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);
Review Comment:
Kept the order on purpose. Directory-first was tried in #10627 and reverted
in #10729, and the `OptimisticLockException` branch (#13792) relies on
entity-first: when the delete conflicts, the job's files are kept and the next
run re-checks retention. Deleting the directory first would remove the files of
a job that is then not deleted. Reordering without losing that is a separate
change, so I left it out of this PR.
_🤖 Addressed by [Claude Code](https://claude.com/claude-code)_
##########
core/src/main/java/org/apache/gravitino/job/JobManager.java:
##########
@@ -1059,6 +985,60 @@ private <T> T updatedValue(T currentValue, Optional<T>
newValue) {
return newValue.orElse(currentValue);
}
+ // Runs a scheduled task on every in-use metalake. An exception escaping a
scheduled task 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 (RuntimeException e) {
+ LOG.error("Failed to list the in-use metalakes to {}", task, e);
+ return;
+ }
+
+ for (String metalake : metalakes) {
+ try {
+ action.accept(metalake);
+ } catch (RuntimeException e) {
+ LOG.error("Failed to {} under metalake {}", task, metalake, e);
Review Comment:
`MetalakeNotInUseException` and `NoSuchMetalakeException` are now caught
separately and logged at WARN with only the message, with a comment that the
race is expected. In 81f54d3.
_🤖 Addressed by [Claude Code](https://claude.com/claude-code)_
##########
core/src/main/java/org/apache/gravitino/job/JobManager.java:
##########
@@ -1059,6 +985,60 @@ private <T> T updatedValue(T currentValue, Optional<T>
newValue) {
return newValue.orElse(currentValue);
}
+ // Runs a scheduled task on every in-use metalake. An exception escaping a
scheduled task 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 (RuntimeException e) {
+ LOG.error("Failed to list the in-use metalakes to {}", task, e);
+ return;
+ }
+
+ for (String metalake : metalakes) {
+ try {
+ action.accept(metalake);
+ } catch (RuntimeException e) {
+ LOG.error("Failed to {} under metalake {}", task, metalake, e);
+ }
+ }
+ }
+
+ private void pullAndUpdateJobStatus(String metalake) {
+ // This unnecessarily lists 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 -> {
+ // A failure on one job must never stop the status pull of the
others.
+ try {
+ // Only the job executor instance owning the job can query its
status. The jobs
+ // owned by other servers are skipped, and the jobs left behind by
a server that has
+ // exited are settled by cleanUpStagingDirs() once they expire.
+ if (!jobExecutor.ownsJob(job.jobExecutionId())) {
Review Comment:
Added `testPullJobStatusContinuesAfterFailingToCheckJobOwner`: `ownsJob`
throws for the first job, and the second job is still polled. In 81f54d3.
_🤖 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]
