[LIVY-1081] SparkKubernetesApp: latch first observed app ID, reject mismatched re-attach under same tag

server/src/main/scala/org/apache/livy/utils/SparkKubernetesApp.scala

@@ -260,6 +260,22 @@ object SparkKubernetesApp extends Logging {
   def getAppSize: Int = appQueue.size()
 
   def clearApps(): Unit = appQueue.clear()
+
+  // Decides whether a newly observed app ID is consistent with the app ID already
+  // latched for this app tag. Returns Right(the app ID to latch going forward), or
+  // Left(an error message) if the tag's app ID changed and monitoring should be rejected
+  // (e.g. a re-attach picked up a different, unrelated Spark app under the same tag).
+  private[utils] def latchAppId(
+      knownAppId: Option[String],
+      observedAppId: String,
+      appTag: String): Either[String, Option[String]] = {
+    knownAppId match {
+      case Some(known) if known != observedAppId =>
+        Left(s"App ID changed for tag $appTag: was $known, now $observedAppId. Rejecting.")
+      case None => Right(Some(observedAppId))
+      case some => Right(some)
+    }
+  }
 }
 
 class SparkKubernetesApp private[utils] (
@@ -282,6 +298,9 @@ class SparkKubernetesApp private[utils] (
   private var kubernetesDiagnostics: IndexedSeq[String] = IndexedSeq.empty[String]
   private var kubernetesAppLog: IndexedSeq[String] = IndexedSeq.empty[String]
 
+  // Latches the first observed app ID; subsequent polls returning a different ID are rejected.
+  @volatile private var knownAppId: Option[String] = appIdOption
+
   private var kubernetesTagToAppIdFailedTimes: Int = _
   private var kubernetesAppMonitorFailedTimes: Int = _
 
@@ -327,9 +346,25 @@ class SparkKubernetesApp private[utils] (
         return
       }
       val app: KubernetesApplication = appOption.get
-      appPromise.trySuccess(app)
       val appId = app.getApplicationId
 
+      latchAppId(knownAppId, appId, appTag) match {
+        case Left(msg) =>
+          error(msg)
+          // Fail the promise so consumers blocked on Await.result(appPromise.future, ...)
+          // surface the mismatch immediately instead of waiting the full appLookupTimeout.
+          appPromise.tryFailure(new IllegalStateException(msg))
+          // Drive the session to FAILED, destroy the spark-submit process, and mark the
+          // tag as leaked so the monitor doesn't keep polling. Mirrors the treatment of
+          // other terminal failures (e.g. failToGetAppId exhausted).
+          kubernetesDiagnostics = IndexedSeq(msg)
+          failToMonitor()
+          return
+        case Right(updatedKnownAppId) =>
+          knownAppId = updatedKnownAppId
+      }
+      appPromise.trySuccess(app)
+
       Thread.currentThread().setName(s"kubernetesAppMonitorThread-$appId")
       listener.foreach(_.appIdKnown(appId))
 

server/src/test/scala/org/apache/livy/utils/SparkKubernetesAppSpec.scala

@@ -271,6 +271,26 @@ class SparkKubernetesAppSpec extends AnyFunSpec with LivyBaseUnitTestSuite with
     }
   }
 
+  describe("latchAppId") {
+    it("should latch the observed app ID when none was known yet") {
+      assertResult(Right(Some("app-1")))(
+        SparkKubernetesApp.latchAppId(None, "app-1", "tag-1"))
+    }
+
+    it("should keep the known app ID when the observed ID matches") {
+      assertResult(Right(Some("app-1")))(
+        SparkKubernetesApp.latchAppId(Some("app-1"), "app-1", "tag-1"))
+    }
+
+    it("should reject a different app ID observed under the same tag") {
+      val result = SparkKubernetesApp.latchAppId(Some("app-1"), "app-2", "tag-1")
+      assert(result.isLeft)
+      assert(result.left.get.contains("tag-1"))
+      assert(result.left.get.contains("app-1"))
+      assert(result.left.get.contains("app-2"))
+    }
+  }
+
   describe("KubernetesClientExtensions") {
     it("should build an ingress from the supplied KubernetesApplication") {
       def test(app: KubernetesApplication, expectedAnnotations: Map[String, String]): Unit = {