dongjoon-hyun opened a new pull request, #58138:
URL: https://github.com/apache/spark/pull/58138
### What changes were proposed in this pull request?
This PR makes a barrier job cancellable while its max-concurrent-tasks slot
check is being
retried, by tracking the deferred submission in `DAGScheduler`:
- Add `deferredBarrierJobs`, a `ConcurrentHashMap[Int, JobListener]` keyed
by job id. It is
concurrent because the re-post runnable reads it from the
`messageScheduler` thread, off the
event loop.
- In the `BarrierJobSlotsNumberCheckFailed` retry branch of
`handleJobSubmitted`, register the
job's listener in the map before scheduling the re-post, and make the
re-post runnable skip
posting `JobSubmitted` when the id is no longer in the map.
- Remove the id from the map at the top of `handleJobSubmitted`, since the
job is being
(re-)processed at that point.
- In `handleJobCancellation`, if the id is removed from the map, fail the
deferred listener
directly with `sparkJobCancelled` (and drop the job's entry in
`barrierJobIdToNumTasksCheckFailures` so the failure counter does not
leak).
- Drain the map in `doCancelAllJobs` (failing each deferred job with the
cancel-all reason) and
in `cleanUpAfterSchedulerStop` (failing each with "SparkContext was shut
down").
Note: https://github.com/apache/spark/pull/58054 implements the same
mechanism restricted to
the executors-held case (`jobsDeferredWhileHeld`); once this PR is merged,
that PR will be
rebased on top of this generalized version.
### Why are the changes needed?
When a barrier job fails the max-concurrent-tasks slot check,
`handleJobSubmitted` re-posts the
`JobSubmitted` event every
`spark.scheduler.barrier.maxConcurrentTasksCheck.interval` (15s) up
to `spark.scheduler.barrier.maxConcurrentTasksCheck.maxFailures` (40) times,
an ~10 minute
window by default. During that window the job is registered nowhere
(`createResultStage` aborts
with the exception before any stage or `ActiveJob` is created), so:
- `sc.cancelJob(jobId)` and `sc.cancelAllJobs()` are silent no-ops
(`handleJobCancellation` finds nothing in `jobIdToStageIds`), and
- no `SparkListenerJobStart` was posted, so the job is invisible in the UI.
The user has no way to cancel the job until the retries are exhausted.
### Does this PR introduce _any_ user-facing change?
Yes, it is a bug fix. Previously, cancelling a barrier job during its
slot-check retry window
was silently ignored and the job kept retrying; now the job fails
immediately with the usual
job-cancelled error, and the pending re-post is skipped.
### How was this patch tested?
Pass the CIs.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Fable 5
--
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]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
