Allows the user to choose whether to add 'batch_request_args' to the …

…returned object of 'bathc'

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

@@ -3248,4 +3248,13 @@ object KyuubiConf {
       .version("1.8.1")
       .booleanConf
       .createWithDefault(false)
+
+  val BATCH_INFORMATION_RESPONSE_MODEL: ConfigEntry[String] =
+    buildConf("kyuubi.batch.information.response.model")
+      .doc("When get batch object by ID use compact or fat model." +
+        "When use fat model will return additional information. for example the request args.")
+      .version("1.8.1")
+      .stringConf
+      .createWithDefault("compact")
+
 }

kyuubi-server/src/main/scala/org/apache/kyuubi/server/api/v1/BatchesResource.scala

@@ -313,18 +313,19 @@ private[v1] class BatchesResource extends ApiRequestContext with Logging {
   def batchInfo(@PathParam("batchId") batchId: String): Batch = {
     val userName = fe.getSessionUser(Map.empty[String, String])
     val sessionHandle = formatSessionHandle(batchId)
+    var batch2return = new Batch()
     sessionManager.getBatchSession(sessionHandle).map { batchSession =>
-      buildBatch(batchSession)
+      batch2return = buildBatch(batchSession)
     }.getOrElse {
       sessionManager.getBatchMetadata(batchId).map { metadata =>
         if (batchV2Enabled(metadata.requestConf) ||
           OperationState.isTerminal(OperationState.withName(metadata.state)) ||
           metadata.kyuubiInstance == fe.connectionUrl) {
-          MetadataManager.buildBatch(metadata)
+          batch2return = MetadataManager.buildBatch(metadata)
         } else {
           val internalRestClient = getInternalRestClient(metadata.kyuubiInstance)
           try {
-            internalRestClient.getBatch(userName, batchId)
+            batch2return = internalRestClient.getBatch(userName, batchId)
           } catch {
             case e: KyuubiRestException =>
               error(s"Error redirecting get batch[$batchId] to ${metadata.kyuubiInstance}", e)
@@ -334,14 +335,22 @@ private[v1] class BatchesResource extends ApiRequestContext with Logging {
                 Some(userName),
                 // prevent that the batch be marked as terminated if application state is NOT_FOUND
                 Some(metadata.engineOpenTime).filter(_ > 0).orElse(Some(System.currentTimeMillis)))
-              buildBatch(metadata, batchAppStatus)
+              batch2return = buildBatch(metadata, batchAppStatus)
           }
         }
       }.getOrElse {
         error(s"Invalid batchId: $batchId")
         throw new NotFoundException(s"Invalid batchId: $batchId")
       }
     }
+
+    val responseModel = ResponseModels
+      .withName(fe.getConf.get(BATCH_INFORMATION_RESPONSE_MODEL).toUpperCase)
+    responseModel match {
+      case ResponseModels.FAT => return batch2return
+      case _ => batch2return.setArgs(new java.util.ArrayList[String]())
+    }
+    batch2return
   }
 
   @ApiResponse(

kyuubi-server/src/main/scala/org/apache/kyuubi/server/api/v1/ResponseModels.scala

@@ -0,0 +1,24 @@
+/*
+ * 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.server.api.v1
+
+object ResponseModels extends Enumeration {
+  type ResponseModel = Value
+
+  val COMPACT, FAT = Value
+}