[KYUUBI #2054] [KYUUBI-1819] Support closing Flink SQL engine process

<!--
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.
-->
Support closing Flink SQL engine process.

### _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 #2054 from SteNicholas/KYUUBI-1819.

Closes #2054

992bd0b [SteNicholas] [KYUUBI-1819] Support closing Flink SQL engine process
6956712 [SteNicholas] [KYUUBI-1819] Support closing Flink SQL engine process

Authored-by: SteNicholas <programgeek@163.com>
Signed-off-by: ulysses-you <ulyssesyou@apache.org>

Author: SteNicholas

kyuubi-server/src/main/scala/org/apache/kyuubi/engine/flink/FlinkProcessBuilder.scala

@@ -18,9 +18,12 @@
 package org.apache.kyuubi.engine.flink
 
 import java.io.{File, FilenameFilter}
+import java.lang.ProcessBuilder.Redirect
 import java.net.URI
 import java.nio.file.{Files, Paths}
 
+import scala.collection.JavaConverters._
+
 import com.google.common.annotations.VisibleForTesting
 
 import org.apache.kyuubi._
@@ -99,6 +102,29 @@ class FlinkProcessBuilder(
 
   override protected def commands: Array[String] = Array(executable)
 
+  override def killApplication(line: String = lastRowsOfLog.toArray.mkString("\n")): String = {
+    "Job ID: .*".r.findFirstIn(line) match {
+      case Some(jobIdLine) =>
+        val jobId = jobIdLine.split("Job ID: ")(1).trim
+        env.get("FLINK_HOME") match {
+          case Some(flinkHome) =>
+            val pb = new ProcessBuilder("/bin/sh", s"$flinkHome/bin/flink", "stop", jobId)
+            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 $jobId, please kill it manually. "
+              case _ => s"Killed Application $jobId successfully. "
+            }
+          case None =>
+            s"FLINK_HOME is not set! Failed to kill Application $jobId, please kill it manually."
+        }
+      case None => ""
+    }
+  }
+
   override def toString: String = commands.map {
     case arg if arg.startsWith("--") => s"\\\n\t$arg"
     case arg => arg

kyuubi-server/src/test/scala/org/apache/kyuubi/engine/flink/FlinkProcessBuilderSuite.scala

@@ -28,4 +28,25 @@ class FlinkProcessBuilderSuite extends KyuubiFunSuite {
     val commands = builder.toString.split(' ')
     assert(commands.exists(_ endsWith "flink-sql-engine.sh"))
   }
+
+  test("kill application") {
+    val processBuilder = new FakeFlinkProcessBuilder(conf) {
+      override protected def env: Map[String, String] = Map("FLINK_HOME" -> "")
+    }
+    val exit1 = processBuilder.killApplication(
+      """
+        |[INFO] SQL update statement has been successfully submitted to the cluster:
+        |Job ID: 6b1af540c0c0bb3fcfcad50ac037c862
+        |""".stripMargin)
+    assert(exit1.contains("6b1af540c0c0bb3fcfcad50ac037c862")
+      && !exit1.contains("FLINK_HOME is not set!"))
+
+    val exit2 = processBuilder.killApplication("unknow")
+    assert(exit2.equals(""))
+  }
+}
+
+class FakeFlinkProcessBuilder(config: KyuubiConf)
+  extends FlinkProcessBuilder("fake", config) {
+  override protected def commands: Array[String] = Array("ls")
 }