[SPARK-59623][SS][PYTHON] Fix streaming query listener removal deadlock

[zifeif2]

What changes were proposed in this pull request?

This PR fixes a deadlock when removing the last Python Spark Connect streaming query listener.

It adds a listener lifecycle lock to serialize append and last-listener remove operations, releases the listener-bus data lock before waiting for the event thread to terminate, and reacquires the data lock for final cleanup. It also adds a deterministic regression test covering a pending event and a concurrent listener append during shutdown.

Why are the changes needed?

Previously, StreamingQueryListenerBus.remove called self._execution_thread.join() while holding self._lock. If the event thread had already received an event and was entering post_to_all, it needed the same lock to dispatch that event before it could terminate. The remover waited for the event thread, while the event thread waited for the remover's lock, causing a deadlock.

The lifecycle lock preserves atomic listener startup/shutdown transitions without preventing the event thread from acquiring the data lock and draining pending events.

Does this PR introduce any user-facing change?

Yes. Removing the final Python Spark Connect streaming query listener no longer hangs when an event is pending. There is no API change.

How was this patch tested?

Added StreamingQueryListenerBusTests.test_remove_last_listener_with_pending_event, which deterministically verifies that:

• a pending event can be dispatched while last-listener removal waits for the event thread;
• a concurrent listener append waits until shutdown completes;
• the new listener starts a fresh event thread after shutdown.

The targeted unit test passed. Ruff lint, Ruff formatting validation, and git diff --check also passed.

Was this patch authored or co-authored using generative AI tooling?

Generated-by: OpenAI Codex (GPT-5)

[uros-b]

Thank you @zifeif2 for working on this, please see a couple of drive-by comments below:

StreamingQueryListenerBus.remove still holds _listeners_state_lock across execute_command.

The join-while-holding deadlock is real, and splitting _lifecycle_lock from the data lock is the right shape: join() now happens after releasing _listeners_state_lock, and lifecycle serializes append vs last-remove.

Last-listener remove still holds the data lock for the whole execute_command RTT. Lifecycle already serializes that RPC, so the data lock is not needed there. Holding it blocks post_to_all for the duration. If the server-side remove ever waits for the client to consume remaining events (or the iterator queue fills), this is the same deadlock with a longer fuse.

The unit test hides it: the mock execute_command is released on event_dispatch_started, which is set before post_to_all.

Please consider dropping _listeners_state_lock before execute_command, not only before join:

with self._lifecycle_lock:
    with self._listeners_state_lock:
        if listener not in self._listener_bus:
            return
        is_last_listener = len(self._listener_bus) == 1
        execution_thread = self._execution_thread if is_last_listener else None
        if not is_last_listener:
            self._listener_bus.remove(listener)
            return
    if is_last_listener:
        try:
            self._sqm._session.client.execute_command(exec_cmd)
        except Exception as e:
            warnings.warn(...)
            return
        if execution_thread is not None:
            execution_thread.join()
        with self._listeners_state_lock:
            ...

[zifeif2]

Thanks @uros-b for the feedback. I've addressed your comment. PTAL.

[uros-b]

Thank you @zifeif2, let's loop in @zhengruifeng for further review here