[KYUUBI #1796] Destroy the process after the engine starts

<!--
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.
-->

Destroy the process after the engine starts. #1796

### _How was this patch tested?_
- [ ] Add some test cases that check the changes thoroughly including negative and positive cases if possible

- [X] Add screenshots for manual tests if appropriate
add conf: `kyuubi.session.engine.startup.waitCompletion  false`
![image](https://user-images.githubusercontent.com/17894939/150467332-6d30a316-494a-4d1a-8c79-a7c9479b16cb.png)

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

Closes #1815 from wForget/KYUUBI-1796.

Closes #1796

12aea19 [Wang Zhen] fix
27cb602 [Wang Zhen] fix
e70e707 [Wang Zhen] add log
ecaf828 [Wang Zhen] regenerate settings.md
a5f6c5d [Wang Zhen] fix desc
09812be [Wang Zhen] Merge remote-tracking branch 'origin/master' into KYUUBI-1796
27fab5e [Wang Zhen] [KYUUBI-1796] Destroy the process after the engine starts

Authored-by: Wang Zhen <wangzhen07@qiyi.com>
Signed-off-by: Cheng Pan <chengpan@apache.org>

Author: wForget

docs/deployment/settings.md

@@ -328,6 +328,7 @@ Key | Default | Meaning | Type | Since
 <code>kyuubi.session.engine.spark.main.resource</code>|<div style='width: 65pt;word-wrap: break-word;white-space: normal'>&lt;undefined&gt;</div>|<div style='width: 170pt;word-wrap: break-word;white-space: normal'>The package used to create Spark SQL engine remote application. If it is undefined, Kyuubi will use the default</div>|<div style='width: 30pt'>string</div>|<div style='width: 20pt'>1.0.0</div>
 <code>kyuubi.session.engine.startup.error.max.size</code>|<div style='width: 65pt;word-wrap: break-word;white-space: normal'>8192</div>|<div style='width: 170pt;word-wrap: break-word;white-space: normal'>During engine bootstrapping, if error occurs, using this config to limit the length error message(characters).</div>|<div style='width: 30pt'>int</div>|<div style='width: 20pt'>1.1.0</div>
 <code>kyuubi.session.engine.startup.maxLogLines</code>|<div style='width: 65pt;word-wrap: break-word;white-space: normal'>10</div>|<div style='width: 170pt;word-wrap: break-word;white-space: normal'>The maximum number of engine log lines when errors occur during engine startup phase. Note that this max lines is for client-side to help track engine startup issue.</div>|<div style='width: 30pt'>int</div>|<div style='width: 20pt'>1.4.0</div>
+<code>kyuubi.session.engine.startup.waitCompletion</code>|<div style='width: 65pt;word-wrap: break-word;white-space: normal'>true</div>|<div style='width: 170pt;word-wrap: break-word;white-space: normal'>Whether to wait for completion after engine starts. If false, the startup process will be destroyed after the engine is started. Note that only use it when the driver is not running locally, such as yarn-cluster mode; Otherwise, the engine will be killed.</div>|<div style='width: 30pt'>boolean</div>|<div style='width: 20pt'>1.5.0</div>
 <code>kyuubi.session.engine.trino.connection.catalog</code>|<div style='width: 65pt;word-wrap: break-word;white-space: normal'>&lt;undefined&gt;</div>|<div style='width: 170pt;word-wrap: break-word;white-space: normal'>The default catalog that trino engine will connect to</div>|<div style='width: 30pt'>string</div>|<div style='width: 20pt'>1.5.0</div>
 <code>kyuubi.session.engine.trino.connection.url</code>|<div style='width: 65pt;word-wrap: break-word;white-space: normal'>&lt;undefined&gt;</div>|<div style='width: 170pt;word-wrap: break-word;white-space: normal'>The server url that trino engine will connect to</div>|<div style='width: 30pt'>string</div>|<div style='width: 20pt'>1.5.0</div>
 <code>kyuubi.session.engine.trino.main.resource</code>|<div style='width: 65pt;word-wrap: break-word;white-space: normal'>&lt;undefined&gt;</div>|<div style='width: 170pt;word-wrap: break-word;white-space: normal'>The package used to create Trino engine remote job. If it is undefined, Kyuubi will use the default</div>|<div style='width: 30pt'>string</div>|<div style='width: 20pt'>1.5.0</div>

kyuubi-common/src/main/scala/org/apache/kyuubi/config/KyuubiConf.scala

@@ -703,6 +703,16 @@ object KyuubiConf {
       .checkValue(_ > 0, "the maximum must be positive integer.")
       .createWithDefault(10)
 
+  val SESSION_ENGINE_STARTUP_WAIT_COMPLETION: ConfigEntry[Boolean] =
+    buildConf("session.engine.startup.waitCompletion")
+      .doc("Whether to wait for completion after engine starts." +
+        " If false, the startup process will be destroyed after the engine is started." +
+        " Note that only use it when the driver is not running locally," +
+        " such as yarn-cluster mode; Otherwise, the engine will be killed.")
+      .version("1.5.0")
+      .booleanConf
+      .createWithDefault(true)
+
   val SESSION_ENGINE_LAUNCH_ASYNC: ConfigEntry[Boolean] =
     buildConf("session.engine.launch.async")
       .doc("When opening kyuubi session, whether to launch backend engine asynchronously." +

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

@@ -96,10 +96,12 @@ trait ProcBuilder {
   @volatile private var error: Throwable = UNCAUGHT_ERROR
 
   private val engineLogMaxLines = conf.get(KyuubiConf.SESSION_ENGINE_STARTUP_MAX_LOG_LINES)
+  private val waitCompletion = conf.get(KyuubiConf.SESSION_ENGINE_STARTUP_WAIT_COMPLETION)
   protected val lastRowsOfLog: EvictingQueue[String] = EvictingQueue.create(engineLogMaxLines)
   // Visible for test
   @volatile private[kyuubi] var logCaptureThreadReleased: Boolean = true
   private var logCaptureThread: Thread = _
+  private var process: Process = _
 
   private[kyuubi] lazy val engineLog: File = ProcBuilder.synchronized {
     val engineLogTimeout = conf.get(KyuubiConf.ENGINE_LOG_TIMEOUT)
@@ -138,8 +140,7 @@ trait ProcBuilder {
   }
 
   final def start: Process = synchronized {
-
-    val proc = processBuilder.start()
+    process = processBuilder.start()
     val reader = Files.newBufferedReader(engineLog.toPath, StandardCharsets.UTF_8)
 
     val redirect: Runnable = { () =>
@@ -181,14 +182,20 @@ trait ProcBuilder {
     logCaptureThreadReleased = false
     logCaptureThread = PROC_BUILD_LOGGER.newThread(redirect)
     logCaptureThread.start()
-    proc
+    process
   }
 
   def killApplication(line: String = lastRowsOfLog.toArray.mkString("\n")): String = ""
 
-  def close(): Unit = {
+  def close(): Unit = synchronized {
     if (logCaptureThread != null) {
       logCaptureThread.interrupt()
+      logCaptureThread = null
+    }
+    if (!waitCompletion && process != null) {
+      info("Destroy the process, since waitCompletion is false.")
+      process.destroyForcibly()
+      process = null
     }
   }