You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by sa...@apache.org on 2026/09/02 13:03:58 UTC
(spark) branch branch-4.3 updated: [SPARK-58000][CONNECT][SS][TEST] Mitigate flakiness between `streaming query listener` and `listener event` in `ClientStreamingQuerySuite`
This is an automated email from the ASF dual-hosted git repository.
sarutak pushed a commit to branch branch-4.3
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.3 by this push:
new fb0a9e8a8627 [SPARK-58000][CONNECT][SS][TEST] Mitigate flakiness between `streaming query listener` and `listener event` in `ClientStreamingQuerySuite`
fb0a9e8a8627 is described below
commit fb0a9e8a8627f86f3976a4a81f18bde557051c68
Author: Kousuke Saruta <sa...@apache.org>
AuthorDate: Wed Sep 2 22:03:10 2026 +0900
[SPARK-58000][CONNECT][SS][TEST] Mitigate flakiness between `streaming query listener` and `listener event` in `ClientStreamingQuerySuite`
### What changes were proposed in this pull request?
This PR mitigates a flakiness in `ClientStreamQuerySuite`.
When tests in the suite `streaming query listener` and `listener events` run consecutively in this order, `listener events` occasionally fails.
The issue happens when a thread (let's say execution thread here) can accidentally remove the StreamingQueryListener added through `spark.streams.addListener()`.
The machanism of this issue is as follows.
1. In `streaming query listener`, the main thread calls [spark.streams.addListener()](https://github.com/apache/spark/blob/af0d5c3abd19ba4bc244da174d3df257d582f970/sql/connect/client/jvm/src/test/scala/org/apache/spark/sql/connect/streaming/ClientStreamingQuerySuite.scala#L507) and [an execution thread is spawned](https://github.com/apache/spark/blob/af0d5c3abd19ba4bc244da174d3df257d582f970/sql/connect/common/src/main/scala/org/apache/spark/sql/connect/StreamingQueryListenerBus.scala#L59).
2. In `streaming query listener`, the main thread calls [spark.streams.removeListener()](https://github.com/apache/spark/blob/af0d5c3abd19ba4bc244da174d3df257d582f970/sql/connect/client/jvm/src/test/scala/org/apache/spark/sql/connect/streaming/ClientStreamingQuerySuite.scala#L556) and [the execution thread is interrupted](https://github.com/apache/spark/blob/af0d5c3abd19ba4bc244da174d3df257d582f970/sql/connect/common/src/main/scala/org/apache/spark/sql/connect/StreamingQueryListenerBu [...]
3. If the execution thread is interrupted while calling [iter.hasNext()](https://github.com/apache/spark/blob/af0d5c3abd19ba4bc244da174d3df257d582f970/sql/connect/common/src/main/scala/org/apache/spark/sql/connect/StreamingQueryListenerBus.scala#L106), [buffer.take()](https://github.com/grpc/grpc-java/blob/d0db12973952b6269a7fa36e2222ed9cd69177b6/stub/src/main/java/io/grpc/stub/ClientCalls.java#L707) in `ClientCalls.waitForNext()` in the gRPC library throws `InterruptedException`. At [...]
4. The execution thread [catches the StatusRuntimeException](https://github.com/apache/spark/blob/af0d5c3abd19ba4bc244da174d3df257d582f970/sql/connect/common/src/main/scala/org/apache/spark/sql/connect/StreamingQueryListenerBus.scala#L123).
5. The main thread finishes `streaming query listener`, starts `listener events` and [adds a listener](https://github.com/apache/spark/blob/af0d5c3abd19ba4bc244da174d3df257d582f970/sql/connect/client/jvm/src/test/scala/org/apache/spark/sql/connect/streaming/ClientStreamingQuerySuite.scala#L562) before the execution thread enters [the synchronized block](https://github.com/apache/spark/blob/af0d5c3abd19ba4bc244da174d3df257d582f970/sql/connect/common/src/main/scala/org/apache/spark/sql/ [...]
6. The execution thread enters the synchronized block and calls [remove()](https://github.com/apache/spark/blob/af0d5c3abd19ba4bc244da174d3df257d582f970/sql/connect/common/src/main/scala/org/apache/spark/sql/connect/StreamingQueryListenerBus.scala#L130), and then calls [sparkSession.execute()](https://github.com/apache/spark/blob/af0d5c3abd19ba4bc244da174d3df257d582f970/sql/connect/common/src/main/scala/org/apache/spark/sql/connect/StreamingQueryListenerBus.scala#L69).
7. Through `sparkSession.execute()`, the execution thread [send a grpc request](https://github.com/apache/spark/blob/af0d5c3abd19ba4bc244da174d3df257d582f970/sql/connect/common/src/main/scala/org/apache/spark/sql/connect/client/ExecutePlanResponseReattachableIterator.scala#L123) which is for removing a server side listener.
8. The execution thread reaches `ClientCalls.waitForNext()`. Like step 3, `buffer.take()` throws `InterruptedException` and a `StatusRuntimeException` is `added to the buffer`. On the other hand, the grpc response corresponding to the request sent in step 7 will also be `added to the buffer` asynchronously. So if the response is added to the buffer first, the next call to `buffer.take()` will succeed, and finally returns from `sparkSession.execute()` with no exception thrown from it.
9. The execution thread [removes a listener](https://github.com/apache/spark/blob/af0d5c3abd19ba4bc244da174d3df257d582f970/sql/connect/common/src/main/scala/org/apache/spark/sql/connect/StreamingQueryListenerBus.scala#L80).
10. The execution thread repeats steps 6 through 9 for all listeners including the one added in step 5.
11. In `listener event`, the main thread evaluates [assert(listener.start.length == 1)](https://github.com/apache/spark/blob/af0d5c3abd19ba4bc244da174d3df257d582f970/sql/connect/client/jvm/src/test/scala/org/apache/spark/sql/connect/streaming/ClientStreamingQuerySuite.scala#L575) but this assertion never succeeds because the listener has already removed. This continues for [30 seconds](https://github.com/apache/spark/blob/af0d5c3abd19ba4bc244da174d3df257d582f970/sql/connect/client/jvm [...]
12. The main thread reaches [the finally block](https://github.com/apache/spark/blob/af0d5c3abd19ba4bc244da174d3df257d582f970/sql/connect/client/jvm/src/test/scala/org/apache/spark/sql/connect/streaming/ClientStreamingQuerySuite.scala#L578) and evaluates [assert(listener.terminate.nonEmpty)](https://github.com/apache/spark/blob/af0d5c3abd19ba4bc244da174d3df257d582f970/sql/connect/client/jvm/src/test/scala/org/apache/spark/sql/connect/streaming/ClientStreamingQuerySuite.scala#L582) but [...]
This issue can be reproduced by modifying the code as follows to control the timing between threads.
```
--- a/sql/connect/client/jvm/src/test/scala/org/apache/spark/sql/connect/streaming/ClientStreamingQuerySuite.scala
+++ b/sql/connect/client/jvm/src/test/scala/org/apache/spark/sql/connect/streaming/ClientStreamingQuerySuite.scala
-495,7 +495,8 class ClientStreamingQuerySuite extends QueryTest with RemoteSparkSession with L
}
test("streaming query listener") {
- testStreamingQueryListener(new EventCollectorV1, "_v1")
+ // Disable the test for v1 to focus on the effect of the test for v2 on "listener event".
+ // testStreamingQueryListener(new EventCollectorV1, "_v1")
testStreamingQueryListener(new EventCollectorV2, "_v2")
}
--- a/sql/connect/common/src/main/scala/org/apache/spark/sql/connect/StreamingQueryListenerBus.scala
+++ b/sql/connect/common/src/main/scala/org/apache/spark/sql/connect/StreamingQueryListenerBus.scala
-114,6 +114,15 class StreamingQueryListenerBus(sparkSession: SparkSession) extends Logging {
postToAll(QueryIdleEvent.fromJson(event.getEventJson))
case StreamingQueryEventType.QUERY_TERMINATED_EVENT =>
postToAll(QueryTerminatedEvent.fromJson(event.getEventJson))
+
+ // Wait for being interrupted and restore the interrupted status to emulate the case
+ // this thread is interrupted during iter.hasNext.
+ try {
+ Thread.sleep(1000)
+ } catch {
+ case e: InterruptedException =>
+ Thread.currentThread.interrupt
+ }
case _ =>
logWarning(log"Unknown StreamingQueryListener event: ${MDC(LogKeys.EVENT, event)}")
}
-125,6 +134,13 class StreamingQueryListenerBus(sparkSession: SparkSession) extends Logging {
"StreamingQueryListenerBus Handler thread received exception, all client" +
" side listeners are removed and handler thread is terminated.",
e)
+
+ // Clear the interrupted status set within iter.hasNext.
+ Thread.interrupted
+ // Sleep here so that another thread can append a StreamingQueryListener first.
+ Thread.sleep(100)
+ // Restore the interrupted status set within iter.hasNext.
+ Thread.currentThread.interrupt
lock.synchronized {
executionThread = Option.empty
listeners.forEach(remove(_))
--- a/sql/connect/common/src/main/scala/org/apache/spark/sql/connect/client/ExecutePlanResponseReattachableIterator.scala
+++ b/sql/connect/common/src/main/scala/org/apache/spark/sql/connect/client/ExecutePlanResponseReattachableIterator.scala
-115,6 +115,8 class ExecutePlanResponseReattachableIterator(
// Visible for testing.
private[connect] var iter: Option[java.util.Iterator[proto.ExecutePlanResponse]] =
Some(stubWithDeadline(reattachableExecutePlanDeadline).executePlan(initialRequest))
+ // Expect initialRequest is sent to the server asynchronously during the sleep below.
+ Thread.sleep(100)
```
```
$ build/sbt -Phive -Phive-thriftserver 'testOnly org.apache.spark.sql.connect.streaming.ClientStreamingQuerySuite -- -z "streaming query listener" -z "listener events"'
-------------------------------------------
Batch: 0
-------------------------------------------
+---------+-----+
|timestamp|value|
+---------+-----+
+---------+-----+
-------------------------------------------
Batch: 1
-------------------------------------------
+--------------------+-----+
| timestamp|value|
+--------------------+-----+
|2026-08-29 10:19:...| 0|
+--------------------+-----+
-------------------------------------------
Batch: 2
-------------------------------------------
+--------------------+-----+
| timestamp|value|
+--------------------+-----+
|2026-08-29 10:19:...| 1|
+--------------------+-----+
...
-------------------------------------------
Batch: 31
-------------------------------------------
+--------------------+-----+
| timestamp|value|
+--------------------+-----+
|2026-08-29 10:19:...| 30|
+--------------------+-----+
[info] - listener events *** FAILED *** (1 minute, 32 seconds)
[info] The code passed to eventually never returned normally. Attempted 59 times over 1.0013067229166668 minutes. Last failure message: List() was empty. (ClientStreamingQuerySuite.scala:581)
[info] org.scalatest.exceptions.TestFailedDueToTimeoutException:
[info] at org.scalatest.enablers.Retrying$$anon$4.tryTryAgain$2(Retrying.scala:219)
[info] at org.scalatest.enablers.Retrying$$anon$4.retry(Retrying.scala:226)
[info] at org.scalatest.concurrent.Eventually.eventually(Eventually.scala:313)
[info] at org.scalatest.concurrent.Eventually.eventually$(Eventually.scala:312)
[info] at org.scalatest.concurrent.Eventually$.eventually(Eventually.scala:457)
[info] at org.apache.spark.sql.connect.streaming.ClientStreamingQuerySuite.$anonfun$new$40(ClientStreamingQuerySuite.scala:581)
...
```
To mitigate this issue, this PR inserts `Thread.sleep()` after `spark.streams.removeListener()` to expect the execution thread finishes before `listener events` starts.
### Why are the changes needed?
For test stability.
### Does this PR introduce _any_ user-facing change?
No.
### How was this patch tested?
Confirmed tests with the following command passed even if `Thread.sleep()`s are inserted like above.
```
$ build/sbt -Phive -Phive-thriftserver 'testOnly org.apache.spark.sql.connect.streaming.ClientStreamingQuerySuite -- -z "streaming query listener" -z "listener events"'
```
### Was this patch authored or co-authored using generative AI tooling?
No.
Closes #58425 from sarutak/fix-flaky-test-in-ClientStreamingQuerySuite-2.
Authored-by: Kousuke Saruta <sa...@apache.org>
Signed-off-by: Kousuke Saruta <sa...@apache.org>
(cherry picked from commit c09f14af975e91b4de51ce42182da1fa92a3e063)
Signed-off-by: Kousuke Saruta <sa...@apache.org>
---
.../apache/spark/sql/connect/streaming/ClientStreamingQuerySuite.scala | 2 ++
1 file changed, 2 insertions(+)
diff --git a/sql/connect/client/jvm/src/test/scala/org/apache/spark/sql/connect/streaming/ClientStreamingQuerySuite.scala b/sql/connect/client/jvm/src/test/scala/org/apache/spark/sql/connect/streaming/ClientStreamingQuerySuite.scala
index d9d2498393c6..b4cafbc15bd4 100644
--- a/sql/connect/client/jvm/src/test/scala/org/apache/spark/sql/connect/streaming/ClientStreamingQuerySuite.scala
+++ b/sql/connect/client/jvm/src/test/scala/org/apache/spark/sql/connect/streaming/ClientStreamingQuerySuite.scala
@@ -555,6 +555,8 @@ class ClientStreamingQuerySuite extends QueryTest with RemoteSparkSession with L
// Remove the listener, length should be 1.
spark.streams.removeListener(listener)
assert(spark.streams.listListeners().length == 0)
+ // Wait for the executionThread to complete. See SPARK-58000.
+ Thread.sleep(5000)
}
test("listener events") {
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org