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 <[email protected]>
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 <[email protected]>
Co-authored-by: reviewer <reviewer@local>
Signed-off-by: yangjie01 <[email protected]>
(cherry picked from commit 38422aefaf599ffb4cfa579d98898d95b1f02081)
Signed-off-by: yangjie01 <[email protected]>
---
.../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: [email protected]
For additional commands, e-mail: [email protected]
