[SPARK-50534][SPARK-50535][TEST][CONNECT] Fix sporadic test failures

### What changes were proposed in this pull request?

Fix sporadic Spark Connect test failures.
1. SPARK-50534: VerifyEvents.this.listener.executeHolder was not declared "volatile", causing the thread to repeatedly read potentially outdated value. The data structure is only used by the test suite.
2. SPARK-50535: org.apache.spark.sql.connect.service.SparkConnectSessionManager.invalidateAllSessions is susceptible to system time synchronization (e.g., NTP), leaving stale sessions. invalidateAllSessions is only used by test suites.

### Why are the changes needed?

Fix sporadic test failures.

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

No.

### How was this patch tested?

Repeatedly ran testOnly org.apache.spark.sql.connect.planner.SparkConnectServiceSuite and org.apache.spark.sql.connect.service.SparkConnectServiceE2ESuite.

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

No.

Closes #49253 from changgyoopark-db/SPARK-50534.

Authored-by: changgyoopark-db <[email protected]>
Signed-off-by: Hyukjin Kwon <[email protected]>

Author: changgyoopark-db

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

@@ -289,8 +289,10 @@ class SparkConnectSessionManager extends Logging {
    * Used for testing
    */
   private[connect] def invalidateAllSessions(): Unit = {
-    periodicMaintenance(defaultInactiveTimeoutMs = 0L, ignoreCustomTimeout = true)
-    assert(sessionStore.isEmpty)
+    sessionStore.forEach((key, sessionHolder) => {
+      removeSessionHolder(key)
+      shutdownSessionHolder(sessionHolder)
+    })
     closedSessionsCache.invalidateAll()
   }
 

sql/connect/server/src/test/scala/org/apache/spark/sql/connect/planner/SparkConnectServiceSuite.scala

@@ -919,7 +919,8 @@ class SparkConnectServiceSuite
   }
   class MockSparkListener() extends SparkListener {
     val semaphoreStarted = new Semaphore(0)
-    var executeHolder = Option.empty[ExecuteHolder]
+    // Accessed by multiple threads in parallel.
+    @volatile var executeHolder = Option.empty[ExecuteHolder]
     override def onOtherEvent(event: SparkListenerEvent): Unit = {
       event match {
         case e: SparkListenerConnectOperationStarted =>