LuciferYang commented on code in PR #58518:
URL: https://github.com/apache/spark/pull/58518#discussion_r3995114972
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/SQLExecution.scala:
##########
@@ -252,67 +309,69 @@ object SQLExecution extends Logging {
ex = Some(e)
throw e
} finally {
- val endTime = System.nanoTime()
- val errorMessage = ex.map {
- case e: SparkThrowable =>
- SparkThrowableHelper.getMessage(e, ErrorMessageFormat.PRETTY)
- case e =>
- Utils.exceptionString(e)
- }
- if (queryExecution.shuffleCleanupMode != DoNotCleanup
- && isExecutedPlanAvailable) {
- val shuffleIds = queryExecution.executedPlan match {
- case command: V2CommandExec =>
- command.children.flatMap(extractShuffleIds)
- case dataWritingCommand: DataWritingCommandExec =>
- extractShuffleIds(dataWritingCommand.child)
- case plan =>
- extractShuffleIds(plan)
- }
- shuffleIds.foreach { shuffleId =>
- queryExecution.shuffleCleanupMode match {
- case RemoveShuffleFiles =>
- // Same as what we do in
ContextCleaner.doCleanupShuffle, but do not
- // unregister the shuffle on MapOutputTracker, so that
stage retries would be
- // triggered.
- // Set blocking to Utils.isTesting to deflake unit tests.
- sc.shuffleDriverComponents.removeShuffle(shuffleId,
Utils.isTesting)
- case SkipMigration =>
-
SparkEnv.get.blockManager.migratableResolver.addShuffleToSkip(shuffleId)
- case _ => // this should not happen
+ // `SparkContext.stop()` nulls `dagScheduler` before it stops
the listener bus, so the
+ // end event may still be posted after the scheduler is
unavailable. Keep observation
+ // completion in a `finally` so an error in this block never
leaves a waiter hung.
+ try {
+ val endTime = System.nanoTime()
+ val errorMessage = ex.map { e =>
+ try {
+ e match {
+ case st: SparkThrowable =>
+ SparkThrowableHelper.getMessage(st,
ErrorMessageFormat.PRETTY)
+ case _ =>
+ Utils.exceptionString(e)
+ }
+ } catch {
+ // Rendering a user throwable can itself throw (e.g. a
custom `getMessage`).
+ // Fall back to a safe value so the query's real failure
is still surfaced and
+ // the cleanup, event post and observation completion
below still run.
+ case NonFatal(t) =>
+ logWarning(log"Failed to render the error message for
execution " +
+ log"${MDC(EXECUTION_ID, executionId)}.", t)
+ e.getClass.getName
}
}
- }
- val event = SparkListenerSQLExecutionEnd(
- executionId,
- System.currentTimeMillis(),
- // Use empty string to indicate no error, as None may mean
events generated by old
- // versions of Spark.
- errorMessage.orElse(Some("")),
- Some(queryId))
- // Currently only `Dataset.withAction` and
`DataFrameWriter.runCommand` specify the
- // `name` parameter. The `ExecutionListenerManager` only watches
SQL executions with
- // name. We can specify the execution name in more places in the
future, so that
- // `QueryExecutionListener` can track more cases.
- event.executionName = name
- event.duration = endTime - startTime
- event.qe = queryExecution
- event.executionFailure = ex
- if (Utils.isTesting) {
- import scala.jdk.CollectionConverters._
- event.jobIds =
Option(sc.dagScheduler.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)
+ if (queryExecution.shuffleCleanupMode != DoNotCleanup &&
isExecutedPlanAvailable) {
+ cleanupShuffleDependencies(queryExecution, executionId)
+ }
+ val event = SparkListenerSQLExecutionEnd(
+ executionId,
+ System.currentTimeMillis(),
+ // Use empty string to indicate no error, as None may mean
events generated by old
+ // versions of Spark.
+ errorMessage.orElse(Some("")),
+ Some(queryId))
+ // Currently only `Dataset.withAction` and
`DataFrameWriter.runCommand` specify the
+ // `name` parameter. The `ExecutionListenerManager` only
watches SQL executions with
+ // name. We can specify the execution name in more places in
the future, so that
+ // `QueryExecutionListener` can track more cases.
+ event.executionName = name
+ event.duration = endTime - startTime
+ event.qe = queryExecution
+ event.executionFailure = ex
+ // Snapshot the `@volatile` `dagScheduler` once and share it
across both reads
+ // below; it is null once `SparkContext.stop()` has run.
+ val dagSchedulerOpt = Option(sc.dagScheduler)
+ if (Utils.isTesting) {
+ import scala.jdk.CollectionConverters._
+ // Only runs under `Utils.isTesting`; hits the same teardown
race as the job
+ // cleanup below.
+ event.jobIds = dagSchedulerOpt
+ .flatMap(ds =>
Option(ds.activeQueryToJobs.get(executionId)))
+ .map(_.asScala.map(_.jobId).toSet)
+ .getOrElse(Set.empty)
+ }
- sc.listenerBus.post(event)
+ // Clean up jobs tracked by DAGScheduler for this query
execution.
+ dagSchedulerOpt.foreach(_.cleanupQueryJobs(executionId))
- // Observation.tryComplete is called here to ensure the
observation is completed,
- // but it is not high priority, so it is fine to call it later.
- sparkSession.observationManager.tryComplete(queryExecution)
+ sc.listenerBus.post(event)
+ } finally {
+ // Complete the observation whatever the block above threw, so
an `Observation.get`
+ // waiter is never left hung. `promise.tryComplete` is
idempotent.
+ sparkSession.observationManager.tryComplete(queryExecution)
Review Comment:
This observation-completing `tryComplete` now sits in the inner `finally`,
but the call itself has no try/catch, and `ObservationManager.tryComplete` is
not wholesale exception-safe: only `qe.observedMetrics` is `Try`-wrapped, while
the `foreachWithSubqueriesAndPruning` plan walk and `setMetricsAndNotify` are
not. From here a NonFatal throw would mask both the body's `ex` and any
exception the inner `try` raised, which is exactly the masking this PR removes
elsewhere.
It can't throw on the teardown path (metric collection is Try-wrapped,
promise completion is idempotent, and it only walks an already-built plan), so
this is latent rather than a live bug. But since the error-message rendering
above is wrapped in a try/catch fallback for this very reason, wrapping this
call in `catch NonFatal` (or having `ObservationManager.tryComplete` swallow
NonFatal internally) would line the two up.
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/SQLExecutionSuite.scala:
##########
@@ -423,6 +427,261 @@ 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
+ }
+ }
+
+ /**
+ * Runs `f` with `spark`'s `BlockManagerMaster.driverEndpoint` nulled out,
standing in for a
+ * `SparkContext` whose `SparkEnv` has been stopped. `SparkContext.stop()`
stops `SparkEnv` (which
+ * nulls that endpoint) after nulling `dagScheduler`, so the shuffle cleanup
in
+ * `withNewExecutionId`'s `finally` -- which runs before the `dagScheduler`
cleanup -- really can
+ * hit a stopped `BlockManagerMaster` and NPE while a query unwinds during
teardown.
+ */
+ private def withStoppedBlockManagerMaster[T](spark: SparkSession)(f: => T):
T = {
+ val master = spark.sparkContext.env.blockManager.master
+ val savedEndpoint = master.driverEndpoint
+ master.driverEndpoint = null
+ try {
+ f
+ } finally {
+ master.driverEndpoint = savedEndpoint
+ }
+ }
+
+ private def withUnavailableSparkEnv[T](f: => T): T = {
+ val savedEnv = SparkEnv.get
+ SparkEnv.set(null)
Review Comment:
`SparkEnv.set(null)` here simulates teardown, but `SparkEnv` is a
process-wide singleton, so nulling it affects every thread in the JVM, not just
this one. During the window the end event posted from `withNewExecutionId0` is
handled asynchronously by the listener bus, and that thread can observe
`SparkEnv.get == null`. A listener exception is swallowed per-listener by the
bus and won't fail the test, and the window is short (restored in the
`finally`), so the worst case is occasional error-log noise, not a failure.
Not blocking. If you want it tighter, null it only around the narrow call
that needs it, or assert no extra error logs appear during the window.
--
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]
