HwangDongJun opened a new pull request, #58216: URL: https://github.com/apache/spark/pull/58216
### 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 -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
