app

Author: turboFei

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

@@ -252,8 +252,13 @@ class KubernetesApplicationOperation extends ApplicationOperation with Logging {
     try {
       // need to initialize the kubernetes client if not exists
       getOrCreateKubernetesClient(appMgrInfo.kubernetesInfo)
-      val (_, appInfo) =
-        appInfoStore.getOrDefault(tag, appMgrInfo.kubernetesInfo -> ApplicationInfo.NOT_FOUND)
+      val appInfo = appInfoStore.get(tag) match {
+        case (_, info) => info
+        case _ =>
+          // try to get the application info from kubernetes metadata
+          metadataManager.flatMap(
+            _.getKubernetesApplicationInfo(tag)).getOrElse(ApplicationInfo.NOT_FOUND)
+      }
       (appInfo.state, submitTime) match {
         // Kyuubi should wait second if pod is not be created
         case (NOT_FOUND, Some(_submitTime)) =>

kyuubi-server/src/main/scala/org/apache/kyuubi/server/metadata/MetadataManager.scala

@@ -26,6 +26,7 @@ import org.apache.kyuubi.{KyuubiException, Logging}
 import org.apache.kyuubi.client.api.v1.dto.Batch
 import org.apache.kyuubi.config.KyuubiConf
 import org.apache.kyuubi.config.KyuubiConf.METADATA_MAX_AGE
+import org.apache.kyuubi.engine.{ApplicationInfo, ApplicationState}
 import org.apache.kyuubi.metrics.{MetricsConstants, MetricsSystem}
 import org.apache.kyuubi.operation.OperationState
 import org.apache.kyuubi.server.metadata.api.{KubernetesMetadata, Metadata, MetadataFilter}
@@ -134,6 +135,11 @@ class MetadataManager extends AbstractService("MetadataManager") {
       .filter(_.sessionType == SessionType.BATCH)
   }
 
+  def getKubernetesApplicationInfo(identifier: String): Option[ApplicationInfo] = {
+    Option(withMetadataRequestMetrics(
+      _metadataStore.getKubernetesMetadata(identifier))).map(buildApplicationInfo)
+  }
+
   def getBatches(
       filter: MetadataFilter,
       from: Int,
@@ -374,4 +380,12 @@ object MetadataManager extends Logging {
       batchMetadata.endTime,
       Map.empty[String, String].asJava)
   }
+
+  def buildApplicationInfo(metadata: KubernetesMetadata): ApplicationInfo = {
+    ApplicationInfo(
+      id = metadata.appId,
+      name = metadata.podName,
+      state = ApplicationState.withName(metadata.appState),
+      error = metadata.appError)
+  }
 }

kyuubi-server/src/main/scala/org/apache/kyuubi/server/metadata/jdbc/JDBCMetadataStore.scala

@@ -432,7 +432,7 @@ class JDBCMetadataStore(conf: KyuubiConf) extends MetadataStore with Logging {
          |app_error,
          |create_time
          |)
-         |VALUES(?, ?, ?, ?, ?, ?, ?)
+         |VALUES(?, ?, ?, ?, ?, ?, ?, ?)
          |WHERE NOT EXISTS (
          |  SELECT 1 FROM $KUBERNETES_METADATA_TABLE WHERE identifier = ?
          |)