Copilot commented on code in PR #14278:
URL: https://github.com/apache/gravitino/pull/14278#discussion_r4399357567
##########
plugins/k8s-job-executor/src/main/java/org/apache/gravitino/job/k8s/K8sJobExecutor.java:
##########
@@ -203,7 +204,8 @@ public String submitJob(JobContext context, JobTemplate
jobTemplate) {
if (existing == null
|| existing.getMetadata().getDeletionTimestamp() != null
|| !K8sJobResourceUtils.isOfJob(existing, context.jobId())) {
- throw new IllegalStateException(
+ // Nothing was created for this job, so the submission is rejected for
sure.
+ throw new JobSubmissionRejectedException(
Review Comment:
A same-job SparkApplication with a deletion timestamp may still have running
pods while foreground deletion finishes. Throwing
`JobSubmissionRejectedException` makes `JobManager` delete its registration and
staging directory before cleanup is confirmed. Keep this case on the
unknown-outcome path, reserve confirmed rejection for missing or different-job
resources, and update `testAdoptNoResourceBeingDeleted` accordingly.
##########
core/src/main/java/org/apache/gravitino/job/JobManager.java:
##########
@@ -612,52 +630,299 @@ public JobEntity runJob(String metalake, String
jobTemplateName, Map<String, Str
.withRuntimeJobTemplate(runtimeJobTemplateJson)
.build();
+ jobsBeingSubmitted.add(jobId);
try {
- entityStore.put(jobEntity, false /* overwrite */);
+ return registerAndSubmit(metalake, pendingJobEntity, jobTemplate,
jobStagingDir);
+ } finally {
+ jobsBeingSubmitted.remove(jobId);
+ }
+ }
+
+ private JobEntity registerAndSubmit(
+ String metalake, JobEntity pendingJobEntity, JobTemplate jobTemplate,
File jobStagingDir) {
+ long jobId = pendingJobEntity.id();
+ String jobTemplateName = pendingJobEntity.jobTemplateName();
+ try {
+ entityStore.put(pendingJobEntity, false /* overwrite */);
} catch (NoSuchEntityException e) {
- LOG.error(
- "Job {} was submitted as execution {} but could not be registered
because its template "
- + "{} or metalake {} no longer exists",
- jobEntity.name(),
- jobExecutionId,
- jobTemplateName,
- metalake,
- e);
+ deleteStagingDirOfUnsubmittedJob(jobStagingDir, jobId);
throw new NoSuchJobTemplateException(
e,
"Job template with name %s under metalake %s does not exist",
jobTemplateName,
metalake);
- } catch (OptimisticLockException e) {
- // Cancellation is asynchronous and cannot undo work that already
finished. Request it
- // best-effort, but do not expose a retryable conflict after submitting
an execution.
- LOG.warn(
- "Job {} was submitted as execution {} but could not be registered
because of a "
- + "concurrent operation, cancelling the execution",
- jobEntity.name(),
- jobExecutionId,
- e);
+ } catch (RuntimeException e) {
+ // Nothing is submitted yet, so a conflict is reported as is and the
caller can retry.
+ deleteStagingDirOfUnsubmittedJob(jobStagingDir, jobId);
+ throw e;
+ } catch (IOException e) {
+ deleteStagingDirOfUnsubmittedJob(jobStagingDir, jobId);
+ throw new RuntimeException("Failed to register the job entity " +
pendingJobEntity, e);
+ }
+
+ // Ownership must be durable before the external submission: recovery does
not run the
+ // original request's post hook. A failure here has no external side
effect to recover.
+ try {
+ OwnerDispatcher owners =
GravitinoEnv.getInstance().internalOwnerDispatcher();
+ if (owners != null) {
+ owners.setOwner(
+ metalake,
+ NameIdentifierUtil.toMetadataObject(
+ pendingJobEntity.nameIdentifier(), Entity.EntityType.JOB),
+ PrincipalUtils.getCurrentUserName(),
+ Owner.Type.USER);
+ }
+ } catch (RuntimeException e) {
+ deleteUnsubmittedJob(metalake, pendingJobEntity, jobStagingDir, e);
+ throw e;
+ }
+
+ // Submit the job template to the job executor
+ String jobExecutionId;
+ try {
+ jobExecutionId =
+ jobExecutor.submitJob(new JobContext(jobId, metalake,
jobStagingDir), jobTemplate);
+ } catch (IllegalArgumentException | JobSubmissionRejectedException e) {
+ // The job executor rejects the job because it cannot be launched, for
example, a required
+ // configuration is missing. Rethrow it as is so the caller gets the
original reason.
+ deleteUnsubmittedJob(metalake, pendingJobEntity, jobStagingDir, e);
+ throw e;
+ } catch (Exception e) {
+ if (!findsSubmittedJobs) {
+ // The job executor can't find an execution the submission may have
left, so the
+ // registration couldn't recover it. The job executor removes what it
created instead.
+ RuntimeException failure =
+ new RuntimeException(
+ String.format(
+ "Failed to submit job %s of job template %s for execution:
%s",
+ pendingJobEntity.name(), jobTemplateName, e.getMessage()),
+ e);
+ deleteUnsubmittedJob(metalake, pendingJobEntity, jobStagingDir,
failure);
+ throw failure;
+ }
+
+ RuntimeException failure =
+ new RuntimeException(
+ String.format(
+ "Failed to submit job %s of job template %s for execution:
%s. The outcome is"
+ + " unknown, so the job is kept and cancelled, and any
execution found for it"
+ + " in the job executor is cancelled too. Check the job
before retrying",
+ pendingJobEntity.name(), jobTemplateName, e.getMessage()),
+ e);
+ // A failed request is not proof that nothing was submitted. Keep the
recovery anchor
+ // and persist cancellation before returning the failure. Never replay
the submission.
+ markFailedSubmissionForCancellation(metalake, pendingJobEntity, failure);
+ throw failure;
+ }
+
+ return recordSubmission(metalake, pendingJobEntity, jobExecutionId);
+ }
+
+ private void markFailedSubmissionForCancellation(
+ String metalake, JobEntity job, Exception failure) {
+ NameIdentifier jobIdent = NameIdentifierUtil.ofJob(metalake, job.name());
+ for (int attempt = 1; attempt <= SUBMISSION_WRITE_MAX_ATTEMPTS; attempt++)
{
try {
- jobExecutor.cancelJob(jobExecutionId);
- } catch (Exception cancelFailure) {
+ TreeLockUtils.doWithTreeLock(
+ jobIdent,
+ LockType.WRITE,
+ () ->
+ entityStore.update(
+ jobIdent, JobEntity.class, Entity.EntityType.JOB,
this::toCancellingJobEntity));
+ return;
+ } catch (OptimisticLockException e) {
+ failure.addSuppressed(e);
+ } catch (NoSuchEntityException e) {
+ // E.g. its metalake was dropped. Nothing is left to recover the
execution with.
LOG.error(
- "Failed to cancel the unregistered execution {} of job {}",
- jobExecutionId,
- jobEntity.name(),
- cancelFailure);
- e.addSuppressed(cancelFailure);
+ "Job {} under metalake {} has an unknown submission outcome and no
longer exists, so"
+ + " an execution of it can't be found and cancelled. Check the
job executor for"
+ + " an execution of job id {}",
+ job.name(),
+ metalake,
+ job.id(),
+ failure);
+ return;
+ } catch (Exception e) {
+ failure.addSuppressed(e);
+ break;
}
- throw new RuntimeException(
+ }
+ // The registration is still retained if the store cannot record the
cancellation. Recovery
+ // can at least track a found execution instead of leaving it unknown to
Gravitino.
+ LOG.error(
+ "Job {} under metalake {} has an unknown submission outcome and its
cancellation could"
+ + " not be recorded. The job registration is retained; check the
job before retrying",
+ job.name(),
+ metalake,
+ failure);
+ }
+
+ private JobEntity recordSubmission(
+ String metalake, JobEntity pendingJobEntity, String jobExecutionId) {
+ NameIdentifier jobIdent = NameIdentifierUtil.ofJob(metalake,
pendingJobEntity.name());
+ Exception failure = null;
+ JobEntity recorded = null;
+ // entityStore.update() re-reads the job, so a conflict with a concurrent
write, e.g. a
+ // cancellation, succeeds when retried. The retries are bounded, a job
that still can't be
+ // recorded is settled by the status pull.
+ for (int attempt = 1; attempt <= SUBMISSION_WRITE_MAX_ATTEMPTS; attempt++)
{
+ try {
+ recorded =
+ TreeLockUtils.doWithTreeLock(
+ jobIdent,
+ LockType.WRITE,
+ () ->
+ entityStore.update(
+ jobIdent,
+ JobEntity.class,
+ Entity.EntityType.JOB,
+ latest -> withSubmittedExecution(latest,
jobExecutionId)));
+ break;
+ } catch (OptimisticLockException e) {
+ if (failure == null) {
+ failure = e;
+ } else if (failure != e) {
+ failure.addSuppressed(e);
+ }
+ } catch (Exception e) {
+ // The job is gone, e.g. its metalake was dropped, or the store failed.
+ if (failure != null && failure != e) {
+ e.addSuppressed(failure);
+ }
+ failure = e;
+ break;
+ }
+ }
+
+ if (recorded != null && jobExecutionId.equals(recorded.jobExecutionId())) {
+ if (recorded.status() == JobHandle.Status.CANCELLING) {
+ // The job was cancelled while it was being submitted, before it could
be cancelled in the
+ // job executor.
+ cancelSubmittedExecution(recorded, jobExecutionId, null);
+ }
+ return recorded;
+ }
+
+ // The submission could not be recorded, or another server settled the job
as its submission
+ // took too long. Cancellation is asynchronous and cannot undo work that
already finished, so
+ // request it best-effort, but do not expose a retryable conflict after
submitting an execution.
+ String reason;
+ if (recorded == null) {
+ reason = "the submission could not be recorded";
+ } else {
+ String settledAs =
+ isPendingSubmission(recorded)
+ ? "marked it as " + recorded.status()
+ : "recorded execution " + recorded.jobExecutionId() + " for it";
+ reason =
String.format(
- "Failed to register job %s after submitting execution %s;
cancellation was attempted "
- + "but the execution may already have run. Check the
execution before retrying",
- jobEntity.name(), jobExecutionId),
+ "another Gravitino server %s, as its submission took longer than
%d ms. Make sure"
+ + " the job executor submits a job within three"
+ + " gravitino.job.statusPullIntervalInMs, and that all
servers use the same"
+ + " interval and have their clocks in sync",
+ settledAs, submissionTimeoutInMs);
+ }
+ RuntimeException registrationFailure =
+ new RuntimeException(
+ String.format(
+ "Failed to register job %s after submitting execution %s: %s.
Cancellation was"
+ + " attempted but the execution may already have run.
Check the execution"
+ + " before retrying",
+ pendingJobEntity.name(), jobExecutionId, reason),
+ failure);
+ LOG.warn(
+ "Job {} was submitted as execution {} but {}, cancelling the
execution",
+ pendingJobEntity.name(),
+ jobExecutionId,
+ reason,
+ failure);
+ if (recorded == null && !(failure instanceof NoSuchEntityException)) {
+ // The registration still anchors recovery even when recording the id
conflicted. Persist
+ // the cancellation intent too, so a failed immediate cancellation is
retried after lookup.
+ markFailedSubmissionForCancellation(metalake, pendingJobEntity,
registrationFailure);
+ }
Review Comment:
If the three execution-id writes conflict but this cancellation-intent write
succeeds, the row remains a pending placeholder:
`markFailedSubmissionForCancellation` only changes its status. For an executor
using the default `findSubmittedJob`, a transient failure of the immediate
cancel cannot then be retried—the poller gets an empty lookup result and marks
the job `CANCELLED` even though its accepted execution may still be running.
Preserve the already-known execution id with the cancellation intent, or
otherwise make that execution recoverable, and cover the no-lookup case.
--
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]
