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/09 10:47:35 UTC
[incubator-kyuubi] branch branch-1.6 updated (3c855c4d6 -> 4617af613)
This is an automated email from the ASF dual-hosted git repository.
feiwang pushed a change to branch branch-1.6
in repository https://gitbox.apache.org/repos/asf/incubator-kyuubi.git
discard 3c855c4d6 [KYUUBI #3950] Fix the batch metadata in-consistent issue on open batch session failure
discard 97247e6d0 [KYUUBI #3922] Only the ApplicationInfo with non-empty id is valid for BatchJobSubmission
new 1288c2d04 [KYUUBI #3922] Only the ApplicationInfo with non-empty id is valid for BatchJobSubmission
new 4617af613 [KYUUBI #3950] Fix the batch metadata in-consistent issue on open batch session failure
This update added new revisions after undoing existing revisions.
That is to say, some revisions that were in the old version of the
branch are not in the new version. This situation occurs
when a user --force pushes a change and generates a repository
containing something like this:
* -- * -- B -- O -- O -- O (3c855c4d6)
\
N -- N -- N refs/heads/branch-1.6 (4617af613)
You should already have received notification emails for all of the O
revisions, and so the following emails describe only the N revisions
from the common base, B.
Any revisions marked "omit" are not gone; other references still
refer to them. Any revisions marked "discard" are gone forever.
The 2 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../src/test/scala/org/apache/kyuubi/RestClientTestHelper.scala | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
[incubator-kyuubi] 02/02: [KYUUBI #3950] Fix the batch metadata in-consistent issue on open batch session failure
Posted by fe...@apache.org.
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
commit 4617af6134c78a4bda7ea753751e4f5fb11279e0
Author: fwang12 <fw...@ebay.com>
AuthorDate: Fri Dec 9 18:39:38 2022 +0800
[KYUUBI #3950] Fix the batch metadata in-consistent issue on open batch session failure
### _Why are the changes needed?_
I found that, there is some confused batch metadata.
The metadata state in mysql is PENDING, but the batch session state is `ERROR`.
I think the RC is that:
- the metadata insert is in KyuubiBatchSessionImpl
- If open batch session failed, the BatchJobSubmission opHandle has not been added into batch session operation set.
- then the close session will not update the batch metadata
In this pr, I re-order the code when opening batch session.
### _How was this patch tested?_
- [ ] 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 #3950 from turboFei/update_batch_state.
Closes #3950
49a8e073c [fwang12] try to kill
fd495bcd7 [fwang12] fix ut
6d3bd69f3 [fwang12] add ut
bb2bf89a7 [fwang12] add more log
0c9f32b68 [fwang12] check recovery metadata if failed before run batch op
23f8fef29 [fwang12] comments
e45b1924c [fwang12] re-order
0f0ad8422 [fwang12] set launchEngineOp to lazy to prevent the OP CONN metrics failure
0eed14dff [fwang12] add batch op
c39da31a2 [fwang12] Revert "insert metadata when run operation"
61b40c468 [fwang12] insert metadata when run operation
Authored-by: fwang12 <fw...@ebay.com>
Signed-off-by: fwang12 <fw...@ebay.com>
(cherry picked from commit e86df1acbb3055489147fadb779120da1db35444)
Signed-off-by: fwang12 <fw...@ebay.com>
---
.../kyuubi/operation/BatchJobSubmission.scala | 11 +++++++++
.../kyuubi/session/KyuubiBatchSessionImpl.scala | 3 ++-
.../kyuubi/session/KyuubiSessionManager.scala | 7 +++---
.../server/api/v1/BatchesResourceSuite.scala | 28 +++++++++++++++++++---
4 files changed, 42 insertions(+), 7 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 6936cd5ed..d56ac1959 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
@@ -314,6 +314,17 @@ class BatchJobSubmission(
}
MetricsSystem.tracing(_.decCount(MetricRegistry.name(OPERATION_OPEN, opType)))
+
+ if (!builder.processLaunched) {
+ builder.close()
+ if (recoveryMetadata.isDefined) {
+ killMessage = killBatchApplication()
+ }
+ setState(OperationState.CANCELED)
+ updateBatchMetadata()
+ return
+ }
+
// fast fail
if (isTerminalState(state)) {
killMessage = (false, s"batch $batchId is already terminal so can not kill it.")
diff --git a/kyuubi-server/src/main/scala/org/apache/kyuubi/session/KyuubiBatchSessionImpl.scala b/kyuubi-server/src/main/scala/org/apache/kyuubi/session/KyuubiBatchSessionImpl.scala
index dd2afc9f1..5773198db 100644
--- a/kyuubi-server/src/main/scala/org/apache/kyuubi/session/KyuubiBatchSessionImpl.scala
+++ b/kyuubi-server/src/main/scala/org/apache/kyuubi/session/KyuubiBatchSessionImpl.scala
@@ -139,7 +139,7 @@ class KyuubiBatchSessionImpl(
checkSessionAccessPathURIs()
- // we should call super.open before running batch job submission operation
+ // create the operation root directory before running batch job submission operation
super.open()
runOperation(batchJobSubmissionOp)
@@ -147,6 +147,7 @@ class KyuubiBatchSessionImpl(
override def close(): Unit = {
super.close()
+ batchJobSubmissionOp.close()
waitMetadataRequestsRetryCompletion()
sessionEvent.endTime = System.currentTimeMillis()
EventBus.post(sessionEvent)
diff --git a/kyuubi-server/src/main/scala/org/apache/kyuubi/session/KyuubiSessionManager.scala b/kyuubi-server/src/main/scala/org/apache/kyuubi/session/KyuubiSessionManager.scala
index ff0510199..54b4faf35 100644
--- a/kyuubi-server/src/main/scala/org/apache/kyuubi/session/KyuubiSessionManager.scala
+++ b/kyuubi-server/src/main/scala/org/apache/kyuubi/session/KyuubiSessionManager.scala
@@ -136,8 +136,8 @@ class KyuubiSessionManager private (name: String) extends SessionManager(name) {
private[kyuubi] def openBatchSession(batchSession: KyuubiBatchSessionImpl): SessionHandle = {
val user = batchSession.user
val ipAddress = batchSession.ipAddress
+ val handle = batchSession.handle
try {
- val handle = batchSession.handle
batchSession.open()
setSession(handle, batchSession)
info(s"$user's batch session with $handle is opened, current opening sessions" +
@@ -149,14 +149,15 @@ class KyuubiSessionManager private (name: String) extends SessionManager(name) {
batchSession.close()
} catch {
case t: Throwable =>
- warn(s"Error closing batch session for $user client ip: $ipAddress", t)
+ warn(s"Error closing batch session[$handle] for $user client ip: $ipAddress", t)
}
MetricsSystem.tracing { ms =>
ms.incCount(CONN_FAIL)
ms.incCount(MetricRegistry.name(CONN_FAIL, user))
}
throw KyuubiSQLException(
- s"Error opening batch session for $user client ip $ipAddress, due to ${e.getMessage}",
+ s"Error opening batch session[$handle] for $user client ip $ipAddress," +
+ s" due to ${e.getMessage}",
e)
}
}
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 05d70d411..e8ea61513 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
@@ -18,6 +18,7 @@
package org.apache.kyuubi.server.api.v1
import java.net.InetAddress
+import java.nio.file.Paths
import java.util.{Base64, UUID}
import javax.ws.rs.client.Entity
import javax.ws.rs.core.MediaType
@@ -49,6 +50,9 @@ class BatchesResourceSuite extends KyuubiFunSuite with RestFrontendTestHelper wi
.set(
KyuubiConf.ENGINE_SECURITY_SECRET_PROVIDER,
classOf[UserDefinedEngineSecuritySecretProvider].getName)
+ .set(
+ KyuubiConf.SESSION_LOCAL_DIR_ALLOW_LIST,
+ Seq(Paths.get(sparkBatchTestResource.get).getParent.toString))
override def afterEach(): Unit = {
val sessionManager = fe.be.sessionManager.asInstanceOf[KyuubiSessionManager]
@@ -213,7 +217,7 @@ class BatchesResourceSuite extends KyuubiFunSuite with RestFrontendTestHelper wi
Map.empty,
newBatchRequest(
"spark",
- "",
+ sparkBatchTestResource.get,
"",
""))
sessionManager.openSession(
@@ -235,7 +239,7 @@ class BatchesResourceSuite extends KyuubiFunSuite with RestFrontendTestHelper wi
Map.empty,
newBatchRequest(
"spark",
- "",
+ sparkBatchTestResource.get,
"",
""))
sessionManager.openBatchSession(
@@ -245,7 +249,7 @@ class BatchesResourceSuite extends KyuubiFunSuite with RestFrontendTestHelper wi
Map.empty,
newBatchRequest(
"spark",
- "",
+ sparkBatchTestResource.get,
"",
""))
@@ -622,4 +626,22 @@ class BatchesResourceSuite extends KyuubiFunSuite with RestFrontendTestHelper wi
val counterName = s"${MetricsConstants.OPERATION_STATE}.$opType.${state.toString.toLowerCase}"
MetricsSystem.meterValue(counterName).getOrElse(0L)
}
+
+ test("the batch session should be consistent on open session failure") {
+ val sessionManager = server.frontendServices.head
+ .be.sessionManager.asInstanceOf[KyuubiSessionManager]
+
+ val e = intercept[Exception] {
+ sessionManager.openBatchSession(
+ "kyuubi",
+ "kyuubi",
+ InetAddress.getLocalHost.getCanonicalHostName,
+ Map.empty,
+ newSparkBatchRequest(Map("spark.jars" -> "disAllowPath")))
+ }
+ val sessionHandleRegex = "\\[[\\S]*\\]".r
+ val batchId = sessionHandleRegex.findFirstMatchIn(e.getMessage).get.group(0)
+ .replaceAll("\\[", "").replaceAll("\\]", "")
+ assert(sessionManager.getBatchMetadata(batchId).state == "CANCELED")
+ }
}
[incubator-kyuubi] 01/02: [KYUUBI #3922] Only the ApplicationInfo with non-empty id is valid for BatchJobSubmission
Posted by fe...@apache.org.
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
commit 1288c2d047268a115d6851f570e257b2013e5a93
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
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)
}
```
- [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>
(cherry picked from commit 8eac51385b42c42a29d34e2208d512a7bc2d86bf)
Signed-off-by: fwang12 <fw...@ebay.com>
---
.../scala/org/apache/kyuubi/operation/BatchJobSubmission.scala | 10 +++++++++-
.../test/scala/org/apache/kyuubi/RestClientTestHelper.scala | 1 +
.../test/scala/org/apache/kyuubi/WithKyuubiServerOnYarn.scala | 9 +++++++--
.../org/apache/kyuubi/server/api/v1/BatchesResourceSuite.scala | 2 +-
.../org/apache/kyuubi/server/rest/client/BatchCliSuite.scala | 8 ++++----
5 files changed, 22 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 bcaca6237..6936cd5ed 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
@@ -99,7 +99,8 @@ class BatchJobSubmission(
}
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 = {
@@ -117,6 +118,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/RestClientTestHelper.scala b/kyuubi-server/src/test/scala/org/apache/kyuubi/RestClientTestHelper.scala
index d15993797..1d3281736 100644
--- a/kyuubi-server/src/test/scala/org/apache/kyuubi/RestClientTestHelper.scala
+++ b/kyuubi-server/src/test/scala/org/apache/kyuubi/RestClientTestHelper.scala
@@ -56,6 +56,7 @@ trait RestClientTestHelper extends RestFrontendTestHelper with KerberizedTestHel
.set(
KyuubiConf.AUTHENTICATION_CUSTOM_CLASS,
classOf[UserDefineAuthenticationProviderImpl].getCanonicalName)
+ .set(KyuubiConf.BATCH_APPLICATION_CHECK_INTERVAL, 100L)
}
}
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 efc53375d..05d70d411 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
@@ -436,7 +436,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..18e583e9a 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,7 @@ 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.ctl.{CtlConf, TestPrematureExit}
import org.apache.kyuubi.metrics.{MetricsConstants, MetricsSystem}
import org.apache.kyuubi.session.KyuubiSessionManager
@@ -234,12 +234,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") {