[SPARK-50710][CONNECT] Add support for optional client reconnection to sessions after release

[vicennial]

What changes were proposed in this pull request?

Adds a new boolean allow_reconnect field to ReleaseSessionRequest.
When set to true in the request, the server will not place the session in the closedSessionsCache of SparkConnectSessionManager.
The session's clean-up process is unmodified.

Why are the changes needed?

Currently, the connect server will, by default, tombstone all sessions that have either been released explicitly (through a ReleaseSession request) or cleaned up due to inactivity/idleness in periodic checks.

Tombstoning prevents clients from reconnecting with the same userId and sessionId. This mechanism ensures that clients do not accidentally end up with a 'fresh' server-side session, which may be disastrous/fatal as all previously held state is lost (e.g., Temporary views, temporary UDFs, modified configs, current catalog, etc.).

Consider a client that runs simple non-state dependant queries (e.g select count from ...), perhaps sparsely during the lifetime of the application. Such a client may prefer to opt out of tombstoning for the following reasons:

• Queries do not depend on any custom server-side state.
• Modifying userId/sessionId on each reconnect may be inconvenient for tracking/observability purposes.
• On resource-constrained servers, clients may want to minimize their memory footprint by explicitly releasing their state, especially when they believe their requests are sparsely spread out.

Currently, the only way to allow clients to reconnect is to set spark.connect.session.manager.closedSessionsTombstonesSize to 0. However, this is not ideal as it would allow all clients to reconnect, which as previously pointed out, may be dangerous.

As an improvement, allowing specific clients to explicitly signal/request the reconnection possibility addresses the needs mentioned earlier.

Does this PR introduce any user-facing change?

Yes.

When the client releases a session with allow_reconnect set to true, a reconnection will lead to the server generation a fresh session and not result in an error like [INVALID_HANDLE.SESSION_CLOSED] The handle 271dab46-a9a0-4458-ad3a-71442eaa9a21 is invalid. Session was closed. SQLSTATE: HY000

Full example (gRPC based):

Default/allow_reconnect set to false

Create a session via a Config request:

{
    "operation": {
        "get": {
            "keys": ["spark.sql.ansi.enabled"]
        }
    },
    "session_id": "271dab46-a9a0-4458-ad3a-71442eaa9a21",
    "user_context": {
        "user_id": "vicennial",
        "user_name": "Akhil"
    }
}

Release session via ReleaseSession request:

{
    "session_id": "271dab46-a9a0-4458-ad3a-71442eaa9a21",
    "user_context": {
        "user_id": "vicennial",
        "user_name": "Akhil"
    },
    "allow_reconnect": false
}

Retry the earlier config request, the error [INVALID_HANDLE.SESSION_CLOSED] The handle 271dab46-a9a0-4458-ad3a-71442eaa9a21 is invalid. Session was closed. SQLSTATE: HY000 is hit.

Default/allow_reconnect set to true

Create a session via a Config request:

{
    "operation": {
        "get": {
            "keys": ["spark.sql.ansi.enabled"]
        }
    },
    "session_id": "ff1410b9-0a75-4634-820a-b46c14f30896",
    "user_context": {
        "user_id": "vicennial",
        "user_name": "Akhil"
    }
}

Release session via ReleaseSession request:

{
    "session_id": "ff1410b9-0a75-4634-820a-b46c14f30896",
    "user_context": {
        "user_id": "vicennial",
        "user_name": "Akhil"
    },
    "allow_reconnect": true
}

Retry the earlier config request, the request goes through and it can be noted the server_side_session_id in the response of the last config request is different from the first one as a new server side session was generated.

How was this patch tested?

New unit test + existing tests.

Was this patch authored or co-authored using generative AI tooling?

No.

[vicennial]

PTAL @hvanhovell / @HyukjinKwon / @xupefei

[HyukjinKwon]

Merged to master.