You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by we...@apache.org on 2026/08/12 02:43:05 UTC
(spark) branch master updated: [SPARK-58616][CORE][PYTHON] Attribute a reason for context-wide job cancellation in cancelAllJobs
This is an automated email from the ASF dual-hosted git repository.
cloud-fan pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/master by this push:
new b82f473dbdd2 [SPARK-58616][CORE][PYTHON] Attribute a reason for context-wide job cancellation in cancelAllJobs
b82f473dbdd2 is described below
commit b82f473dbdd2f876ceac06f7bd7b4858a7c8f7d8
Author: Ganesha S <ga...@gmail.com>
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 <ga...@gmail.com>
Co-authored-by: Ganesha S <ga...@databricks.com>
Signed-off-by: Wenchen Fan <we...@databricks.com>
---
.../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: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org