[KYUUBI #7034] [KUBERNETES] Prefer to use pod spark-app-name label as application name than pod name

### Why are the changes needed?

After apache/spark#34460 (Since Spark 3.3.0), the `spark-app-name` is available.

We shall use it as the application name if it exists.

### How was this patch tested?

Minor change.
### Was this patch authored or co-authored using generative AI tooling?

No.

Closes #7034 from turboFei/k8s_app_name.

Closes #7034

bfa88a4 [Wang, Fei] Get pod app name

Authored-by: Wang, Fei <[email protected]>
Signed-off-by: Wang, Fei <[email protected]>

Author: turboFei

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

@@ -23,7 +23,7 @@ import io.fabric8.kubernetes.api.model.Pod
 
 import org.apache.kyuubi.Logging
 import org.apache.kyuubi.config.KyuubiConf.KubernetesApplicationStateSource.KubernetesApplicationStateSource
-import org.apache.kyuubi.engine.KubernetesApplicationOperation.{toApplicationStateAndError, LABEL_KYUUBI_UNIQUE_KEY, SPARK_APP_ID_LABEL}
+import org.apache.kyuubi.engine.KubernetesApplicationOperation.{getPodAppId, getPodAppName, toApplicationStateAndError, LABEL_KYUUBI_UNIQUE_KEY}
 import org.apache.kyuubi.engine.KubernetesResourceEventTypes.KubernetesResourceEventType
 
 object KubernetesApplicationAuditLogger extends Logging {
@@ -49,7 +49,8 @@ object KubernetesApplicationAuditLogger extends Logging {
       s"${containerState.getName}->${containerState.getState}"
     }.mkString("[", ",", "]")
     sb.append(s"containers=$containerStatuses").append("\t")
-    sb.append(s"appId=${pod.getMetadata.getLabels.get(SPARK_APP_ID_LABEL)}").append("\t")
+    sb.append(s"appId=${getPodAppId(pod)}").append("\t")
+    sb.append(s"appName=${getPodAppName(pod)}").append("\t")
     val (appState, appError) =
       toApplicationStateAndError(pod, appStateSource, appStateContainer, eventType)
     sb.append(s"appState=$appState").append("\t")

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

@@ -408,16 +408,16 @@ class KubernetesApplicationOperation extends ApplicationOperation with Logging {
         appInfoStore.put(
           kyuubiUniqueKey,
           kubernetesInfo -> appInfo.copy(
-            id = pod.getMetadata.getLabels.get(SPARK_APP_ID_LABEL),
-            name = pod.getMetadata.getName,
+            id = getPodAppId(pod),
+            name = getPodAppName(pod),
             state = appState,
             error = appError))
       }.getOrElse {
         appInfoStore.put(
           kyuubiUniqueKey,
           kubernetesInfo -> ApplicationInfo(
-            id = pod.getMetadata.getLabels.get(SPARK_APP_ID_LABEL),
-            name = pod.getMetadata.getName,
+            id = getPodAppId(pod),
+            name = getPodAppName(pod),
             state = appState,
             error = appError))
       }
@@ -506,7 +506,8 @@ class KubernetesApplicationOperation extends ApplicationOperation with Logging {
 
 object KubernetesApplicationOperation extends Logging {
   val LABEL_KYUUBI_UNIQUE_KEY = "kyuubi-unique-tag"
-  val SPARK_APP_ID_LABEL = "spark-app-selector"
+  private val SPARK_APP_ID_LABEL = "spark-app-selector"
+  private val SPARK_APP_NAME_LABEL = "spark-app-name"
   val KUBERNETES_SERVICE_HOST = "KUBERNETES_SERVICE_HOST"
   val KUBERNETES_SERVICE_PORT = "KUBERNETES_SERVICE_PORT"
   val SPARK_UI_PORT_NAME = "spark-ui"
@@ -634,4 +635,12 @@ object KubernetesApplicationOperation extends Logging {
       .replace("{{KUBERNETES_NAMESPACE}}", kubernetesNamespace)
       .replace("{{SPARK_UI_PORT}}", sparkUiPort.toString)
   }
+
+  def getPodAppId(pod: Pod): String = {
+    pod.getMetadata.getLabels.get(SPARK_APP_ID_LABEL)
+  }
+
+  def getPodAppName(pod: Pod): String = {
+    Option(pod.getMetadata.getLabels.get(SPARK_APP_NAME_LABEL)).getOrElse(pod.getMetadata.getName)
+  }
 }