Skip to content

[KYUUBI #7290] Prevent engine session leak after Kyuubi session closed #7294

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
ruanwenjun wants to merge 5 commits into from
+17 −1
Closed

[KYUUBI #7290] Prevent engine session leak after Kyuubi session closed #7294

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
3b2ee39
[KYUUBI #7290] Fix the engine session might still alive when kyuubi s…
ruanwenjun Dec 31, 2025
866126d
If the session already closed, then return
ruanwenjun Dec 31, 2025
d3c8608
remove unused comment
ruanwenjun Jan 4, 2026
9745206
set shouldRetry to false
ruanwenjun Jan 4, 2026
142a210
make sure the client closed
ruanwenjun Jan 4, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
7 changes: 7 additions & 0 deletions kyuubi-common/src/main/scala/org/apache/kyuubi/session/AbstractSession.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -52,6 +52,8 @@ abstract class AbstractSession(
@volatile private var _lastIdleTime: Long = _createTime
override def lastIdleTime: Long = _lastIdleTime

@volatile private var _closed: Boolean = false

override def getNoOperationTime: Long = {
if (lastIdleTime > 0) System.currentTimeMillis() - _lastIdleTime else 0
}
Expand Down Expand Up @@ -87,6 +89,8 @@ abstract class AbstractSession(
}

override def close(): Unit = withAcquireRelease() {
_closed = true
info(s"Mark session $handle closed")
opHandleSet.forEach { opHandle =>
try {
sessionManager.operationManager.closeOperation(opHandle)
Expand All @@ -95,8 +99,11 @@ abstract class AbstractSession(
warn(s"Error closing operation $opHandle during closing $handle for", e)
}
}
info(s"Closed all operations for session $handle")
}

override def isClosed: Boolean = _closed

protected def runOperation(operation: Operation): OperationHandle = {
try {
val opHandle = operation.getHandle
Expand Down
2 changes: 2 additions & 0 deletions kyuubi-common/src/main/scala/org/apache/kyuubi/session/Session.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -44,6 +44,8 @@ trait Session {
def open(): Unit
def close(): Unit

def isClosed: Boolean

def getInfo(infoType: TGetInfoType): TGetInfoValue

def executeStatement(
Expand Down
5 changes: 5 additions & 0 deletions kyuubi-server/src/main/scala/org/apache/kyuubi/session/KyuubiSessionImpl.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -180,6 +180,11 @@ class KyuubiSessionImpl(
_engineSessionHandle =
engineClient.openSession(protocol, user, passwd, openEngineSessionConf)
_client = engineClient
// Since the _client might be null when we close session
// so we need to check isClosed here
if (isClosed) {
throw KyuubiSQLException(s"KyuubiSession $handle has been closed")
}
Comment on lines 183 to 186
Copy link

Copilot AI Jan 3, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This critical race condition fix lacks automated test coverage. Consider adding a test that simulates closing a Kyuubi session while the engine session is being initialized to verify that the engine session is properly cleaned up. This could be done by using a mock or by introducing a delay in the engine session opening process during the test.

Copilot uses AI. Check for mistakes.
logSessionInfo(s"Connected to engine [$host:$port]/[${client.engineId.getOrElse("")}]" +
s" with ${_engineSessionHandle}]")
shouldRetry = false
Expand Down
Loading