[incubator-kyuubi] branch master updated: [KYUUBI #2548] Prevent dead loop if the batch job submission process it not alive

[fe...@apache.org]

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 672e8e95c [KYUUBI #2548] Prevent dead loop if the batch job submission process it not alive
672e8e95c is described below

commit 672e8e95c3bc5a9cf0965d1e8ab2c44a27fc09b7
Author: Fei Wang <fw...@ebay.com>
AuthorDate: Thu May 5 09:51:01 2022 +0800

    [KYUUBI #2548] Prevent dead loop if the batch job submission process it not alive
    
    ### _Why are the changes needed?_
    
    If the batch job submission process is not alive, out of the get applicationStatus loop
    ### _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 #2548 from turboFei/process_yarn.
    
    Closes #2548
    
    0b444f02 [Fei Wang] add ut
    5115effe [Fei Wang] If the process is not alive, break the get application status loop
    
    Authored-by: Fei Wang <fw...@ebay.com>
    Signed-off-by: Fei Wang <fw...@ebay.com>
---
 .../kyuubi/operation/BatchJobSubmission.scala      |  7 ++--
 .../operation/KyuubiBatchYarnClusterSuite.scala    | 43 ++++++++++++++++++++--
 2 files changed, 43 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 30023710d..ce3b81a12 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
@@ -104,12 +104,13 @@ class BatchJobSubmission(session: KyuubiBatchSessionImpl, batchRequest: BatchReq
       info(s"Submitting ${batchRequest.batchType} batch job: $builder")
       val process = builder.start
       var applicationStatus = currentApplicationState
-      while (applicationStatus.isEmpty) {
+      while (applicationStatus.isEmpty && process.isAlive) {
         applicationStatus = currentApplicationState
         Thread.sleep(applicationCheckInterval)
       }
-      val state = applicationStatus.get(ApplicationOperation.APP_STATE_KEY)
-      if (state == "KILLED" || state == "FAILED") {
+
+      if (applicationStatus.map(_.get(ApplicationOperation.APP_STATE_KEY)).filter(s =>
+          s == Some("KILLED") || s == Some("FAILED")).isDefined) {
         process.destroyForcibly()
         throw new RuntimeException("Batch job failed:" + applicationStatus.get.mkString(","))
       } else {
diff --git a/kyuubi-server/src/test/scala/org/apache/kyuubi/operation/KyuubiBatchYarnClusterSuite.scala b/kyuubi-server/src/test/scala/org/apache/kyuubi/operation/KyuubiBatchYarnClusterSuite.scala
index df47de86b..fda38fc36 100644
--- a/kyuubi-server/src/test/scala/org/apache/kyuubi/operation/KyuubiBatchYarnClusterSuite.scala
+++ b/kyuubi-server/src/test/scala/org/apache/kyuubi/operation/KyuubiBatchYarnClusterSuite.scala
@@ -32,9 +32,11 @@ import org.apache.kyuubi.session.{KyuubiBatchSessionImpl, KyuubiSessionManager}
 
 class KyuubiBatchYarnClusterSuite extends WithKyuubiServerOnYarn {
 
+  private val preDefinedAppName = "kyuubi-batch-job"
+
   override protected val conf: KyuubiConf = {
-    new KyuubiConf().set(s"$KYUUBI_BATCH_CONF_PREFIX.spark.spark.master", "yarn")
-      .set(BATCH_CONF_IGNORE_LIST, Seq("spark.master"))
+    new KyuubiConf().set(s"$KYUUBI_BATCH_CONF_PREFIX.spark.spark.app.name", preDefinedAppName)
+      .set(BATCH_CONF_IGNORE_LIST, Seq("spark.app.name"))
   }
 
   private def sessionManager(): KyuubiSessionManager =
@@ -47,9 +49,10 @@ class KyuubiBatchYarnClusterSuite extends WithKyuubiServerOnYarn {
       "spark",
       sparkProcessBuilder.mainResource.get,
       sparkProcessBuilder.mainClass,
-      "spark-batch-submission",
+      null,
       Map(
-        "spark.master" -> "local",
+        "spark.master" -> "yarn",
+        "spark.app.name" -> "customName",
         s"spark.${ENGINE_SPARK_MAX_LIFETIME.key}" -> "5000",
         s"spark.${ENGINE_CHECK_INTERVAL.key}" -> "1000"),
       Seq.empty[String])
@@ -70,6 +73,7 @@ class KyuubiBatchYarnClusterSuite extends WithKyuubiServerOnYarn {
       val state = batchJobSubmissionOp.currentApplicationState
       assert(state.nonEmpty)
       assert(state.exists(_("id").startsWith("application_")))
+      assert(state.exists(_("name") == preDefinedAppName))
     }
 
     val killResponse = yarnOperation.killApplicationByTag(sessionHandle.identifier.toString)
@@ -104,4 +108,35 @@ class KyuubiBatchYarnClusterSuite extends WithKyuubiServerOnYarn {
     assert(appError === state2("error"))
     sessionManager.closeSession(sessionHandle)
   }
+
+  test("prevent dead loop if the batch job submission process it not alive") {
+    val sparkProcessBuilder = new SparkProcessBuilder("kyuubi", conf)
+
+    val batchRequest = BatchRequest(
+      "spark",
+      sparkProcessBuilder.mainResource.get,
+      sparkProcessBuilder.mainClass,
+      "spark-batch-submission",
+      Map(
+        "spark.master" -> "invalid",
+        s"spark.${ENGINE_SPARK_MAX_LIFETIME.key}" -> "5000",
+        s"spark.${ENGINE_CHECK_INTERVAL.key}" -> "1000"),
+      Seq.empty[String])
+
+    val sessionHandle = sessionManager().openBatchSession(
+      TProtocolVersion.HIVE_CLI_SERVICE_PROTOCOL_V1,
+      "kyuubi",
+      "passwd",
+      "localhost",
+      batchRequest.conf,
+      batchRequest)
+
+    val session = sessionManager.getSession(sessionHandle).asInstanceOf[KyuubiBatchSessionImpl]
+    val batchJobSubmissionOp = session.batchJobSubmissionOp
+
+    eventually(timeout(3.minutes), interval(50.milliseconds)) {
+      assert(batchJobSubmissionOp.currentApplicationState.isEmpty)
+      assert(batchJobSubmissionOp.getStatus.state === OperationState.ERROR)
+    }
+  }
 }