LuciferYang commented on code in PR #58518:
URL: https://github.com/apache/spark/pull/58518#discussion_r3950141150
##########
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:
The one consequence in the description that hangs a user thread is
`tryComplete` being skipped so `Observation.get` never returns
(Observation.scala:135). Both new tests only assert `thrown eq bodyFailure` and
the return value, so nothing pins down whether the observation was completed.
`withStoppedDagScheduler` nulls `dagScheduler` alone; `SparkEnv` and the
listener bus stay alive. Add any read between :316 and :322 that is only null
under a real teardown (`SparkEnv.get` is exactly that) and these tests stay
green while the observation hangs again.
Adding a `df.observe(...)` to the second test and asserting that
`observation.get` returns would cover it. Assert only that it returns, not the
value, since no job runs.
##########
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:
`dagScheduler` is a `@volatile var` (SparkContext.scala:229), and it is now
read once at :305 and again at :316, each with its own comment saying the same
thing. Both reads are null-checked so there is no NPE, but the two can see
different values.
Taking one `val dagScheduler = sc.dagScheduler` before :301 and sharing it
would collapse both the reads and the comments into one. Worth noting there
that the :305 block only runs with `SPARK_TESTING` or `-Dspark.testing` set, so
the next reader does not have to go look up `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:
This comment says that without the guard the NPE comes from the DAGScheduler
cleanup in the `finally`, but `Utils.isTesting` is true under test, so the read
at :305 (`activeQueryToJobs`) throws first and :316 is never reached.
Spelling that order out in the comment would be enough, e.g. that :305 NPEs
first and :316 would do the same.
##########
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:
`sc.shuffleDriverComponents.removeShuffle` at :279 runs before both guards
and NPEs once `_env.stop()` (SparkContext.scala:2676) has nulled
`BlockManagerMaster.driverEndpoint`; `SparkEnv.get` at :281 is the same once
`SparkEnv.set(null)` (:2678) has run. So the `finally` still replaces the
query's own exception.
`spark.sql.classic.shuffleDependency.fileCleanup.enabled` defaults to
`Utils.isTesting`, so the production default never reaches those two lines, but
every test run does, and so does production with the flag on. A shuffled query
unwinding after teardown then NPEs at :279, and none of the consequences this
PR targets are avoided.
Could the cleanup block at :262-285 get the same treatment, or the whole
block a `Utils.tryLogNonFatalError` (already used elsewhere in sql/core)? That
turns a genuine cleanup failure into a log line, and an `InterruptedException`
is not `NonFatal`, so it would still escape.
--
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]
