comments

Author: turboFei

kyuubi-server/src/main/scala/org/apache/kyuubi/engine/ApplicationOperation.scala

@@ -85,6 +85,12 @@ trait ApplicationOperation {
       tag: String,
       proxyUser: Option[String] = None,
       submitTime: Option[Long] = None): ApplicationInfo
+
+  /**
+   * Whether the application state can be persisted and retrieved after finished.
+   * @return true if the application state can be persisted
+   */
+  def supportPersistedAppState: Boolean
 }
 
 object ApplicationState extends Enumeration {
@@ -96,12 +102,7 @@ object ApplicationState extends Enumeration {
       appOperation: Option[ApplicationOperation]): Boolean = state match {
     case FAILED => true
     case KILLED => true
-    case NOT_FOUND => appOperation match {
-        // For YARN and Kubernetes operations, if the application is not found, treat it as failed
-        // to prevent mistakenly set unsuccessful applications to the finished state.
-        case Some(_: YarnApplicationOperation) | Some(_: KubernetesApplicationOperation) => true
-        case _ => false
-      }
+    case NOT_FOUND => appOperation.exists(_.supportPersistedAppState)
     case _ => false
   }
 

kyuubi-server/src/main/scala/org/apache/kyuubi/engine/JpsApplicationOperation.scala

@@ -105,5 +105,7 @@ class JpsApplicationOperation extends ApplicationOperation {
     // TODO check if the process is zombie
   }
 
+  override def supportPersistedAppState: Boolean = false
+
   override def stop(): Unit = {}
 }

kyuubi-server/src/main/scala/org/apache/kyuubi/engine/KubernetesApplicationOperation.scala

@@ -283,6 +283,8 @@ class KubernetesApplicationOperation extends ApplicationOperation with Logging {
     }
   }
 
+  override def supportPersistedAppState: Boolean = true
+
   override def stop(): Unit = {
     enginePodInformers.asScala.foreach { case (_, informer) =>
       Utils.tryLogNonFatalError(informer.stop())

kyuubi-server/src/main/scala/org/apache/kyuubi/engine/YarnApplicationOperation.scala

@@ -165,6 +165,8 @@ class YarnApplicationOperation extends ApplicationOperation with Logging {
     }
   }
 
+  override def supportPersistedAppState: Boolean = true
+
   override def stop(): Unit = adminYarnClient.foreach { yarnClient =>
     Utils.tryLogNonFatalError(yarnClient.stop())
   }

kyuubi-server/src/main/scala/org/apache/kyuubi/operation/BatchJobSubmission.scala

@@ -216,7 +216,7 @@ class BatchJobSubmission(
           case Some(metadata) if metadata.peerInstanceClosed =>
             setState(OperationState.CANCELED)
           case Some(metadata)
-              // in case it has been updated by peer kyuubi instance, see KYUUBI-6278
+              // in case it has been updated by peer kyuubi instance, see KYUUBI #6278
               if StringUtils.isNotBlank(metadata.engineState) &&
                 ApplicationState.isTerminated(ApplicationState.withName(metadata.engineState)) =>
             _applicationInfo = Some(new ApplicationInfo(