jerryshao commented on code in PR #14047:
URL: https://github.com/apache/gravitino/pull/14047#discussion_r4334666688
##########
core/src/main/java/org/apache/gravitino/job/JobManager.java:
##########
@@ -110,6 +110,14 @@ public class JobManager implements JobOperationDispatcher {
private static final long JOB_STATUS_PULL_MIN_INTERVAL_IN_MS = 60 * 1000L;
// 1 minute
+ // A live owner of a node local job cancels it at its next status pull, and
records it as
+ // CANCELLED at the pull after the process exits, so it answers a
cancellation within about two
+ // pull intervals, as long as the process exits within one, e.g. it is
killed forcibly within
+ // gravitino.jobExecutor.local.cancelForceKillDelayInMs. A job still
CANCELLING after this many
+ // pull intervals is considered left behind by an owner that has exited.
This assumes all servers
+ // use the same pull interval and have their clocks in sync.
+ private static final int CANCELLATION_TIMEOUT_IN_PULL_INTERVALS = 3;
Review Comment:
Tightened the comment in f832aa79bb: it now states the assumption that
`cancelForceKillDelayInMs` is shorter than the pull interval (true for the
recommended 1 minute or more; the docs don't recommend 10s), and that otherwise
a job may be marked CANCELLED while its process is still exiting, after the
owner has already asked it to stop and scheduled the force kill. I kept the
timeout independent of the local executor's config rather than reading an
executor-specific key in JobManager.
_🤖 Addressed by [Claude Code](https://claude.com/claude-code)_
##########
core/src/main/java/org/apache/gravitino/job/JobManager.java:
##########
@@ -1443,6 +1459,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()));
+ }
+
+ private boolean isCancellationUnanswered(JobEntity job, long now) {
+ return job.status() == JobHandle.Status.CANCELLING
+ && now - lastUpdatedTimeInMs(job) > cancellationTimeoutInMs;
Review Comment:
Added `testPullJobStatusSettlesCancellationOnlyAfterTimeout` in f832aa79bb:
a cancellation 10s within the 3-interval timeout is left alone and one 10s
beyond it is settled. Testing exact equality would need a clock seam the pull
doesn't have, so the test brackets the boundary instead.
_🤖 Addressed by [Claude Code](https://claude.com/claude-code)_
##########
core/src/main/java/org/apache/gravitino/job/JobManager.java:
##########
@@ -1031,17 +1043,21 @@ private void pullAndUpdateJobStatus(String metalake) {
|| job.status() == JobHandle.Status.CANCELLING)
.toList();
+ long now = System.currentTimeMillis();
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())) {
- return;
+ // 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);
Review Comment:
Added both in f832aa79bb:
`testPullJobStatusContinuesAfterFailingToSettleCancellation` (the update throws
for one job, the next is still settled) and
`testPullJobStatusContinuesAfterFailingToCheckNodeLocalJobState`
(`isJobStateNodeLocal()` throws for one job, the next is still settled).
_🤖 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]
