[SPARK-51445][SQL][SS][CONNECT] Change the never changed var to val

### What changes were proposed in this pull request?
This PR replaces unchanged `var` with `val`.

### Why are the changes needed?
Use `val` instead of `var` when possible.

### 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 #50219 from LuciferYang/var-to-val.

Authored-by: yangjie01 <yangjie01@baidu.com>
Signed-off-by: yangjie01 <yangjie01@baidu.com>

Author: LuciferYang

sql/connect/server/src/main/scala/org/apache/spark/sql/connect/service/SparkConnectExecutionManager.scala

@@ -74,10 +74,10 @@ private[connect] class SparkConnectExecutionManager() extends Logging {
     .build[ExecuteKey, ExecuteInfo]()
 
   /** The time when the last execution was removed. */
-  private var lastExecutionTimeNs: AtomicLong = new AtomicLong(System.nanoTime())
+  private val lastExecutionTimeNs: AtomicLong = new AtomicLong(System.nanoTime())
 
   /** Executor for the periodic maintenance */
-  private var scheduledExecutor: AtomicReference[ScheduledExecutorService] =
+  private val scheduledExecutor: AtomicReference[ScheduledExecutorService] =
     new AtomicReference[ScheduledExecutorService]()
 
   /**

sql/connect/server/src/main/scala/org/apache/spark/sql/connect/service/SparkConnectListenerBusListener.scala

@@ -78,7 +78,7 @@ private[sql] class ServerSideListenerHolder(val sessionHolder: SessionHolder) {
    * final ResultComplete response.
    */
   def cleanUp(): Unit = {
-    var listener = streamingQueryServerSideListener.getAndSet(null)
+    val listener = streamingQueryServerSideListener.getAndSet(null)
     if (listener != null) {
       sessionHolder.session.streams.removeListener(listener)
       listener.sendResultComplete()

sql/connect/server/src/main/scala/org/apache/spark/sql/connect/service/SparkConnectSessionManager.scala

@@ -49,7 +49,7 @@ class SparkConnectSessionManager extends Logging {
       .build[SessionKey, SessionHolderInfo]()
 
   /** Executor for the periodic maintenance */
-  private var scheduledExecutor: AtomicReference[ScheduledExecutorService] =
+  private val scheduledExecutor: AtomicReference[ScheduledExecutorService] =
     new AtomicReference[ScheduledExecutorService]()
 
   private def validateSessionId(

sql/connect/server/src/main/scala/org/apache/spark/sql/connect/service/SparkConnectStreamingQueryCache.scala

@@ -186,7 +186,7 @@ private[connect] class SparkConnectStreamingQueryCache(
   private[service] val taggedQueries: ConcurrentMap[String, QueryCacheKeySet] =
     new ConcurrentHashMap[String, QueryCacheKeySet]
 
-  private var scheduledExecutor: AtomicReference[ScheduledExecutorService] =
+  private val scheduledExecutor: AtomicReference[ScheduledExecutorService] =
     new AtomicReference[ScheduledExecutorService]()
 
   /** Schedules periodic checks if it is not already scheduled */
@@ -218,7 +218,7 @@ private[connect] class SparkConnectStreamingQueryCache(
       (k, v) => {
         if (v == null || !v.addKey(queryKey)) {
           // Create a new QueryCacheKeySet if the entry is absent or being removed.
-          var keys = mutable.HashSet.empty[QueryCacheKey]
+          val keys = mutable.HashSet.empty[QueryCacheKey]
           keys.add(queryKey)
           new QueryCacheKeySet(keys = keys)
         } else {

sql/core/src/main/scala/org/apache/spark/sql/execution/python/streaming/TransformWithStateInPandasStateServer.scala

@@ -114,7 +114,7 @@ class TransformWithStateInPandasStateServer(
   // A map to store the iterator id -> Iterator[(Row, Row)] mapping. This is to keep track of the
   // current key-value iterator position for each iterator id in a map state for a grouping key in
   // case user tries to fetch another state variable before the current iterator is exhausted.
-  private var keyValueIterators = if (keyValueIteratorMapForTest != null) {
+  private val keyValueIterators = if (keyValueIteratorMapForTest != null) {
     keyValueIteratorMapForTest
   } else {
     new mutable.HashMap[String, Iterator[(Row, Row)]]()

sql/core/src/test/scala/org/apache/spark/sql/VariantSuite.scala

@@ -749,7 +749,7 @@ class VariantSuite extends QueryTest with SharedSparkSession with ExpressionEval
     // The initial size of the buffer backing a cached dataframe column is 128KB.
     // See `ColumnBuilder`.
     val numKeys = 128 * 1024
-    var keyIterator = (0 until numKeys).iterator
+    val keyIterator = (0 until numKeys).iterator
     val entries = Array.fill(numKeys)(s"""\"${keyIterator.next()}\": \"test\"""")
     val jsonStr = s"{${entries.mkString(", ")}}"
     val query = s"""select parse_json('${jsonStr}') v from range(0, 10)"""
@@ -792,7 +792,7 @@ class VariantSuite extends QueryTest with SharedSparkSession with ExpressionEval
     // The initial size of the buffer backing a cached dataframe column is 128KB.
     // See `ColumnBuilder`.
     val numKeys = 128 * 1024
-    var keyIterator = (0 until numKeys).iterator
+    val keyIterator = (0 until numKeys).iterator
     val entries = Array.fill(numKeys)(s"""\"${keyIterator.next()}\": \"test\"""")
     val jsonStr = s"{${entries.mkString(", ")}}"
     val query = s"""select array(parse_json('${jsonStr}')) v from range(0, 10)"""
@@ -839,7 +839,7 @@ class VariantSuite extends QueryTest with SharedSparkSession with ExpressionEval
        // The initial size of the buffer backing a cached dataframe column is 128KB.
        // See `ColumnBuilder`.
       val numKeys = 128 * 1024
-      var keyIterator = (0 until numKeys).iterator
+      val keyIterator = (0 until numKeys).iterator
       val entries = Array.fill(numKeys)(s"""\"${keyIterator.next()}\": \"test\"""")
       val jsonStr = s"{${entries.mkString(", ")}}"
       val query = s"""select named_struct(