dongjoon-hyun commented on code in PR #58138:
URL: https://github.com/apache/spark/pull/58138#discussion_r3817137621
##########
core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala:
##########
@@ -4284,7 +4348,18 @@ private[spark] class DAGScheduler(
}
private[scheduler] def handleJobCancellation(jobId: Int, reason:
Option[String]): Unit = {
- if (!jobIdToStageIds.contains(jobId)) {
+ val deferred = deferredBarrierJobs.get(jobId)
+ if (deferred != null) {
+ if (deferred ne deferredJobCancelledMarker) {
+ // A barrier job deferred for a slot-check retry is registered nowhere
else, so fail its
+ // listener directly. Leave the marker in place instead of removing
the entry: the
+ // pending re-post always fires, and handleJobSubmitted drops it on
finding the marker.
+ deferredBarrierJobs.put(jobId, deferredJobCancelledMarker)
+ barrierJobIdToNumTasksCheckFailures.remove(jobId)
+ deferred.listener.jobFailed(
+ SparkCoreErrors.sparkJobCancelled(jobId, reason.getOrElse(""), null))
Review Comment:
Good catch — confirmed. With an ordinary shuffle upstream of the barrier
stage, `createShuffleMapStage` registers the ancestor stage (and the job id in
`jobIdToStageIds`) before the barrier slot check throws, so a second
cancellation of the abandoned job id reached `jobIdToActiveJob(jobId)` and
crashed the event loop.
Fixed in 5b1eedd6848: the stage-unregistration part of
`cleanupStateForJobAndIndependentStages` is extracted into a job-id-keyed
`cleanupStagesForJob`, which the deferred cancellation now invokes — dropping
the job from its registered stages and unregistering the stages no other job
needs, before failing the listener. Added the regression test `SPARK-58887:
cancelling a deferred barrier job drops its partial stage registrations` using
the ordinary-shuffle -> barrier-shuffle -> result graph; it verifies the
registrations are gone after the cancellation and that a later cancellation of
the same job id (after the re-post is dropped) is a harmless no-op.
##########
core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala:
##########
@@ -1942,17 +1981,27 @@ private[spark] class DAGScheduler(
// Cancel all jobs that have all provided tags.
// First finds all active jobs with this group id, and then kill stages
for them.
val jobsToBeCancelled = activeJobs.filter { activeJob =>
- Option(activeJob.properties).exists { properties =>
-
Option(properties.getProperty(SparkContext.SPARK_JOB_TAGS)).getOrElse("")
-
.split(SparkContext.SPARK_JOB_TAGS_SEP).filter(!_.isEmpty).toSet.contains(tag)
- }
+ hasJobTag(activeJob.properties, tag)
}
+ // A barrier job deferred for a slot-check retry is not in `activeJobs`
yet (and thus is not
+ // reported through `cancelledJobs` either), so match it by the properties
captured at
+ // submission.
+ val deferredTagged = deferredJobsMatching(hasJobTag(_, tag))
val updatedReason =
reason.getOrElse("part of cancelled job tags %s".format(tag))
- jobsToBeCancelled.map(_.jobId).foreach(handleJobCancellation(_,
Option(updatedReason)))
+ (jobsToBeCancelled.map(_.jobId) ++ deferredTagged)
+ .foreach(handleJobCancellation(_, Option(updatedReason)))
cancelledJobs.map(_.success(jobsToBeCancelled.toSeq))
Review Comment:
Fixed in 5b1eedd6848: the promise payload of `cancelJobsWithTagWithFuture`
is now `Seq[CancelledJobInfo]` — a new `private[spark] case class
CancelledJobInfo(jobId, properties)` — and `handleJobTagCancelled` reports both
active jobs and deferred barrier jobs through it, so classic
`SparkSession.interruptTag` / `interruptOperation` / `interruptAll` now return
the SQL execution ids of deferred jobs they cancelled. `doInterruptTag` and the
existing `SparkSessionJobTaggingAndCancellationSuite` only read `.properties`,
so they work unchanged (the erased signature is unchanged as well, and every
touched API is `private[spark]`). Added the regression test `SPARK-58887: tag
cancellation reports a deferred barrier job in its promise`.
--
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]
