[SPARK-58935][CORE] Expose numDroppedExecutorManagementEvents metric

[HwangDongJun]

What changes were proposed in this pull request?

LiveListenerBus delivers the same logical event to several independent, separately-capacity-limited queues. ExecutorAllocationManager's listener is registered on the executorManagement queue, while the listener that drives the Spark UI is registered on a completely separate appStatus queue. Each queue independently drops events once its bounded capacity is exceeded, and a dropped event is never redelivered or resynced.

If a SparkListenerStageSubmitted event is dropped specifically from the executorManagement queue, ExecutorAllocationManager never learns about that stage's tasks, and its "executors needed" calculation permanently omits them for the lifetime of the stage -- even though the Spark UI/REST API (backed by the unaffected appStatus queue) continues to show the stage as RUNNING normally. There is currently no way to observe this happening: the only existing signal is a generic, easy-to-miss WARN log line shared by every queue, logged at most once per 60 seconds, with no indication of which downstream component is affected.

This PR adds LiveListenerBus.numDroppedExecutorManagementEvents, which exposes the executorManagement queue's dropped-event counter, and a corresponding ExecutorAllocationManager delegate method. It registers a new numDroppedExecutorManagementEvents gauge on ExecutorAllocationManagerSource, matching the existing pattern used by that source's other gauges (e.g. numberMaxNeededExecutors), so operators can alert on it and correlate "dynamic allocation stopped requesting executors" with "the executorManagement queue actually dropped an event."

A full self-healing fix (e.g., periodically reconciling ExecutorAllocationManager's bookkeeping against the ground-truth stage/task state already tracked by AppStatusStore) is a larger, more invasive change that needs broader design discussion. This PR is intentionally scoped to making the problem observable, not to fixing the underlying event-drop behavior.

Why are the changes needed?

Dynamic allocation can silently stop requesting new executors for an application that, from the UI/REST API and logs, looks completely healthy and busy, with no error, warning, or other signal indicating which component was affected. This was observed in production on a long-running Spark Connect driver, where ExecutorAllocationManager's JMX metrics showed numberMaxNeededExecutors = 0 and numberTargetExecutors = 0 with pending tasks on an active stage, while the Spark UI simultaneously reported the job as RUNNING. The only workaround was restarting the driver process.

SPARK-32597 previously identified that event drops in the async listener bus can cause general inconsistent application state, and proposed a more invasive VariableLinkedBlockingQueue approach (closed unmerged in 2020). This PR documents a specific, reproducible manifestation of that general class of problem, with a much narrower first fix.

SPARK-58446 reports a similarly-surfacing symptom (dynamic allocation stuck at zero needed/target executors), but from a distinct mechanism: a late TaskStart/SpeculativeTaskSubmitted event arriving after onStageCompleted corrupts the pending-task count. That fix does not touch onStageSubmitted and would not prevent or fix the issue described here, where stageAttemptToNumTasks is never populated for the affected stage attempt in the first place because the SparkListenerStageSubmitted event never reaches the listener.

Does this PR introduce any user-facing change?

Yes. A new numDroppedExecutorManagementEvents gauge metric is exposed under the ExecutorAllocationManagerSource metrics namespace. This PR does not change any existing behavior.

How was this patch tested?

• Added an ExecutorAllocationManagerSuite regression test that deterministically forces a SparkListenerStageSubmitted event to be dropped from the executorManagement queue (by setting its capacity to 1 and occupying its single dispatch thread with a blocking listener), and asserts that numDroppedExecutorManagementEvents reflects the drop and that maxNumExecutorsNeededPerResourceProfile is left at 0 for that stage, even though it has pending tasks.
• Added a SparkListenerSuite test verifying, in both directions, that numDroppedExecutorManagementEvents tracks drops on the executorManagement queue only, and is unaffected by drops on the shared queue.
• Ran the full ExecutorAllocationManagerSuite (38/38), SparkListenerSuite (24/24), and ExecutorMonitorSuite (16/16, regression check): all passed.
• scalastyle: no violations.

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

Generated-by: Claude Sonnet 5

[uros-b]

Thank you @HwangDongJun!

[dongjoon-hyun]

Thanks for the detailed writeup and for including a repro test. The failure mode you describe -- the executorManagement queue drops a SparkListenerStageSubmitted, dynamic allocation silently stops requesting executors, and the UI keeps showing the stage as RUNNING -- is real and worth documenting.

That said, I don't think we need a new metric for it, because the per-queue dropped-event counter is already exported.

AsyncEventQueue registers queue.<name>.numDroppedEvents on the LiveListenerBusMetrics registry when the queue is created:

https://github.com/apache/spark/blob/master/core/src/main/scala/org/apache/spark/scheduler/AsyncEventQueue.scala#L85

and that source (sourceName = "LiveListenerBus") is registered with the MetricsSystem in LiveListenerBus.start():

https://github.com/apache/spark/blob/master/core/src/main/scala/org/apache/spark/scheduler/LiveListenerBus.scala#L182

So <app-id>.driver.LiveListenerBus.queue.executorManagement.numDroppedEvents.count is already available today in JMX/Prometheus/any configured sink. The proposed ExecutorAllocationManagerSource gauge returns exactly that same counter under a second name.

It is also strictly less available than the existing one: ExecutorAllocationManagerSource is only registered when spark.dynamicAllocation.enabled=true

https://github.com/apache/spark/blob/master/core/src/main/scala/org/apache/spark/SparkContext.scala#L747-L749

whereas the LiveListenerBus source is always registered.

What I believe is the actual gap here is documentation. In docs/monitoring.md, the namespace=LiveListenerBus list spells out all three metrics for appStatus and eventLog, but stops at queue.executorManagement.listenerProcessingTime. queue.executorManagement.numDroppedEvents.count, queue.executorManagement.size, and the entire queue.shared.* group are missing:

https://github.com/apache/spark/blob/master/docs/monitoring.md?plain=1#L1327-L1338

If an operator could not find this signal, that omission seems the likely reason. Would you consider re-scoping this PR to fill in those missing lines? That makes exactly the signal you needed discoverable, with no new API surface and no duplicated metric.

A few other notes, in case parts of the change are kept:

• The description says the only existing signal is "a generic, easy-to-miss WARN log line". AsyncEventQueue also logs an ERROR on the first drop, including the queue name: https://github.com/apache/spark/blob/master/core/src/main/scala/org/apache/spark/scheduler/AsyncEventQueue.scala#L182-L187

• metrics.metricRegistry.counter(...) has get-or-add semantics, so the new accessor registers a permanent all-zero counter as a side effect if the executorManagement queue does not exist yet. It also duplicates the queue.<name>.numDroppedEvents name string in a second place, so a rename in AsyncEventQueue would make it silently return 0. Reading through metricRegistry.getCounters.get(...), or delegating to the queue itself, avoids both.

• The new ExecutorAllocationManagerSuite test looks racy. new ResourceProfileManager(conf, customBus) posts SparkListenerResourceProfileAdded from its constructor (ResourceProfileManager.scala L59-L60 and L159). With capacity=1, if the dispatch thread has not taken that event before post(SparkListenerJobStart(0, ...)) runs, JobStart(0) is the event that gets dropped and blockStarted.acquire() blocks forever. A customBus.waitUntilEmpty() right after creating the ResourceProfileManager would make it deterministic.

• assert(manager.maxNumExecutorsNeededPerResourceProfile(defaultProfile.id) === 0) passes both when the drop caused the stall and when nothing happened at all. A control case with a large capacity (expecting 2) would make the causality explicit. Note also that this assertion pins the current buggy behavior, so it has to be removed once the underlying issue is actually fixed; worth calling out in a comment.

• If a gauge does stay, please follow the naming used by the rest of ExecutorAllocationManagerSource (numberMaxNeededExecutors, numberDecommissioningExecutors, ...), and add it to the namespace=ExecutorAllocationManager list in docs/monitoring.md since the description marks this as a user-facing change.

[dongjoon-hyun]

Please consider the above comment and double-check, @HwangDongJun .