This is an automated email from the ASF dual-hosted git repository.
cloud-fan pushed a commit to branch branch-4.x
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.x by this push:
new 3ad3d2c29917 [SPARK-58616][CORE][PYTHON] Attribute a reason for
context-wide job cancellation in cancelAllJobs
3ad3d2c29917 is described below
commit 3ad3d2c299170d82878d1c51d99750c7835a663d
Author: Ganesha S <[email protected]>
AuthorDate: Wed Aug 12 10:42:50 2026 +0800
[SPARK-58616][CORE][PYTHON] Attribute a reason for context-wide job
cancellation in cancelAllJobs
### What changes were proposed in this pull request?
`cancelAllJobs` is the only cancellation entry point with no way to say why
the cancel happened. Every job it aborts fails with the same fixed message:
[SPARK_JOB_CANCELLED] Job 7 cancelled as part of cancellation of all
jobs
This threads an optional reason through the path
`SparkContext.cancelAllJobs` -> `DAGScheduler.cancelAllJobs` ->
`AllJobsCancelled` -> `doCancelAllJobs` -> `handleJobCancellation`, so it lands
in the `SPARK_JOB_CANCELLED` error of every cancelled job. This follows the
shape SPARK-48900 established for `cancelJobGroup` and `cancelJobsWithTag`.
- `AllJobsCancelled` gains `reason: Option[String]` (default `None`),
matching the sibling events `JobCancelled` / `JobGroupCancelled` /
`JobTagCancelled`.
- New `cancelAllJobs(reason: String)` overloads on `SparkContext` and
JavaSparkContext`; PySpark's `cancelAllJobs` gains an optional `reason`. The
existing no-arg forms are untouched, so the change is purely additive.
- When no reason is supplied, `doCancelAllJobs` falls back to
`DAGScheduler.DEFAULT_CANCEL_ALL_JOBS_REASON` — the historical wording
verbatim, so the default-path message is byte-identical.
- `cancelAllJobs` now logs that it was invoked, which its siblings already
did.
- Internal callers now pass a reason: the REPL SIGINT handler, the Spark
SQL CLI SIGINT handler, and the `DAGScheduler` event-loop `onError` shutdown
path.
The message template is already `Job <jobId> cancelled <reason>`, so
`error-conditions.json` is unchanged.
### Why are the changes needed?
The fixed message says what happened but not who triggered it or why, and
it is reported identically on jobs that were perfectly healthy and are only
being aborted as collateral of someone else's cancel. When reading logs there
is no way to tell a context-wide cancellation from a genuine failure of the job
in question, which makes this class of failure slow to investigate.
### Does this PR introduce _any_ user-facing change?
Yes, additively:
1. New `cancelAllJobs(reason)` in Scala and Java, and a new optional
`reason` parameter in PySpark.
2. The reason in `SPARK_JOB_CANCELLED` becomes caller-dependent for the
internal callers that now supply one. A shell interrupt previously produced
`Job 7 cancelled as part of cancellation of all jobs`; it now produces `Job 7
cancelled because the driver process received an interrupt signal (SIGINT)`.
Callers that supply no reason still produce the original text, and the
error condition (`SPARK_JOB_CANCELLED`) and SQLSTATE (`XXKDA`) are unchanged,
so consumers matching on the error class are unaffected. A consumer
string-matching the full message of an attributed cancel would see the new
wording.
### How was this patch tested?
- Two new cases in `DAGSchedulerSuite` covering both branches of the
fallback, asserting via `checkError` on condition, SQLSTATE, and message
parameters. `DAGSchedulerSuite` and `JobCancellationSuite` pass in full (234
tests).
- A new case in `python/pyspark/tests/test_context.py` that cancels a real
running job with a reason and asserts it reaches the job's error while the
generic wording does not appear, this covers the py4j round trip that
unit-level stubs cannot. Confirmed to fail when the reason is not forwarded to
the JVM.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code (Opus 5)
Closes #57824 from ganeshashree/SPARK-58616.
Lead-authored-by: Ganesha S <[email protected]>
Co-authored-by: Ganesha S <[email protected]>
Signed-off-by: Wenchen Fan <[email protected]>
(cherry picked from commit b82f473dbdd2f876ceac06f7bd7b4858a7c8f7d8)
Signed-off-by: Wenchen Fan <[email protected]>
---
.../main/scala/org/apache/spark/SparkContext.scala | 14 ++++++++++
.../apache/spark/api/java/JavaSparkContext.scala | 9 +++++++
.../org/apache/spark/scheduler/DAGScheduler.scala | 30 +++++++++++++++-------
.../apache/spark/scheduler/DAGSchedulerEvent.scala | 3 ++-
.../apache/spark/scheduler/DAGSchedulerSuite.scala | 28 ++++++++++++++++++++
python/pyspark/core/context.py | 16 ++++++++++--
python/pyspark/tests/test_context.py | 27 +++++++++++++++++++
.../scala/org/apache/spark/repl/Signaling.scala | 2 +-
.../sql/hive/thriftserver/SparkSQLCLIDriver.scala | 3 ++-
9 files changed, 118 insertions(+), 14 deletions(-)
diff --git a/core/src/main/scala/org/apache/spark/SparkContext.scala
b/core/src/main/scala/org/apache/spark/SparkContext.scala
index c0203aa032ba..96b326e107af 100644
--- a/core/src/main/scala/org/apache/spark/SparkContext.scala
+++ b/core/src/main/scala/org/apache/spark/SparkContext.scala
@@ -2777,6 +2777,20 @@ class SparkContext(config: SparkConf) extends Logging {
dagScheduler.cancelAllJobs()
}
+ /**
+ * Cancel all jobs that have been scheduled or are running.
+ *
+ * @param reason reason for cancellation. It is surfaced in the error of
every cancelled job, so
+ * that a job aborted as collateral of a context-wide
cancellation can be told
+ * apart from one that failed on its own.
+ *
+ * @since 4.4.0
+ */
+ def cancelAllJobs(reason: String): Unit = {
+ assertNotStopped()
+ dagScheduler.cancelAllJobs(Option(reason))
+ }
+
/**
* Cancel a given job if it's scheduled or running.
*
diff --git
a/core/src/main/scala/org/apache/spark/api/java/JavaSparkContext.scala
b/core/src/main/scala/org/apache/spark/api/java/JavaSparkContext.scala
index bb8b02616c93..35de8f531250 100644
--- a/core/src/main/scala/org/apache/spark/api/java/JavaSparkContext.scala
+++ b/core/src/main/scala/org/apache/spark/api/java/JavaSparkContext.scala
@@ -819,6 +819,15 @@ class JavaSparkContext(val sc: SparkContext) extends
Closeable {
/** Cancel all jobs that have been scheduled or are running. */
def cancelAllJobs(): Unit = sc.cancelAllJobs()
+ /**
+ * Cancel all jobs that have been scheduled or are running.
+ *
+ * @param reason reason for cancellation
+ *
+ * @since 4.4.0
+ */
+ def cancelAllJobs(reason: String): Unit = sc.cancelAllJobs(reason)
+
/**
* Returns a Java map of JavaRDDs that have marked themselves as persistent
via cache() call.
*
diff --git a/core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala
b/core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala
index 05c142558a43..652d38b7a4e8 100644
--- a/core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala
+++ b/core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala
@@ -1628,9 +1628,14 @@ private[spark] class DAGScheduler(
/**
* Cancel all jobs that are running or waiting in the queue.
+ *
+ * @param reason reason for cancellation. It is surfaced in the error of
every cancelled job, so
+ * that a job aborted as collateral of a context-wide
cancellation can be told
+ * apart from one that failed on its own.
*/
- def cancelAllJobs(): Unit = {
- eventProcessLoop.post(AllJobsCancelled)
+ def cancelAllJobs(reason: Option[String] = None): Unit = {
+ logInfo(log"Asked to cancel all jobs${MDC(REASON, reason.map(" " +
_).getOrElse(""))}")
+ eventProcessLoop.post(AllJobsCancelled(reason))
}
/**
@@ -1641,10 +1646,12 @@ private[spark] class DAGScheduler(
eventProcessLoop.post(CleanupQueryJobs(executionId))
}
- private[scheduler] def doCancelAllJobs(): Unit = {
- // Cancel all running jobs.
- runningStages.map(_.firstJobId).foreach(handleJobCancellation(_,
- Option("as part of cancellation of all jobs")))
+ private[scheduler] def doCancelAllJobs(reason: Option[String] = None): Unit
= {
+ // Cancel all running jobs. A job caught here was healthy and is being
aborted as collateral of
+ // a context-wide cancellation, not because it failed on its own, so
attribute the reason when
+ // the caller supplied one.
+ val updatedReason =
reason.getOrElse(DAGScheduler.DEFAULT_CANCEL_ALL_JOBS_REASON)
+ runningStages.map(_.firstJobId).foreach(handleJobCancellation(_,
Option(updatedReason)))
activeJobs.clear() // These should already be empty by this point,
jobIdToActiveJob.clear() // but just in case we lost track of some jobs...
}
@@ -4675,8 +4682,8 @@ private[scheduler] class
DAGSchedulerEventProcessLoop(dagScheduler: DAGScheduler
case JobTagCancelled(tag, reason, cancelledJobs) =>
dagScheduler.handleJobTagCancelled(tag, reason, cancelledJobs)
- case AllJobsCancelled =>
- dagScheduler.doCancelAllJobs()
+ case AllJobsCancelled(reason) =>
+ dagScheduler.doCancelAllJobs(reason)
case CleanupQueryJobs(executionId) =>
dagScheduler.doCleanupQueryJobs(executionId)
@@ -4735,7 +4742,8 @@ private[scheduler] class
DAGSchedulerEventProcessLoop(dagScheduler: DAGScheduler
override def onError(e: Throwable): Unit = {
logError("DAGSchedulerEventProcessLoop failed; shutting down
SparkContext", e)
try {
- dagScheduler.doCancelAllJobs()
+ dagScheduler.doCancelAllJobs(
+ Option("because the DAGScheduler event loop failed and the
SparkContext is shutting down"))
} catch {
case t: Throwable => logError("DAGScheduler failed to cancel all jobs.",
t)
}
@@ -4753,6 +4761,10 @@ private[spark] object DAGScheduler {
// this is a simplistic way to avoid resubmitting tasks in the non-fetchable
map stage one by one
// as more failure events come in
val RESUBMIT_TIMEOUT = 200
+
+ // Fallback reason used when a context-wide cancellation does not supply a
more specific one.
+ // Kept as the historical wording so existing log and error consumers are
unaffected.
+ val DEFAULT_CANCEL_ALL_JOBS_REASON = "as part of cancellation of all jobs"
}
/**
diff --git
a/core/src/main/scala/org/apache/spark/scheduler/DAGSchedulerEvent.scala
b/core/src/main/scala/org/apache/spark/scheduler/DAGSchedulerEvent.scala
index cc788e5c65bc..44811a721483 100644
--- a/core/src/main/scala/org/apache/spark/scheduler/DAGSchedulerEvent.scala
+++ b/core/src/main/scala/org/apache/spark/scheduler/DAGSchedulerEvent.scala
@@ -76,7 +76,8 @@ private[scheduler] case class JobTagCancelled(
reason: Option[String],
cancelledJobs: Option[Promise[Seq[ActiveJob]]]) extends DAGSchedulerEvent
-private[scheduler] case object AllJobsCancelled extends DAGSchedulerEvent
+private[scheduler] case class AllJobsCancelled(reason: Option[String] = None)
+ extends DAGSchedulerEvent
private[scheduler] case class CleanupQueryJobs(executionId: Long) extends
DAGSchedulerEvent
diff --git
a/core/src/test/scala/org/apache/spark/scheduler/DAGSchedulerSuite.scala
b/core/src/test/scala/org/apache/spark/scheduler/DAGSchedulerSuite.scala
index 7d51ce58449c..d18b260d7b24 100644
--- a/core/src/test/scala/org/apache/spark/scheduler/DAGSchedulerSuite.scala
+++ b/core/src/test/scala/org/apache/spark/scheduler/DAGSchedulerSuite.scala
@@ -2334,6 +2334,34 @@ class DAGSchedulerSuite extends SparkFunSuite with
TempLocalSparkContext with Ti
assert(scheduler.waitingStages.isEmpty)
}
+ test("SPARK-58616: cancelAllJobs surfaces the supplied reason in the job
failure") {
+ val jobId = submit(new MyRDD(sc, 1, Nil), Array(0))
+ assert(scheduler.runningStages.size === 1)
+ runEvent(AllJobsCancelled(Some("because the user requested cancellation")))
+ checkError(
+ exception = failure.asInstanceOf[SparkException],
+ condition = "SPARK_JOB_CANCELLED",
+ sqlState = "XXKDA",
+ parameters = scala.collection.immutable.Map(
+ "jobId" -> jobId.toString,
+ "reason" -> "because the user requested cancellation"))
+ assertDataStructuresEmpty()
+ }
+
+ test("SPARK-58616: cancelAllJobs falls back to the default reason when none
is supplied") {
+ val jobId = submit(new MyRDD(sc, 1, Nil), Array(0))
+ assert(scheduler.runningStages.size === 1)
+ runEvent(AllJobsCancelled())
+ checkError(
+ exception = failure.asInstanceOf[SparkException],
+ condition = "SPARK_JOB_CANCELLED",
+ sqlState = "XXKDA",
+ parameters = scala.collection.immutable.Map(
+ "jobId" -> jobId.toString,
+ "reason" -> "as part of cancellation of all jobs"))
+ assertDataStructuresEmpty()
+ }
+
test("misbehaved accumulator should not crash DAGScheduler and
SparkContext") {
val acc = new LongAccumulator {
override def add(v: java.lang.Long): Unit = throw new
DAGSchedulerSuiteDummyException
diff --git a/python/pyspark/core/context.py b/python/pyspark/core/context.py
index 2a6d8ab68e5f..339c94f54e6f 100644
--- a/python/pyspark/core/context.py
+++ b/python/pyspark/core/context.py
@@ -2491,19 +2491,31 @@ class SparkContext:
"""
return self._jsc.cancelJobsWithTag(tag)
- def cancelAllJobs(self) -> None:
+ def cancelAllJobs(self, reason: Optional[str] = None) -> None:
"""
Cancel all jobs that have been scheduled or are running.
.. versionadded:: 1.1.0
+ Parameters
+ ----------
+ reason : str, optional
+ Reason for cancellation. It is surfaced in the error of every
cancelled job, so that
+ a job aborted as collateral of a context-wide cancellation can be
told apart from one
+ that failed on its own.
+
+ .. versionadded:: 4.4.0
+
See Also
--------
:meth:`SparkContext.cancelJobGroup`
:meth:`SparkContext.cancelJobsWithTag`
:meth:`SparkContext.runJob`
"""
- self._jsc.sc().cancelAllJobs()
+ if reason is None:
+ self._jsc.sc().cancelAllJobs()
+ else:
+ self._jsc.sc().cancelAllJobs(reason)
def statusTracker(self) -> StatusTracker:
"""
diff --git a/python/pyspark/tests/test_context.py
b/python/pyspark/tests/test_context.py
index 66bb0ebdd097..1c7de86b3b3e 100644
--- a/python/pyspark/tests/test_context.py
+++ b/python/pyspark/tests/test_context.py
@@ -321,6 +321,33 @@ class ContextTests(unittest.TestCase):
with SparkContext("local-cluster[3, 1, 1024]") as sc:
sc.range(2).foreach(lambda _: create_spark_context())
+ def test_cancel_all_jobs_reason_reaches_the_job_failure(self):
+ # SPARK-58616: the reason must survive the trip to the JVM and land in
the cancelled
+ # job's error, instead of the generic "as part of cancellation of all
jobs".
+ with SparkContext() as sc:
+ errors = []
+
+ def run_job():
+ try:
+ sc.parallelize(range(4), 4).map(lambda x:
time.sleep(60)).collect()
+ except Exception as e:
+ errors.append(str(e))
+
+ job = threading.Thread(target=run_job)
+ job.start()
+ # Wait for the job to reach the scheduler before cancelling it.
+ deadline = time.time() + 60
+ while not sc.statusTracker().getActiveJobsIds():
+ self.assertLess(time.time(), deadline, "job never reached the
scheduler")
+ time.sleep(0.1)
+
+ sc.cancelAllJobs(reason="because the test asked for it")
+ job.join(60)
+
+ self.assertEqual(len(errors), 1)
+ self.assertIn("because the test asked for it", errors[0])
+ self.assertNotIn("as part of cancellation of all jobs", errors[0])
+
class ContextTestsWithResources(unittest.TestCase):
def setUp(self):
diff --git a/repl/src/main/scala/org/apache/spark/repl/Signaling.scala
b/repl/src/main/scala/org/apache/spark/repl/Signaling.scala
index 9577e0ecaa2e..81cd0794edf8 100644
--- a/repl/src/main/scala/org/apache/spark/repl/Signaling.scala
+++ b/repl/src/main/scala/org/apache/spark/repl/Signaling.scala
@@ -33,7 +33,7 @@ private[repl] object Signaling extends Logging {
if (!ctx.statusTracker.getActiveJobIds().isEmpty) {
logWarning("Cancelling all active jobs, this can take a while. " +
"Press Ctrl+C again to exit now.")
- ctx.cancelAllJobs()
+ ctx.cancelAllJobs("because the driver process received an interrupt
signal (SIGINT)")
true
} else {
false
diff --git
a/sql/hive-thriftserver/src/main/scala/org/apache/spark/sql/hive/thriftserver/SparkSQLCLIDriver.scala
b/sql/hive-thriftserver/src/main/scala/org/apache/spark/sql/hive/thriftserver/SparkSQLCLIDriver.scala
index e1553ce9b368..ab322d2cb407 100644
---
a/sql/hive-thriftserver/src/main/scala/org/apache/spark/sql/hive/thriftserver/SparkSQLCLIDriver.scala
+++
b/sql/hive-thriftserver/src/main/scala/org/apache/spark/sql/hive/thriftserver/SparkSQLCLIDriver.scala
@@ -72,7 +72,8 @@ private[hive] object SparkSQLCLIDriver extends Logging {
def installSignalHandler(): Unit = {
HiveInterruptUtils.add(() => {
if (SparkSQLEnv.sparkContext != null) {
- SparkSQLEnv.sparkContext.cancelAllJobs()
+ SparkSQLEnv.sparkContext.cancelAllJobs(
+ "because the user interrupted the Spark SQL CLI with Ctrl+C")
}
})
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
