You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kyuubi.apache.org by fe...@apache.org on 2022/03/25 16:23:51 UTC
[incubator-kyuubi] branch master updated: [KYUUBI #2102] Support to retry the internal thrift request call and add engine liveness probe to enable fast fail before retry
This is an automated email from the ASF dual-hosted git repository.
feiwang pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-kyuubi.git
The following commit(s) were added to refs/heads/master by this push:
new e8445b7 [KYUUBI #2102] Support to retry the internal thrift request call and add engine liveness probe to enable fast fail before retry
e8445b7 is described below
commit e8445b7f28920c35aa2bd8e53c545143cd902dd2
Author: Fei Wang <fw...@ebay.com>
AuthorDate: Sat Mar 26 00:23:44 2022 +0800
[KYUUBI #2102] Support to retry the internal thrift request call and add engine liveness probe to enable fast fail before retry
### _Why are the changes needed?_
To close #2102
Support to retry all the internal thrift request calls(except RenewDelegationToken now), and fast fail if the remote engine is not stable or not alive.
In this PR, it supports engine liveness probe.
If it is enabled, a companion thrift client will be created and open a liveness probe session when opening remote engine session.
It will send some simple thrift request(GetInfo) to check whether the remote engine is alive, and fast fail before retry if remote engine is not connectable.
#### Why not use the same thrift client to check engine liveness before retry?
I tried that, but met `out of resp sequence` error.
For example:
1. send getOperationStatus request
2. read time out
3. send GetInfoType request
4. receive getOperationStatus response (out of resp sequence)
### _How was this patch tested?_
- [x] Add some test cases that check the changes thoroughly including negative and positive cases if possible
- [ ] Add screenshots for manual tests if appropriate
- [x] [Run test](https://kyuubi.apache.org/docs/latest/develop_tools/testing.html#running-tests) locally before make a pull request
Closes #2122 from turboFei/retry_rpc.
Closes #2102
3926ba04 [Fei Wang] adress comments
ade4ede6 [Fei Wang] add timeout
1b7a64f9 [Fei Wang] Only check remote engine alive before retry
98e03f8e [Fei Wang] refactor
fac388cf [Fei Wang] remove unused import
9c6d8737 [Fei Wang] add ut
9b595650 [Fei Wang] Support to retry the thrift request and engine alive probe
Authored-by: Fei Wang <fw...@ebay.com>
Signed-off-by: Fei Wang <fw...@ebay.com>
---
docs/deployment/settings.md | 6 +-
.../org/apache/kyuubi/config/KyuubiConf.scala | 36 ++++-
.../kyuubi/client/KyuubiSyncThriftClient.scala | 175 ++++++++++++++++++---
.../apache/kyuubi/operation/ExecuteStatement.scala | 34 +---
.../KyuubiOperationPerConnectionSuite.scala | 21 +++
5 files changed, 213 insertions(+), 59 deletions(-)
diff --git a/docs/deployment/settings.md b/docs/deployment/settings.md
index ce3b93c..2ada309 100644
--- a/docs/deployment/settings.md
+++ b/docs/deployment/settings.md
@@ -301,8 +301,9 @@ Key | Default | Meaning | Type | Since
<code>kyuubi.operation.plan.only.mode</code>|<div style='width: 65pt;word-wrap: break-word;white-space: normal'>NONE</div>|<div style='width: 170pt;word-wrap: break-word;white-space: normal'>Whether to perform the statement in a PARSE, ANALYZE, OPTIMIZE, PHYSICAL, EXECUTION only way without executing the query. When it is NONE, the statement will be fully executed</div>|<div style='width: 30pt'>string</div>|<div style='width: 20pt'>1.4.0</div>
<code>kyuubi.operation.query.timeout</code>|<div style='width: 65pt;word-wrap: break-word;white-space: normal'><undefined></div>|<div style='width: 170pt;word-wrap: break-word;white-space: normal'>Timeout for query executions at server-side, take affect with client-side timeout(`java.sql.Statement.setQueryTimeout`) together, a running query will be cancelled automatically if timeout. It's off by default, which means only client-side take fully control whether the query should timeo [...]
<code>kyuubi.operation.scheduler.pool</code>|<div style='width: 65pt;word-wrap: break-word;white-space: normal'><undefined></div>|<div style='width: 170pt;word-wrap: break-word;white-space: normal'>The scheduler pool of job. Note that, this config should be used after change Spark config spark.scheduler.mode=FAIR.</div>|<div style='width: 30pt'>string</div>|<div style='width: 20pt'>1.1.1</div>
-<code>kyuubi.operation.status.polling.max.attempts</code>|<div style='width: 65pt;word-wrap: break-word;white-space: normal'>5</div>|<div style='width: 170pt;word-wrap: break-word;white-space: normal'>Max attempts for long polling asynchronous running sql query's status on raw transport failures, e.g. TTransportException</div>|<div style='width: 30pt'>int</div>|<div style='width: 20pt'>1.4.0</div>
+<code>kyuubi.operation.status.polling.max.attempts</code>|<div style='width: 65pt;word-wrap: break-word;white-space: normal'>5</div>|<div style='width: 170pt;word-wrap: break-word;white-space: normal'>(deprecated) - Using kyuubi.operation.thrift.client.request.max.attempts instead</div>|<div style='width: 30pt'>int</div>|<div style='width: 20pt'>1.4.0</div>
<code>kyuubi.operation.status.polling.timeout</code>|<div style='width: 65pt;word-wrap: break-word;white-space: normal'>PT5S</div>|<div style='width: 170pt;word-wrap: break-word;white-space: normal'>Timeout(ms) for long polling asynchronous running sql query's status</div>|<div style='width: 30pt'>duration</div>|<div style='width: 20pt'>1.0.0</div>
+<code>kyuubi.operation.thrift.client.request.max.attempts</code>|<div style='width: 65pt;word-wrap: break-word;white-space: normal'>5</div>|<div style='width: 170pt;word-wrap: break-word;white-space: normal'>Max attempts for operation thrift request call at server-side on raw transport failures, e.g. TTransportException</div>|<div style='width: 30pt'>int</div>|<div style='width: 20pt'>1.6.0</div>
### Server
@@ -321,6 +322,9 @@ Key | Default | Meaning | Type | Since
<code>kyuubi.session.conf.advisor</code>|<div style='width: 65pt;word-wrap: break-word;white-space: normal'><undefined></div>|<div style='width: 170pt;word-wrap: break-word;white-space: normal'>A config advisor plugin for Kyuubi Server. This plugin can provide some custom configs for different user or session configs and overwrite the session configs before open a new session. This config value should be a class which is a child of 'org.apache.kyuubi.plugin.SessionConfAdvisor' whic [...]
<code>kyuubi.session.conf.ignore.list</code>|<div style='width: 65pt;word-wrap: break-word;white-space: normal'></div>|<div style='width: 170pt;word-wrap: break-word;white-space: normal'>A comma separated list of ignored keys. If the client connection contains any of them, the key and the corresponding value will be removed silently during engine bootstrap and connection setup. Note that this rule is for server-side protection defined via administrators to prevent some essential configs [...]
<code>kyuubi.session.conf.restrict.list</code>|<div style='width: 65pt;word-wrap: break-word;white-space: normal'></div>|<div style='width: 170pt;word-wrap: break-word;white-space: normal'>A comma separated list of restricted keys. If the client connection contains any of them, the connection will be rejected explicitly during engine bootstrap and connection setup. Note that this rule is for server-side protection defined via administrators to prevent some essential configs from tamperin [...]
+<code>kyuubi.session.engine.alive.probe.enabled</code>|<div style='width: 65pt;word-wrap: break-word;white-space: normal'>false</div>|<div style='width: 170pt;word-wrap: break-word;white-space: normal'>Whether to enable the engine alive probe, it true, we will create a companion thrift client that sends simple request to check whether the engine is keep alive.</div>|<div style='width: 30pt'>boolean</div>|<div style='width: 20pt'>1.6.0</div>
+<code>kyuubi.session.engine.alive.probe.interval</code>|<div style='width: 65pt;word-wrap: break-word;white-space: normal'>PT10S</div>|<div style='width: 170pt;word-wrap: break-word;white-space: normal'>The interval for engine alive probe.</div>|<div style='width: 30pt'>duration</div>|<div style='width: 20pt'>1.6.0</div>
+<code>kyuubi.session.engine.alive.timeout</code>|<div style='width: 65pt;word-wrap: break-word;white-space: normal'>PT2M</div>|<div style='width: 170pt;word-wrap: break-word;white-space: normal'>The timeout for engine alive. If there is no alive probe success in the last timeout window, the engine will be marked as no-alive.</div>|<div style='width: 30pt'>duration</div>|<div style='width: 20pt'>1.6.0</div>
<code>kyuubi.session.engine.check.interval</code>|<div style='width: 65pt;word-wrap: break-word;white-space: normal'>PT1M</div>|<div style='width: 170pt;word-wrap: break-word;white-space: normal'>The check interval for engine timeout</div>|<div style='width: 30pt'>duration</div>|<div style='width: 20pt'>1.0.0</div>
<code>kyuubi.session.engine.flink.main.resource</code>|<div style='width: 65pt;word-wrap: break-word;white-space: normal'><undefined></div>|<div style='width: 170pt;word-wrap: break-word;white-space: normal'>The package used to create Flink SQL engine remote job. If it is undefined, Kyuubi will use the default</div>|<div style='width: 30pt'>string</div>|<div style='width: 20pt'>1.4.0</div>
<code>kyuubi.session.engine.flink.max.rows</code>|<div style='width: 65pt;word-wrap: break-word;white-space: normal'>1000000</div>|<div style='width: 170pt;word-wrap: break-word;white-space: normal'>Max rows of Flink query results. For batch queries, rows that exceeds the limit would be ignored. For streaming queries, the query would be canceled if the limit is reached.</div>|<div style='width: 30pt'>int</div>|<div style='width: 20pt'>1.5.0</div>
diff --git a/kyuubi-common/src/main/scala/org/apache/kyuubi/config/KyuubiConf.scala b/kyuubi-common/src/main/scala/org/apache/kyuubi/config/KyuubiConf.scala
index 513c840..09e87f3 100644
--- a/kyuubi-common/src/main/scala/org/apache/kyuubi/config/KyuubiConf.scala
+++ b/kyuubi-common/src/main/scala/org/apache/kyuubi/config/KyuubiConf.scala
@@ -668,6 +668,29 @@ object KyuubiConf {
.timeConf
.createWithDefault(Duration.ofSeconds(60).toMillis)
+ val ENGINE_ALIVE_PROBE_ENABLED: ConfigEntry[Boolean] =
+ buildConf("session.engine.alive.probe.enabled")
+ .doc("Whether to enable the engine alive probe, it true, we will create a companion thrift" +
+ " client that sends simple request to check whether the engine is keep alive.")
+ .version("1.6.0")
+ .booleanConf
+ .createWithDefault(false)
+
+ val ENGINE_ALIVE_PROBE_INTERVAL: ConfigEntry[Long] =
+ buildConf("session.engine.alive.probe.interval")
+ .doc("The interval for engine alive probe.")
+ .version("1.6.0")
+ .timeConf
+ .createWithDefault(Duration.ofSeconds(10).toMillis)
+
+ val ENGINE_ALIVE_TIMEOUT: ConfigEntry[Long] =
+ buildConf("session.engine.alive.timeout")
+ .doc("The timeout for engine alive. If there is no alive probe success in the last timeout" +
+ " window, the engine will be marked as no-alive.")
+ .version("1.6.0")
+ .timeConf
+ .createWithDefault(Duration.ofSeconds(120).toMillis)
+
val ENGINE_INIT_TIMEOUT: ConfigEntry[Long] = buildConf("session.engine.initialize.timeout")
.doc("Timeout for starting the background engine, e.g. SparkSQLEngine.")
.version("1.0.0")
@@ -825,14 +848,23 @@ object KyuubiConf {
.timeConf
.createWithDefault(Duration.ofSeconds(5).toMillis)
+ @deprecated(s"using kyuubi.operation.thrift.client.request.max.attempts instead", "1.6.0")
val OPERATION_STATUS_POLLING_MAX_ATTEMPTS: ConfigEntry[Int] =
buildConf("operation.status.polling.max.attempts")
- .doc("Max attempts for long polling asynchronous running sql query's status on raw" +
- " transport failures, e.g. TTransportException")
+ .doc(s"(deprecated) - Using kyuubi.operation.thrift.client.request.max.attempts instead")
.version("1.4.0")
.intConf
.createWithDefault(5)
+ val OPERATION_THRIFT_CLIENT_REQUEST_MAX_ATTEMPTS: ConfigEntry[Int] =
+ buildConf("operation.thrift.client.request.max.attempts")
+ .doc("Max attempts for operation thrift request call at server-side on raw transport" +
+ " failures, e.g. TTransportException")
+ .version("1.6.0")
+ .intConf
+ .checkValue(_ > 0, "must be positive number")
+ .createWithDefault(5)
+
val OPERATION_FORCE_CANCEL: ConfigEntry[Boolean] =
buildConf("operation.interrupt.on.cancel")
.doc("When true, all running tasks will be interrupted if one cancels a query. " +
diff --git a/kyuubi-server/src/main/scala/org/apache/kyuubi/client/KyuubiSyncThriftClient.scala b/kyuubi-server/src/main/scala/org/apache/kyuubi/client/KyuubiSyncThriftClient.scala
index 1ec359b..4202bea 100644
--- a/kyuubi-server/src/main/scala/org/apache/kyuubi/client/KyuubiSyncThriftClient.scala
+++ b/kyuubi-server/src/main/scala/org/apache/kyuubi/client/KyuubiSyncThriftClient.scala
@@ -17,24 +17,31 @@
package org.apache.kyuubi.client
+import java.util.concurrent.{ScheduledExecutorService, TimeUnit}
import java.util.concurrent.locks.ReentrantLock
import scala.collection.JavaConverters._
import org.apache.hive.service.rpc.thrift._
+import org.apache.thrift.TException
import org.apache.thrift.protocol.{TBinaryProtocol, TProtocol}
import org.apache.thrift.transport.TSocket
-import org.apache.kyuubi.{KyuubiSQLException, Logging}
+import org.apache.kyuubi.{KyuubiSQLException, Logging, Utils}
import org.apache.kyuubi.config.KyuubiConf
import org.apache.kyuubi.config.KyuubiConf.{ENGINE_LOGIN_TIMEOUT, ENGINE_REQUEST_TIMEOUT}
import org.apache.kyuubi.operation.FetchOrientation
import org.apache.kyuubi.operation.FetchOrientation.FetchOrientation
import org.apache.kyuubi.service.authentication.PlainSASLHelper
import org.apache.kyuubi.session.SessionHandle
-import org.apache.kyuubi.util.ThriftUtils
+import org.apache.kyuubi.util.{ThreadUtils, ThriftUtils}
-class KyuubiSyncThriftClient private (protocol: TProtocol)
+class KyuubiSyncThriftClient private (
+ protocol: TProtocol,
+ maxAttempts: Int,
+ engineAliveProbeProtocol: Option[TProtocol],
+ engineAliveProbeInterval: Long,
+ engineAliveTimeout: Long)
extends TCLIService.Client(protocol) with Logging {
@volatile private var _remoteSessionHandle: TSessionHandle = _
@@ -42,6 +49,49 @@ class KyuubiSyncThriftClient private (protocol: TProtocol)
private val lock = new ReentrantLock()
+ @volatile private var _aliveProbeSessionHandle: TSessionHandle = _
+ @volatile private var remoteEngineBroken: Boolean = false
+ private val engineAliveProbeClient = engineAliveProbeProtocol.map(new TCLIService.Client(_))
+ private var engineAliveThreadPool: ScheduledExecutorService = _
+ private var engineLastAlive: Long = _
+
+ private def startEngineAliveProbe(): Unit = {
+ engineAliveThreadPool = ThreadUtils.newDaemonSingleThreadScheduledExecutor(
+ "engine-alive-probe-" + _aliveProbeSessionHandle)
+ val task = new Runnable {
+ override def run(): Unit = {
+ if (!remoteEngineBroken) {
+ engineAliveProbeClient.foreach { client =>
+ val tGetInfoReq = new TGetInfoReq()
+ tGetInfoReq.setSessionHandle(_aliveProbeSessionHandle)
+ tGetInfoReq.setInfoType(TGetInfoType.CLI_DBMS_VER)
+
+ try {
+ client.GetInfo(tGetInfoReq).getInfoValue.getStringValue
+ engineLastAlive = System.currentTimeMillis()
+ remoteEngineBroken = false
+ } catch {
+ case e: Throwable =>
+ warn(s"The engine alive probe fails", e)
+ val now = System.currentTimeMillis()
+ if (now - engineLastAlive > engineAliveTimeout) {
+ error("Mark the engine not alive with no recent alive probe success:" +
+ s" ${now - engineLastAlive} ms exceeds timeout $engineAliveTimeout ms")
+ remoteEngineBroken = true
+ }
+ }
+ }
+ }
+ }
+ }
+ engineLastAlive = System.currentTimeMillis()
+ engineAliveThreadPool.scheduleAtFixedRate(
+ task,
+ engineAliveProbeInterval,
+ engineAliveProbeInterval,
+ TimeUnit.MILLISECONDS)
+ }
+
/**
* Lock every rpc call to send them sequentially
*/
@@ -55,6 +105,27 @@ class KyuubiSyncThriftClient private (protocol: TProtocol)
} finally lock.unlock()
}
+ private def withRetryingRequest[T](block: => T, request: String): T = withLockAcquired {
+ var attemptCount = 1
+
+ var resp: T = null.asInstanceOf[T]
+ while (attemptCount <= maxAttempts && resp == null) {
+ try {
+ resp = block
+ remoteEngineBroken = false
+ } catch {
+ case e: TException if attemptCount < maxAttempts && isConnectionValid() =>
+ warn(s"Failed to execute $request after $attemptCount/$maxAttempts times, retrying", e)
+ attemptCount += 1
+ Thread.sleep(100)
+ case e: Throwable =>
+ error(s"Failed to execute $request after $attemptCount/$maxAttempts times, aborting", e)
+ throw e
+ }
+ }
+ resp
+ }
+
def engineId: Option[String] = _engineId
/**
@@ -69,12 +140,22 @@ class KyuubiSyncThriftClient private (protocol: TProtocol)
req.setUsername(user)
req.setPassword(password)
req.setConfiguration(configs.asJava)
- val resp = withLockAcquired(OpenSession(req))
+ val resp = withRetryingRequest(OpenSession(req), "OpenSession")
ThriftUtils.verifyTStatus(resp.getStatus)
_remoteSessionHandle = resp.getSessionHandle
_engineId = Option(resp.getConfiguration)
.filter(_.containsKey("kyuubi.engine.id"))
.map(_.get("kyuubi.engine.id"))
+
+ engineAliveProbeClient.foreach { aliveProbeClient =>
+ Utils.tryLogNonFatalError {
+ val resp = aliveProbeClient.OpenSession(req)
+ ThriftUtils.verifyTStatus(resp.getStatus)
+ _aliveProbeSessionHandle = resp.getSessionHandle
+ startEngineAliveProbe()
+ }
+ }
+
SessionHandle(_remoteSessionHandle, protocol)
}
@@ -82,14 +163,26 @@ class KyuubiSyncThriftClient private (protocol: TProtocol)
try {
if (_remoteSessionHandle != null) {
val req = new TCloseSessionReq(_remoteSessionHandle)
- val resp = withLockAcquired(CloseSession(req))
+ val resp = withRetryingRequest(CloseSession(req), "CloseSession")
ThriftUtils.verifyTStatus(resp.getStatus)
}
} catch {
case e: Exception =>
throw KyuubiSQLException("Error while cleaning up the engine resources", e)
} finally {
- if (protocol.getTransport.isOpen) protocol.getTransport.close()
+ Option(engineAliveThreadPool).foreach(_.shutdown())
+ if (_aliveProbeSessionHandle != null && !remoteEngineBroken) {
+ engineAliveProbeClient.foreach { client =>
+ Utils.tryLogNonFatalError {
+ val req = new TCloseSessionReq(_aliveProbeSessionHandle)
+ val resp = client.CloseSession(req)
+ ThriftUtils.verifyTStatus(resp.getStatus)
+ }
+ }
+ }
+ Seq(protocol).union(engineAliveProbeProtocol.toSeq).foreach { tProtocol =>
+ if (tProtocol.getTransport.isOpen) tProtocol.getTransport.close()
+ }
}
}
@@ -104,21 +197,21 @@ class KyuubiSyncThriftClient private (protocol: TProtocol)
req.setConfOverlay(confOverlay.asJava)
req.setRunAsync(shouldRunAsync)
req.setQueryTimeout(queryTimeout)
- val resp = withLockAcquired(ExecuteStatement(req))
+ val resp = withRetryingRequest(ExecuteStatement(req), "ExecuteStatement")
ThriftUtils.verifyTStatus(resp.getStatus)
resp.getOperationHandle
}
def getTypeInfo: TOperationHandle = {
val req = new TGetTypeInfoReq(_remoteSessionHandle)
- val resp = withLockAcquired(GetTypeInfo(req))
+ val resp = withRetryingRequest(GetTypeInfo(req), "GetTypeInfo")
ThriftUtils.verifyTStatus(resp.getStatus)
resp.getOperationHandle
}
def getCatalogs: TOperationHandle = {
val req = new TGetCatalogsReq(_remoteSessionHandle)
- val resp = withLockAcquired(GetCatalogs(req))
+ val resp = withRetryingRequest(GetCatalogs(req), "GetCatalogs")
ThriftUtils.verifyTStatus(resp.getStatus)
resp.getOperationHandle
}
@@ -128,7 +221,7 @@ class KyuubiSyncThriftClient private (protocol: TProtocol)
req.setSessionHandle(_remoteSessionHandle)
req.setCatalogName(catalogName)
req.setSchemaName(schemaName)
- val resp = withLockAcquired(GetSchemas(req))
+ val resp = withRetryingRequest(GetSchemas(req), "GetSchemas")
ThriftUtils.verifyTStatus(resp.getStatus)
resp.getOperationHandle
}
@@ -144,14 +237,14 @@ class KyuubiSyncThriftClient private (protocol: TProtocol)
req.setSchemaName(schemaName)
req.setTableName(tableName)
req.setTableTypes(tableTypes)
- val resp = withLockAcquired(GetTables(req))
+ val resp = withRetryingRequest(GetTables(req), "GetTables")
ThriftUtils.verifyTStatus(resp.getStatus)
resp.getOperationHandle
}
def getTableTypes: TOperationHandle = {
val req = new TGetTableTypesReq(_remoteSessionHandle)
- val resp = withLockAcquired(GetTableTypes(req))
+ val resp = withRetryingRequest(GetTableTypes(req), "GetTableTypes")
ThriftUtils.verifyTStatus(resp.getStatus)
resp.getOperationHandle
}
@@ -166,7 +259,7 @@ class KyuubiSyncThriftClient private (protocol: TProtocol)
req.setSchemaName(schemaName)
req.setTableName(tableName)
req.setColumnName(columnName)
- val resp = withLockAcquired(GetColumns(req))
+ val resp = withRetryingRequest(GetColumns(req), "GetColumns")
ThriftUtils.verifyTStatus(resp.getStatus)
resp.getOperationHandle
}
@@ -178,7 +271,7 @@ class KyuubiSyncThriftClient private (protocol: TProtocol)
val req = new TGetFunctionsReq(_remoteSessionHandle, functionName)
req.setCatalogName(catalogName)
req.setSchemaName(schemaName)
- val resp = withLockAcquired(GetFunctions(req))
+ val resp = withRetryingRequest(GetFunctions(req), "GetFunctions")
ThriftUtils.verifyTStatus(resp.getStatus)
resp.getOperationHandle
}
@@ -192,7 +285,7 @@ class KyuubiSyncThriftClient private (protocol: TProtocol)
req.setCatalogName(catalogName)
req.setSchemaName(schemaName)
req.setTableName(tableName)
- val resp = withLockAcquired(GetPrimaryKeys(req))
+ val resp = withRetryingRequest(GetPrimaryKeys(req), "GetPrimaryKeys")
ThriftUtils.verifyTStatus(resp.getStatus)
resp.getOperationHandle
}
@@ -212,20 +305,20 @@ class KyuubiSyncThriftClient private (protocol: TProtocol)
req.setForeignCatalogName(foreignCatalog)
req.setForeignSchemaName(foreignSchema)
req.setForeignTableName(foreignTable)
- val resp = withLockAcquired(GetCrossReference(req))
+ val resp = withRetryingRequest(GetCrossReference(req), "GetCrossReference")
ThriftUtils.verifyTStatus(resp.getStatus)
resp.getOperationHandle
}
def getOperationStatus(operationHandle: TOperationHandle): TGetOperationStatusResp = {
val req = new TGetOperationStatusReq(operationHandle)
- val resp = withLockAcquired(GetOperationStatus(req))
+ val resp = withRetryingRequest(GetOperationStatus(req), "GetOperationStatus")
resp
}
def cancelOperation(operationHandle: TOperationHandle): Unit = {
val req = new TCancelOperationReq(operationHandle)
- val resp = withLockAcquired(CancelOperation(req))
+ val resp = withRetryingRequest(CancelOperation(req), "CancelOperation")
if (resp.getStatus.getStatusCode == TStatusCode.SUCCESS_STATUS) {
info(s"$req succeed on engine side")
} else {
@@ -235,7 +328,7 @@ class KyuubiSyncThriftClient private (protocol: TProtocol)
def closeOperation(operationHandle: TOperationHandle): Unit = {
val req = new TCloseOperationReq(operationHandle)
- val resp = withLockAcquired(CloseOperation(req))
+ val resp = withRetryingRequest(CloseOperation(req), "CloseOperation")
if (resp.getStatus.getStatusCode == TStatusCode.SUCCESS_STATUS) {
info(s"$req succeed on engine side")
} else {
@@ -245,7 +338,7 @@ class KyuubiSyncThriftClient private (protocol: TProtocol)
def getResultSetMetadata(operationHandle: TOperationHandle): TTableSchema = {
val req = new TGetResultSetMetadataReq(operationHandle)
- val resp = withLockAcquired(GetResultSetMetadata(req))
+ val resp = withRetryingRequest(GetResultSetMetadata(req), "GetResultSetMetadata")
ThriftUtils.verifyTStatus(resp.getStatus)
resp.getSchema
}
@@ -259,7 +352,7 @@ class KyuubiSyncThriftClient private (protocol: TProtocol)
val req = new TFetchResultsReq(operationHandle, or, maxRows)
val fetchType = if (fetchLog) 1.toShort else 0.toShort
req.setFetchType(fetchType)
- val resp = withLockAcquired(FetchResults(req))
+ val resp = withRetryingRequest(FetchResults(req), "FetchResults")
ThriftUtils.verifyTStatus(resp.getStatus)
resp.getResults
}
@@ -281,9 +374,26 @@ class KyuubiSyncThriftClient private (protocol: TProtocol)
case e: Exception => warn(s"$req failed on engine side", e)
}
}
+
+ def isConnectionValid(): Boolean = {
+ !remoteEngineBroken && protocol.getTransport.isOpen
+ }
}
private[kyuubi] object KyuubiSyncThriftClient {
+ private def createTProtocol(
+ user: String,
+ passwd: String,
+ host: String,
+ port: Int,
+ socketTimeout: Int,
+ connectionTimeout: Int): TProtocol = {
+ val tSocket = new TSocket(host, port, socketTimeout, connectionTimeout)
+ val tTransport = PlainSASLHelper.getPlainTransport(user, passwd, tSocket)
+ tTransport.open()
+ new TBinaryProtocol(tTransport)
+ }
+
def createClient(
user: String,
password: String,
@@ -293,10 +403,23 @@ private[kyuubi] object KyuubiSyncThriftClient {
val passwd = Option(password).filter(_.nonEmpty).getOrElse("anonymous")
val loginTimeout = conf.get(ENGINE_LOGIN_TIMEOUT).toInt
val requestTimeout = conf.get(ENGINE_REQUEST_TIMEOUT).toInt
- val tSocket = new TSocket(host, port, requestTimeout, loginTimeout)
- val tTransport = PlainSASLHelper.getPlainTransport(user, passwd, tSocket)
- tTransport.open()
- val tProtocol = new TBinaryProtocol(tTransport)
- new KyuubiSyncThriftClient(tProtocol)
+ val requestMaxAttempts = conf.get(KyuubiConf.OPERATION_THRIFT_CLIENT_REQUEST_MAX_ATTEMPTS)
+ val aliveProbeEnabled = conf.get(KyuubiConf.ENGINE_ALIVE_PROBE_ENABLED)
+ val aliveProbeInterval = conf.get(KyuubiConf.ENGINE_ALIVE_PROBE_INTERVAL).toInt
+ val aliveTimeout = conf.get(KyuubiConf.ENGINE_ALIVE_TIMEOUT)
+
+ val tProtocol = createTProtocol(user, passwd, host, port, requestTimeout, loginTimeout)
+ val aliveProbeProtocol =
+ if (aliveProbeEnabled) {
+ Option(createTProtocol(user, passwd, host, port, aliveProbeInterval, loginTimeout))
+ } else {
+ None
+ }
+ new KyuubiSyncThriftClient(
+ tProtocol,
+ requestMaxAttempts,
+ aliveProbeProtocol,
+ aliveProbeInterval,
+ aliveTimeout)
}
}
diff --git a/kyuubi-server/src/main/scala/org/apache/kyuubi/operation/ExecuteStatement.scala b/kyuubi-server/src/main/scala/org/apache/kyuubi/operation/ExecuteStatement.scala
index 1dc8d47..3a6dede 100644
--- a/kyuubi-server/src/main/scala/org/apache/kyuubi/operation/ExecuteStatement.scala
+++ b/kyuubi-server/src/main/scala/org/apache/kyuubi/operation/ExecuteStatement.scala
@@ -21,10 +21,8 @@ import scala.collection.JavaConverters._
import org.apache.hive.service.rpc.thrift.{TGetOperationStatusResp, TProtocolVersion}
import org.apache.hive.service.rpc.thrift.TOperationState._
-import org.apache.thrift.TException
import org.apache.kyuubi.KyuubiSQLException
-import org.apache.kyuubi.config.KyuubiConf
import org.apache.kyuubi.events.{EventLogging, KyuubiOperationEvent}
import org.apache.kyuubi.operation.FetchOrientation.FETCH_NEXT
import org.apache.kyuubi.operation.OperationState.OperationState
@@ -47,10 +45,6 @@ class ExecuteStatement(
null
}
- private val maxStatusPollOnFailure = {
- session.sessionManager.getConf.get(KyuubiConf.OPERATION_STATUS_POLLING_MAX_ATTEMPTS)
- }
-
override def getOperationLog: Option[OperationLog] = Option(_operationLog)
override def beforeRun(): Unit = {
@@ -77,31 +71,11 @@ class ExecuteStatement(
try {
setState(OperationState.RUNNING)
var statusResp: TGetOperationStatusResp = null
- var currentAttempts = 0
-
- def fetchOperationStatusWithRetry(): Unit = {
- try {
- statusResp = client.getOperationStatus(_remoteOpHandle)
- currentAttempts = 0 // reset attempts whenever get touch with engine again
- } catch {
- case e: TException if currentAttempts >= maxStatusPollOnFailure =>
- error(
- s"Failed to get ${session.user}'s query[$getHandle] status after" +
- s" $maxStatusPollOnFailure times, aborting",
- e)
- throw e
- case e: TException =>
- currentAttempts += 1
- warn(
- s"Failed to get ${session.user}'s query[$getHandle] status" +
- s" ($currentAttempts / $maxStatusPollOnFailure)",
- e)
- Thread.sleep(100)
- }
- }
// initialize operation status
- while (statusResp == null) { fetchOperationStatusWithRetry() }
+ while (statusResp == null) {
+ statusResp = client.getOperationStatus(_remoteOpHandle)
+ }
var isComplete = false
while (!isComplete) {
@@ -113,7 +87,7 @@ class ExecuteStatement(
remoteState match {
case INITIALIZED_STATE | PENDING_STATE | RUNNING_STATE =>
isComplete = false
- fetchOperationStatusWithRetry()
+ statusResp = client.getOperationStatus(_remoteOpHandle)
case FINISHED_STATE =>
setState(OperationState.FINISHED)
diff --git a/kyuubi-server/src/test/scala/org/apache/kyuubi/operation/KyuubiOperationPerConnectionSuite.scala b/kyuubi-server/src/test/scala/org/apache/kyuubi/operation/KyuubiOperationPerConnectionSuite.scala
index de811e8..5cbf38c 100644
--- a/kyuubi-server/src/test/scala/org/apache/kyuubi/operation/KyuubiOperationPerConnectionSuite.scala
+++ b/kyuubi-server/src/test/scala/org/apache/kyuubi/operation/KyuubiOperationPerConnectionSuite.scala
@@ -216,6 +216,27 @@ class KyuubiOperationPerConnectionSuite extends WithKyuubiServer with HiveJDBCTe
assert(kyuubiConnection.isClosed)
}
}
+
+ test("KYUUBI #2102 - support engine alive probe to fast fail on engine broken") {
+ withSessionConf(Map(
+ KyuubiConf.ENGINE_ALIVE_PROBE_ENABLED.key -> "true",
+ KyuubiConf.ENGINE_ALIVE_PROBE_INTERVAL.key -> "100",
+ KyuubiConf.ENGINE_ALIVE_TIMEOUT.key -> "3000",
+ KyuubiConf.OPERATION_THRIFT_CLIENT_REQUEST_MAX_ATTEMPTS.key -> "10000",
+ KyuubiConf.ENGINE_REQUEST_TIMEOUT.key -> "100"))(Map.empty)(Map.empty) {
+ withSessionHandle { (client, handle) =>
+ val executeStmtReq = new TExecuteStatementReq()
+ executeStmtReq.setStatement("select java_method('java.lang.System', 'exit', 1)")
+ executeStmtReq.setSessionHandle(handle)
+ executeStmtReq.setRunAsync(false)
+ val startTime = System.currentTimeMillis()
+ val executeStmtResp = client.ExecuteStatement(executeStmtReq)
+ assert(executeStmtResp.getStatus.getStatusCode === TStatusCode.ERROR_STATUS)
+ val elapsedTime = System.currentTimeMillis() - startTime
+ assert(elapsedTime > 3 * 1000 && elapsedTime < 20 * 1000)
+ }
+ }
+ }
}
class TestSessionConfAdvisor extends SessionConfAdvisor {