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/09/01 04:28:33 UTC
[incubator-kyuubi] branch branch-1.6 updated: [KYUUBI #3315] Fast fail the kyuubi connection if the engine application has been terminated
This is an automated email from the ASF dual-hosted git repository.
feiwang pushed a commit to branch branch-1.6
in repository https://gitbox.apache.org/repos/asf/incubator-kyuubi.git
The following commit(s) were added to refs/heads/branch-1.6 by this push:
new d1b6426bf [KYUUBI #3315] Fast fail the kyuubi connection if the engine application has been terminated
d1b6426bf is described below
commit d1b6426bf93bc6f5200950b9e2b4b763502c5e3e
Author: Fei Wang <fw...@ebay.com>
AuthorDate: Tue Aug 30 14:52:34 2022 +0800
[KYUUBI #3315] Fast fail the kyuubi connection if the engine application has been terminated
### _Why are the changes needed?_
We met the case that the LaunchEngine operation is succeed.
But the spark engine met exception and terminated with SUCCESS state.
And then the kyuubi connection stuck until reach the `kyuubi.session.engine.initialize.timeout`.
The engine application terminated in 1.5 minutes and the engine init timeout is 10m.
FYI:
So, we shall check the engine application state when waiting the engine initialization, and fast fail the connection if the application has been in terminated state.
### _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
- [ ] [Run test](https://kyuubi.apache.org/docs/latest/develop_tools/testing.html#running-tests) locally before make a pull request
Closes #3315 from turboFei/fast_fail_on_terminate.
Closes #3315
03333623 [Fei Wang] add comments
f1632dbb [Fei Wang] fix ut
4059a4fe [Fei Wang] combine the state
33dc4a88 [Fei Wang] only check it for exit 0
d5746af1 [Fei Wang] fix ut
854f3e3b [Fei Wang] add metrics
62001610 [Fei Wang] revert
3fb1b630 [Fei Wang] remove engine ref check, meaning less
d2c2c794 [Fei Wang] add ut
ddafa24a [Fei Wang] fast fail the connection on engine termiante
Authored-by: Fei Wang <fw...@ebay.com>
Signed-off-by: Fei Wang <fw...@ebay.com>
---
.../kyuubi/engine/ApplicationOperation.scala | 16 ++++++++++++++++
.../scala/org/apache/kyuubi/engine/EngineRef.scala | 22 ++++++++++++++++++++++
.../kyuubi/operation/BatchJobSubmission.scala | 14 ++------------
.../org/apache/kyuubi/WithKyuubiServerOnYarn.scala | 17 +++++++++++++++++
4 files changed, 57 insertions(+), 12 deletions(-)
diff --git a/kyuubi-server/src/main/scala/org/apache/kyuubi/engine/ApplicationOperation.scala b/kyuubi-server/src/main/scala/org/apache/kyuubi/engine/ApplicationOperation.scala
index 9d8d4237d..93d495895 100644
--- a/kyuubi-server/src/main/scala/org/apache/kyuubi/engine/ApplicationOperation.scala
+++ b/kyuubi-server/src/main/scala/org/apache/kyuubi/engine/ApplicationOperation.scala
@@ -64,6 +64,22 @@ trait ApplicationOperation {
object ApplicationState extends Enumeration {
type ApplicationState = Value
val PENDING, RUNNING, FINISHED, KILLED, FAILED, ZOMBIE, NOT_FOUND, UNKNOWN = Value
+
+ def isFailed(state: ApplicationState): Boolean = state match {
+ case FAILED => true
+ case KILLED => true
+ case _ => false
+ }
+
+ def isTerminated(state: ApplicationState): Boolean = {
+ state match {
+ case FAILED => true
+ case KILLED => true
+ case FINISHED => true
+ case NOT_FOUND => true
+ case _ => false
+ }
+ }
}
case class ApplicationInfo(
diff --git a/kyuubi-server/src/main/scala/org/apache/kyuubi/engine/EngineRef.scala b/kyuubi-server/src/main/scala/org/apache/kyuubi/engine/EngineRef.scala
index f3e2f3acd..10ffb87c9 100644
--- a/kyuubi-server/src/main/scala/org/apache/kyuubi/engine/EngineRef.scala
+++ b/kyuubi-server/src/main/scala/org/apache/kyuubi/engine/EngineRef.scala
@@ -205,6 +205,28 @@ private[kyuubi] class EngineRef(
throw error
}
}
+
+ // even the submit process succeeds, the application might meet failure when initializing,
+ // check the engine application state from engine manager and fast fail on engine terminate
+ if (exitValue == Some(0)) {
+ Option(engineManager).foreach { engineMgr =>
+ engineMgr.getApplicationInfo(builder.clusterManager(), engineRefId).foreach { appInfo =>
+ if (ApplicationState.isTerminated(appInfo.state)) {
+ MetricsSystem.tracing { ms =>
+ ms.incCount(MetricRegistry.name(ENGINE_FAIL, appUser))
+ ms.incCount(MetricRegistry.name(ENGINE_FAIL, "ENGINE_TERMINATE"))
+ }
+ throw new KyuubiSQLException(
+ s"""
+ |The engine application has been terminated. Please check the engine log.
+ |ApplicationInfo: ${appInfo.toMap.mkString("(\n", ",\n", "\n)")}
+ |""".stripMargin,
+ builder.getError)
+ }
+ }
+ }
+ }
+
if (started + timeout <= System.currentTimeMillis()) {
val killMessage = engineManager.killApplication(builder.clusterManager(), engineRefId)
process.destroyForcibly()
diff --git a/kyuubi-server/src/main/scala/org/apache/kyuubi/operation/BatchJobSubmission.scala b/kyuubi-server/src/main/scala/org/apache/kyuubi/operation/BatchJobSubmission.scala
index c570c2d65..bcaca6237 100644
--- a/kyuubi-server/src/main/scala/org/apache/kyuubi/operation/BatchJobSubmission.scala
+++ b/kyuubi-server/src/main/scala/org/apache/kyuubi/operation/BatchJobSubmission.scala
@@ -339,20 +339,10 @@ class BatchJobSubmission(
object BatchJobSubmission {
def applicationFailed(applicationStatus: Option[ApplicationInfo]): Boolean = {
- applicationStatus.map(_.state).exists {
- case ApplicationState.FAILED => true
- case ApplicationState.KILLED => true
- case _ => false
- }
+ applicationStatus.map(_.state).exists(ApplicationState.isFailed)
}
def applicationTerminated(applicationStatus: Option[ApplicationInfo]): Boolean = {
- applicationStatus.map(_.state).exists {
- case ApplicationState.FAILED => true
- case ApplicationState.KILLED => true
- case ApplicationState.FINISHED => true
- case ApplicationState.NOT_FOUND => true
- case _ => false
- }
+ applicationStatus.map(_.state).exists(ApplicationState.isTerminated)
}
}
diff --git a/kyuubi-server/src/test/scala/org/apache/kyuubi/WithKyuubiServerOnYarn.scala b/kyuubi-server/src/test/scala/org/apache/kyuubi/WithKyuubiServerOnYarn.scala
index 514c89b17..cb55450ad 100644
--- a/kyuubi-server/src/test/scala/org/apache/kyuubi/WithKyuubiServerOnYarn.scala
+++ b/kyuubi-server/src/test/scala/org/apache/kyuubi/WithKyuubiServerOnYarn.scala
@@ -174,4 +174,21 @@ class KyuubiOperationYarnClusterSuite extends WithKyuubiServerOnYarn with HiveJD
assert(batchJobSubmissionOp.getStatus.state === OperationState.ERROR)
}
}
+
+ test("fast fail the kyuubi connection on engine terminated") {
+ withSessionConf(Map.empty)(Map(
+ "spark.master" -> "yarn",
+ "spark.submit.deployMode" -> "cluster",
+ "spark.sql.defaultCatalog=spark_catalog" -> "spark_catalog",
+ "spark.sql.catalog.spark_catalog.type" -> "invalid_type",
+ "kyuubi.session.engine.initialize.timeout" -> "PT10m"))(Map.empty) {
+ val startTime = System.currentTimeMillis()
+ val exception = intercept[Exception] {
+ withJdbcStatement() { _ => }
+ }
+ val elapsedTime = System.currentTimeMillis() - startTime
+ assert(elapsedTime < 60 * 1000)
+ assert(exception.getMessage contains "The engine application has been terminated.")
+ }
+ }
}