HyukjinKwon commented on code in PR #59200:
URL: https://github.com/apache/spark/pull/59200#discussion_r4164066852
##########
sql/connect/server/src/test/scala/org/apache/spark/sql/connect/planner/StreamingForeachBatchHelperSuite.scala:
##########
@@ -114,6 +116,60 @@ class StreamingForeachBatchHelperSuite extends
SharedSparkSession with MockitoSu
assert(spark.streams.listListeners().length == listenersBefore)
}
+ test("ForeachBatchSessionManager: close is terminal, and a closing root
session is refused") {
+ // The cloned holder is registered with customInactiveTimeoutMs = -1, so
it never expires by
+ // inactivity: once the cleaner ran, a later batch must not be able to
register another one.
+ // Use a child session, not the shared one -- closing the holder tears its
session down.
+ val batchDf = spark.newSession().range(1).toDF()
+
+ val manager = new StreamingForeachBatchHelper.ForeachBatchSessionManager(
+ SparkConnectTestUtils.createDummySessionHolder(spark),
+ useClonedSession = true)
+ 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)
+
+ manager.close()
Review Comment:
**Nit (P3):** The two tests added for the cloned-session cleanup path verify
terminality and that a mock closeable runs, but neither asserts the side effect
this code exists to guarantee: that `close()` actually removes the cloned
`SessionHolder` from `SparkConnectService.sessionManager`. Because `close()`
sets the private `closed` flag independently of the `closeSession` call, a
future change that drops or breaks the unregister step would still pass both
tests while re-introducing the never-expiring-holder leak. Consider asserting
the cloned `SessionKey` is present in `SparkConnectService.sessionManager`
before `close()` and absent after, so the leak-prevention contract is
regression-sensitive.
**Verification:**
- **Regression:** Reproduce the leak scenario: after a cloned SessionHolder
is created and registered via getOrCreateClonedSessionHolder,
ForeachBatchSessionManager.close() must remove it from
SparkConnectService.sessionManager; the assertion fails if the
closeSession/unregister step is removed or broken.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
