This is an automated email from the ASF dual-hosted git repository.
gaogaotiantian pushed a commit to branch branch-4.x
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.x by this push:
new 5bb865ad31b6 [SPARK-59847][CONNECT][SS][PYTHON] Fix deadlock in
StreamingQueryListenerBus when removing the last listener
5bb865ad31b6 is described below
commit 5bb865ad31b6390afc3125cc373a7e7e185996de
Author: Tian Gao <[email protected]>
AuthorDate: Tue Sep 29 14:26:11 2026 -0700
[SPARK-59847][CONNECT][SS][PYTHON] Fix deadlock in
StreamingQueryListenerBus when removing the last listener
### What changes were proposed in this pull request?
This PR fixes a deadlock in the PySpark Spark Connect client's
`StreamingQueryListenerBus`
(`python/pyspark/sql/connect/streaming/query.py`) that can occur when the
last streaming query
listener is removed while listener events are still being delivered.
The bus previously used a single lock (`_lock`) for two distinct
responsibilities:
- serializing listener lifecycle operations (`append` / `remove`), and
- guarding `_listener_bus` during event delivery in `post_to_all()`.
`remove()` held that lock across `self._execution_thread.join()`. When an
event was in flight,
the event-handling thread was blocked in `post_to_all()` trying to acquire
the same lock, so it
could never finish draining — while `remove()` waited forever for it via
`join()`. That is a
classic lock-ordering deadlock: the thread holding the lock waits for a
thread that needs the
lock to make progress.
This PR splits the single lock into two:
- `_thread_lock` — serializes `append` / `remove` end-to-end, held across
the `join()`. This
guarantees at most one event-handling thread exists at a time, so no
second handler thread can
be started while `remove()` is joining.
- `_bus_lock` — guards `_listener_bus` and is acquired by the handler
thread in `post_to_all()`.
It is intentionally released before the `join()`, so the handler thread
can keep draining and
delivering remaining events and then exit, allowing the join to complete.
Lock ordering is consistent (`_thread_lock` outer, `_bus_lock` inner) and
the handler thread only
ever acquires `_bus_lock`, so there is no circular wait. `remove()` also
snapshots the thread
reference before joining and guards the final bus removal, since the
handler's error path may
clear the bus concurrently.
### Why are the changes needed?
Calling `spark.streams.removeListener(listener)` on the last remaining
listener could hang
indefinitely if a streaming query event was being delivered at the same
time. This is a real
user-facing hang, not merely a test problem; it has also been observed as
an intermittent CI
timeout in the `pyspark.sql.connect.dataframe` doctest, where the process
had to be killed after
the test timeout with a thread dump showing the main thread blocked in
`remove()` -> `join()`
and the handler thread blocked in `post_to_all()` -> `with self._lock`.
### Does this PR introduce _any_ user-facing change?
No behavior change other than fixing the hang. `removeListener` still
blocks until remaining
events are processed, but no longer deadlocks.
### How was this patch tested?
The affected path is exercised by existing Spark Connect streaming listener
tests — notably the
`pyspark.sql.connect.dataframe` doctest (which had been intermittently
timing out on this path)
and the streaming listener parity tests. The fix preserves the documented
`removeListener`
semantics (blocks until remaining events are processed) while removing the
deadlock, so those
existing tests continue to cover the behavior. Relying on CI to validate
across the test matrix.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code
Closes #59120 from gaogaotiantian/fix-connect-streaming-listener-deadlock.
Authored-by: Tian Gao <[email protected]>
Signed-off-by: Tian Gao <[email protected]>
(cherry picked from commit b2dd3e443894d46b923cc6e68d70392fdd2d3eda)
Signed-off-by: Tian Gao <[email protected]>
---
python/pyspark/sql/connect/streaming/query.py | 48 +++++++++++++++++++--------
1 file changed, 34 insertions(+), 14 deletions(-)
diff --git a/python/pyspark/sql/connect/streaming/query.py
b/python/pyspark/sql/connect/streaming/query.py
index ea1a7b07bfd6..d7c1d30ef866 100644
--- a/python/pyspark/sql/connect/streaming/query.py
+++ b/python/pyspark/sql/connect/streaming/query.py
@@ -278,7 +278,13 @@ class StreamingQueryListenerBus:
self._sqm = sqm
self._listener_bus: List[StreamingQueryListener] = []
self._execution_thread: Optional[Thread] = None
- self._lock = Lock()
+ # _thread_lock serializes listener lifecycle ops (append/remove) so at
most one
+ # event-handling thread exists at a time. It is held across the thread
join in
+ # remove(), so the handler thread must never acquire it.
+ self._thread_lock = Lock()
+ # _bus_lock guards _listener_bus and is acquired by the handler thread
in
+ # post_to_all(); it must stay free while remove() joins, to avoid
deadlock.
+ self._bus_lock = Lock()
def close(self) -> None:
for listener in self._listener_bus:
@@ -290,10 +296,12 @@ class StreamingQueryListenerBus:
the first listener, request the server to create the server side
listener
and start a thread to handle query events.
"""
- with self._lock:
- self._listener_bus.append(listener)
+ with self._thread_lock:
+ with self._bus_lock:
+ self._listener_bus.append(listener)
+ is_first_listener = len(self._listener_bus) == 1
- if len(self._listener_bus) == 1:
+ if is_first_listener:
assert self._execution_thread is None
try:
result_iter = self._register_server_side_listener()
@@ -302,7 +310,8 @@ class StreamingQueryListenerBus:
f"Failed to add the listener because of exception:
{e}\n"
f"The listener is not added, please add it again."
)
- self._listener_bus.remove(listener)
+ with self._bus_lock:
+ self._listener_bus.remove(listener)
return
self._execution_thread = Thread(
target=self._query_event_handler, args=(result_iter,)
@@ -321,11 +330,13 @@ class StreamingQueryListenerBus:
will return after processing remaining listener events. This function
blocks until
all events are processed.
"""
- with self._lock:
- if listener not in self._listener_bus:
- return
+ with self._thread_lock:
+ with self._bus_lock:
+ if listener not in self._listener_bus:
+ return
+ is_last_listener = len(self._listener_bus) == 1
- if len(self._listener_bus) == 1:
+ if is_last_listener:
cmd = pb2.StreamingQueryListenerBusCommand()
cmd.remove_listener_bus_listener = True
exec_cmd = pb2.Command()
@@ -338,11 +349,20 @@ class StreamingQueryListenerBus:
f"The listener is not removed, please remove it again."
)
return
- if self._execution_thread is not None:
- self._execution_thread.join()
+ # Join without holding _bus_lock so the handler thread can keep
+ # acquiring it in post_to_all() to drain remaining events and
exit.
+ # _thread_lock still blocks a concurrent append() from
starting a
+ # second handler thread. Snapshot the reference first, since
the
+ # handler's error path may clear it concurrently.
+ execution_thread = self._execution_thread
+ if execution_thread is not None:
+ execution_thread.join()
self._execution_thread = None
- self._listener_bus.remove(listener)
+ with self._bus_lock:
+ # The handler's error path may have cleared the bus while we
joined.
+ if listener in self._listener_bus:
+ self._listener_bus.remove(listener)
@staticmethod
def _iter_listener_events(
@@ -400,7 +420,7 @@ class StreamingQueryListenerBus:
"StreamingQueryListenerBus Handler thread received exception,
all client side "
f"listeners are removed and handler thread is terminated. The
error is: {e}"
)
- with self._lock:
+ with self._bus_lock:
self._execution_thread = None
self._listener_bus.clear()
return
@@ -431,7 +451,7 @@ class StreamingQueryListenerBus:
Post listener events to all active listeners, note that if one
listener throws,
it should not affect other listeners.
"""
- with self._lock:
+ with self._bus_lock:
for listener in self._listener_bus:
try:
if isinstance(event, QueryStartedEvent):
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
