(spark) branch branch-4.x updated: [SPARK-44462][SS][CONNECT][FOLLOWUP] Assert foreachBatch close() unregisters the cloned session

[ya...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

LuciferYang pushed a commit to branch branch-4.x
in repository https://gitbox.apache.org/repos/asf/spark.git


The following commit(s) were added to refs/heads/branch-4.x by this push:
     new b47c14462307 [SPARK-44462][SS][CONNECT][FOLLOWUP] Assert foreachBatch close() unregisters the cloned session
b47c14462307 is described below

commit b47c1446230724c3b631b227c4fbcd117b5eda85
Author: YangJie <ya...@baidu.com>
AuthorDate: Fri Oct 2 19:45:40 2026 +0800

    [SPARK-44462][SS][CONNECT][FOLLOWUP] Assert foreachBatch close() unregisters the cloned session
    
    ### What changes were proposed in this pull request?
    
    Followup to #55410. The two `ForeachBatchSessionManager` cleanup tests in `StreamingForeachBatchHelperSuite` checked that `close()` is terminal and that the registered cleaner runs, but neither asserted the side effect `close()` exists to guarantee: that the cloned `SessionHolder` is removed from `SparkConnectService.sessionManager`. Because `close()` sets its `closed` flag independently of the `closeSession` call, a future change that dropped or broke the unregister step would still  [...]
    
    ### Why are the changes needed?
    
    To make the cloned-session leak-prevention contract regression-sensitive. Spotted in review of #59200, the branch-4.3 backport.
    
    ### 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?
    
    Generated-by: Claude Code
    
    Closes #59202 from LuciferYang/SPARK-44462-assert-cloned-session-unregister.
    
    Lead-authored-by: YangJie <ya...@baidu.com>
    Co-authored-by: reviewer <re...@local>
    Signed-off-by: yangjie01 <ya...@baidu.com>
    (cherry picked from commit 38422aefaf599ffb4cfa579d98898d95b1f02081)
    Signed-off-by: yangjie01 <ya...@baidu.com>
---
 .../sql/connect/planner/StreamingForeachBatchHelperSuite.scala    | 8 +++++++-
 1 file changed, 7 insertions(+), 1 deletion(-)

diff --git a/sql/connect/server/src/test/scala/org/apache/spark/sql/connect/planner/StreamingForeachBatchHelperSuite.scala b/sql/connect/server/src/test/scala/org/apache/spark/sql/connect/planner/StreamingForeachBatchHelperSuite.scala
index 9cad09887f3d..0243771bcf27 100644
--- a/sql/connect/server/src/test/scala/org/apache/spark/sql/connect/planner/StreamingForeachBatchHelperSuite.scala
+++ b/sql/connect/server/src/test/scala/org/apache/spark/sql/connect/planner/StreamingForeachBatchHelperSuite.scala
@@ -27,7 +27,7 @@ import org.scalatestplus.mockito.MockitoSugar
 
 import org.apache.spark.SparkIllegalStateException
 import org.apache.spark.sql.connect.SparkConnectTestUtils
-import org.apache.spark.sql.connect.service.SessionHolder
+import org.apache.spark.sql.connect.service.{SessionHolder, SparkConnectService}
 import org.apache.spark.sql.streaming.StreamingQuery
 import org.apache.spark.sql.streaming.StreamingQueryListener
 import org.apache.spark.sql.test.SharedSparkSession
@@ -128,8 +128,14 @@ class StreamingForeachBatchHelperSuite extends SharedSparkSession with MockitoSu
     val clonedHolder = manager.getOrCreateClonedSessionHolder(batchDf)
     // The id is pinned for the whole query: every batch resolves to the same holder.
     assert(manager.getOrCreateClonedSessionHolder(batchDf) eq clonedHolder)
+    // The holder is registered in the global session manager and never expires by inactivity, so
+    // close() must unregister it; a broken unregister step would leak it until server shutdown.
+    assert(
+      SparkConnectService.sessionManager.getIsolatedSessionIfPresent(clonedHolder.key).isDefined)
 
     manager.close()
+    assert(
+      SparkConnectService.sessionManager.getIsolatedSessionIfPresent(clonedHolder.key).isEmpty)
     checkError(
       exception =
         intercept[SparkIllegalStateException](manager.getOrCreateClonedSessionHolder(batchDf)),


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org