gaogaotiantian opened a new pull request, #59120:
URL: https://github.com/apache/spark/pull/59120
### 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
--
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]
