yuqi1129 commented on code in PR #14278:
URL: https://github.com/apache/gravitino/pull/14278#discussion_r4399999112
##########
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:
Fixed in 6ae3ccb687. When recording the execution id conflicts, the fallback
write now stores the known execution id together with `CANCELLING`, so the row
is no longer a pending placeholder. The status pull then retries cancellation
of that execution instead of reconciling it through `findSubmittedJob`.
`testRecordingFailureRetriesCancellationWithoutLookup` covers the case where a
lookup finds nothing and the first cancellation fails (H2, MySQL and
PostgreSQL).
##########
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:
Fixed in 6ae3ccb687. A same-job SparkApplication that is being deleted now
fails the submission with an `IllegalStateException`, so `JobManager` keeps the
registration on the unknown-outcome path, and `findSubmittedJob` still finds
the resource until deletion completes. Confirmed rejection is now used only for
missing resources or resources of another job.
`testAdoptNoResourceBeingDeleted` now asserts that the failure is not a
rejection and that the resource can still be found.
--
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]
