[KYUUBI #2628][FOLLOWUP] Refine kyuubi-ctl batch commands

### _Why are the changes needed?_

Refine kyuubi-ctl batch commands

### _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 #2889 from turboFei/refine_batch_command.

Closes #2628

56bb7e3 [Fei Wang] print current state
d7be6b7 [Fei Wang] print batch report
3c94fb2 [Fei Wang] set debug for doRequest
7c5fd65 [Fei Wang] fix ut
bcd28a7 [Fei Wang] exit if batch job fail
5d73226 [Fei Wang] refactor
4a2bad8 [Fei Wang] refactor the state
9c98219 [Fei Wang] revert --conf
c2b7008 [Fei Wang] fix ut
10778df [Fei Wang] print close resp first
a453084 [Fei Wang] support --conf

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

Author: turboFei

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

@@ -45,7 +45,8 @@ object RestClientFactory {
       map: JMap[String, Object],
       conf: KyuubiConf): KyuubiRestClient = {
     val version = getApiVersion(map)
-    val hostUrl = getRestConfig("hostUrl", conf.get(CTL_REST_CLIENT_BASE_URL).get, cliConfig, map)
+    val hostUrl =
+      getRestConfig("hostUrl", conf.get(CTL_REST_CLIENT_BASE_URL).orNull, cliConfig, map)
     val authSchema =
       getRestConfig("authSchema", conf.get(CTL_REST_CLIENT_AUTH_SCHEMA), cliConfig, map)
 
@@ -67,7 +68,7 @@ object RestClientFactory {
           .build()
       case "spnego" =>
         val spnegoHost =
-          getRestConfig("spnegoHost", conf.get(CTL_REST_CLIENT_SPNEGO_HOST).get, cliConfig, map)
+          getRestConfig("spnegoHost", conf.get(CTL_REST_CLIENT_SPNEGO_HOST).orNull, cliConfig, map)
         kyuubiRestClient = KyuubiRestClient.builder(hostUrl)
           .apiVersion(KyuubiRestClient.ApiVersion.valueOf(version))
           .authHeaderMethod(KyuubiRestClient.AuthHeaderMethod.SPNEGO)

kyuubi-ctl/src/main/scala/org/apache/kyuubi/ctl/cmd/delete/DeleteBatchCommand.scala

@@ -18,24 +18,32 @@ package org.apache.kyuubi.ctl.cmd.delete
 
 import org.apache.kyuubi.client.BatchRestApi
 import org.apache.kyuubi.client.util.JsonUtil
-import org.apache.kyuubi.ctl.CliConfig
+import org.apache.kyuubi.ctl.{CliConfig, ControlCliException}
 import org.apache.kyuubi.ctl.RestClientFactory.withKyuubiRestClient
 import org.apache.kyuubi.ctl.cmd.Command
+import org.apache.kyuubi.ctl.util.BatchUtil
 
 class DeleteBatchCommand(cliConfig: CliConfig) extends Command(cliConfig) {
-
-  var result: String = null
-
   def validate(): Unit = {}
 
   def run(): Unit = {
     withKyuubiRestClient(normalizedCliConfig, null, conf) { kyuubiRestClient =>
       val batchRestApi: BatchRestApi = new BatchRestApi(kyuubiRestClient)
+      val batchId = normalizedCliConfig.batchOpts.batchId
+
+      val result = batchRestApi.deleteBatch(batchId, normalizedCliConfig.batchOpts.hs2ProxyUser)
 
-      val result = batchRestApi.deleteBatch(
-        normalizedCliConfig.batchOpts.batchId,
-        normalizedCliConfig.batchOpts.hs2ProxyUser)
       info(JsonUtil.toJson(result))
+
+      if (!result.isSuccess) {
+        val batch = batchRestApi.getBatchById(batchId)
+        if (!BatchUtil.isTerminalState(batch.getState)) {
+          error(s"Failed to delete batch $batchId, its current state is ${batch.getState}")
+          throw ControlCliException(1)
+        } else {
+          warn(s"Batch $batchId is already in terminal state ${batch.getState}.")
+        }
+      }
     }
   }
 

kyuubi-ctl/src/main/scala/org/apache/kyuubi/ctl/cmd/log/LogBatchCommand.scala

@@ -23,6 +23,7 @@ import org.apache.kyuubi.client.api.v1.dto.{Batch, OperationLog}
 import org.apache.kyuubi.ctl.CliConfig
 import org.apache.kyuubi.ctl.RestClientFactory.withKyuubiRestClient
 import org.apache.kyuubi.ctl.cmd.Command
+import org.apache.kyuubi.ctl.util.BatchUtil
 
 class LogBatchCommand(cliConfig: CliConfig) extends Command(cliConfig) {
 
@@ -59,8 +60,7 @@ class LogBatchCommand(cliConfig: CliConfig) extends Command(cliConfig) {
           Thread.sleep(DEFAULT_LOG_QUERY_INTERVAL)
 
           batch = batchRestApi.getBatchById(batchId)
-          if (log.getLogRowSet.size() == 0 && batch.getState() != "PENDING"
-            && batch.getState() != "RUNNING") {
+          if (log.getLogRowSet.size() == 0 && BatchUtil.isTerminalState(batch.getState)) {
             done = true
           }
         }

kyuubi-ctl/src/main/scala/org/apache/kyuubi/ctl/cmd/submit/SubmitBatchCommand.scala

@@ -22,10 +22,11 @@ import scala.collection.JavaConverters._
 
 import org.apache.kyuubi.client.BatchRestApi
 import org.apache.kyuubi.client.api.v1.dto.{Batch, BatchRequest, OperationLog}
-import org.apache.kyuubi.ctl.CliConfig
+import org.apache.kyuubi.client.util.JsonUtil
+import org.apache.kyuubi.ctl.{CliConfig, ControlCliException}
 import org.apache.kyuubi.ctl.RestClientFactory.withKyuubiRestClient
 import org.apache.kyuubi.ctl.cmd.Command
-import org.apache.kyuubi.ctl.util.{CtlUtils, Validator}
+import org.apache.kyuubi.ctl.util.{BatchUtil, CtlUtils, Validator}
 
 class SubmitBatchCommand(cliConfig: CliConfig) extends Command(cliConfig) {
 
@@ -65,12 +66,17 @@ class SubmitBatchCommand(cliConfig: CliConfig) extends Command(cliConfig) {
         Thread.sleep(DEFAULT_LOG_QUERY_INTERVAL)
 
         batch = batchRestApi.getBatchById(batchId)
-        if (log.getLogRowSet.size() == 0 && batch.getState() != "PENDING"
-          && batch.getState() != "RUNNING") {
+        if (log.getLogRowSet.size() == 0 && BatchUtil.isTerminalState(batch.getState)) {
           done = true
         }
       }
+
+      if (BatchUtil.isTerminalState(batch.getState) && !BatchUtil.isFinishedState(batch.getState)) {
+        error(s"Batch $batchId failed: ${JsonUtil.toJson(batch)}")
+        throw ControlCliException(1)
+      } else {
+        info(s"Batch report for $batchId: ${JsonUtil.toJson(batch)}")
+      }
     }
   }
-
 }

kyuubi-ctl/src/main/scala/org/apache/kyuubi/ctl/util/BatchUtil.scala

@@ -0,0 +1,45 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.kyuubi.ctl.util
+
+import java.util.Locale
+
+object BatchUtil {
+  private val PENDING_STATE = "PENDING"
+  private val RUNNING_STATE = "RUNNING"
+  private val FINISHED_STATE = "FINISHED"
+  private val ERROR_STATE = "ERROR"
+  private val CANCELED_STATE = "CANCELED"
+  private val terminalBatchStates = Seq(FINISHED_STATE, ERROR_STATE, CANCELED_STATE)
+
+  def isPendingState(state: String): Boolean = {
+    PENDING_STATE.equalsIgnoreCase(state)
+  }
+
+  def isRunningState(state: String): Boolean = {
+    RUNNING_STATE.equalsIgnoreCase(state)
+  }
+
+  def isFinishedState(state: String): Boolean = {
+    FINISHED_STATE.equalsIgnoreCase(state)
+  }
+
+  def isTerminalState(state: String): Boolean = {
+    state != null && terminalBatchStates.contains(state.toUpperCase(Locale.ROOT))
+  }
+}

kyuubi-rest-client/src/main/java/org/apache/kyuubi/client/RestClient.java

@@ -111,7 +111,7 @@ private String doRequest(URI uri, String authHeader, RequestBuilder requestBuild
               .setHeader(HttpHeaders.CONTENT_TYPE, "application/json")
               .build();
 
-      LOG.info("Executing {} request: {}", httpRequest.getMethod(), uri);
+      LOG.debug("Executing {} request: {}", httpRequest.getMethod(), uri);
 
       ResponseHandler<String> responseHandler =
           resp -> {
@@ -126,7 +126,7 @@ private String doRequest(URI uri, String authHeader, RequestBuilder requestBuild
           };
 
       response = httpclient.execute(httpRequest, responseHandler);
-      LOG.info("Response: {}", response);
+      LOG.debug("Response: {}", response);
     } catch (ConnectException | ConnectTimeoutException e) {
       // net exception can be retried by connecting to other Kyuubi server
       throw new KyuubiRetryableException("Api request failed for " + uri.toString(), e);

kyuubi-rest-client/src/main/java/org/apache/kyuubi/client/util/JsonUtil.java

@@ -28,15 +28,17 @@ public static String toJson(Object object) {
     try {
       return MAPPER.writeValueAsString(object);
     } catch (Exception e) {
-      throw new KyuubiRestException("Failed to convert object to json", e);
+      throw new KyuubiRestException(
+          String.format("Failed to convert object(%s) to json", object), e);
     }
   }
 
   public static <T> T toObject(String json, Class<T> clazz) {
     try {
       return MAPPER.readValue(json, clazz);
     } catch (Exception e) {
-      throw new KyuubiRestException("Failed to convert json string to object", e);
+      throw new KyuubiRestException(
+          String.format("Failed to convert json string(%s) to %s", json, clazz.getName()), e);
     }
   }
 }