[SPARK-45948][K8S] Make single-pod spark jobs respect spark.app.id

resource-managers/kubernetes/core/src/main/scala/org/apache/spark/scheduler/cluster/k8s/KubernetesClusterManager.scala

@@ -63,7 +63,10 @@ private[spark] class KubernetesClusterManager extends ExternalClusterManager wit
       }
       logInfo(s"Running Spark with ${sc.conf.get(KUBERNETES_DRIVER_MASTER_URL)}")
       val schedulerImpl = scheduler.asInstanceOf[TaskSchedulerImpl]
-      val backend = new LocalSchedulerBackend(sc.conf, schedulerImpl, threadCount)
+      // KubernetesClusterSchedulerBackend respects `spark.app.id` while LocalSchedulerBackend
+      // does not. Propagate `spark.app.id` via `spark.test.appId` to match the behavior.
+      val conf = sc.conf.getOption("spark.app.id").map(sc.conf.set("spark.test.appId", _))
+      val backend = new LocalSchedulerBackend(conf.getOrElse(sc.conf), schedulerImpl, threadCount)
       schedulerImpl.initialize(backend)
       return backend
     }

resource-managers/kubernetes/core/src/test/scala/org/apache/spark/scheduler/cluster/k8s/KubernetesClusterManagerSuite.scala

@@ -20,10 +20,13 @@ import io.fabric8.kubernetes.client.KubernetesClient
 import org.mockito.{Mock, MockitoAnnotations}
 import org.mockito.Mockito.when
 import org.scalatest.BeforeAndAfter
+import org.scalatestplus.mockito.MockitoSugar.mock
 
 import org.apache.spark._
 import org.apache.spark.deploy.k8s.Config._
 import org.apache.spark.internal.config._
+import org.apache.spark.scheduler.TaskSchedulerImpl
+import org.apache.spark.scheduler.local.LocalSchedulerBackend
 
 class KubernetesClusterManagerSuite extends SparkFunSuite with BeforeAndAfter {
 
@@ -59,4 +62,22 @@ class KubernetesClusterManagerSuite extends SparkFunSuite with BeforeAndAfter {
       manager.makeExecutorPodsAllocator(sc, kubernetesClient, null)
     }
   }
+
+  test("SPARK-45948: Single-pod Spark jobs respect spark.app.id") {
+    val conf = new SparkConf()
+    conf.set(KUBERNETES_DRIVER_MASTER_URL, "local[2]")
+    when(sc.conf).thenReturn(conf)
+    val scheduler = mock[TaskSchedulerImpl]
+    when(scheduler.sc).thenReturn(sc)
+    val manager = new KubernetesClusterManager()
+
+    val backend1 = manager.createSchedulerBackend(sc, "", scheduler)
+    assert(backend1.isInstanceOf[LocalSchedulerBackend])
+    assert(backend1.applicationId().startsWith("local-"))
+
+    conf.set("spark.app.id", "user-app-id")
+    val backend2 = manager.createSchedulerBackend(sc, "", scheduler)
+    assert(backend2.isInstanceOf[LocalSchedulerBackend])
+    assert(backend2.applicationId() === "user-app-id")
+  }
 }