Copilot commented on code in PR #14047:
URL: https://github.com/apache/gravitino/pull/14047#discussion_r4332899512
##########
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:
The new settling path can legitimately hit optimistic concurrency (OCC)
conflicts when multiple servers attempt to settle the same job. If an
`OptimisticLockException` (or the project’s OCC exception type) bubbles up
here, it will be logged as an error, creating noisy logs for an expected race.
Consider adding a specific catch for the OCC exception *before* the generic
`RuntimeException` catch and log it at a lower level (debug/info) with a 'will
retry on next pull' message.
##########
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:
The implementation checks `now - lastUpdatedTimeInMs(job) >
cancellationTimeoutInMs` (strictly greater than 3×pull interval), so the
behavior is 'more than three intervals' rather than 'three intervals'. To avoid
ambiguity, consider updating the doc phrasing to match the strict comparison
(e.g., 'more than three times') or adjust the code to use `>=` if the intent is
exactly three intervals.
##########
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:
This warning logs `job.name()` / `job.jobExecutionId()` from the potentially
stale `listJobs()` snapshot, while the operation actually applies to
`settledJob` (the re-read entity). For accuracy (especially if any fields can
differ between snapshot and latest), prefer logging identifiers from
`settledJob` (or `latestJobEntity`) rather than `job`.
--
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]
