venkata91 commented on code in PR #59277:
URL: https://github.com/apache/spark/pull/59277#discussion_r4251883142
##########
sql/connect/server/src/main/scala/org/apache/spark/sql/connect/config/Connect.scala:
##########
@@ -150,6 +150,23 @@ object Connect {
.intConf
.createWithDefaultString("1000")
+ val CONNECT_SESSION_ALLOW_RECONNECT_AFTER_CLOSE =
+ buildConf("spark.connect.session.allowReconnectAfterClose")
+ .doc(
+ "When true, the session id can be reused after this session is closed,
for example " +
+ "on inactivity expiry. The id stays reserved, and requests fail with
" +
+ "INVALID_HANDLE.SESSION_CLOSED, until the close finishes. After
that, a client that " +
+ "used the closed session gets INVALID_HANDLE.SESSION_CHANGED and
must create a new " +
Review Comment:
Fixed in 42cc9b6fbd. `getIsolatedSession` now reports `SESSION_CHANGED` from
its not-found branch when the client sent an observed server-side session id,
after the `SESSION_CLOSED` check, and `cloneSession` passes the observed id for
the source. The conf doc and the `docs/configuration.md` row now say the signal
relies on the client sending its observed id, as the built-in Python and Scala
clients do, and that a client that doesn't silently gets a new session. A new
test covers the observed and non-observed lookups after an opted-in expiry,
plus clone; the in-close test now checks `SESSION_CLOSED` with and without an
observed id.
##########
sql/connect/server/src/main/scala/org/apache/spark/sql/connect/service/SparkConnectSessionManager.scala:
##########
@@ -97,6 +109,11 @@ class SparkConnectSessionManager extends Logging {
key,
Some(() => {
validateSessionCreate(key)
+ // A client that observed a previous session expects that session, so
do not create a
+ // new one that it cannot use. Throwing here leaves sessionStore
unchanged.
+ if (previouslyObservedSesssionId.isDefined) {
+ throw sessionChangedError(key)
Review Comment:
Fixed in 42cc9b6fbd. The message is now "The server-side session has
changed, for example because the server restarted or the session was closed.
Please create a new session." The condition name and SQL state are unchanged;
`SparkThrowableSuite` passes.
##########
sql/connect/server/src/main/scala/org/apache/spark/sql/connect/service/SparkConnectSessionManager.scala:
##########
@@ -224,46 +241,75 @@ class SparkConnectSessionManager extends Logging {
session
}
- // Removes session from sessionStore and returns it.
+ // Removes the session with the given key from sessionStore, if any. Returns
the removed
+ // session and whether its tombstone is dropped after the close, see
tryRemoveSessionHolder.
private def removeSessionHolder(
key: SessionKey,
- allowReconnect: Boolean = false): Option[SessionHolder] = {
- var sessionHolder: Option[SessionHolder] = None
-
- // The session holder should remain in the session store until it is added
to the closed session
- // cache, because of a subtle data race: a new session with the same key
can be created if the
- // closed session cache does not contain the key right after the key has
been removed from the
- // session store.
- sessionHolder = Option(sessionStore.get(key))
-
- sessionHolder.foreach { s =>
- if (!allowReconnect) {
- // Put into closedSessionsCache to prevent the same session from being
recreated by
- // getOrCreateIsolatedSession when reconnection isn't allowed.
- closedSessionsCache.put(s.key, s.getSessionHolderInfo)
- }
-
- // Then, remove the session holder from the session store.
- sessionStore.remove(key)
+ allowReconnect: Boolean): Option[(SessionHolder, Boolean)] = {
+ Option(sessionStore.get(key)).flatMap { sessionHolder =>
+ tryRemoveSessionHolder(sessionHolder, allowReconnect).map(sessionHolder
-> _)
}
- sessionHolder
+ }
+
+ // Removes the given session from sessionStore if it is still stored there.
Returns
+ // Some(dropTombstone) only for the one caller that removed it; that caller
must then call
+ // shutdownSessionHolder with it. Returns None if the session was not
removed.
+ //
+ // The session id is fenced before the session is removed, and the fence is
kept until
+ // shutdownSessionHolder has finished closing it. Otherwise a session with
the same key could be
+ // created while the old session is still closing, and the key-based cleanup
in
+ // SessionHolder.close() would tear it down. dropTombstone is decided here,
once: if true, the
+ // fence is an entry in closingSessions that is dropped after the close, so
the session id can
+ // be reused; otherwise it is a tombstone in closedSessionsCache that is
kept.
+ private[service] def tryRemoveSessionHolder(
+ sessionHolder: SessionHolder,
+ allowReconnect: Boolean = false): Option[Boolean] = {
+ // Fence and remove the session in one atomic step for the key, matched by
identity, so that
+ // two callers cannot both remove the same session, a caller that loses
cannot re-put a fence
+ // that the owner already dropped, and a later session with the same key
is never removed.
+ var removed: Option[Boolean] = None
+ sessionStore.computeIfPresent(
+ sessionHolder.key,
+ (_, current) => {
+ if (current eq sessionHolder) {
+ val dropTombstone = allowReconnect ||
current.allowReconnectAfterClose
+ if (dropTombstone) {
+ closingSessions.put(current.key, current)
+ } else {
+ closedSessionsCache.put(current.key, current.getSessionHolderInfo)
+ }
+ removed = Some(dropTombstone)
+ null
+ } else {
+ current
+ }
+ })
+ removed
}
// Shuts down the session after removing.
private def shutdownSessionHolder(
sessionHolder: SessionHolder,
- allowReconnect: Boolean = false): Unit = {
- sessionHolder.close()
- if (!allowReconnect) {
- // Update in closedSessionsCache: above it wasn't updated with
closedTime etc. yet.
- closedSessionsCache.put(sessionHolder.key,
sessionHolder.getSessionHolderInfo)
+ dropTombstone: Boolean): Unit = {
+ try {
+ sessionHolder.close()
+ } finally {
Review Comment:
Added in 42cc9b6fbd: `a close that throws still releases or tombstones the
session id`, and `cloning into a session id that is still closing fails until
the close completes`. I checked that moving the fence release out of the
`finally`, or reverting the clone-target check to the cache-only version, each
fails the matching test.
--
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]
