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 all commits
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
10 changes: 10 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,11 @@ abstract class AbstractSession(
}

override def close(): Unit = withAcquireRelease() {
if (_closed) {
return
}
_closed = true
info(s"Mark session $handle closed")
opHandleSet.forEach { opHandle =>
try {
sessionManager.operationManager.closeOperation(opHandle)
Expand All @@ -95,8 +102,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
6 changes: 5 additions & 1 deletion 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,10 @@ class KyuubiSessionImpl(
_engineSessionHandle =
engineClient.openSession(protocol, user, passwd, openEngineSessionConf)
_client = engineClient
if (isClosed) {
shouldRetry = false
throw KyuubiSQLException(s"KyuubiSession $handle has been closed")
}
logSessionInfo(s"Connected to engine [$host:$port]/[${client.engineId.getOrElse("")}]" +
s" with ${_engineSessionHandle}]")
shouldRetry = false
Expand Down Expand Up @@ -210,7 +214,7 @@ class KyuubiSessionImpl(
throw e
} finally {
attempt += 1
if (shouldRetry && engineClient != null) {
if ((isClosed || shouldRetry) && engineClient != null) {
try {
engineClient.closeSession()
} catch {
Expand Down
Loading