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/12/08 16:22:04 UTC
[incubator-kyuubi] branch master updated: [KYUUBI #3922] Only the ApplicationInfo with non-empty id is valid for BatchJobSubmission
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 8eac51385 [KYUUBI #3922] Only the ApplicationInfo with non-empty id is valid for BatchJobSubmission
8eac51385 is described below
commit 8eac51385b42c42a29d34e2208d512a7bc2d86bf
Author: fwang12 <fw...@ebay.com>
AuthorDate: Fri Dec 9 00:21:54 2022 +0800
[KYUUBI #3922] Only the ApplicationInfo with non-empty id is valid for BatchJobSubmission
### _Why are the changes needed?_
The batch state is update to running, but app state is NOT_FOUND.
We shall check the ApplicationInfo return by applicationManager.
Because for YarnApplicationOperation, the result of getApplicationInfoByTag is always non-empty.
```
if (reports.isEmpty) {
debug(s"Application with tag $tag not found")
ApplicationInfo(id = null, name = null, state = ApplicationState.NOT_FOUND)
}
```
### _How was this patch tested?_
- [x] Add some test cases that check the changes thoroughly including negative and positive cases if possible
- [x] 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 #3922 from turboFei/not_found.
Closes #3922
f83c67c2 [fwang12] set not found state if finally state is empty
62e619b0 [fwang12] fix ut
53c76d29 [fwang12] fix ut
ff5c36b3 [fwang12] check id none empty
7b1180c9 [fwang12] add ut
4843f8a7 [fwang12] do not update not found as batch state
Authored-by: fwang12 <fw...@ebay.com>
Signed-off-by: fwang12 <fw...@ebay.com>
---
.../org/apache/kyuubi/operation/BatchJobSubmission.scala | 10 +++++++++-
.../scala/org/apache/kyuubi/WithKyuubiServerOnYarn.scala | 9 +++++++--
.../apache/kyuubi/server/api/v1/BatchesResourceSuite.scala | 2 +-
.../apache/kyuubi/server/rest/client/BatchCliSuite.scala | 13 +++++++++----
4 files changed, 26 insertions(+), 8 deletions(-)
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 d981b80cb..c6161a831 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
@@ -95,7 +95,8 @@ class BatchJobSubmission(
}
override private[kyuubi] def currentApplicationInfo: Option[ApplicationInfo] = {
- applicationManager.getApplicationInfo(builder.clusterManager(), batchId)
+ // only the ApplicationInfo with non-empty id is valid for the operation
+ applicationManager.getApplicationInfo(builder.clusterManager(), batchId).filter(_.id != null)
}
private[kyuubi] def killBatchApplication(): KillResponse = {
@@ -113,6 +114,13 @@ class BatchJobSubmission(
0L
}
+ if (isTerminalState(state)) {
+ if (applicationInfo.isEmpty) {
+ applicationInfo =
+ Option(ApplicationInfo(id = null, name = null, state = ApplicationState.NOT_FOUND))
+ }
+ }
+
applicationInfo.foreach { status =>
val metadataToUpdate = Metadata(
identifier = batchId,
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 cb55450ad..727c5545e 100644
--- a/kyuubi-server/src/test/scala/org/apache/kyuubi/WithKyuubiServerOnYarn.scala
+++ b/kyuubi-server/src/test/scala/org/apache/kyuubi/WithKyuubiServerOnYarn.scala
@@ -122,6 +122,12 @@ class KyuubiOperationYarnClusterSuite extends WithKyuubiServerOnYarn with HiveJD
assert(appInfo.exists(_.id.startsWith("application_")))
}
+ eventually(timeout(10.seconds)) {
+ val metadata = session.sessionManager.getBatchMetadata(session.handle.identifier.toString)
+ assert(metadata.state === "RUNNING")
+ assert(metadata.engineId.startsWith("application_"))
+ }
+
val killResponse = yarnOperation.killApplicationByTag(sessionHandle.identifier.toString)
assert(killResponse._1)
assert(killResponse._2 startsWith "Succeeded to terminate:")
@@ -169,8 +175,7 @@ class KyuubiOperationYarnClusterSuite extends WithKyuubiServerOnYarn with HiveJD
val batchJobSubmissionOp = session.batchJobSubmissionOp
eventually(timeout(3.minutes), interval(50.milliseconds)) {
- assert(batchJobSubmissionOp.currentApplicationInfo.isDefined)
- assert(batchJobSubmissionOp.currentApplicationInfo.get.id == null)
+ assert(batchJobSubmissionOp.currentApplicationInfo.isEmpty)
assert(batchJobSubmissionOp.getStatus.state === OperationState.ERROR)
}
}
diff --git a/kyuubi-server/src/test/scala/org/apache/kyuubi/server/api/v1/BatchesResourceSuite.scala b/kyuubi-server/src/test/scala/org/apache/kyuubi/server/api/v1/BatchesResourceSuite.scala
index 1f59a7dc6..8a2635035 100644
--- a/kyuubi-server/src/test/scala/org/apache/kyuubi/server/api/v1/BatchesResourceSuite.scala
+++ b/kyuubi-server/src/test/scala/org/apache/kyuubi/server/api/v1/BatchesResourceSuite.scala
@@ -442,7 +442,7 @@ class BatchesResourceSuite extends KyuubiFunSuite with RestFrontendTestHelper wi
assert(session1.createTime === batchMetadata.createTime)
assert(session2.createTime === batchMetadata2.createTime)
- eventually(timeout(5.seconds)) {
+ eventually(timeout(10.seconds)) {
assert(session1.batchJobSubmissionOp.getStatus.state === OperationState.RUNNING ||
session1.batchJobSubmissionOp.getStatus.state === OperationState.FINISHED)
assert(session1.batchJobSubmissionOp.builder.processLaunched)
diff --git a/kyuubi-server/src/test/scala/org/apache/kyuubi/server/rest/client/BatchCliSuite.scala b/kyuubi-server/src/test/scala/org/apache/kyuubi/server/rest/client/BatchCliSuite.scala
index ff52bf7fb..9d0a9b15a 100644
--- a/kyuubi-server/src/test/scala/org/apache/kyuubi/server/rest/client/BatchCliSuite.scala
+++ b/kyuubi-server/src/test/scala/org/apache/kyuubi/server/rest/client/BatchCliSuite.scala
@@ -28,7 +28,8 @@ import org.apache.hive.service.rpc.thrift.TProtocolVersion
import org.scalatest.time.SpanSugar.convertIntToGrainOfTime
import org.apache.kyuubi.{BatchTestHelper, RestClientTestHelper, Utils}
-import org.apache.kyuubi.ctl.TestPrematureExit
+import org.apache.kyuubi.config.KyuubiConf
+import org.apache.kyuubi.ctl.{CtlConf, TestPrematureExit}
import org.apache.kyuubi.metrics.{MetricsConstants, MetricsSystem}
import org.apache.kyuubi.session.KyuubiSessionManager
@@ -37,6 +38,10 @@ class BatchCliSuite extends RestClientTestHelper with TestPrematureExit with Bat
val basePath: String = Utils.getCodeSourceLocation(getClass)
val batchFile: String = s"${basePath}/batch.yaml"
+ override protected val otherConfigs: Map[String, String] = {
+ Map(KyuubiConf.BATCH_APPLICATION_CHECK_INTERVAL.key -> "100")
+ }
+
override def beforeAll(): Unit = {
super.beforeAll()
@@ -234,12 +239,12 @@ class BatchCliSuite extends RestClientTestHelper with TestPrematureExit with Bat
"--password",
ldapUserPasswd,
"--waitCompletion",
- "false")
+ "false",
+ "--conf",
+ s"${CtlConf.CTL_BATCH_LOG_QUERY_INTERVAL.key}=100")
val result = testPrematureExitForControlCli(submitArgs, "")
assert(result.contains(s"/bin/spark-submit"))
assert(!result.contains("ShutdownHookManager: Shutdown hook called"))
- val numberOfRows = result.split("\n").length
- assert(numberOfRows <= 100)
}
test("list batch test") {