From 18c422b6ba4505537295d71a698270da718a2625 Mon Sep 17 00:00:00 2001
From: SteNicholas <programgeek@163.com>
Date: Thu, 10 Mar 2022 15:00:29 +0800
Subject: [PATCH] [KYUUBI #2078] `logCaptureThread` does not catch sparksubmit
 exception

### _Why are the changes needed?_

`logCaptureThread` does not catch sparksubmit exception.

### _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 #2090 from SteNicholas/KYUUBI-2078.

Closes #2078

11ff6824 [SteNicholas] [KYUUBI #2078] logCaptureThread does not catch sparksubmit exception

Authored-by: SteNicholas <programgeek@163.com>
Signed-off-by: Kent Yao <yao@apache.org>
(cherry picked from commit cf014ee282d28e717acb6e243522944e7bbb07cd)
Signed-off-by: Kent Yao <yao@apache.org>
---
 .../scala/org/apache/kyuubi/engine/ProcBuilder.scala  | 11 +++++++----
 .../engine/spark/SparkProcessBuilderSuite.scala       | 10 ++++++++++
 2 files changed, 17 insertions(+), 4 deletions(-)

diff --git a/kyuubi-server/src/main/scala/org/apache/kyuubi/engine/ProcBuilder.scala b/kyuubi-server/src/main/scala/org/apache/kyuubi/engine/ProcBuilder.scala
index bf1ca92c1..eda8cbc38 100644
--- a/kyuubi-server/src/main/scala/org/apache/kyuubi/engine/ProcBuilder.scala
+++ b/kyuubi-server/src/main/scala/org/apache/kyuubi/engine/ProcBuilder.scala
@@ -123,14 +123,14 @@ trait ProcBuilder {
         val maxErrorSize = conf.get(KyuubiConf.ENGINE_ERROR_MAX_SIZE)
         while (true) {
           if (reader.ready()) {
-            var line: String = reader.readLine
-            if (containsIgnoreCase(line, "Exception:") &&
+            var line: String = reader.readLine.trim
+            if (containsException(line) &&
               !line.contains("at ") && !line.startsWith("Caused by:")) {
               val sb = new StringBuilder(line)
               error = KyuubiSQLException(sb.toString() + s"\n See more: $engineLog")
-              line = reader.readLine()
+              line = reader.readLine().trim
               while (sb.length < maxErrorSize && line != null &&
-                (containsIgnoreCase(line, "Exception:") ||
+                (containsException(line) ||
                   line.startsWith("\tat ") ||
                   line.startsWith("Caused by: "))) {
                 sb.append("\n" + line)
@@ -201,6 +201,9 @@ trait ProcBuilder {
       case other => other
     }
   }
+
+  private def containsException(log: String): Boolean =
+    containsIgnoreCase(log, "Exception:") || containsIgnoreCase(log, "Exception in thread")
 }
 
 object ProcBuilder extends Logging {
diff --git a/kyuubi-server/src/test/scala/org/apache/kyuubi/engine/spark/SparkProcessBuilderSuite.scala b/kyuubi-server/src/test/scala/org/apache/kyuubi/engine/spark/SparkProcessBuilderSuite.scala
index d50b75aad..80c6a89f1 100644
--- a/kyuubi-server/src/test/scala/org/apache/kyuubi/engine/spark/SparkProcessBuilderSuite.scala
+++ b/kyuubi-server/src/test/scala/org/apache/kyuubi/engine/spark/SparkProcessBuilderSuite.scala
@@ -95,6 +95,16 @@ class SparkProcessBuilderSuite extends KerberizedTestHelper {
       assert(error1.getMessage.contains("See more: "))
       assert(!error1.getMessage.contains(msg), "stack trace shall be truncated")
     }
+
+    val pb3 =
+      new SparkProcessBuilder("kentyao", conf.set("spark.kerberos.principal", testPrincipal))
+    pb3.start
+    eventually(timeout(90.seconds), interval(500.milliseconds)) {
+      val error1 = pb3.getError
+      assert(!error1.getMessage.contains("Failed to detect the root cause"))
+      assert(error1.getMessage.contains("See more: "))
+      assert(error1.getMessage.contains("Exception in thread"))
+    }
   }
 
   test("proxy user or keytab") {