jerryshao opened a new pull request, #13147:
URL: https://github.com/apache/gravitino/pull/13147
### What changes were proposed in this pull request?
Make each server only track the local jobs it runs itself, so that servers
sharing the same metadata store no longer overwrite each other's job statuses.
- `LocalJobExecutor`
- Generates a random 8-hex-digit executor id on initialization and embeds
it in the job execution ids: `local-job-<executorId>-<uuid>`.
- Implements the new `JobExecutor` methods `ownsJob(jobId)` and
`isJobStateNodeLocal()`.
- Fixes the thread pool to run up to `maxRunningJobs` jobs concurrently
(core size was 0 with an unbounded queue, so only one job ran at a time).
- `JobExecutor`: adds the default methods `ownsJob` (default `true`) and
`isJobStateNodeLocal` (default `false`), so external job executors keep their
current behavior.
- `JobManager`
- The status pull only queries the jobs owned by this server's executor,
and skips the others.
- `cancelJob` on a job owned by another server only marks it as
`CANCELLING`. The owning server cancels it on its next status pull. This only
applies to node-local job state.
- `cleanUpStagingDirs` marks active jobs not updated for
`gravitino.job.stagingDirKeepTimeInMs` as `FAILED` (or `CANCELLED` if
cancelling), e.g. jobs left behind by a server that exited. They are then kept
for another keep time and cleaned up like other finished jobs.
- Docs: describe the multi-node behavior of the local job executor and its
limitations.
### Why are the changes needed?
`LocalJobExecutor` keeps job status only in the memory of the server that
submitted the job, but every server pulls the status of all active jobs from
the shared store. A server that doesn't own a job gets `NoSuchJobException`
from its own executor and marks the job as `FAILED`, even if the job succeeded
on the other server. Cancelling a job through a non-owning server also failed
with a 500 error.
Fix: #13146
### Does this PR introduce _any_ user-facing change?
- Job execution ids of the local job executor change from `local-job-<uuid>`
to `local-job-<executorId>-<uuid>`.
- Cancelling a job through a server that doesn't run it now returns
`CANCELLING` instead of failing. The cancellation takes effect on the owning
server's next status pull (up to `gravitino.job.statusPullIntervalInMs`).
- `gravitino.job.stagingDirKeepTimeInMs` now also expires active jobs whose
status has not changed for that long, for all job executors.
- `gravitino.jobExecutor.local.maxRunningJobs` now takes effect.
- New default methods on the `JobExecutor` developer API: `ownsJob` and
`isJobStateNodeLocal`.
Known limitations, documented in `manage-jobs-in-gravitino.md`:
- A job running, or staying queued, longer than
`gravitino.job.stagingDirKeepTimeInMs` without a status change is marked as
`FAILED`.
- The executor id changes on every restart, so a restarted server doesn't
recognize the jobs it ran before. These jobs are only marked as `FAILED` once
they expire, up to about 7.7 days with the default config. Active jobs created
before upgrading, with the old id format, are handled the same way.
- During a rolling upgrade, servers still running the old version may mark
jobs run by upgraded servers as `FAILED`.
### How was this patch tested?
- Unit tests
- `TestLocalJobExecutor`: job ownership by executor id, and running jobs
concurrently up to `maxRunningJobs`.
- `TestJobManager`: skipping jobs owned by other executors, cross-node
cancellation, not re-cancelling jobs of non-node-local executors, and expiring
stale active jobs in the cleanup (including concurrent updates and failures).
- New `TestJobManagerMultiNode`: two `JobManager`s with real
`LocalJobExecutor`s sharing a real relational store (H2, plus MySQL/PostgreSQL
with `dockerTest=true`) cover owner-only tracking, cancelling through the other
node, and expiring jobs left by an exited node.
- Integration test `JobIT` (embedded mode) simulates another server's job in
the metadata store: it is not marked as `FAILED`, cancelling it returns
`CANCELLING`, and a stale one is marked as `FAILED`.
🤖 Generated with [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]
