sunchao commented on code in PR #58138:
URL: https://github.com/apache/spark/pull/58138#discussion_r3816907251
##########
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:
[P2] Include deferred cancellations in the interruption result
The PR notes that deferred jobs are omitted from this promise, but that has
a public-facing consequence: [classic
SparkSession.doInterruptTag](https://github.com/apache/spark/blob/b093c6c7d71e378cc0cd7e4fd2c44d6f3b58378a/sql/core/src/main/scala/org/apache/spark/sql/classic/SparkSession.scala#L808-L813)
derives all returned SQL execution IDs from it. A three-partition
`mapInPandas(..., barrier=True)` query on `local[2]` can therefore be cancelled
by `interruptOperation`, `interruptTag`, or `interruptAll`, while the API
returns an empty sequence.
The captured properties already contain the SQL root execution ID. A focused
scheduler regression confirms that the listener is failed but the returned ID
list is empty. Please return cancellation metadata for deferred jobs too,
rather than restricting the result to `ActiveJob`. This concern is specific to
classic SQL; Connect builds its returned IDs from its operation holders.
##########
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:
[P1] Clean up partially registered stages before abandoning this job
A deferred barrier job is not necessarily registered nowhere else. On
`local[2]`, an ordinary shuffle -> three-partition barrier shuffle -> result
graph causes `getOrCreateShuffleMapStage` to create and register the ordinary
ancestor before the barrier stage's slot check throws. This branch fails the
listener, but leaves the job in `jobIdToStageIds` and the ancestor's `jobIds`,
as well as the stage/shuffle maps. The pending repost then consumes the marker
and permanently abandons those registrations.
After that, another cancellation of the old job reaches
`jobIdToActiveJob(jobId)` and throws `NoSuchElementException`; the production
event-loop error handler shuts down the SparkContext. Reusing the ancestor can
also leave a dead job ID pinning the stage.
The partial registration itself predates this PR. The introduced case is
group/tag cancellation becoming terminal without cleaning it. An identical
regression that allows capacity to recover passes on the base (the ignored
cancellation lets the job finish and clean up) and fails on this head at line
4366. Please remove the cancelled job's partial stage registrations before
dropping its retry, while preserving stages needed by other jobs.
--
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]
