LuciferYang commented on code in PR #59200:
URL: https://github.com/apache/spark/pull/59200#discussion_r4164217426
##########
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:
[686c566](https://github.com/apache/spark/pull/59200/commits/686c56632f6ddb8e14c203327bf466eb3445418d)
fix this and opened followup for master/branch-4.x:
https://github.com/apache/spark/pull/59202
--
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]
