Fix kubernetes container state

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

@@ -607,6 +607,7 @@ object KubernetesApplicationOperation extends Logging {
   val KUBERNETES_SERVICE_HOST = "KUBERNETES_SERVICE_HOST"
   val KUBERNETES_SERVICE_PORT = "KUBERNETES_SERVICE_PORT"
   val SPARK_UI_PORT_NAME = "spark-ui"
+  private val PENDING_WAITING_REASONS: Set[String] = Set("ContainerCreating", "PodInitializing")
 
   def toLabel(tag: String): String = s"label: $LABEL_KYUUBI_UNIQUE_KEY=$tag"
 
@@ -685,7 +686,8 @@ object KubernetesApplicationOperation extends Logging {
   def containerStateToApplicationState(containerState: ContainerState): ApplicationState = {
     // https://kubernetes.io/docs/concepts/workloads/pods/pod-lifecycle/#container-states
     if (containerState.getWaiting != null) {
-      PENDING
+      val reasonOpt = Option(containerState.getWaiting.getReason).map(_.trim).filter(_.nonEmpty)
+      if (reasonOpt.isEmpty || PENDING_WAITING_REASONS.contains(reasonOpt.get)) PENDING else FAILED
     } else if (containerState.getRunning != null) {
       RUNNING
     } else if (containerState.getTerminated == null) {

kyuubi-server/src/test/scala/org/apache/kyuubi/engine/KubernetesApplicationOperationSuite.scala

@@ -17,8 +17,11 @@
 
 package org.apache.kyuubi.engine
 
+import io.fabric8.kubernetes.api.model.{ContainerState, ContainerStateWaiting}
+
 import org.apache.kyuubi.{KyuubiException, KyuubiFunSuite}
 import org.apache.kyuubi.config.KyuubiConf
+import org.apache.kyuubi.engine.ApplicationState.{FAILED, PENDING}
 
 class KubernetesApplicationOperationSuite extends KyuubiFunSuite {
 
@@ -127,4 +130,46 @@ class KubernetesApplicationOperationSuite extends KyuubiFunSuite {
         KubernetesInfo(Some("c1"), None),
         KubernetesInfo(None, Some("ns1"))))
   }
+
+  test("containerStateToApplicationState waiting reasons") {
+    // Only valid pending reasons: ContainerCreating and PodInitializing
+    val pendingWaitingReasons = Set("ContainerCreating", "PodInitializing")
+
+    pendingWaitingReasons.foreach { reason =>
+      val containerState = new ContainerState()
+      val waiting = new ContainerStateWaiting()
+      waiting.setReason(reason)
+      containerState.setWaiting(waiting)
+
+      val result = KubernetesApplicationOperation.containerStateToApplicationState(containerState)
+      assert(result === PENDING)
+    }
+  }
+
+  test("containerStateToApplicationState failure reasons and empty reason") {
+    val failureReasons = Set(
+      "ErrImagePull",
+      "ImagePullBackOff",
+      "CrashLoopBackOff",
+      "CreateContainerConfigError")
+
+    failureReasons.foreach { reason =>
+      val containerState = new ContainerState()
+      val waiting = new ContainerStateWaiting()
+      waiting.setReason(reason)
+      containerState.setWaiting(waiting)
+
+      val result = KubernetesApplicationOperation.containerStateToApplicationState(containerState)
+      assert(result === FAILED)
+    }
+
+    // Empty/null reason should be treated as PENDING (still initializing)
+    val containerStateEmpty = new ContainerState()
+    val waitingEmpty = new ContainerStateWaiting()
+    waitingEmpty.setReason(null)
+    containerStateEmpty.setWaiting(waitingEmpty)
+    val resultEmpty =
+      KubernetesApplicationOperation.containerStateToApplicationState(containerStateEmpty)
+    assert(resultEmpty === PENDING)
+  }
 }