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

[ch...@apache.org]

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

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


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

commit c4e3c1ff3df0a5793cd47c72b0e221cf14e5196b
Author: zwangsheng <bi...@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 <bi...@apache.org>
    Co-authored-by: Cheng Pan <pa...@gmail.com>
    Co-authored-by: Binjie Yang <bi...@apache.org>
    Signed-off-by: Cheng Pan <ch...@apache.org>
    (cherry picked from commit feb9d164f68875a178d7eaa993d7218b257c951a)
    Signed-off-by: Cheng Pan <ch...@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)