[SPARK-49876][CONNECT] Get rid of global locks from Spark Connect Service

[ghost]

What changes were proposed in this pull request?

Get rid of global locks from Spark Connect Service.

• ServerSideListenerHolder: AtomicReference replaces the global lock.
• SparkConnectStreamingQueryCache: two global locks are replaced with ConcurrentHashMap and a mutex-protected per-tag data structure, i.e., global locks -> a per-tag lock.

Why are the changes needed?

Spark Connect Service doesn't limit the number of threads, susceptible to priority inversion because of heavy use of global locks.

Does this PR introduce any user-facing change?

No.

How was this patch tested?

Existing tests + modified an existing test.

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

No.

[ghost]

To whom it may concern - I'm pretty sure that this condition was wrong, so fixed it.

[ghost]

@HyukjinKwon can you please check if the condition was right? as far as I understood, the key was supposed to be removed if no associated entry was found in queryCache. thanks!

[HyukjinKwon]

Hm, it has to remove the element within value instead of the whole entry IIRC.

[wvwwvwwv]

You're right, and the code actually does that, though it's not quite obvious; value.filter will return true only when value is empty, so semantically the piece of code means,

1. Remove elements that do not satisfy queryCache.containsKey(k) from value.
2. If value is empty after filtering those elements, value will be removed from taggedQueries.

[HyukjinKwon]

ah okie, then lgtm

[ghost]

@rangadi Hello, nice to meet you here! This is a minor code optimization as part of an attempt to get rid of the use of global locks in the spark connect service code. Could you review the code please? The code semantics should be the same as before. Thanks.

[HyukjinKwon]

Merged to master.