[kyuubi] branch master updated: [KYUUBI #5513][BATCH] Always redirect delete batch request to Kyuubi instance that owns batch session

[chengpan]

This is an automated email from the ASF dual-hosted git repository.

chengpan pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/kyuubi.git


The following commit(s) were added to refs/heads/master by this push:
     new feb9d164f [KYUUBI #5513][BATCH] Always redirect delete batch request 
to Kyuubi instance that owns batch session
feb9d164f is described below

commit feb9d164f68875a178d7eaa993d7218b257c951a
Author: zwangsheng <binjiey...@apache.org>
AuthorDate: Wed Oct 25 21:18:39 2023 +0800

    [KYUUBI #5513][BATCH] Always redirect delete batch request to Kyuubi 
instance that owns batch session
    
    ### _Why are the changes needed?_
    
    Fixed an issue where spark submit(BatchJobSubmission) could not be closed 
when invoking the delete interface to delete a batch in the process of 
submission with batch v2 feature.
    
    ### _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.readthedocs.io/en/master/contributing/code/testing.html#running-tests)
 locally before make a pull request
    
    - [x] Run exist unit test.
    
    ### _Was this patch authored or co-authored using generative AI tooling?_
    No
    
    Closes #5514 from zwangsheng/KYUUBI#5513.
    
    Closes #5513
    
    868c02638 [Cheng Pan] Update 
kyuubi-server/src/main/scala/org/apache/kyuubi/server/api/v1/BatchesResource.scala
    1e44acf89 [Cheng Pan] Update 
kyuubi-server/src/main/scala/org/apache/kyuubi/server/api/v1/BatchesResource.scala
    f0b3a53db [Binjie Yang] Update 
kyuubi-server/src/main/scala/org/apache/kyuubi/server/api/v1/BatchesResource.scala
    a238e4520 [Binjie Yang] Update 
kyuubi-server/src/main/scala/org/apache/kyuubi/server/api/v1/BatchesResource.scala
    565b93857 [zwangsheng] modify comments
    86ef016bb [zwangsheng] add comment and catch spec case
    4c4086327 [zwangsheng] fix comments
    73f89f051 [zwangsheng] [KYUUBI #5513][Improvement][BATCH] Redirect delete 
batch request when enabled v2 and current kyuubi not submitted this batch
    
    Lead-authored-by: zwangsheng <binjiey...@apache.org>
    Co-authored-by: Cheng Pan <pan3...@gmail.com>
    Co-authored-by: Binjie Yang <binjiey...@apache.org>
    Signed-off-by: Cheng Pan <cheng...@apache.org>
---
 .../apache/kyuubi/server/api/v1/BatchesResource.scala | 19 +++++++++----------
 1 file changed, 9 insertions(+), 10 deletions(-)

diff --git 
a/kyuubi-server/src/main/scala/org/apache/kyuubi/server/api/v1/BatchesResource.scala
 
b/kyuubi-server/src/main/scala/org/apache/kyuubi/server/api/v1/BatchesResource.scala
index 76d913a98..c0a3b0ed9 100644
--- 
a/kyuubi-server/src/main/scala/org/apache/kyuubi/server/api/v1/BatchesResource.scala
+++ 
b/kyuubi-server/src/main/scala/org/apache/kyuubi/server/api/v1/BatchesResource.scala
@@ -477,16 +477,15 @@ private[v1] class BatchesResource extends 
ApiRequestContext with Logging {
         checkPermission(userName, metadata.username)
         if 
(OperationState.isTerminal(OperationState.withName(metadata.state))) {
           new CloseBatchResponse(false, s"The batch[$metadata] has been 
terminated.")
-        } else if (batchV2Enabled(metadata.requestConf) && metadata.state == 
"INITIALIZED") {
-          if (batchService.get.cancelUnscheduledBatch(batchId)) {
-            new CloseBatchResponse(true, s"Unscheduled batch $batchId is 
canceled.")
-          } else if 
(OperationState.isTerminal(OperationState.withName(metadata.state))) {
-            new CloseBatchResponse(false, s"The batch[$metadata] has been 
terminated.")
-          } else {
-            info(s"Cancel batch[$batchId] with state ${metadata.state} by 
killing application")
-            val (killed, msg) = forceKill(metadata.appMgrInfo, batchId, 
userName)
-            new CloseBatchResponse(killed, msg)
-          }
+        } else if (batchV2Enabled(metadata.requestConf) && metadata.state == 
"INITIALIZED" &&
+          // there is a chance that metadata is outdated, then 
`cancelUnscheduledBatch` fails
+          // and returns false
+          batchService.get.cancelUnscheduledBatch(batchId)) {
+          new CloseBatchResponse(true, s"Unscheduled batch $batchId is 
canceled.")
+        } else if (batchV2Enabled(metadata.requestConf) && 
metadata.kyuubiInstance == null) {
+          // code goes here indicates metadata is outdated, recursively calls 
itself to refresh
+          // the metadata
+          closeBatchSession(batchId, hs2ProxyUser)
         } else if (metadata.kyuubiInstance != fe.connectionUrl) {
           info(s"Redirecting delete batch[$batchId] to 
${metadata.kyuubiInstance}")
           val internalRestClient = 
getInternalRestClient(metadata.kyuubiInstance)