(kyuubi) branch branch-1.11 updated: [KYUUBI #7290] Prevent engine session leak after Kyuubi session closed

[chengpan]

This is an automated email from the ASF dual-hosted git repository.

chengpan pushed a commit to branch branch-1.11
in repository https://gitbox.apache.org/repos/asf/kyuubi.git


The following commit(s) were added to refs/heads/branch-1.11 by this push:
     new d151dce2f0 [KYUUBI #7290] Prevent engine session leak after Kyuubi 
session closed
d151dce2f0 is described below

commit d151dce2f092d366fff98c9e1ef141479c53abcb
Author: ruanwenjun <wen...@apache.org>
AuthorDate: Mon Jan 5 14:22:44 2026 +0800

    [KYUUBI #7290] Prevent engine session leak after Kyuubi session closed
    
    ### Why are the changes needed?
    close #7290
    When we close a kyuubi session, if the engine session(thrift client) has 
not initialized, then the kyuubi session will be closed, but the engine session 
might be alive, then will result in a resource leak.
    
    ### How was this patch tested?
    Can be tested by when the engine pod is pending, and then kill the jdbc 
client, the driver pod should be killed after initialized.
    
    ### Was this patch authored or co-authored using generative AI tooling?
    NO
    
    Closes #7294 from ruanwenjun/dev_wenjun_fix7290.
    
    Closes #7290
    
    142a21088 [ruanwenjun] make sure the client closed
    974520645 [ruanwenjun] set shouldRetry to false
    d3c860874 [ruanwenjun] remove unused comment
    866126d32 [ruanwenjun] If the session already closed, then return
    3b2ee3921 [ruanwenjun] [KYUUBI #7290] Fix the engine session might still 
alive when kyuubi session has been closed
    
    Authored-by: ruanwenjun <wen...@apache.org>
    Signed-off-by: Cheng Pan <cheng...@apache.org>
    (cherry picked from commit 8bed9de894f1b4ae0f510cc58daf7199555f6e98)
    Signed-off-by: Cheng Pan <cheng...@apache.org>
---
 .../main/scala/org/apache/kyuubi/session/AbstractSession.scala | 10 ++++++++++
 .../src/main/scala/org/apache/kyuubi/session/Session.scala     |  2 ++
 .../scala/org/apache/kyuubi/session/KyuubiSessionImpl.scala    |  6 +++++-
 3 files changed, 17 insertions(+), 1 deletion(-)

diff --git 
a/kyuubi-common/src/main/scala/org/apache/kyuubi/session/AbstractSession.scala 
b/kyuubi-common/src/main/scala/org/apache/kyuubi/session/AbstractSession.scala
index 1fe5188ad6..e518132898 100644
--- 
a/kyuubi-common/src/main/scala/org/apache/kyuubi/session/AbstractSession.scala
+++ 
b/kyuubi-common/src/main/scala/org/apache/kyuubi/session/AbstractSession.scala
@@ -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
   }
@@ -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)
@@ -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
diff --git 
a/kyuubi-common/src/main/scala/org/apache/kyuubi/session/Session.scala 
b/kyuubi-common/src/main/scala/org/apache/kyuubi/session/Session.scala
index dd9f69fb94..49d0c819ad 100644
--- a/kyuubi-common/src/main/scala/org/apache/kyuubi/session/Session.scala
+++ b/kyuubi-common/src/main/scala/org/apache/kyuubi/session/Session.scala
@@ -44,6 +44,8 @@ trait Session {
   def open(): Unit
   def close(): Unit
 
+  def isClosed: Boolean
+
   def getInfo(infoType: TGetInfoType): TGetInfoValue
 
   def executeStatement(
diff --git 
a/kyuubi-server/src/main/scala/org/apache/kyuubi/session/KyuubiSessionImpl.scala
 
b/kyuubi-server/src/main/scala/org/apache/kyuubi/session/KyuubiSessionImpl.scala
index 0336f67862..436b610861 100644
--- 
a/kyuubi-server/src/main/scala/org/apache/kyuubi/session/KyuubiSessionImpl.scala
+++ 
b/kyuubi-server/src/main/scala/org/apache/kyuubi/session/KyuubiSessionImpl.scala
@@ -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
@@ -210,7 +214,7 @@ class KyuubiSessionImpl(
               throw e
           } finally {
             attempt += 1
-            if (shouldRetry && engineClient != null) {
+            if ((isClosed || shouldRetry) && engineClient != null) {
               try {
                 engineClient.closeSession()
               } catch {