HwangDongJun opened a new pull request, #58861: URL: https://github.com/apache/spark/pull/58861
### What changes were proposed in this pull request? `LiveListenerBus` delivers events to independent, capacity-bounded queues. If the `executorManagement` queue is full when a `SparkListenerStageSubmitted` event arrives, the event is silently dropped and never redelivered. `ExecutorAllocationManager` only listens on that queue, so it never learns the stage exists, and dynamic allocation never scales up for it -- permanently, even after the stage finishes. Meanwhile the Spark UI, backed by `AppStatusStore` via the separate `appStatus` queue, can still correctly show the stage as `RUNNING`. This adds a lightweight, best-effort reconciliation on top of `ExecutorAllocationManager`'s existing ~100ms polling loop: - `schedule()` cheaply checks `LiveListenerBus`'s existing `executorManagement` drop counter (a single `Long` comparison) and, only when it has increased, calls `reconcile()`. - `reconcile()` diffs `AppStatusStore.activeStages()` against the manager's own bookkeeping and registers any stage AppStatusStore knows about but the manager doesn't, through the same code path as a normal `onStageSubmitted`. A recovered stage's ongoing pending/running counts are then read fresh from `AppStatusStore` (rather than reconstructing the manager's normal per-task-index bookkeeping) for as long as it remains "recovered," until it completes. - The same congestion that drops a stage's `StageSubmitted` could later also drop its `StageCompleted`. `reconcile()` also sweeps recovered attempts whose ground-truth status has since become terminal and cleans them up the same way a real `StageCompleted` would, so this can't leak for the remaining lifetime of the application. - The per-call ground-truth cache is a `TrieMap`: the `numberMaxNeededExecutors` gauge already calls into this same code from the metrics-reporting thread with no lock, concurrently with the allocation thread, so a plain `mutable.HashMap` would not be safe here. This intentionally does not introduce a new pub/sub framework analogous to K8s's `ExecutorPodsSnapshotsStore` (there is only one consumer here), and does not attempt a locality-optimized placement for recovered stages (their locality preferences aren't available from `AppStatusStore`'s lightweight `StageData`; the goal is a correct executor count, not optimal placement). Known limitation: a task that was speculatively submitted but not yet started during the drop window can't be recovered, since `AppStatusStore`'s lightweight path has no ground truth for "requested but not yet running" speculative tasks. This self-heals once the task actually starts and is picked up by normal running-task accounting. Related: SPARK-32597 previously proposed resizing the `executorManagement` queue dynamically to reduce drops, but did not reach consensus, and as noted there, no queue-capacity approach can fully eliminate drops. This takes a complementary angle: instead of preventing drops, it detects them cheaply and repairs their effect on dynamic allocation specifically, using ground truth Spark already maintains elsewhere. Follow-up to the documentation-only fix already merged for SPARK-58935. ### Why are the changes needed? A single dropped `SparkListenerStageSubmitted` event can permanently stall dynamic allocation for an application's remaining lifetime, with no self-recovery and no obvious symptom beyond `numberMaxNeededExecutors` staying at 0. This can only currently be worked around by increasing `spark.scheduler.listenerbus.eventqueue.executorManagement.capacity`, which lowers the probability of a drop but cannot eliminate it. ### Does this PR introduce _any_ user-facing change? No public API changes. Behavior only changes for applications that were already hitting this bug: dynamic allocation now self-heals instead of permanently under-requesting executors for the affected stage(s). ### How was this patch tested? Added unit tests to `ExecutorAllocationManagerSuite` covering: recovery of a dropped stage (single and multiple in a burst); a stage already completed on the manager's side is never resurrected by a stale `AppStatusStore` snapshot; in-flight task progress is reflected correctly without double-counting; a failed task on a recovered stage is re-counted as pending on retry; a delayed (not actually dropped) event arriving after recovery doesn't double-count; a failed reconciliation attempt is retried on the next tick; a recovered stage is attributed to its own (non-default) resource profile; the shrink path (`updateAndSyncNumExecutorsTarget`) correctly reflects a recovered-then-completed stage; and a recovered stage whose own `StageCompleted` is also dropped is still cleaned up. ### 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]
