This is an automated email from the ASF dual-hosted git repository.
zhouyuan pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/gluten.git
The following commit(s) were added to refs/heads/main by this push:
new d1aa092ebf [GLUTEN-13222][VL] Fix race in WithQueryPlanListener
causing flaky NPE (#13250)
d1aa092ebf is described below
commit d1aa092ebfa4a7115077a58be1cb94e3abed58be
Author: Shivadarshan Devadiga <[email protected]>
AuthorDate: Wed Oct 14 00:12:32 2026 +0530
[GLUTEN-13222][VL] Fix race in WithQueryPlanListener causing flaky NPE
(#13250)
---
.../apache/spark/sql/WithQueryPlanListener.scala | 34 ++++++++++++++++++----
1 file changed, 28 insertions(+), 6 deletions(-)
diff --git
a/gluten-substrait/src/test/scala/org/apache/spark/sql/WithQueryPlanListener.scala
b/gluten-substrait/src/test/scala/org/apache/spark/sql/WithQueryPlanListener.scala
index 4fa381b532..412439b9a6 100644
---
a/gluten-substrait/src/test/scala/org/apache/spark/sql/WithQueryPlanListener.scala
+++
b/gluten-substrait/src/test/scala/org/apache/spark/sql/WithQueryPlanListener.scala
@@ -28,6 +28,8 @@ import org.scalactic.source.Position
import org.scalatest.Tag
import org.scalatest.funsuite.AnyFunSuiteLike
+import java.util.concurrent.TimeUnit
+
import scala.collection.mutable
import scala.reflect.ClassTag
// format: off
@@ -53,6 +55,8 @@ trait WithQueryPlanListener extends SharedSparkSession with
AnyFunSuiteLike {
pos: Position): Unit = {
super.test(testName, testTags: _*) {
testFun
+ // Query execution listeners are notified asynchronously; wait for all
the plans.
+ waitForListenerBus()
listeners().foreach(_.invokeAll())
}(pos)
}
@@ -63,10 +67,19 @@ trait WithQueryPlanListener extends SharedSparkSession with
AnyFunSuiteLike {
}
override def afterEach(): Unit = {
- listeners().foreach(listener =>
spark.sessionState.listenerManager.unregister(listener))
- super.afterEach()
+ // Let any in-flight notifications finish before unregistering, e.g. when
the test failed.
+ try {
+ waitForListenerBus()
+ } finally {
+ listeners().foreach(listener =>
spark.sessionState.listenerManager.unregister(listener))
+ super.afterEach()
+ }
}
+ // The no-arg waitUntilEmpty() gives up after 10s, which a slow CI runner
can exceed.
+ private def waitForListenerBus(): Unit =
+
spark.sparkContext.listenerBus.waitUntilEmpty(TimeUnit.SECONDS.toMillis(30))
+
private def listeners() = {
spark.sessionState.listenerManager
.listListeners()
@@ -83,20 +96,29 @@ object WithQueryPlanListener {
with AdaptiveSparkPlanHelper {
private val plans = mutable.ArrayBuffer[SparkPlan]()
+ // Called on the listener bus thread, while invokeAll() runs on the test
thread.
override def onSuccess(funcName: String, qe: QueryExecution, durationNs:
Long): Unit = {
val plan = qe.executedPlan
- plans += plan
+ plans.synchronized {
+ plans += plan
+ }
}
override def onFailure(funcName: String, qe: QueryExecution, exception:
Exception): Unit = {}
def invokeAll(): Unit = {
- while (plans.nonEmpty) {
- val plansCopy = Seq[SparkPlan](plans.toSeq: _*)
- plans.clear()
+ var plansCopy = takePlans()
+ while (plansCopy.nonEmpty) {
plansCopy.foreach(plan => listeners.foreach(listener =>
listener.apply(plan)))
+ plansCopy = takePlans()
}
}
+
+ private def takePlans(): Seq[SparkPlan] = plans.synchronized {
+ val plansCopy = plans.toList
+ plans.clear()
+ plansCopy
+ }
}
class QueryPlanListeners private[WithQueryPlanListener] {
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
