[KYUUBI #2021] Command OptionParser for launching Hive Backend Engine

### _Why are the changes needed?_

### _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 #2356 from zhaomin1423/option_parser.

Closes #2021

e07abe55 [Min Zhao] fix unit test
a9bc522b [Min Zhao] fix unit test
be628c3c [Min Zhao] [KYUUBI #2021] Command OptionParser for launching Hive Backend Engine
db88bfef [Min Zhao] [KYUUBI #2021] Command OptionParser for launching Hive Backend Engine
b4f50cee [Min Zhao] [KYUUBI #2021] Command OptionParser for launching Hive Backend Engine

Authored-by: Min Zhao <zhaomin1423@163.com>
Signed-off-by: Kent Yao <yao@apache.org>

externals/kyuubi-hive-sql-engine/src/main/scala/org/apache/kyuubi/engine/hive/HiveSQLEngine.scala

@@ -125,6 +125,7 @@ object HiveSQLEngine extends Logging {
   def main(args: Array[String]): Unit = {
     SignalRegister.registerLogger(logger)
     try {
+      Utils.fromCommandLineArgs(args, kyuubiConf)
       startEngine()
     } catch {
       case t: Throwable => currentEngine match {

kyuubi-common/src/main/scala/org/apache/kyuubi/Utils.scala

@@ -31,6 +31,7 @@ import org.apache.commons.lang3.time.DateFormatUtils
 import org.apache.hadoop.security.UserGroupInformation
 import org.apache.hadoop.util.ShutdownHookManager
 
+import org.apache.kyuubi.config.KyuubiConf
 import org.apache.kyuubi.config.internal.Tests.IS_TESTING
 
 object Utils extends Logging {
@@ -272,4 +273,19 @@ object Utils extends Logging {
   def getCodeSourceLocation(clazz: Class[_]): String = {
     new File(clazz.getProtectionDomain.getCodeSource.getLocation.toURI).getPath
   }
+
+  def fromCommandLineArgs(args: Array[String], conf: KyuubiConf): Unit = {
+    require(args.length % 2 == 0, s"Illegal size of arguments.")
+    for (i <- args.indices by 2) {
+      require(
+        args(i) == "--conf",
+        s"Unrecognized main arguments prefix ${args(i)}," +
+          s"the argument format is '--conf k=v'.")
+
+      args(i + 1).split("=", 2).map(_.trim) match {
+        case seq if seq.length == 2 => conf.set(seq.head, seq.last)
+        case _ => throw new IllegalArgumentException(s"Illegal argument: ${args(i + 1)}.")
+      }
+    }
+  }
 }

kyuubi-common/src/test/scala/org/apache/kyuubi/UtilsSuite.scala

@@ -25,6 +25,8 @@ import java.util.Properties
 
 import org.apache.hadoop.security.UserGroupInformation
 
+import org.apache.kyuubi.config.KyuubiConf
+
 class UtilsSuite extends KyuubiFunSuite {
 
   test("build information check") {
@@ -140,4 +142,20 @@ class UtilsSuite extends KyuubiFunSuite {
     val path2 = "/tmp/path2"
     assert(Utils.getAbsolutePathFromWork(path2).toString === path2)
   }
+
+  test("test args parser") {
+    val args = Array[String]("--conf", "k1=v1", "--conf", " k2 = v2")
+    val conf = new KyuubiConf()
+    Utils.fromCommandLineArgs(args, conf)
+    assert(conf.getOption("k1").get == "v1")
+    assert(conf.getOption("k2").get == "v2")
+
+    val args1 = Array[String]("--conf", "k1=v1", "--conf")
+    val exception1 = intercept[IllegalArgumentException](Utils.fromCommandLineArgs(args1, conf))
+    assert(exception1.getMessage.contains("Illegal size of arguments"))
+
+    val args2 = Array[String]("--conf", "k1=v1", "--conf", "a")
+    val exception2 = intercept[IllegalArgumentException](Utils.fromCommandLineArgs(args2, conf))
+    assert(exception2.getMessage.contains("Illegal argument: a"))
+  }
 }

kyuubi-server/src/main/scala/org/apache/kyuubi/engine/hive/HiveProcessBuilder.scala

@@ -61,10 +61,6 @@ class HiveProcessBuilder(
     }
     // -Xmx5g
     // java options
-    for ((k, v) <- conf.getAll) {
-      buffer += s"-D$k=$v"
-    }
-
     buffer += "-cp"
     val classpathEntries = new LinkedHashSet[String]
     // hive engine runtime jar
@@ -99,6 +95,10 @@ class HiveProcessBuilder(
     }
     buffer += classpathEntries.asScala.mkString(File.pathSeparator)
     buffer += mainClass
+    for ((k, v) <- conf.getAll) {
+      buffer += "--conf"
+      buffer += s"$k=$v"
+    }
     buffer.toArray
   }
 

kyuubi-server/src/test/scala/org/apache/kyuubi/engine/hive/HiveProcessBuilderSuite.scala

@@ -29,8 +29,8 @@ class HiveProcessBuilderSuite extends KyuubiFunSuite {
     val commands = builder.toString.split('\n')
     assert(commands.head.endsWith("bin/java"), "wrong exec")
     assert(commands.contains("-Dkyuubi.session.user=kyuubi"))
-    assert(commands.contains("-Dkyuubi.on=off"))
     assert(commands.exists(ss => ss.contains("kyuubi-hive-sql-engine")), "wrong classpath")
+    assert(builder.toString.contains("--conf\nkyuubi.on=off"))
   }
 
   test("default engine memory") {