[KYUUBI #2859][SUB-TASK][KPIP-4] Support --conf for kyuubi-ctl

### _Why are the changes needed?_

To close #2859

### _How was this patch tested?_
- [x] 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 #2875 from turboFei/kyuubi_2859_retry_delete.

Closes #2859

7908d82 [Fei Wang] revert commands related
23d357d [Fei Wang] fix ut
7673001 [Fei Wang] print close resp first
76449da [Fei Wang] support --conf

Authored-by: Fei Wang <fwang12@ebay.com>
Signed-off-by: ulysses-you <ulyssesyou@apache.org>

Author: turboFei

docs/deployment/settings.md

@@ -189,6 +189,8 @@ Key | Default | Meaning | Type | Since
 --- | --- | --- | --- | ---
 kyuubi.ctl.rest.auth.schema|basic|The authentication schema. Valid values are: basic, spnego.|string|1.6.0
 kyuubi.ctl.rest.base.url|<undefined>|The REST API base URL, which contains the scheme (http:// or https://), host name, port number|string|1.6.0
+kyuubi.ctl.rest.request.attempt.wait|PT3S|How long to wait between attempts of ctl rest request.|duration|1.6.0
+kyuubi.ctl.rest.request.max.attempts|3|The max attempts number for ctl rest request.|int|1.6.0
 kyuubi.ctl.rest.spnego.host|<undefined>|When auth schema is spnego, need to config spnego host.|string|1.6.0
 
 

kyuubi-ctl/src/main/scala/org/apache/kyuubi/ctl/CliConfig.scala

@@ -36,7 +36,8 @@ case class CliConfig(
     createOpts: CreateOpts = CreateOpts(),
     logOpts: LogOpts = LogOpts(),
     batchOpts: BatchOpts = BatchOpts(),
-    engineOpts: EngineOpts = EngineOpts())
+    engineOpts: EngineOpts = EngineOpts(),
+    conf: Map[String, String] = Map.empty)
 
 case class CommonOpts(
     zkQuorum: String = null,

kyuubi-ctl/src/main/scala/org/apache/kyuubi/ctl/CommandLine.scala

@@ -18,7 +18,7 @@ package org.apache.kyuubi.ctl
 
 import scopt.{OParser, OParserBuilder}
 
-import org.apache.kyuubi.KYUUBI_VERSION
+import org.apache.kyuubi.{KYUUBI_VERSION, KyuubiException}
 import org.apache.kyuubi.ctl.util.DateTimeUtils._
 
 object CommandLine {
@@ -83,7 +83,15 @@ object CommandLine {
         .text("Password for basic authentication."),
       opt[String]("spnegoHost")
         .action((v, c) => c.copy(commonOpts = c.commonOpts.copy(spnegoHost = v)))
-        .text("Spnego host for spnego authentication."))
+        .text("Spnego host for spnego authentication."),
+      opt[String]("conf")
+        .action((v, c) => {
+          v.split("=", 2).toSeq match {
+            case Seq(k, v) => c.copy(conf = c.conf ++ Map(k -> v))
+            case _ => throw new KyuubiException(s"Kyuubi config without '=': $v")
+          }
+        })
+        .text("Kyuubi config property pair, formatted key=value."))
   }
 
   private def create(builder: OParserBuilder[CliConfig]): OParser[_, CliConfig] = {

kyuubi-ctl/src/main/scala/org/apache/kyuubi/ctl/CtlConf.scala

@@ -17,6 +17,8 @@
 
 package org.apache.kyuubi.ctl
 
+import java.time.Duration
+
 import org.apache.kyuubi.config.{ConfigBuilder, ConfigEntry, KyuubiConf, OptionalConfigEntry}
 
 object CtlConf {
@@ -45,4 +47,17 @@ object CtlConf {
       .stringConf
       .createOptional
 
+  val CTL_REST_CLIENT_REQUEST_MAX_ATTEMPTS =
+    buildConf("kyuubi.ctl.rest.request.max.attempts")
+      .doc("The max attempts number for ctl rest request.")
+      .version("1.6.0")
+      .intConf
+      .createWithDefault(3)
+
+  val CTL_REST_CLIENT_REQUEST_ATTEMPT_WAIT =
+    buildConf("kyuubi.ctl.rest.request.attempt.wait")
+      .doc("How long to wait between attempts of ctl rest request.")
+      .version("1.6.0")
+      .timeConf
+      .createWithDefault(Duration.ofSeconds(3).toMillis)
 }

kyuubi-ctl/src/main/scala/org/apache/kyuubi/ctl/RestClientFactory.scala

@@ -49,6 +49,9 @@ object RestClientFactory {
     val authSchema =
       getRestConfig("authSchema", conf.get(CTL_REST_CLIENT_AUTH_SCHEMA), cliConfig, map)
 
+    val maxAttempts = conf.get(CTL_REST_CLIENT_REQUEST_MAX_ATTEMPTS)
+    val attemptWaitTime = conf.get(CTL_REST_CLIENT_REQUEST_ATTEMPT_WAIT).toInt
+
     var kyuubiRestClient: KyuubiRestClient = null
     authSchema.toLowerCase match {
       case "basic" =>
@@ -59,6 +62,8 @@ object RestClientFactory {
           .authHeaderMethod(KyuubiRestClient.AuthHeaderMethod.BASIC)
           .username(username)
           .password(password)
+          .maxAttempts(maxAttempts)
+          .attemptWaitTime(attemptWaitTime)
           .build()
       case "spnego" =>
         val spnegoHost =
@@ -67,6 +72,8 @@ object RestClientFactory {
           .apiVersion(KyuubiRestClient.ApiVersion.valueOf(version))
           .authHeaderMethod(KyuubiRestClient.AuthHeaderMethod.SPNEGO)
           .spnegoHost(spnegoHost)
+          .maxAttempts(maxAttempts)
+          .attemptWaitTime(attemptWaitTime)
           .build()
       case _ => throw new KyuubiException(s"Unsupported auth schema: $authSchema")
     }

kyuubi-ctl/src/main/scala/org/apache/kyuubi/ctl/cmd/Command.scala

@@ -34,6 +34,10 @@ abstract class Command(cliConfig: CliConfig) extends Logging {
 
   val conf = KyuubiConf().loadFileDefaults()
 
+  cliConfig.conf.foreach { case (key, value) =>
+    conf.set(key, value)
+  }
+
   val verbose = cliConfig.commonOpts.verbose
 
   val normalizedCliConfig: CliConfig = useDefaultPropertyValueIfMissing()

kyuubi-ctl/src/test/scala/org/apache/kyuubi/ctl/ControlCliArgumentsSuite.scala

@@ -353,6 +353,7 @@ class ControlCliArgumentsSuite extends KyuubiFunSuite with TestPrematureExit {
          |  --username <value>       Username for basic authentication.
          |  --password <value>       Password for basic authentication.
          |  --spnegoHost <value>     Spnego host for spnego authentication.
+         |  --conf <value>           Kyuubi config property pair, formatted key=value.
          |
          |Command: create [batch|server] [options]
          |${"\t"}Create a resource.
@@ -439,4 +440,27 @@ class ControlCliArgumentsSuite extends KyuubiFunSuite with TestPrematureExit {
 
     testHelpExit(Array("--help"), helpString)
   }
+
+  test("test kyuubi conf property") {
+    val args = Seq(
+      "delete",
+      "batch",
+      "123",
+      "--conf",
+      s"${CtlConf.CTL_REST_CLIENT_REQUEST_MAX_ATTEMPTS.key}=10")
+    val opArgs = new ControlCliArguments(args)
+    assert(opArgs.cliConfig.action.toString.equalsIgnoreCase("DELETE"))
+    assert(opArgs.cliConfig.resource.toString.equalsIgnoreCase("BATCH"))
+    assert(opArgs.cliConfig.batchOpts.batchId === "123")
+    assert(opArgs.cliConfig.conf ===
+      Map(CtlConf.CTL_REST_CLIENT_REQUEST_MAX_ATTEMPTS.key -> "10"))
+
+    val args2 = Seq(
+      "delete",
+      "batch",
+      "123",
+      "--conf",
+      s"${CtlConf.CTL_REST_CLIENT_REQUEST_MAX_ATTEMPTS.key}")
+    testPrematureExitForControlCliArgs(args2.toArray, "Kyuubi config without '='")
+  }
 }