[SPARK-58616][CORE][PYTHON] Attribute a reason for context-wide job cancellation in cancelAllJobs

[ganeshashree]

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 cancelAllJobsgains an optionalreason`. 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)