[PR] [SPARK-59561][CORE] Reconcile ExecutorAllocationManager against AppStatusStore after a dropped StageSubmitted event [spark]

["HwangDongJun (via GitHub)" <gi...@apache.org>]

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: reviews-unsubscribe@spark.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org

Re: [PR] [SPARK-59561][CORE] Reconcile ExecutorAllocationManager against AppStatusStore after a dropped StageSubmitted event [spark]

["HwangDongJun (via GitHub)" <gi...@apache.org>]

HwangDongJun commented on PR #58861:
URL: https://github.com/apache/spark/pull/58861#issuecomment-6511753066

   @dongjoon-hyun This is the follow-up to the stall we discussed on #58216, where we only documented the existing drop metric. Since my first request on 9/29 I have added a check for each repair piece and fixed two tests that did not actually cover what they claimed to, and the CI is green. I also ran it on a real `local-cluster` with dynamic allocation: a stage whose submission event was dropped needed 0 executors before the change and 4 after it. The change is larger than the documentation one (about 190 lines of non-test code), so my question is whether a change like this is something you would want in `ExecutorAllocationManager`, and if so, how you would prefer it to be split.


-- 
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: reviews-unsubscribe@spark.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org

Re: [PR] [SPARK-59561][CORE] Reconcile ExecutorAllocationManager against AppStatusStore after a dropped StageSubmitted event [spark]

["HwangDongJun (via GitHub)" <gi...@apache.org>]

HwangDongJun commented on PR #58861:
URL: https://github.com/apache/spark/pull/58861#issuecomment-5883488136

   @dongjoon-hyun Would you be able to take a look at this when you have a chance?


-- 
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: reviews-unsubscribe@spark.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org