[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 868c026 [Cheng Pan] Update kyuubi-server/src/main/scala/org/apache/kyuubi/server/api/v1/BatchesResource.scala 1e44acf [Cheng Pan] Update kyuubi-server/src/main/scala/org/apache/kyuubi/server/api/v1/BatchesResource.scala f0b3a53 [Binjie Yang] Update kyuubi-server/src/main/scala/org/apache/kyuubi/server/api/v1/BatchesResource.scala a238e45 [Binjie Yang] Update kyuubi-server/src/main/scala/org/apache/kyuubi/server/api/v1/BatchesResource.scala 565b938 [zwangsheng] modify comments 86ef016 [zwangsheng] add comment and catch spec case 4c40863 [zwangsheng] fix comments 73f89f0 [zwangsheng] [KYUUBI #5513][Improvement][BATCH] Redirect delete batch request when enabled v2 and current kyuubi not submitted this batch Lead-authored-by: zwangsheng <[email protected]> Co-authored-by: Cheng Pan <[email protected]> Co-authored-by: Binjie Yang <[email protected]> Signed-off-by: Cheng Pan <[email protected]>
apache · Oct 25, 2023 · feb9d16 · feb9d16
1 parent 2b20f7b
commit feb9d16
Showing 1 changed file with 9 additions and 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
@@ -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)