jerryshao commented on code in PR #14047:
URL: https://github.com/apache/gravitino/pull/14047#discussion_r4332929728
##########
core/src/main/java/org/apache/gravitino/job/JobManager.java:
##########
@@ -770,18 +782,22 @@ void pullAndUpdateJobStatus() {
|| job.status() == JobHandle.Status.CANCELLING)
.toList();
+ long now = System.currentTimeMillis();
activeJobs.forEach(
job -> {
- // 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())) {
- return;
- }
// An exception escaping this scheduled task would cancel all its
later runs, so a
// failure on one job must never stop the status pull of the
others.
try {
- pullAndUpdateOwnedJobStatus(metalake, job);
+ // Only the job executor instance owning the job can query its
status. The jobs
+ // owned by other servers are skipped, except a node local job
whose cancellation
+ // its owner has not answered in time: the owner has likely
exited, e.g. the server
+ // restarted, so nobody else would ever settle it. The other
jobs left behind by a
+ // server that has exited are settled by cleanUpStagingDirs()
once they expire.
+ if (jobExecutor.ownsJob(job.jobExecutionId())) {
+ pullAndUpdateOwnedJobStatus(metalake, job);
+ } else if (jobExecutor.isJobStateNodeLocal() &&
isCancellationUnanswered(job, now)) {
+ settleUnansweredCancellation(metalake, job, now);
+ }
} catch (RuntimeException e) {
LOG.error(
"Failed to update the status of job {} with execution id {}
under metalake {}",
Review Comment:
No change needed: `updateJobEntity()` already catches
`OptimisticLockException` before it can reach this catch, logs it at INFO and
returns empty, so the next pull re-reads the job (covered by
`testPullJobStatusSkipsCancellationSettledConcurrentlyByAnotherServer`). Only
unexpected failures are logged as errors here.
_🤖 Addressed by [Claude Code](https://claude.com/claude-code)_
##########
docs/manage-jobs-in-gravitino.md:
##########
@@ -441,6 +441,10 @@ only tracks the jobs it runs itself:
`CANCELLED` if it was being cancelled. Like other finished jobs, it is then
kept for another
`gravitino.job.stagingDirKeepTimeInMs` before being cleaned up together with
its staging
directory.
+- To clear such a job sooner, cancel it. If the job is still `CANCELLING`
three times
+ `gravitino.job.statusPullIntervalInMs` after the cancel request, the server
running it is
+ considered gone, and another server marks the job as `CANCELLED` when it
pulls job statuses. Its
Review Comment:
Changed the docs to "more than three times" to match the strict comparison
(b5d3f8204c).
_🤖 Addressed by [Claude Code](https://claude.com/claude-code)_
##########
core/src/main/java/org/apache/gravitino/job/JobManager.java:
##########
@@ -1410,6 +1426,39 @@ private boolean isStaleActiveJob(JobEntity job, long
now) {
&& lastUpdatedTimeInMs(job) + jobStagingDirKeepTimeInMs < now;
}
+ private void settleUnansweredCancellation(String metalake, JobEntity job,
long now) {
+ AtomicBoolean settled = new AtomicBoolean(false);
+ updateJobEntity(
+ metalake,
+ job,
+ latestJobEntity -> {
+ // The owner may have answered the cancellation since the
listJobs() snapshot.
+ if (!isCancellationUnanswered(latestJobEntity, now)) {
+ return latestJobEntity;
+ }
+ settled.set(true);
+ return toUpdatedStatusJobEntity(
+ latestJobEntity,
JobExecutionInfo.of(JobHandle.Status.CANCELLED));
+ })
+ .filter(settledJob -> settled.get())
+ .ifPresent(
+ settledJob ->
+ LOG.warn(
+ "Job {} with execution id {} under metalake {} has been
cancelling for more "
+ + "than {} ms without its owner answering, marking it
as {}. This could be "
+ + "due to the Gravitino server running the job having
exited.",
+ job.name(),
+ job.jobExecutionId(),
+ metalake,
+ cancellationTimeoutInMs,
+ settledJob.status()));
Review Comment:
Kept as is: the job name is derived from its id and the execution id never
changes, so the snapshot and the re-read entity always have the same
identifiers. This also matches `expireStaleActiveJob()`. The logged status
already comes from `settledJob`.
_🤖 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]
