[KYUUBI #7034][FOLLOWUP] Decouple the kubernetes pod name and app name

integration-tests/kyuubi-kubernetes-it/src/test/scala/org/apache/kyuubi/kubernetes/test/spark/SparkOnKubernetesTestsSuite.scala

@@ -28,7 +28,7 @@ import org.apache.kyuubi._
 import org.apache.kyuubi.client.util.BatchUtils._
 import org.apache.kyuubi.config.KyuubiConf
 import org.apache.kyuubi.config.KyuubiConf._
-import org.apache.kyuubi.engine.{ApplicationInfo, ApplicationManagerInfo, ApplicationOperation, KubernetesApplicationOperation}
+import org.apache.kyuubi.engine.{ApplicationInfo, ApplicationManagerInfo, ApplicationOperation, KubernetesApplicationInfo, KubernetesApplicationOperation}
 import org.apache.kyuubi.engine.ApplicationState.{FAILED, NOT_FOUND, RUNNING}
 import org.apache.kyuubi.engine.spark.SparkProcessBuilder
 import org.apache.kyuubi.kubernetes.test.MiniKube
@@ -230,7 +230,8 @@ class KyuubiOperationKubernetesClusterClusterModeSuite
         appMgrInfo,
         sessionHandle.identifier.toString)
       assert(appInfo.state == RUNNING)
-      assert(appInfo.name.startsWith(driverPodNamePrefix))
+      assert(
+        appInfo.asInstanceOf[KubernetesApplicationInfo].podName.startsWith(driverPodNamePrefix))
     }
 
     val killResponse = k8sOperation.killApplicationByTag(

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

@@ -129,6 +129,33 @@ object ApplicationInfo {
   val UNKNOWN: ApplicationInfo = ApplicationInfo(null, null, ApplicationState.UNKNOWN)
 }
 
+class KubernetesApplicationInfo(
+    id: String,
+    name: String,
+    state: ApplicationState,
+    url: Option[String] = None,
+    error: Option[String] = None,
+    val podName: String) extends ApplicationInfo(id, name, state, url, error) {
+  override def toMap: Map[String, String] = super.toMap ++ Map("podName" -> podName)
+}
+
+object KubernetesApplicationInfo {
+  val NOT_FOUND: KubernetesApplicationInfo = new KubernetesApplicationInfo(
+    null,
+    null,
+    ApplicationState.NOT_FOUND,
+    None,
+    None,
+    null)
+  val UNKNOWN: KubernetesApplicationInfo = new KubernetesApplicationInfo(
+    null,
+    null,
+    ApplicationState.UNKNOWN,
+    None,
+    None,
+    null)
+}
+
 object ApplicationOperation {
   val NOT_FOUND = "APPLICATION_NOT_FOUND"
 }

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

@@ -66,8 +66,8 @@ class KubernetesApplicationOperation extends ApplicationOperation with Logging {
     kyuubiConf.get(KyuubiConf.KUBERNETES_APPLICATION_STATE_CONTAINER)
 
   // key is kyuubi_unique_key
-  private val appInfoStore: ConcurrentHashMap[String, (KubernetesInfo, ApplicationInfo)] =
-    new ConcurrentHashMap[String, (KubernetesInfo, ApplicationInfo)]
+  private val appInfoStore: ConcurrentHashMap[String, (KubernetesInfo, KubernetesApplicationInfo)] =
+    new ConcurrentHashMap[String, (KubernetesInfo, KubernetesApplicationInfo)]
   // key is kyuubi_unique_key
   private var cleanupTerminatedAppInfoTrigger: Cache[String, ApplicationState] = _
 
@@ -140,7 +140,7 @@ class KubernetesApplicationOperation extends ApplicationOperation with Logging {
             case COMPLETED => !ApplicationState.isFailed(notification.getValue)
           }
           if (shouldDelete) {
-            deletePod(kubernetesInfo, removed.name, appLabel)
+            deletePod(kubernetesInfo, removed.podName, appLabel)
           }
           info(s"Remove terminated application $removed with ${toLabel(appLabel)}")
         }
@@ -220,7 +220,7 @@ class KubernetesApplicationOperation extends ApplicationOperation with Logging {
                 s"[$kubernetesInfo] Target application[${toLabel(tag)}] is in ${info.state} state")
             case _ =>
               (
-                !kubernetesClient.pods.withName(info.name).delete().isEmpty,
+                !kubernetesClient.pods.withName(info.podName).delete().isEmpty,
                 s"[$kubernetesInfo] Operation of deleted" +
                   s" application[appId: ${info.id}, ${toLabel(tag)}] is completed")
           }
@@ -252,30 +252,32 @@ class KubernetesApplicationOperation extends ApplicationOperation with Logging {
       // need to initialize the kubernetes client if not exists
       getOrCreateKubernetesClient(appMgrInfo.kubernetesInfo)
       val (_, appInfo) =
-        appInfoStore.getOrDefault(tag, appMgrInfo.kubernetesInfo -> ApplicationInfo.NOT_FOUND)
+        appInfoStore.getOrDefault(
+          tag,
+          appMgrInfo.kubernetesInfo -> KubernetesApplicationInfo.NOT_FOUND)
       (appInfo.state, submitTime) match {
         // Kyuubi should wait second if pod is not be created
         case (NOT_FOUND, Some(_submitTime)) =>
           val elapsedTime = System.currentTimeMillis - _submitTime
           if (elapsedTime > submitTimeout) {
             error(s"Can't find target driver pod by ${toLabel(tag)}, " +
               s"elapsed time: ${elapsedTime}ms exceeds ${submitTimeout}ms.")
-            ApplicationInfo.NOT_FOUND
+            KubernetesApplicationInfo.NOT_FOUND
           } else {
             warn(s"Waiting for driver pod with ${toLabel(tag)} to be created, " +
               s"elapsed time: ${elapsedTime}ms, return UNKNOWN status")
-            ApplicationInfo.UNKNOWN
+            KubernetesApplicationInfo.UNKNOWN
           }
         case (NOT_FOUND, None) =>
-          ApplicationInfo.NOT_FOUND
+          KubernetesApplicationInfo.NOT_FOUND
         case _ =>
           debug(s"Successfully got application[${toLabel(tag)}]'s info: $appInfo")
           appInfo
       }
     } catch {
       case e: Exception =>
         error(s"Failed to get application by ${toLabel(tag)}, due to ${e.getMessage}")
-        ApplicationInfo.NOT_FOUND
+        KubernetesApplicationInfo.NOT_FOUND
     }
   }
 
@@ -407,19 +409,23 @@ class KubernetesApplicationOperation extends ApplicationOperation with Logging {
       Option(appInfoStore.get(kyuubiUniqueKey)).map { case (_, appInfo) =>
         appInfoStore.put(
           kyuubiUniqueKey,
-          kubernetesInfo -> appInfo.copy(
+          kubernetesInfo -> new KubernetesApplicationInfo(
             id = getPodAppId(pod),
             name = getPodAppName(pod),
             state = appState,
-            error = appError))
+            // inherit the url from the old appInfo
+            url = appInfo.url,
+            error = appError,
+            podName = pod.getMetadata.getName))
       }.getOrElse {
         appInfoStore.put(
           kyuubiUniqueKey,
-          kubernetesInfo -> ApplicationInfo(
+          kubernetesInfo -> new KubernetesApplicationInfo(
             id = getPodAppId(pod),
             name = getPodAppName(pod),
             state = appState,
-            error = appError))
+            error = appError,
+            podName = pod.getMetadata.getName))
       }
     }
   }
@@ -443,7 +449,16 @@ class KubernetesApplicationOperation extends ApplicationOperation with Logging {
         val kyuubiUniqueKey = svc.getSpec.getSelector.get(LABEL_KYUUBI_UNIQUE_KEY)
         appInfoStore.synchronized {
           Option(appInfoStore.get(kyuubiUniqueKey)).foreach { case (_, appInfo) =>
-            appInfoStore.put(kyuubiUniqueKey, kubernetesInfo -> appInfo.copy(url = Some(appUrl)))
+            appInfoStore.put(
+              kyuubiUniqueKey,
+              kubernetesInfo ->
+                new KubernetesApplicationInfo(
+                  id = appInfo.id,
+                  name = appInfo.name,
+                  state = appInfo.state,
+                  url = Some(appUrl),
+                  error = appInfo.error,
+                  podName = appInfo.podName))
           }
         }
     }.getOrElse(warn(s"Spark UI port not found in service ${svc.getMetadata.getName}"))