jerryshao commented on code in PR #14016:
URL: https://github.com/apache/gravitino/pull/14016#discussion_r4332377278
##########
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:
[Nit] Moving `ownsJob` inside the per-job `try` is a real behaviour change -
on `main` an exception from it escaped and cancelled the schedule, now it is
logged and the job is skipped - but no test exercises it. `ownsJob` is only
ever stubbed with `thenReturn` (TestJobManager.java:137, :1652, :1766), never
`thenThrow`.
A fourth small test in the same style as the new ones
(`when(jobExecutor.ownsJob(any())).thenThrow(...)` for the first job, then
assert the second job is still polled) would cover the one new branch the suite
misses.
Verified by: grepped `ownsJob` across
core/src/test/java/org/apache/gravitino/job on this branch - the three stubs
above are all of them.
##########
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:
[Important] Both new guards catch `RuntimeException`, so an `Error` still
escapes the run and `scheduleAtFixedRate` cancels the task forever - the exact
failure mode this PR sets out to remove.
That is not hypothetical in this codebase. The job executor is a plugin
loaded by class name at JobExecutorFactory.java:172-191, and that method's own
comment says "A LinkageError is thrown when the class is there, but a class it
depends on isn't". `checkJobExecutorClass` exists precisely because a stale
executor throws `AbstractMethodError` at call time (javadoc at
JobExecutorFactory.java:117-119), but it only validates `getJobExecutionInfo`
and `submitJob`. `cancelJob` (connector/job/JobExecutor.java:144) is abstract
and unchecked, and `cancelOwnedJob` calls it from inside the poll at
JobManager.java:1054.
Following that path up: `pullAndUpdateOwnedJobStatus` catches `Exception`
(JobManager.java:1074), the per-job guard catches `RuntimeException`
(JobManager.java:1031), and this guard catches `RuntimeException`. A
`NoClassDefFoundError` or `AbstractMethodError` raised inside a plugin executor
therefore passes all three and silently stops every later status poll until the
server restarts.
Suggest catching `Throwable` (or at least `Exception`) here and at
JobManager.java:994. This method is the task boundary, and the invariant it is
documenting is that nothing may cancel the schedule.
Verified by: read JobExecutorFactory.java:85-191 and
connector/job/JobExecutor.java:45-174 on this branch, then traced the catch
chain from `jobExecutor.cancelJob` back up through JobManager.java:1074, :1031
and :1002.
##########
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:
[Question] The entity is deleted before its staging directory. If
`entityStore.delete` succeeds and `deleteJobStagingDir` then fails, the new
catch below logs it and the directory is orphaned: no job entity remains, so no
later cleanup run will ever revisit it. Widening the catch to
`RuntimeException` extends that outcome to runtime failures from
`FileUtils.deleteDirectory` as well as the `IOException` already handled.
Would it be better to delete the directory first and the entity second? Then
a failure leaves the entity in place and the next run retries, and the re-run
is harmless because `deleteJobStagingDir` already no-ops when the directory is
gone (JobManager.java:955-961).
I am raising this as a question rather than a finding because the ordering
predates this PR; only the widened catch is new.
Verified by: read `deleteJobStagingDir` at JobManager.java:955-968 and
`EntityStore.delete`'s contract at EntityStore.java:201-212 on this branch.
##########
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:
[Nit] A metalake disabled or dropped between the listing and the poll is an
expected race - the PR description names it, and
`testPullJobStatusContinuesAfterFailingOnOneMetalake` asserts it with
`MetalakeNotInUseException` - yet it is logged at ERROR with a full stack trace
on every run that loses the race.
This file already treats benign races more quietly:
`OptimisticLockException` at JobManager.java:1327 is INFO, and
`NoSuchEntityException` for a concurrently deleted job at JobManager.java:1337
is WARN, both without a stack trace, with a comment explaining why. Catching
`NotInUseException` / `NoSuchMetalakeException` separately and logging it at
WARN without the throwable would match that, and keep ERROR meaning "something
is actually broken".
Verified by: compared with the log levels and comments at
JobManager.java:1324-1341 in this file, and with the exception the new test
throws at TestJobManager.java:2009.
--
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]
