Skip to content
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

[KYUUBI #6236] Throw connectionClosed KyuubiSQLException when engine died #6237

Open
wants to merge 3 commits into
base: master
Choose a base branch
from
Open

[KYUUBI #6236] Throw connectionClosed KyuubiSQLException when engine died #6237

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
642ed69
throw connectionClosed KyuubiSQLException when engine died
Apr 2, 2024
019fe6b
fix client close, then engineConnectionClosed return true
Apr 2, 2024
a5cae3b
fix unit test
Apr 3, 2024
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
5 changes: 5 additions & 0 deletions kyuubi-common/src/main/scala/org/apache/kyuubi/KyuubiSQLException.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -24,6 +24,7 @@ import scala.annotation.tailrec
import scala.collection.JavaConverters._

import org.apache.kyuubi.Utils.stringifyException
import org.apache.kyuubi.session.SessionHandle
import org.apache.kyuubi.shaded.hive.service.rpc.thrift.{TStatus, TStatusCode}
import org.apache.kyuubi.util.reflect.DynConstructors

Expand Down Expand Up @@ -90,6 +91,10 @@ object KyuubiSQLException {
new KyuubiSQLException("connection does not exist", "08003", 91001, null)
}

def connectionClosed(session: SessionHandle, e: Throwable): KyuubiSQLException = {
new KyuubiSQLException(s"connection for ${session} is closed", "08003", 91001, e)
}

def toTStatus(e: Exception, verbose: Boolean = false): TStatus = e match {
case k: KyuubiSQLException => k.toTStatus
case _ =>
Expand Down
11 changes: 3 additions & 8 deletions kyuubi-server/src/main/scala/org/apache/kyuubi/operation/KyuubiOperation.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,7 +20,6 @@ package org.apache.kyuubi.operation
import java.io.IOException

import com.codahale.metrics.MetricRegistry
import org.apache.commons.lang3.StringUtils

import org.apache.kyuubi.{KyuubiSQLException, Utils}
import org.apache.kyuubi.config.KyuubiReservedKeys.KYUUBI_OPERATION_HANDLE_KEY
Expand Down Expand Up @@ -83,13 +82,9 @@ abstract class KyuubiOperation(session: Session) extends AbstractOperation(sessi
MetricRegistry.name(OPERATION_FAIL, opType, errorType)))
val ke = e match {
case kse: KyuubiSQLException => kse
case te: TTransportException
if te.getType == TTransportException.END_OF_FILE &&
StringUtils.isEmpty(te.getMessage) =>
// https://issues.apache.org/jira/browse/THRIFT-4858
KyuubiSQLException(
s"Error $action $opType: Socket for ${session.handle} is closed",
e)
case te: TTransportException =>
warn(s"Error $action $opType: Socket for ${session.handle} is closed", e)
KyuubiSQLException.connectionClosed(session.handle, e)
Copy link
Member

Choose a reason for hiding this comment

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

socket closed may be more appropriate. and cc @yaooqinn, it was introduced in #375.

Copy link
Member

Choose a reason for hiding this comment

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

I don't have a strong option here, both seem fine to me

case e =>
KyuubiSQLException(s"Error $action $opType: ${Utils.stringifyException(e)}", e)
}
Expand Down
35 changes: 34 additions & 1 deletion ...server/src/test/scala/org/apache/kyuubi/operation/KyuubiOperationPerConnectionSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -80,7 +80,7 @@ class KyuubiOperationPerConnectionSuite extends WithKyuubiServer with HiveJDBCTe
assert(executeStmtResp.getOperationHandle === null)
val errMsg = executeStmtResp.getStatus.getErrorMessage
assert(errMsg.contains("Caused by: java.net.SocketException: Connection reset") ||
errMsg.contains(s"Socket for ${SessionHandle(handle)} is closed") ||
errMsg.contains(s"connection for ${SessionHandle(handle)} is closed") ||
errMsg.contains("Socket is closed by peer") ||
errMsg.contains("SparkContext was shut down"))
}
Expand Down Expand Up @@ -346,6 +346,39 @@ class KyuubiOperationPerConnectionSuite extends WithKyuubiServer with HiveJDBCTe
}
}
}

test("throw connectionClosed KyuubiSQLException when engine died") {
withSessionConf()(Map.empty)(Map.empty) {
withSessionHandle { (client, handle) =>
val preReq = new TExecuteStatementReq()
preReq.setStatement("select engine_name()")
preReq.setSessionHandle(handle)
preReq.setRunAsync(false)
client.ExecuteStatement(preReq)

val sessionHandle = SessionHandle(handle)

val exitReq = new TExecuteStatementReq()
exitReq.setStatement("SELECT java_method('java.lang.Thread', 'sleep', 1000L)," +
"java_method('java.lang.System', 'exit', 1)")
exitReq.setSessionHandle(handle)
exitReq.setRunAsync(true)
client.ExecuteStatement(exitReq)

val executeStmtReq = new TExecuteStatementReq()
executeStmtReq.setSessionHandle(handle)
executeStmtReq.setRunAsync(true)
exitReq.setStatement("SELECT java_method('java.lang.Thread', 'sleep', 30000l)")
exitReq.setSessionHandle(handle)
exitReq.setRunAsync(true)
val status = client.ExecuteStatement(exitReq).getStatus
assert(status.getStatusCode === TStatusCode.ERROR_STATUS)
assert(status.getSqlState.equals("08003"))
assert(status.getErrorCode == 91001)
assert(status.getErrorMessage.contains(s"connection for ${sessionHandle} is closed"))
}
}
}
}

class TestSessionConfAdvisor extends SessionConfAdvisor {
Expand Down
Loading