yuqi1129 opened a new pull request, #14278:
URL: https://github.com/apache/gravitino/pull/14278
### What changes were proposed in this pull request?
`JobManager.runJob` now registers the job before handing it to the job
executor:
1. Insert the job row as `QUEUED` with a reserved placeholder execution id
(`pending-submission:<jobId>`).
2. Submit it to the job executor.
3. Record the returned execution id with a version-checked update (at most 3
attempts on a conflict).
Recovery and races:
- If the submission fails, the row is deleted. If that delete fails, the
status pull later marks the job `FAILED`.
- The status pull reconciles a row whose submission is not recorded within
three status pull intervals, treating the job executor as the source of truth.
It calls a new `JobExecutor#findSubmittedJob(JobContext, JobTemplate)` SPI,
which defaults to empty:
- If an execution is found, it is recorded and tracked from then on.
- Otherwise the job is marked `FAILED` (or `CANCELLED` if it was
cancelling).
- If the submission returns after the job was settled, or the execution id
can't be recorded, the submitting server cancels the execution and reports a
non-retryable error (same as before).
- `cancelJob` on a job being submitted only marks it `CANCELLING`. The
execution is cancelled by whichever side records the submission. `getJob`
returns no output for it.
- `K8sJobExecutor` implements `findSubmittedJob`. The SparkApplication is
named after the Gravitino job id, so it is looked up the same way `submitJob`
names it, and only a resource of that job that is not being deleted counts.
The placeholder keeps the API and schema unchanged: the execution id is not
part of `JobDTO`, and `JobHandle.Status` gains no value that older clients
couldn't parse.
### Why are the changes needed?
The job was submitted before `entityStore.put`. If registration failed or
the server exited after the submission, the executor ran a job Gravitino didn't
know about, and a retry submitted it twice. Store OCC can't cover this window
because the executor is not part of the store transaction.
Fixes #13180
### Does this PR introduce _any_ user-facing change?
- New `JobExecutor#findSubmittedJob` default method (developer API),
documented in `docs/development/custom-job-executor.md`.
- A job whose submission was interrupted is now visible as `QUEUED` until it
is reconciled, instead of being unknown.
- A registration conflict before the submission is now reported as a
retryable conflict, since nothing has run yet.
### How was this patch tested?
- `TestJobManager`: register-before-submit order, conflict before
submission, bounded retries of the recording, cancel when the job was settled
or cancelled during the submission, cancel of a job being submitted (including
the submission being recorded concurrently), reconciliation (found / not found
/ cancelling / recorded concurrently / lookup failure).
- `TestJobManagerMultiNode` (two servers, shared relational store):
- server exits after the executor accepts the job, and the other node
settles it as `FAILED`;
- the same, with an executor that finds the execution, so it is recorded
and tracked to `SUCCEEDED`;
- the other node settles the job during a slow submission, and the
execution is cancelled;
- cancel from the other node during submission, and the execution is
cancelled.
- `TestK8sJobExecutor`: `findSubmittedJob` before/after submission, another
job's resource, a cancelled job, and a non-Spark template.
- Related job tests pass locally: `TestJobManager`,
`TestJobManagerMultiNode` (H2), `TestLocalJobExecutor`,
`TestBuiltInJobTemplateEventListener`, `TestJobTemplateValidationDispatcher`,
`TestJobEventDispatcher`, `TestJobHookDispatcher`, `TestJobMetaService` and
`TestK8sJobExecutor`.
--
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]
