[SPARK-33737][K8S] Support getting pod state using Informers + Listers

[littlexyw]

What changes were proposed in this pull request?

Add an opt-in SharedIndexInformer + Lister based path for tracking executor pod state, as an alternative to ExecutorPodsWatchSnapshotSource + ExecutorPodsPollingSnapshotSource. A new trait ExecutorPodsSnapshotSource is extracted so the existing watch/poll sources and the new informer/lister sources share one interface. The two paths are mutually exclusive and selected by a new config; the default is unchanged.

Why are the changes needed?

ExecutorPodsPollingSnapshotSource issues a full pods().list() against the apiserver every spark.kubernetes.executor.apiPollingInterval (30s default) as a safety net for missed watch events. LIST hits the apiserver's in-memory watch cache and does not go to etcd, but the watch cache indexes only certain fields (namespace, name, nodeName…) and not labels, so the label selector is applied by scanning every Pod in the namespace and matching in memory. Per-request cost therefore scales with the namespace's total pod count rather than with the result size. When a shared K8s cluster hosts many concurrent Spark applications, the aggregate steady-state LIST QPS from all drivers is material.

SharedIndexInformer does one initial LIST and then keeps a local Indexer in sync via a single long-lived WATCH, resuming from the last observed resourceVersion on disconnect (only re-listing on HTTP 410). Lister.list() reads that local Indexer, so the periodic snapshot has zero apiserver cost. Optional informerResyncInterval replays from the local cache; it does not re-list against the apiserver.

Does this PR introduce any user-facing change?

No behavior change by default. Adds spark.kubernetes.executor.enableInformer (off by default) plus spark.kubernetes.executor.listerPollingInterval (default 30s) and spark.kubernetes.executor.informerResyncInterval (default 0s, disabled).

How was this patch tested?

New unit tests: ExecutorPodsInformerSnapshotSourceSuite, ExecutorPodsListerSnapshotSourceSuite, InformerManagerSuite. Manually verified on an internal Spark on K8s cluster.

Was this patch authored or co-authored using generative AI tooling?

Assisted-by: Claude Opus 4.7

[littlexyw]

@LuciferYang @dongjoon-hyun Could you please help to review this?

[LuciferYang]

Thanks for the PR — the direction (one LIST + a long-lived watch instead of a periodic full LIST) makes sense to me. My main concern is the startup/exception semantics of the informer, covered by comments 1-3, which I think need to land together: switching run() to the non-blocking start() alone would trade a startup crash for a silently dead informer. Comments are numbered and ordered by severity (1-3 must-fix, 4-5 should follow in this PR, 6-11 minor).

[LuciferYang]

1. startInformer() runs the blocking, unbounded run() on the SparkContext creation thread

InformerManager.startInformer() calls informer.run() directly; in fabric8 7.x run() blocks
until the initial LIST completes and the watch is established, and this runs on the SparkContext
creation thread, whereas the legacy watch/polling sources start fully asynchronously. From the
v7.8.0 Reflector source, the default exception handler declines to retry any error before the
first successful sync, so an apiserver hiccup at startup (throttling, transient error) makes
run() throw on the calling thread and SparkContext creation fail outright; a slow-but-progressing
initial LIST (large namespace) blocks startup with no timeout and no log. A cold start lists ~0
matching pods, so manual verification wouldn't show it.

Please switch to start() (which doesn't block the caller), but this must come with an
exceptionHandler that forces retries (see my other comment) — otherwise you've only traded a
startup crash for an informer that dies silently in the background. The lister poll also needs
a hasSynced() check.

[LuciferYang]

2. No exceptionHandler: startup errors are not retried, and a mid-run informer death is silent

InformerManager sets no exceptionHandler and never consumes stopped(). In fabric8 7.x the
ExceptionHandler decides retry-vs-stop, and the default chooses stop for any error before the
first successful sync (it already logs internally — what's missing is the retry decision), and
for non-GONE WatcherExceptions after startup too. So with the current code, a mid-run informer
death is silent: events stop, the lister keeps replacing snapshots from a cache that no longer
updates, and dead executors wait for RPC timeouts to be noticed. And if start becomes async per
my startup comment without a handler, any startup error turns into a silent background death
that the hasSynced check would then skip past forever; strictly worse than crashing.

Please set exceptionHandler((b, t) => { logError(...); true }) in initInformer (it can only
be set before start) to log and force a retry, or consume the exceptional completion of
stopped() and fail the driver.

[LuciferYang]

3. The lister poll doesn't check hasSynced(), so an unsynced empty cache would wipe the snapshot store

PollRunnable unconditionally calls replaceSnapshot(lister.list()), but the informer's local
cache is empty until the initial LIST finishes. replaceSnapshot replaces wholesale with a fresh
fullSnapshotTs, so ExecutorPodsLifecycleManager's missing-pod reconcile fires on every poll
and removes every executor registered more than missingPodDetectDelta (30s) ago via
doRemoveExecutor, while the allocator re-requests a full batch seeing zero known executors.
Today the window is masked by the blocking run() (by the time backend.start() returns, the
cache is synced), but it opens the moment start becomes async per my startup comment: any
namespace where the initial sync is slower than listerPollingInterval hits it.

Skipping the round when !informer.hasSynced() fixes it; the "Empty list of pods" test should
then assert no replacement happens while unsynced. Note this check needs the force-retry
exceptionHandler from my exceptionHandler comment as well — a never-synced, dead informer would
otherwise mean polls skipped forever.

[LuciferYang]

4. The three new user-facing configs are undocumented

The three new user-facing configs (spark.kubernetes.executor.enableInformer,
listerPollingInterval, informerResyncInterval) have no entries anywhere under docs/, while
the closest precedent spark.kubernetes.executor.apiPollingInterval is documented in the config
table in docs/running-on-kubernetes.md. The mutual exclusion with the two legacy switches
currently lives only in the config doc string, so users can't discover the switch or the
migration notes from the docs; it's also worth noting there that the informer path requires
both list and watch permissions on pods. Could you add the three entries to
running-on-kubernetes.md?

[LuciferYang]

5. The mode selection in makeSnapshotSources has no test, and the method is private

makeSnapshotSources is private, while the sibling makeExecutorPodsAllocator in the same
file is private[k8s] for testability, and KubernetesClusterManagerSuite never references it
or enableInformer. So the core wiring of this PR (which pair of sources the flag selects) has
no coverage, and the informer-mode sources are never run through
KubernetesClusterSchedulerBackend's start/stop; a wrong branch or a missing listerExecutor
would pass every existing test. Could you make it private[k8s] and add a test asserting the
selected source types per flag value?

[LuciferYang]

7. resync > 0 replays N snapshots per round (churn only; off by default)

With informerResyncInterval > 0, each resync replays onUpdate for every pod; the handler calls
updatePod unconditionally and ExecutorPodsSnapshot.withUpdate doesn't dedup by
resourceVersion, so N executors produce N snapshot objects per round and both subscribers rescan
everything. It's idempotent (fullSnapshotTs is preserved), so this is churn rather than a
correctness issue, and the default resync=0 avoids it — just noting a driver-side cost
proportional to the executor count when the interval is set low. Comparing resourceVersion
before updatePod in the handler would cap it.

[LuciferYang]

8. start() has no double-start guard, and a misleading INFO fires on every normal startup

Neither new source's start() has the double-start guard the legacy ones have ("Cannot start
the watcher twice." / "Cannot start polling more than once."); starting the informer source
twice would add a second handler and duplicate events, and a second lister start would overwrite
and leak the first pollingFuture. Also, since the informer source runs the informer first and
the lister source then sees isRunning=true, the logInfo("Informer is already running.") in
InformerManager fires on every normal startup and reads like something went wrong. Adding the
require guards and demoting that log to debug would match the existing sources.

[LuciferYang]

9. The three new test suites diverge from the conventions of the neighboring suites

A few conventions diverge from the neighboring suites: all three use the deprecated
MockitoAnnotations.initMocks while every existing suite here uses openMocks(this).close();
the Mockito.spy[InformerManager] in the informer suite uses no spy feature (the lister suite
just calls new); with MockitoSugar is unused; handlerCaptor is a class-level val instead of
being rebuilt in beforeEach; InformerManagerSuite has a redundant import org.mockito.Mockito.verify (the wildcard import already provides it); and the five-line
label-filter mock chain is copy-pasted verbatim in all three suites, which would suit a small
shared helper. Also, "getInformer should throw if the informer has not been initialized"
actually exercises the init→start→stop path rather than a fresh manager. None of this blocks —
fine as a quick cleanup.

[LuciferYang]

10. The new config toggle lands ~90 lines away from its family, and this PR splits the legacy family too

KUBERNETES_EXECUTOR_ENABLE_INFORMER lands at line 579 (between POD_DELETION_COST and
ALLOCATION_BATCH_SIZE), while its semantic siblings ENABLE_API_POLLING/ENABLE_API_WATCHER
and the two new interval configs sit ~90 lines further down. The two new intervals are also
inserted between the legacy toggles and API_POLLING_INTERVAL, so this PR splits the legacy
family that used to be contiguous as well. Moving the toggle down next to the two intervals,
and placing the new intervals after API_POLLING_INTERVAL, would keep both families together.

[LuciferYang]

11. The four new types are public with no annotation, inconsistent with adjacent types

ExecutorPodsSnapshotSource, InformerManager, and the two new sources are all public with no
annotation, while the types they sit next to differ: ExecutorPodsSnapshotsStore is
private[spark], and the two legacy sources carry @Stable @DeveloperApi on the classes and
@Since on the methods. The new
trait now sits in the public hierarchy of two @DeveloperAPI classes without being @DeveloperAPI
itself, and InformerManager is pure internal wiring with no need to be public. Marking the
trait @DeveloperAPI and narrowing InformerManager to private[spark] (or the whole group) would
avoid widening the API surface first and paying a breaking change to shrink it later.