diff --git a/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/Config.scala b/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/Config.scala
index 7fd21dfdcb0fd..e5e13f2558b0b 100644
--- a/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/Config.scala
+++ b/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/Config.scala
@@ -576,6 +576,19 @@ private[spark] object Config extends Logging {
       .intConf
       .createOptional
 
+  val KUBERNETES_EXECUTOR_ENABLE_INFORMER =
+    ConfigBuilder("spark.kubernetes.executor.enableInformer")
+      .doc("If true, use a shared Kubernetes informer (list + watch) to track executor pod " +
+        "state, driven by ExecutorPodsInformerSnapshotSource (event-driven) and " +
+        "ExecutorPodsListerSnapshotSource (periodic refresh of the informer cache). If " +
+        "false (default), use the legacy path backed by ExecutorPodsWatchSnapshotSource and " +
+        "ExecutorPodsPollingSnapshotSource. The two modes are mutually exclusive; " +
+        "`spark.kubernetes.executor.enableApiWatcher` and " +
+        "`spark.kubernetes.executor.enableApiPolling` only apply when this is false.")
+      .version("4.4.0")
+      .booleanConf
+      .createWithDefault(false)
+
   val KUBERNETES_ALLOCATION_BATCH_SIZE =
     ConfigBuilder("spark.kubernetes.allocation.batch.size")
       .doc("Number of pods to launch at once in each round of executor allocation.")
@@ -668,6 +681,24 @@ private[spark] object Config extends Logging {
       .booleanConf
       .createWithDefault(true)
 
+  val KUBERNETES_EXECUTOR_LISTER_POLLING_INTERVAL =
+    ConfigBuilder("spark.kubernetes.executor.listerPollingInterval")
+      .doc("Interval between polls against the Kubernetes informer cache to inspect the " +
+        "state of executors.")
+      .version("4.4.0")
+      .timeConf(TimeUnit.MILLISECONDS)
+      .checkValue(interval => interval > 0,
+        "Informer lister polling interval must be a positive time value.")
+      .createWithDefaultString("30s")
+
+  val KUBERNETES_EXECUTOR_INFORMER_RESYNC_INTERVAL =
+    ConfigBuilder("spark.kubernetes.executor.informerResyncInterval")
+      .doc("Interval between informer cache resync.")
+      .version("4.4.0")
+      .timeConf(TimeUnit.MILLISECONDS)
+      .checkValue(interval => interval >= 0,
+        "Informer resync interval must not be a negative time value.")
+      .createWithDefaultString("0s")
 
   val KUBERNETES_EXECUTOR_API_POLLING_INTERVAL =
     ConfigBuilder("spark.kubernetes.executor.apiPollingInterval")
diff --git a/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/scheduler/cluster/k8s/ExecutorPodsInformerSnapshotSource.scala b/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/scheduler/cluster/k8s/ExecutorPodsInformerSnapshotSource.scala
new file mode 100644
index 0000000000000..56716b9bab5a8
--- /dev/null
+++ b/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/scheduler/cluster/k8s/ExecutorPodsInformerSnapshotSource.scala
@@ -0,0 +1,63 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.spark.scheduler.cluster.k8s
+
+import io.fabric8.kubernetes.api.model.Pod
+import io.fabric8.kubernetes.client.informers.ResourceEventHandler
+
+import org.apache.spark.internal.Logging
+import org.apache.spark.util.Utils
+
+/**
+ * Publishes executor pod updates to [[ExecutorPodsSnapshotsStore]] using the shared informer
+ * owned by [[InformerManager]]. Event-driven counterpart of [[ExecutorPodsListerSnapshotSource]],
+ * which periodically snapshots the same informer's local cache.
+ */
+class ExecutorPodsInformerSnapshotSource(
+    snapshotsStore: ExecutorPodsSnapshotsStore,
+    informerManager: InformerManager)
+  extends ExecutorPodsSnapshotSource with Logging {
+
+  override def start(applicationId: String): Unit = {
+    informerManager.initInformer(applicationId)
+    informerManager.getInformer().addEventHandler(new ExecutorPodsInformer())
+    informerManager.startInformer()
+  }
+
+  override def stop(): Unit = {
+    Utils.tryLogNonFatalError {
+      informerManager.stopInformer()
+    }
+  }
+
+  private class ExecutorPodsInformer extends ResourceEventHandler[Pod] {
+    override def onAdd(pod: Pod): Unit = {
+      logDebug(s"Received add executor pod event for pod named ${pod.getMetadata.getName}")
+      snapshotsStore.updatePod(pod)
+    }
+
+    override def onUpdate(oldPod: Pod, newPod: Pod): Unit = {
+      logDebug(s"Received update executor pod event for pod named ${newPod.getMetadata.getName}")
+      snapshotsStore.updatePod(newPod)
+    }
+
+    override def onDelete(pod: Pod, deletedFinalStateUnknown: Boolean): Unit = {
+      logDebug(s"Received delete executor pod event for pod named ${pod.getMetadata.getName}")
+      snapshotsStore.updatePod(pod)
+    }
+  }
+}
diff --git a/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/scheduler/cluster/k8s/ExecutorPodsListerSnapshotSource.scala b/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/scheduler/cluster/k8s/ExecutorPodsListerSnapshotSource.scala
new file mode 100644
index 0000000000000..34191af21521c
--- /dev/null
+++ b/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/scheduler/cluster/k8s/ExecutorPodsListerSnapshotSource.scala
@@ -0,0 +1,76 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.spark.scheduler.cluster.k8s
+
+import java.util.concurrent.{Future, ScheduledExecutorService, TimeUnit}
+
+import scala.jdk.CollectionConverters._
+
+import io.fabric8.kubernetes.api.model.Pod
+import io.fabric8.kubernetes.client.KubernetesClient
+import io.fabric8.kubernetes.client.informers.cache.Lister
+
+import org.apache.spark.SparkConf
+import org.apache.spark.deploy.k8s.Config.KUBERNETES_EXECUTOR_LISTER_POLLING_INTERVAL
+import org.apache.spark.internal.Logging
+import org.apache.spark.util.{ThreadUtils, Utils}
+
+/**
+ * Periodically snapshots the local cache of the shared [[InformerManager]] and replaces the
+ * contents of the [[ExecutorPodsSnapshotsStore]] with the result. Companion to
+ * [[ExecutorPodsInformerSnapshotSource]], which pushes updates as informer events arrive.
+ */
+class ExecutorPodsListerSnapshotSource(
+    conf: SparkConf,
+    kubernetesClient: KubernetesClient,
+    snapshotsStore: ExecutorPodsSnapshotsStore,
+    informerManager: InformerManager,
+    pollingExecutor: ScheduledExecutorService)
+  extends ExecutorPodsSnapshotSource with Logging {
+
+  private val pollingInterval = conf.get(KUBERNETES_EXECUTOR_LISTER_POLLING_INTERVAL)
+
+  private var pollingFuture: Future[_] = _
+
+  override def start(applicationId: String): Unit = {
+    informerManager.initInformer(applicationId)
+    informerManager.startInformer()
+    val lister = new Lister[Pod](
+      informerManager.getInformer().getIndexer, kubernetesClient.getNamespace)
+    pollingFuture = pollingExecutor.scheduleWithFixedDelay(
+      new PollRunnable(lister), pollingInterval, pollingInterval, TimeUnit.MILLISECONDS)
+  }
+
+  override def stop(): Unit = {
+    if (pollingFuture != null) {
+      pollingFuture.cancel(true)
+      pollingFuture = null
+    }
+    Utils.tryLogNonFatalError {
+      informerManager.stopInformer()
+    }
+    ThreadUtils.shutdown(pollingExecutor)
+  }
+
+  private class PollRunnable(lister: Lister[Pod]) extends Runnable {
+    override def run(): Unit = Utils.tryLogNonFatalError {
+      // The informer is already scoped server-side to app-id + role=executor + non-inactive
+      // pods, so we can hand its snapshot to the store as-is.
+      snapshotsStore.replaceSnapshot(lister.list().asScala.toSeq)
+    }
+  }
+}
diff --git a/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/scheduler/cluster/k8s/ExecutorPodsPollingSnapshotSource.scala b/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/scheduler/cluster/k8s/ExecutorPodsPollingSnapshotSource.scala
index 3d2822e5eb518..7d489e48abc73 100644
--- a/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/scheduler/cluster/k8s/ExecutorPodsPollingSnapshotSource.scala
+++ b/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/scheduler/cluster/k8s/ExecutorPodsPollingSnapshotSource.scala
@@ -43,7 +43,7 @@ class ExecutorPodsPollingSnapshotSource(
     conf: SparkConf,
     kubernetesClient: KubernetesClient,
     snapshotsStore: ExecutorPodsSnapshotsStore,
-    pollingExecutor: ScheduledExecutorService) extends Logging {
+    pollingExecutor: ScheduledExecutorService) extends ExecutorPodsSnapshotSource with Logging {
 
   private val pollingInterval = conf.get(KUBERNETES_EXECUTOR_API_POLLING_INTERVAL)
   private val pollingEnabled = conf.get(KUBERNETES_EXECUTOR_ENABLE_API_POLLING)
diff --git a/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/scheduler/cluster/k8s/ExecutorPodsSnapshotSource.scala b/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/scheduler/cluster/k8s/ExecutorPodsSnapshotSource.scala
new file mode 100644
index 0000000000000..8392e96742de8
--- /dev/null
+++ b/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/scheduler/cluster/k8s/ExecutorPodsSnapshotSource.scala
@@ -0,0 +1,27 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.spark.scheduler.cluster.k8s
+
+/**
+ * Publishes snapshots of the set of executor pods that Kubernetes reports as running for an
+ * application. Built-in implementations are chosen by
+ * [[org.apache.spark.deploy.k8s.Config.KUBERNETES_EXECUTOR_ENABLE_INFORMER]].
+ */
+trait ExecutorPodsSnapshotSource {
+  def start(applicationId: String): Unit
+  def stop(): Unit
+}
diff --git a/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/scheduler/cluster/k8s/ExecutorPodsWatchSnapshotSource.scala b/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/scheduler/cluster/k8s/ExecutorPodsWatchSnapshotSource.scala
index 0d9f19ee11b71..22765b957da6b 100644
--- a/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/scheduler/cluster/k8s/ExecutorPodsWatchSnapshotSource.scala
+++ b/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/scheduler/cluster/k8s/ExecutorPodsWatchSnapshotSource.scala
@@ -42,7 +42,7 @@ import org.apache.spark.util.Utils
 class ExecutorPodsWatchSnapshotSource(
     snapshotsStore: ExecutorPodsSnapshotsStore,
     kubernetesClient: KubernetesClient,
-    conf: SparkConf) extends Logging {
+    conf: SparkConf) extends ExecutorPodsSnapshotSource with Logging {
 
   private var watchConnection: Closeable = _
   private val enableWatching = conf.get(KUBERNETES_EXECUTOR_ENABLE_API_WATCHER)
diff --git a/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/scheduler/cluster/k8s/InformerManager.scala b/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/scheduler/cluster/k8s/InformerManager.scala
new file mode 100644
index 0000000000000..044932d7bb38d
--- /dev/null
+++ b/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/scheduler/cluster/k8s/InformerManager.scala
@@ -0,0 +1,86 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.spark.scheduler.cluster.k8s
+
+import io.fabric8.kubernetes.api.model.Pod
+import io.fabric8.kubernetes.client.KubernetesClient
+import io.fabric8.kubernetes.client.informers.SharedIndexInformer
+
+import org.apache.spark.SparkConf
+import org.apache.spark.deploy.k8s.Config.KUBERNETES_EXECUTOR_INFORMER_RESYNC_INTERVAL
+import org.apache.spark.deploy.k8s.Constants.{SPARK_APP_ID_LABEL, SPARK_EXECUTOR_INACTIVE_LABEL, SPARK_POD_EXECUTOR_ROLE, SPARK_ROLE_LABEL}
+import org.apache.spark.internal.Logging
+import org.apache.spark.util.Utils
+
+/**
+ * Owns the shared [[SharedIndexInformer]] used by executor pod snapshot sources when the
+ * informer-based mode is enabled. The informer is scoped server-side to the current
+ * application's executor pods that are not marked inactive, matching the filter set used by
+ * [[ExecutorPodsWatchSnapshotSource]] and [[ExecutorPodsPollingSnapshotSource]].
+ */
+class InformerManager(kubernetesClient: KubernetesClient, conf: SparkConf)
+  extends Logging {
+
+  private val resyncInterval = conf.get(KUBERNETES_EXECUTOR_INFORMER_RESYNC_INTERVAL)
+  // VisibleForTesting
+  private[k8s] var informer: SharedIndexInformer[Pod] = _
+  private var stopped = false
+
+  def initInformer(applicationId: String): Unit = {
+    if (informer == null) {
+      logInfo(s"Initializing executor pods informer for application $applicationId")
+      informer = kubernetesClient.pods()
+        .withLabel(SPARK_APP_ID_LABEL, applicationId)
+        .withLabel(SPARK_ROLE_LABEL, SPARK_POD_EXECUTOR_ROLE)
+        .withoutLabel(SPARK_EXECUTOR_INACTIVE_LABEL, "true")
+        .runnableInformer(resyncInterval)
+    }
+  }
+
+  def getInformer(): SharedIndexInformer[Pod] = {
+    if (informer == null) {
+      throw new IllegalStateException(
+        "Informer has not been initialized. Call initInformer() first.")
+    }
+    informer
+  }
+
+  def startInformer(): Unit = {
+    if (informer == null) {
+      throw new IllegalStateException(
+        "Informer has not been initialized. Call initInformer() first.")
+    }
+    if (stopped) {
+      throw new IllegalStateException("Cannot run informer after stopInformer() has been called.")
+    }
+    if (!informer.isRunning) {
+      informer.run()
+    } else {
+      logInfo("Informer is already running.")
+    }
+  }
+
+  def stopInformer(): Unit = {
+    if (informer != null) {
+      Utils.tryLogNonFatalError {
+        informer.close()
+      }
+      informer = null
+      stopped = true
+    }
+  }
+}
diff --git a/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/scheduler/cluster/k8s/KubernetesClusterManager.scala b/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/scheduler/cluster/k8s/KubernetesClusterManager.scala
index 782fac670fa88..6aa4b077d8323 100644
--- a/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/scheduler/cluster/k8s/KubernetesClusterManager.scala
+++ b/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/scheduler/cluster/k8s/KubernetesClusterManager.scala
@@ -137,15 +137,12 @@ private[spark] class KubernetesClusterManager extends ExternalClusterManager wit
     val executorPodsAllocator = makeExecutorPodsAllocator(
       sc, kubernetesClient, snapshotsStore, Some(executorPodsLifecycleManager))
 
-    val podsWatchEventSource = new ExecutorPodsWatchSnapshotSource(
-      snapshotsStore,
-      kubernetesClient,
-      sc.conf)
-
-    val eventsPollingExecutor = ThreadUtils.newDaemonSingleThreadScheduledExecutor(
-      "kubernetes-executor-pod-polling-sync")
-    val podsPollingEventSource = new ExecutorPodsPollingSnapshotSource(
-      sc.conf, kubernetesClient, snapshotsStore, eventsPollingExecutor)
+    val snapshotSources = {
+      val sources = makeSnapshotSources(sc.conf, kubernetesClient, snapshotsStore)
+      logInfo(s"Executor pods snapshot sources: " +
+        sources.map(_.getClass.getSimpleName).mkString(", "))
+      sources
+    }
 
     new KubernetesClusterSchedulerBackend(
       scheduler.asInstanceOf[TaskSchedulerImpl],
@@ -155,8 +152,7 @@ private[spark] class KubernetesClusterManager extends ExternalClusterManager wit
       snapshotsStore,
       executorPodsAllocator,
       executorPodsLifecycleManager,
-      podsWatchEventSource,
-      podsPollingEventSource)
+      snapshotSources)
   }
 
   private[k8s] def makeExecutorPodsAllocator(
@@ -204,6 +200,28 @@ private[spark] class KubernetesClusterManager extends ExternalClusterManager wit
     allocatorInstance
   }
 
+  private def makeSnapshotSources(
+      conf: SparkConf,
+      kubernetesClient: KubernetesClient,
+      snapshotsStore: ExecutorPodsSnapshotsStore): Seq[ExecutorPodsSnapshotSource] = {
+    if (conf.get(KUBERNETES_EXECUTOR_ENABLE_INFORMER)) {
+      val informerManager = new InformerManager(kubernetesClient, conf)
+      val listerExecutor = ThreadUtils.newDaemonSingleThreadScheduledExecutor(
+        "kubernetes-executor-pod-lister-sync")
+      Seq(
+        new ExecutorPodsInformerSnapshotSource(snapshotsStore, informerManager),
+        new ExecutorPodsListerSnapshotSource(
+          conf, kubernetesClient, snapshotsStore, informerManager, listerExecutor))
+    } else {
+      val eventsPollingExecutor = ThreadUtils.newDaemonSingleThreadScheduledExecutor(
+        "kubernetes-executor-pod-polling-sync")
+      Seq(
+        new ExecutorPodsWatchSnapshotSource(snapshotsStore, kubernetesClient, conf),
+        new ExecutorPodsPollingSnapshotSource(
+          conf, kubernetesClient, snapshotsStore, eventsPollingExecutor))
+    }
+  }
+
   override def initialize(scheduler: TaskScheduler, backend: SchedulerBackend): Unit = {
     scheduler.asInstanceOf[TaskSchedulerImpl].initialize(backend)
   }
diff --git a/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/scheduler/cluster/k8s/KubernetesClusterSchedulerBackend.scala b/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/scheduler/cluster/k8s/KubernetesClusterSchedulerBackend.scala
index 27a0b320cbf9a..2406a825b1f1a 100644
--- a/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/scheduler/cluster/k8s/KubernetesClusterSchedulerBackend.scala
+++ b/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/scheduler/cluster/k8s/KubernetesClusterSchedulerBackend.scala
@@ -52,8 +52,7 @@ private[spark] class KubernetesClusterSchedulerBackend(
     snapshotsStore: ExecutorPodsSnapshotsStore,
     podAllocator: AbstractPodsAllocator,
     lifecycleManager: ExecutorPodsLifecycleManager,
-    watchEvents: ExecutorPodsWatchSnapshotSource,
-    pollEvents: ExecutorPodsPollingSnapshotSource)
+    snapshotSources: Seq[ExecutorPodsSnapshotSource])
     extends CoarseGrainedSchedulerBackend(scheduler, sc.env.rpcEnv) {
 
   protected override val minRegisteredRatio =
@@ -121,8 +120,7 @@ private[spark] class KubernetesClusterSchedulerBackend(
     val initExecs = Map(defaultProfile -> initialExecutors)
     podAllocator.setTotalExpectedExecutors(initExecs)
     lifecycleManager.start(this)
-    watchEvents.start(applicationId())
-    pollEvents.start(applicationId())
+    snapshotSources.foreach(source => source.start(applicationId()))
   }
 
   override def stop(): Unit = {
@@ -136,12 +134,10 @@ private[spark] class KubernetesClusterSchedulerBackend(
       snapshotsStore.stop()
     }
 
-    Utils.tryLogNonFatalError {
-      watchEvents.stop()
-    }
-
-    Utils.tryLogNonFatalError {
-      pollEvents.stop()
+    snapshotSources.foreach { source =>
+      Utils.tryLogNonFatalError {
+        source.stop()
+      }
     }
 
     if (conf.get(KUBERNETES_DRIVER_SERVICE_DELETE_ON_TERMINATION)) {
diff --git a/resource-managers/kubernetes/core/src/test/scala/org/apache/spark/scheduler/cluster/k8s/ExecutorPodsInformerSnapshotSourceSuite.scala b/resource-managers/kubernetes/core/src/test/scala/org/apache/spark/scheduler/cluster/k8s/ExecutorPodsInformerSnapshotSourceSuite.scala
new file mode 100644
index 0000000000000..c0d3746857de4
--- /dev/null
+++ b/resource-managers/kubernetes/core/src/test/scala/org/apache/spark/scheduler/cluster/k8s/ExecutorPodsInformerSnapshotSourceSuite.scala
@@ -0,0 +1,114 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.spark.scheduler.cluster.k8s
+
+import io.fabric8.kubernetes.api.model.{Pod, PodBuilder}
+import io.fabric8.kubernetes.client.KubernetesClient
+import io.fabric8.kubernetes.client.informers.{ResourceEventHandler, SharedIndexInformer}
+import org.mockito.{ArgumentCaptor, Mock, Mockito, MockitoAnnotations}
+import org.mockito.Mockito._
+import org.scalatest.BeforeAndAfterEach
+import org.scalatestplus.mockito.MockitoSugar
+
+import org.apache.spark.{SparkConf, SparkFunSuite}
+import org.apache.spark.deploy.k8s.Config.KUBERNETES_EXECUTOR_INFORMER_RESYNC_INTERVAL
+import org.apache.spark.deploy.k8s.Constants.{SPARK_APP_ID_LABEL, SPARK_EXECUTOR_INACTIVE_LABEL, SPARK_POD_EXECUTOR_ROLE, SPARK_ROLE_LABEL}
+import org.apache.spark.deploy.k8s.Fabric8Aliases.{LABELED_PODS, PODS}
+import org.apache.spark.scheduler.cluster.k8s.ExecutorLifecycleTestUtils.{runningExecutor, TEST_SPARK_APP_ID}
+
+class ExecutorPodsInformerSnapshotSourceSuite
+  extends SparkFunSuite
+  with BeforeAndAfterEach
+  with MockitoSugar {
+
+  private var snapshotSource: ExecutorPodsInformerSnapshotSource = _
+  private var informerManager: InformerManager = _
+
+  private val sparkConf = new SparkConf()
+  private val resyncInterval = sparkConf.get(KUBERNETES_EXECUTOR_INFORMER_RESYNC_INTERVAL)
+  private val handlerCaptor: ArgumentCaptor[ResourceEventHandler[Pod]] =
+    ArgumentCaptor.forClass(classOf[ResourceEventHandler[Pod]])
+
+  @Mock
+  private var kubernetesClient: KubernetesClient = _
+
+  @Mock
+  private var snapshotsStore: ExecutorPodsSnapshotsStore = _
+
+  @Mock
+  private var informer: SharedIndexInformer[Pod] = _
+
+  @Mock
+  private var podOperations: PODS = _
+
+  @Mock
+  private var scopedPods: LABELED_PODS = _
+
+  override def beforeEach(): Unit = {
+    MockitoAnnotations.initMocks(this)
+
+    when(kubernetesClient.pods()).thenReturn(podOperations)
+    when(podOperations.withLabel(SPARK_APP_ID_LABEL, TEST_SPARK_APP_ID)).thenReturn(scopedPods)
+    when(scopedPods.withLabel(SPARK_ROLE_LABEL, SPARK_POD_EXECUTOR_ROLE)).thenReturn(scopedPods)
+    when(scopedPods.withoutLabel(SPARK_EXECUTOR_INACTIVE_LABEL, "true")).thenReturn(scopedPods)
+    when(scopedPods.runnableInformer(resyncInterval)).thenReturn(informer)
+    when(informer.isRunning).thenReturn(false)
+
+    informerManager = Mockito.spy[InformerManager](
+      new InformerManager(kubernetesClient, sparkConf))
+    snapshotSource = new ExecutorPodsInformerSnapshotSource(snapshotsStore, informerManager)
+  }
+
+  test("Informer should be run when snapshot source is started") {
+    snapshotSource.start(TEST_SPARK_APP_ID)
+    verify(informer, times(1)).run()
+  }
+
+  test("Informer should stop running when snapshot source is stopped") {
+    snapshotSource.start(TEST_SPARK_APP_ID)
+    snapshotSource.stop()
+    verify(informer, times(1)).close()
+  }
+
+  test("Informer onAdd/onUpdate/onDelete should push updates to the snapshots store") {
+    snapshotSource.start(TEST_SPARK_APP_ID)
+    verify(informer).addEventHandler(handlerCaptor.capture())
+
+    val exec1 = runningExecutor(1)
+    val exec2 = runningExecutor(2)
+    val exec2ResourceVersionChanged = withNewResourceVersion(exec2, "1")
+    val exec3 = runningExecutor(3)
+
+    val handler = handlerCaptor.getValue
+
+    handler.onAdd(exec1)
+    handler.onUpdate(exec2, exec2ResourceVersionChanged)
+    handler.onDelete(exec3, false)
+
+    verify(snapshotsStore).updatePod(exec1)
+    verify(snapshotsStore).updatePod(exec2ResourceVersionChanged)
+    verify(snapshotsStore).updatePod(exec3)
+  }
+
+  def withNewResourceVersion(pod: Pod, version: String): Pod = {
+    new PodBuilder(pod)
+      .editMetadata()
+      .withResourceVersion(version)
+      .endMetadata()
+      .build()
+  }
+}
diff --git a/resource-managers/kubernetes/core/src/test/scala/org/apache/spark/scheduler/cluster/k8s/ExecutorPodsListerSnapshotSourceSuite.scala b/resource-managers/kubernetes/core/src/test/scala/org/apache/spark/scheduler/cluster/k8s/ExecutorPodsListerSnapshotSourceSuite.scala
new file mode 100644
index 0000000000000..52481f598c94f
--- /dev/null
+++ b/resource-managers/kubernetes/core/src/test/scala/org/apache/spark/scheduler/cluster/k8s/ExecutorPodsListerSnapshotSourceSuite.scala
@@ -0,0 +1,100 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.spark.scheduler.cluster.k8s
+
+import java.util.concurrent.TimeUnit
+
+import io.fabric8.kubernetes.api.model.{Pod, PodListBuilder}
+import io.fabric8.kubernetes.client.KubernetesClient
+import io.fabric8.kubernetes.client.informers.SharedIndexInformer
+import io.fabric8.kubernetes.client.informers.cache.Indexer
+import org.jmock.lib.concurrent.DeterministicScheduler
+import org.mockito.{Mock, MockitoAnnotations}
+import org.mockito.Mockito.{verify, when}
+import org.scalatest.BeforeAndAfterEach
+
+import org.apache.spark.{SparkConf, SparkFunSuite}
+import org.apache.spark.deploy.k8s.Config._
+import org.apache.spark.deploy.k8s.Constants.{SPARK_APP_ID_LABEL, SPARK_EXECUTOR_INACTIVE_LABEL, SPARK_POD_EXECUTOR_ROLE, SPARK_ROLE_LABEL}
+import org.apache.spark.deploy.k8s.Fabric8Aliases.{LABELED_PODS, PODS}
+import org.apache.spark.scheduler.cluster.k8s.ExecutorLifecycleTestUtils._
+
+class ExecutorPodsListerSnapshotSourceSuite extends SparkFunSuite with BeforeAndAfterEach {
+
+  private val testNamespace = "test-namespace"
+  private val sparkConf = new SparkConf
+  private val pollingInterval = sparkConf.get(KUBERNETES_EXECUTOR_LISTER_POLLING_INTERVAL)
+  private val resyncInterval = sparkConf.get(KUBERNETES_EXECUTOR_INFORMER_RESYNC_INTERVAL)
+  private val pollingExecutor = new DeterministicScheduler
+
+  private var informerManager: InformerManager = _
+  private var snapshotSource: ExecutorPodsListerSnapshotSource = _
+
+  @Mock
+  private var informer: SharedIndexInformer[Pod] = _
+
+  @Mock
+  private var indexer: Indexer[Pod] = _
+
+  @Mock
+  private var snapshotsStore: ExecutorPodsSnapshotsStore = _
+
+  @Mock
+  private var kubernetesClient: KubernetesClient = _
+
+  @Mock
+  private var podOperations: PODS = _
+
+  @Mock
+  private var scopedPods: LABELED_PODS = _
+
+  override def beforeEach(): Unit = {
+    MockitoAnnotations.initMocks(this)
+
+    when(kubernetesClient.getNamespace).thenReturn(testNamespace)
+    when(kubernetesClient.pods()).thenReturn(podOperations)
+    when(podOperations.withLabel(SPARK_APP_ID_LABEL, TEST_SPARK_APP_ID)).thenReturn(scopedPods)
+    when(scopedPods.withLabel(SPARK_ROLE_LABEL, SPARK_POD_EXECUTOR_ROLE)).thenReturn(scopedPods)
+    when(scopedPods.withoutLabel(SPARK_EXECUTOR_INACTIVE_LABEL, "true")).thenReturn(scopedPods)
+    when(scopedPods.runnableInformer(resyncInterval)).thenReturn(informer)
+    when(informer.isRunning).thenReturn(false)
+    when(informer.getIndexer).thenReturn(indexer)
+
+    informerManager = new InformerManager(kubernetesClient, sparkConf)
+    snapshotSource = new ExecutorPodsListerSnapshotSource(
+      sparkConf, kubernetesClient, snapshotsStore, informerManager, pollingExecutor)
+    snapshotSource.start(TEST_SPARK_APP_ID)
+  }
+
+  test("Lister snapshot source pushes all current pods to snapshot store") {
+    val exec1 = runningExecutor(1)
+    val exec2 = runningExecutor(2)
+    val podList = new PodListBuilder().addToItems(exec1, exec2).build().getItems
+    when(indexer.byIndex("namespace", testNamespace)).thenReturn(podList)
+    pollingExecutor.tick(pollingInterval, TimeUnit.MILLISECONDS)
+    verify(snapshotsStore).replaceSnapshot(Seq(exec1, exec2))
+  }
+
+  test("Empty list of pods results in empty snapshot replacement") {
+    when(indexer.byIndex("namespace", testNamespace))
+      .thenReturn(new PodListBuilder().build().getItems)
+
+    pollingExecutor.tick(pollingInterval, TimeUnit.MILLISECONDS)
+
+    verify(snapshotsStore).replaceSnapshot(Seq.empty)
+  }
+}
diff --git a/resource-managers/kubernetes/core/src/test/scala/org/apache/spark/scheduler/cluster/k8s/InformerManagerSuite.scala b/resource-managers/kubernetes/core/src/test/scala/org/apache/spark/scheduler/cluster/k8s/InformerManagerSuite.scala
new file mode 100644
index 0000000000000..6ce05e7e8857c
--- /dev/null
+++ b/resource-managers/kubernetes/core/src/test/scala/org/apache/spark/scheduler/cluster/k8s/InformerManagerSuite.scala
@@ -0,0 +1,123 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.spark.scheduler.cluster.k8s
+
+import io.fabric8.kubernetes.api.model.Pod
+import io.fabric8.kubernetes.client.KubernetesClient
+import io.fabric8.kubernetes.client.informers.SharedIndexInformer
+import org.mockito.{Mock, MockitoAnnotations}
+import org.mockito.Mockito._
+import org.mockito.Mockito.verify
+import org.scalatest.BeforeAndAfter
+
+import org.apache.spark.{SparkConf, SparkFunSuite}
+import org.apache.spark.deploy.k8s.Config.KUBERNETES_EXECUTOR_INFORMER_RESYNC_INTERVAL
+import org.apache.spark.deploy.k8s.Constants.{SPARK_APP_ID_LABEL, SPARK_EXECUTOR_INACTIVE_LABEL, SPARK_POD_EXECUTOR_ROLE, SPARK_ROLE_LABEL}
+import org.apache.spark.deploy.k8s.Fabric8Aliases.{LABELED_PODS, PODS}
+
+class InformerManagerSuite extends SparkFunSuite with BeforeAndAfter {
+
+  @Mock
+  private var kubernetesClient: KubernetesClient = _
+
+  @Mock
+  private var informer: SharedIndexInformer[Pod] = _
+
+  @Mock
+  private var podOperations: PODS = _
+
+  @Mock
+  private var scopedPods: LABELED_PODS = _
+
+  private var conf: SparkConf = _
+  private val applicationId = "test-app-id"
+
+  before {
+    MockitoAnnotations.initMocks(this)
+    conf = new SparkConf().set(KUBERNETES_EXECUTOR_INFORMER_RESYNC_INTERVAL, 10000L)
+
+    // The informer is scoped server-side to app-id + role=executor + non-inactive pods.
+    // Chain all filter calls into the same mock so the final .runnableInformer(...) hits.
+    when(kubernetesClient.pods()).thenReturn(podOperations)
+    when(podOperations.withLabel(SPARK_APP_ID_LABEL, applicationId)).thenReturn(scopedPods)
+    when(scopedPods.withLabel(SPARK_ROLE_LABEL, SPARK_POD_EXECUTOR_ROLE)).thenReturn(scopedPods)
+    when(scopedPods.withoutLabel(SPARK_EXECUTOR_INACTIVE_LABEL, "true")).thenReturn(scopedPods)
+    when(scopedPods.runnableInformer(10000L)).thenReturn(informer)
+  }
+
+  test("If informer is null, initInformer should initialize it") {
+    val manager = new InformerManager(kubernetesClient, conf)
+    assert(manager.informer == null)
+    manager.initInformer(applicationId)
+    assert(manager.getInformer() == informer)
+  }
+
+  test("initInformer should scope the informer server-side to executor, non-inactive pods") {
+    val manager = new InformerManager(kubernetesClient, conf)
+    manager.initInformer(applicationId)
+    verify(podOperations).withLabel(SPARK_APP_ID_LABEL, applicationId)
+    verify(scopedPods).withLabel(SPARK_ROLE_LABEL, SPARK_POD_EXECUTOR_ROLE)
+    verify(scopedPods).withoutLabel(SPARK_EXECUTOR_INACTIVE_LABEL, "true")
+    verify(scopedPods).runnableInformer(10000L)
+  }
+
+  test("startInformer should not call run if informer is already running") {
+    when(informer.isRunning).thenReturn(true)
+    val manager = new InformerManager(kubernetesClient, conf)
+
+    manager.initInformer(applicationId)
+    manager.getInformer()
+    manager.startInformer()
+
+    verify(informer, times(0)).run()
+  }
+
+  test("stopInformer should close the informer and null it out") {
+    val manager = new InformerManager(kubernetesClient, conf)
+
+    manager.initInformer(applicationId)
+    manager.startInformer()
+    manager.stopInformer()
+
+    verify(informer).close()
+    assert(manager.informer == null)
+  }
+
+  test("getInformer should throw if the informer has not been initialized") {
+    val manager = new InformerManager(kubernetesClient, conf)
+    manager.initInformer(applicationId)
+    manager.startInformer()
+    assert(manager.getInformer() != null)
+    manager.stopInformer()
+    val e = intercept[IllegalStateException] {
+      manager.getInformer()
+    }
+    assert(e.getMessage.contains("Informer has not been initialized"))
+  }
+
+  test("Calling startInformer after stopInformer should throw") {
+    val manager = new InformerManager(kubernetesClient, conf)
+    manager.initInformer(applicationId)
+    manager.startInformer()
+    manager.stopInformer()
+    val e = intercept[IllegalStateException] {
+      manager.initInformer(applicationId)
+      manager.startInformer()
+    }
+    assert(e.getMessage.contains("Cannot run informer after stopInformer() has been called."))
+  }
+}
diff --git a/resource-managers/kubernetes/core/src/test/scala/org/apache/spark/scheduler/cluster/k8s/KubernetesClusterSchedulerBackendSuite.scala b/resource-managers/kubernetes/core/src/test/scala/org/apache/spark/scheduler/cluster/k8s/KubernetesClusterSchedulerBackendSuite.scala
index 64734cc6c612a..5567ff081f350 100644
--- a/resource-managers/kubernetes/core/src/test/scala/org/apache/spark/scheduler/cluster/k8s/KubernetesClusterSchedulerBackendSuite.scala
+++ b/resource-managers/kubernetes/core/src/test/scala/org/apache/spark/scheduler/cluster/k8s/KubernetesClusterSchedulerBackendSuite.scala
@@ -163,8 +163,7 @@ class KubernetesClusterSchedulerBackendSuite extends SparkFunSuite with BeforeAn
       eventQueue,
       podAllocator,
       lifecycleManager,
-      watchEvents,
-      pollEvents)
+      Seq(watchEvents, pollEvents))
   }
 
   private def registerExecutor(
@@ -489,8 +488,7 @@ class KubernetesClusterSchedulerBackendSuite extends SparkFunSuite with BeforeAn
       eventQueue,
       podAllocator,
       lifecycleManager,
-      watchEvents,
-      pollEvents)
+      Seq(watchEvents, pollEvents))
     val id1 = backendWithoutAppId.applicationId()
     val id2 = backendWithoutAppId.applicationId()
     assert(id1 === id2, "applicationId() must return the same value on repeated calls")