[KYUUBI #1872] Make ProcBuilder decouple with YARN

<!--
Thanks for sending a pull request!

Here are some tips for you:
  1. If this is your first time, please read our contributor guidelines: https://kyuubi.readthedocs.io/en/latest/community/contributions.html
  2. If the PR is related to an issue in https://github.com/apache/incubator-kyuubi/issues, add '[KYUUBI #XXXX]' in your PR title, e.g., '[KYUUBI #XXXX] Your PR title ...'.
  3. If the PR is unfinished, add '[WIP]' in your PR title, e.g., '[WIP][KYUUBI #XXXX] Your PR title ...'.
-->

### _Why are the changes needed?_
<!--
Please clarify why the changes are needed. For instance,
  1. If you add a feature, you can talk about the use case of it.
  2. If you fix a bug, you can clarify why it is a bug.
-->

### _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

- [ ] [Run test](https://kyuubi.apache.org/docs/latest/develop_tools/testing.html#running-tests) locally before make a pull request

Closes #1874 from yanghua/KYUUBI-1872.

Closes #1872

e0b1264 [yanghua] [KYUUBI #1872] Make ProcBuilder decouple with YARN

Authored-by: yanghua <yanghua1127@gmail.com>
Signed-off-by: ulysses-you <ulyssesyou@apache.org>

Author: yanghua

kyuubi-server/src/main/scala/org/apache/kyuubi/engine/ProcBuilder.scala

@@ -18,12 +18,10 @@
 package org.apache.kyuubi.engine
 
 import java.io.{File, IOException}
-import java.lang.ProcessBuilder.Redirect
 import java.nio.charset.StandardCharsets
 import java.nio.file.{Files, Path, Paths}
 
 import scala.collection.JavaConverters._
-import scala.util.matching.Regex
 
 import com.google.common.collect.EvictingQueue
 import org.apache.commons.lang3.StringUtils.containsIgnoreCase
@@ -98,7 +96,7 @@ trait ProcBuilder {
   @volatile private var error: Throwable = UNCAUGHT_ERROR
 
   private val engineLogMaxLines = conf.get(KyuubiConf.SESSION_ENGINE_STARTUP_MAX_LOG_LINES)
-  private val lastRowsOfLog: EvictingQueue[String] = EvictingQueue.create(engineLogMaxLines)
+  protected val lastRowsOfLog: EvictingQueue[String] = EvictingQueue.create(engineLogMaxLines)
   // Visible for test
   @volatile private[kyuubi] var logCaptureThreadReleased: Boolean = true
   private var logCaptureThread: Thread = _
@@ -186,28 +184,7 @@ trait ProcBuilder {
     proc
   }
 
-  val YARN_APP_NAME_REGEX: Regex = "application_\\d+_\\d+".r
-
-  def killApplication(line: String = lastRowsOfLog.toArray.mkString("\n")): String =
-    YARN_APP_NAME_REGEX.findFirstIn(line) match {
-      case Some(appId) =>
-        env.get(KyuubiConf.KYUUBI_HOME) match {
-          case Some(kyuubiHome) =>
-            val pb = new ProcessBuilder("/bin/sh", s"$kyuubiHome/bin/stop-application.sh", appId)
-            pb.environment()
-              .putAll(childProcEnv.asJava)
-            pb.redirectError(Redirect.appendTo(engineLog))
-            pb.redirectOutput(Redirect.appendTo(engineLog))
-            val process = pb.start()
-            process.waitFor() match {
-              case id if id != 0 => s"Failed to kill Application $appId, please kill it manually. "
-              case _ => s"Killed Application $appId successfully. "
-            }
-          case None =>
-            s"KYUUBI_HOME is not set! Failed to kill Application $appId, please kill it manually."
-        }
-      case None => ""
-    }
+  def killApplication(line: String = lastRowsOfLog.toArray.mkString("\n")): String = ""
 
   def close(): Unit = {
     if (logCaptureThread != null) {

kyuubi-server/src/main/scala/org/apache/kyuubi/engine/spark/SparkProcessBuilder.scala

@@ -18,10 +18,13 @@
 package org.apache.kyuubi.engine.spark
 
 import java.io.{File, FilenameFilter, IOException}
+import java.lang.ProcessBuilder.Redirect
 import java.net.URI
 import java.nio.file.{Files, Paths}
 
+import scala.collection.JavaConverters._
 import scala.collection.mutable.ArrayBuffer
+import scala.util.matching.Regex
 
 import org.apache.hadoop.security.UserGroupInformation
 
@@ -163,6 +166,8 @@ class SparkProcessBuilder(
 
   override protected def module: String = "kyuubi-spark-sql-engine"
 
+  val YARN_APP_NAME_REGEX: Regex = "application_\\d+_\\d+".r
+
   private def useKeytab(): Boolean = {
     val principal = conf.getOption(PRINCIPAL)
     val keytab = conf.getOption(KEYTAB)
@@ -201,6 +206,27 @@ class SparkProcessBuilder(
     }
   }
 
+  override def killApplication(line: String = lastRowsOfLog.toArray.mkString("\n")): String =
+    YARN_APP_NAME_REGEX.findFirstIn(line) match {
+      case Some(appId) =>
+        env.get(KyuubiConf.KYUUBI_HOME) match {
+          case Some(kyuubiHome) =>
+            val pb = new ProcessBuilder("/bin/sh", s"$kyuubiHome/bin/stop-application.sh", appId)
+            pb.environment()
+              .putAll(childProcEnv.asJava)
+            pb.redirectError(Redirect.appendTo(engineLog))
+            pb.redirectOutput(Redirect.appendTo(engineLog))
+            val process = pb.start()
+            process.waitFor() match {
+              case id if id != 0 => s"Failed to kill Application $appId, please kill it manually. "
+              case _ => s"Killed Application $appId successfully. "
+            }
+          case None =>
+            s"KYUUBI_HOME is not set! Failed to kill Application $appId, please kill it manually."
+        }
+      case None => ""
+    }
+
 }
 
 object SparkProcessBuilder {