[SPARK-54664][CONNECT] Clean up the code related to listenerCache from connect.StreamingQueryManager

### What changes were proposed in this pull request?
#41752 introduced a `listenerCache` and related private methods (`cacheListenerById`, `getIdByListener`, and `removeCachedListener`) for `connect.StreamingQueryManager`. However, in #46287, the usage related to `listenerCache` was replaced by `streamingQueryListenerBus`. As a result, `listenerCache` and its associated private methods are no longer in use, and this current pr cleans them up.

### Why are the changes needed?
Code cleanup.

### Does this PR introduce _any_ user-facing change?
No.

### How was this patch tested?
Pass Github Actions

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

Closes #53420 from LuciferYang/StreamingQueryManager.

Lead-authored-by: yangjie01 <yangjie01@baidu.com>
Co-authored-by: YangJie <yangjie01@baidu.com>
Signed-off-by: Dongjoon Hyun <dongjoon@apache.org>

Author: LuciferYang

sql/connect/common/src/main/scala/org/apache/spark/sql/connect/StreamingQueryManager.scala

@@ -18,14 +18,12 @@
 package org.apache.spark.sql.connect
 
 import java.util.UUID
-import java.util.concurrent.{ConcurrentHashMap, ConcurrentMap}
 
 import scala.jdk.CollectionConverters._
 
 import org.apache.spark.annotation.Evolving
 import org.apache.spark.connect.proto.{Command, StreamingQueryManagerCommand, StreamingQueryManagerCommandResult}
 import org.apache.spark.internal.Logging
-import org.apache.spark.sql.connect.common.InvalidPlanInput
 import org.apache.spark.sql.streaming
 import org.apache.spark.sql.streaming.{StreamingQueryException, StreamingQueryListener}
 
@@ -39,15 +37,6 @@ class StreamingQueryManager private[sql] (sparkSession: SparkSession)
     extends streaming.StreamingQueryManager
     with Logging {
 
-  // Mapping from id to StreamingQueryListener. There's another mapping from id to
-  // StreamingQueryListener on server side. This is used by removeListener() to find the id
-  // of previously added StreamingQueryListener and pass it to server side to find the
-  // corresponding listener on server side. We use id to StreamingQueryListener mapping
-  // here to make sure there's no hash collision as well as handling the case that adds and
-  // removes the same listener instance multiple times properly.
-  private lazy val listenerCache: ConcurrentMap[String, StreamingQueryListener] =
-    new ConcurrentHashMap()
-
   private[spark] val streamingQueryListenerBus = new StreamingQueryListenerBus(sparkSession)
 
   private[spark] def close(): Unit = {
@@ -128,17 +117,4 @@ class StreamingQueryManager private[sql] (sparkSession: SparkSession)
 
     resp.getStreamingQueryManagerCommandResult
   }
-
-  private def cacheListenerById(id: String, listener: StreamingQueryListener): Unit = {
-    listenerCache.putIfAbsent(id, listener)
-  }
-
-  private def getIdByListener(listener: StreamingQueryListener): String = {
-    listenerCache.forEach((k, v) => if (listener.equals(v)) return k)
-    throw InvalidPlanInput(s"No id with listener $listener is found.")
-  }
-
-  private def removeCachedListener(id: String): StreamingQueryListener = {
-    listenerCache.remove(id)
-  }
 }