[ 
https://issues.apache.org/jira/browse/SPARK-60131?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
 ]

Vitthal updated SPARK-60131:
----------------------------
    Affects Version/s: 4.0.0
          Description: 
In the Scala Spark Connect client, removing a StreamingQueryListener that was 
never registered can unregister the shared server-side listener bus when 
exactly one other listener is registered.

Reproduction sequence:
1. Add listener A with spark.streams.addListener(A).
2. Call spark.streams.removeListener(B), where B is a different listener that 
was never added.
3. StreamingQueryListenerBus.remove sees one registration and sends 
remove_listener_bus_listener=true before checking membership. It also 
interrupts the event worker. Removing B from the local list does not remove A.

Expected: removing B is a no-op and A keeps its registration and event handling.
Actual: the client sends the command that unregisters the shared server-side 
bus and interrupts the worker. Later local state depends on whether the worker 
completes normally or enters its exception cleanup path.

A deterministic regression against the production Scala remove method at 
upstream/master commit 215a1460d9d4acdac6f72b42c28ab60b9ff5b7fe fails because 
an unregister command is recorded for B. The registered-listener removal case 
passes. The fixture substitutes remote command execution and worker input; this 
is not a live-server reproduction.

Python Connect already checks listener membership before unregistering the bus. 
SPARK-59847 and SPARK-59623 concern Python worker-join deadlocks and do not 
change the Scala membership check.

Proposed fix: require listeners.contains(listener) in the existing 
single-listener unregister condition. No API, protocol, duplicate-registration, 
or threading-policy change.


Affected-version evidence: the same unguarded removal path is present in the 
official v4.0.0 source tag. The executable regression was run on the current 
5.0.0-SNAPSHOT master checkout. No released-version runtime test was run. 
Priority is Minor because callers can avoid the trigger by removing only 
listeners they registered.

  was:
In the Scala Spark Connect client, removing a StreamingQueryListener that was 
never registered can unregister the shared server-side listener bus when 
exactly one other listener is registered.

Reproduction sequence:
1. Add listener A with spark.streams.addListener(A).
2. Call spark.streams.removeListener(B), where B is a different listener that 
was never added.
3. StreamingQueryListenerBus.remove sees one registration and sends 
remove_listener_bus_listener=true before checking membership. It also 
interrupts the event worker. Removing B from the local list does not remove A.

Expected: removing B is a no-op and A keeps its registration and event handling.
Actual: the client sends the command that unregisters the shared server-side 
bus and interrupts the worker. Later local state depends on whether the worker 
completes normally or enters its exception cleanup path.

A deterministic regression against the production Scala remove method at 
upstream/master commit 215a1460d9d4acdac6f72b42c28ab60b9ff5b7fe fails because 
an unregister command is recorded for B. The registered-listener removal case 
passes. The fixture substitutes remote command execution and worker input; this 
is not a live-server reproduction.

Python Connect already checks listener membership before unregistering the bus. 
SPARK-59847 and SPARK-59623 concern Python worker-join deadlocks and do not 
change the Scala membership check.

Proposed fix: require listeners.contains(listener) in the existing 
single-listener unregister condition. No API, protocol, duplicate-registration, 
or threading-policy change.


             Priority: Minor  (was: Major)

> Removing an unregistered Scala Connect streaming listener stops the shared bus
> ------------------------------------------------------------------------------
>
>                 Key: SPARK-60131
>                 URL: https://issues.apache.org/jira/browse/SPARK-60131
>             Project: Spark
>          Issue Type: Bug
>          Components: Connect
>    Affects Versions: 4.0.0, 5.0.0
>            Reporter: Vitthal
>            Priority: Minor
>
> In the Scala Spark Connect client, removing a StreamingQueryListener that was 
> never registered can unregister the shared server-side listener bus when 
> exactly one other listener is registered.
> Reproduction sequence:
> 1. Add listener A with spark.streams.addListener(A).
> 2. Call spark.streams.removeListener(B), where B is a different listener that 
> was never added.
> 3. StreamingQueryListenerBus.remove sees one registration and sends 
> remove_listener_bus_listener=true before checking membership. It also 
> interrupts the event worker. Removing B from the local list does not remove A.
> Expected: removing B is a no-op and A keeps its registration and event 
> handling.
> Actual: the client sends the command that unregisters the shared server-side 
> bus and interrupts the worker. Later local state depends on whether the 
> worker completes normally or enters its exception cleanup path.
> A deterministic regression against the production Scala remove method at 
> upstream/master commit 215a1460d9d4acdac6f72b42c28ab60b9ff5b7fe fails because 
> an unregister command is recorded for B. The registered-listener removal case 
> passes. The fixture substitutes remote command execution and worker input; 
> this is not a live-server reproduction.
> Python Connect already checks listener membership before unregistering the 
> bus. SPARK-59847 and SPARK-59623 concern Python worker-join deadlocks and do 
> not change the Scala membership check.
> Proposed fix: require listeners.contains(listener) in the existing 
> single-listener unregister condition. No API, protocol, 
> duplicate-registration, or threading-policy change.
> Affected-version evidence: the same unguarded removal path is present in the 
> official v4.0.0 source tag. The executable regression was run on the current 
> 5.0.0-SNAPSHOT master checkout. No released-version runtime test was run. 
> Priority is Minor because callers can avoid the trigger by removing only 
> listeners they registered.



--
This message was sent by Atlassian Jira
(v8.20.10#820010)

---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]

Reply via email to