ivoson commented on code in PR #58518:
URL: https://github.com/apache/spark/pull/58518#discussion_r3954214618
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/SQLExecution.scala:
##########
@@ -300,13 +300,20 @@ object SQLExecution extends Logging {
event.executionFailure = ex
if (Utils.isTesting) {
import scala.jdk.CollectionConverters._
- event.jobIds =
Option(sc.dagScheduler.activeQueryToJobs.get(executionId))
+ // Tolerate a stopped context here too: this runs earlier in
the same `finally`
+ // as the `cleanupQueryJobs` call below, so it hits the same
teardown race.
+ event.jobIds = Option(sc.dagScheduler)
+ .flatMap(ds => Option(ds.activeQueryToJobs.get(executionId)))
.map(_.asScala.map(_.jobId).toSet)
.getOrElse(Set.empty)
}
// Clean up jobs tracked by DAGScheduler for this query
execution.
- sc.dagScheduler.cleanupQueryJobs(executionId)
+ // `SparkContext.stop()` nulls `dagScheduler` before it stops
the listener bus, so a
+ // query unwinding here while the context tears down would NPE.
As this runs in a
+ // `finally`, that NPE would replace the query's real failure
and skip the event post
+ // and observation completion below, so tolerate an
already-stopped context.
+ Option(sc.dagScheduler).foreach(_.cleanupQueryJobs(executionId))
Review Comment:
Good catch, thanks -- confirmed. The shuffle cleanup runs earlier in the
same `finally` and hits the same race: `removeShuffle` reaches
`driverEndpoint.askSync` (no null check) once `_env.stop()` has nulled it, and
`SparkEnv.get` is null after `SparkEnv.set(null)`. With
`spark.sql.classic.shuffleDependency.fileCleanup.enabled` defaulting to
`Utils.isTesting`, that block runs on every test run, so the invariant was
still broken there.
Wrapped the block in `Utils.tryLogNonFatalError`. I chose the catch over
mirroring the `Option(sc.dagScheduler)` null-check because a null-check would
still race -- `driverEndpoint` is nulled at `SparkContext.stop()`:2676 but
`SparkEnv` only at :2678, and the actual deref is buried inside
`removeShuffle`, so a call-site check can't reliably prevent the NPE. The catch
handles it regardless of how far teardown has progressed.
On the `InterruptedException` note: it isn't `NonFatal`, so it would still
escape -- but reaching an interruptible call (`askSync`/`awaitResult`) requires
`driverEndpoint` to be alive, i.e. not the teardown-NPE case, and nothing in
this path interrupts the driver thread anyway (job cancellation fails the
`JobWaiter` with a `SparkException`; `interruptThread` targets executor task
threads). Letting a genuine interrupt propagate is the right behavior, so I
didn't add special handling for it.
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/SQLExecutionSuite.scala:
##########
@@ -423,6 +423,57 @@ class SQLExecutionSuite extends SparkFunSuite with
SQLConfHelper {
spark.stop()
}
}
+
+ /**
+ * Runs `f` with `spark`'s `dagScheduler` nulled out, standing in for a
`SparkContext` that has
+ * already been stopped. `SparkContext.stop()` nulls `_dagScheduler` before
it stops the listener
+ * bus, so a query really can unwind through `withNewExecutionId`'s
`finally` in this state.
+ */
+ private def withStoppedDagScheduler[T](spark: SparkSession)(f: => T): T = {
+ val sc = spark.sparkContext
+ val savedDagScheduler = sc.dagScheduler
+ sc.dagScheduler = null
+ try {
+ f
+ } finally {
+ sc.dagScheduler = savedDagScheduler
+ }
+ }
+
+ test("SPARK-59242: withNewExecutionId surfaces the body's failure when the
SparkContext " +
+ "has been stopped") {
+ val spark =
SparkSession.builder().master("local[*]").appName("test").getOrCreate()
+ try {
+ val qe = spark.range(1, 10).queryExecution
+ val bodyFailure = new IllegalStateException("body failed")
+ // Without the null guard, the DAGScheduler cleanup in the `finally`
throws an NPE that,
+ // because it is thrown from a `finally`, replaces `bodyFailure`
entirely -- destroying the
+ // only record of why the query actually failed.
+ val thrown = intercept[IllegalStateException] {
+ withStoppedDagScheduler(spark) {
+ SQLExecution.withNewExecutionId(qe) {
+ throw bodyFailure
+ }
+ }
+ }
+ assert(thrown eq bodyFailure)
+ } finally {
+ spark.stop()
+ }
+ }
+
+ test("SPARK-59242: withNewExecutionId completes normally when the
SparkContext has been " +
Review Comment:
Done. Added an observation to the completes-normally test and assert it is
completed. I used `Observation.future.isCompleted` rather than
`Observation.get`, so a regression fails fast instead of blocking on the future
until the suite timeout (the observation is completed synchronously in the
`finally`, so it is already set by the time `withNewExecutionId` returns).
I also added a dedicated test that reproduces the full teardown state --
`dagScheduler` nulled *and* `BlockManagerMaster` stopped -- on a query with a
real shuffle, asserting the shuffle-cleanup failure neither masks the body's
exception nor leaves the observation uncompleted. It asserts a shuffle id is
actually produced first, so it can't silently pass without reaching
`removeShuffle`.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/SQLExecution.scala:
##########
@@ -300,13 +300,20 @@ object SQLExecution extends Logging {
event.executionFailure = ex
if (Utils.isTesting) {
import scala.jdk.CollectionConverters._
- event.jobIds =
Option(sc.dagScheduler.activeQueryToJobs.get(executionId))
+ // Tolerate a stopped context here too: this runs earlier in
the same `finally`
+ // as the `cleanupQueryJobs` call below, so it hits the same
teardown race.
+ event.jobIds = Option(sc.dagScheduler)
+ .flatMap(ds => Option(ds.activeQueryToJobs.get(executionId)))
Review Comment:
Done. Snapshot `Option(sc.dagScheduler)` once as `dagSchedulerOpt` and share
it across both reads, collapsing the two comments into one. Also noted there
that the `jobIds` block only runs under `Utils.isTesting`.
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/SQLExecutionSuite.scala:
##########
@@ -423,6 +423,57 @@ class SQLExecutionSuite extends SparkFunSuite with
SQLConfHelper {
spark.stop()
}
}
+
+ /**
+ * Runs `f` with `spark`'s `dagScheduler` nulled out, standing in for a
`SparkContext` that has
+ * already been stopped. `SparkContext.stop()` nulls `_dagScheduler` before
it stops the listener
+ * bus, so a query really can unwind through `withNewExecutionId`'s
`finally` in this state.
+ */
+ private def withStoppedDagScheduler[T](spark: SparkSession)(f: => T): T = {
+ val sc = spark.sparkContext
+ val savedDagScheduler = sc.dagScheduler
+ sc.dagScheduler = null
+ try {
+ f
+ } finally {
+ sc.dagScheduler = savedDagScheduler
+ }
+ }
+
+ test("SPARK-59242: withNewExecutionId surfaces the body's failure when the
SparkContext " +
Review Comment:
Fixed the comment -- it now spells out that under `Utils.isTesting` the
`activeQueryToJobs` read is the first to NPE, and the `cleanupQueryJobs`
cleanup would do the same.
--
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]
