[jira] [Updated] (SPARK-58935) ExecutorAllocationManager silently stalls at zero needed executors when a stage-submitted event is dropped from the executorManagement queue

["ASF GitHub Bot (Jira)" <ji...@apache.org>]

     [ https://issues.apache.org/jira/browse/SPARK-58935?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ]

ASF GitHub Bot updated SPARK-58935:
-----------------------------------
    Labels: pull-request-available  (was: )

> ExecutorAllocationManager silently stalls at zero needed executors when a stage-submitted event is dropped from the executorManagement queue
> --------------------------------------------------------------------------------------------------------------------------------------------
>
>                 Key: SPARK-58935
>                 URL: https://issues.apache.org/jira/browse/SPARK-58935
>             Project: Spark
>          Issue Type: Bug
>          Components: Scheduler, Spark Core
>    Affects Versions: 4.0.1
>         Environment: Kubernetes deployment, Spark Connect (long-running driver), Spark 4.0.1
>            Reporter: dongjunhwang
>            Priority: Major
>              Labels: pull-request-available
>
> h3. Problem
> {{LiveListenerBus}} delivers the same logical event to several independent, separately-capacity-limited queues. {{ExecutorAllocationManager}}'s listener is registered on the {{executorManagement}} queue ({{LiveListenerBus.addToManagementQueue}}, [LiveListenerBus.scala#L70|https://github.com/apache/spark/blob/5abbd34c837571b059de659898bbf699f6c6d3fb/core/src/main/scala/org/apache/spark/scheduler/LiveListenerBus.scala#L70]), while the listener that drives the Spark UI is registered on a completely separate {{appStatus}} queue ({{addToStatusQueue}}, [LiveListenerBus.scala#L75|https://github.com/apache/spark/blob/5abbd34c837571b059de659898bbf699f6c6d3fb/core/src/main/scala/org/apache/spark/scheduler/LiveListenerBus.scala#L75]).
> Each queue independently drops events once its bounded capacity is exceeded ({{AsyncEventQueue.post}}, [AsyncEventQueue.scala#L168|https://github.com/apache/spark/blob/5abbd34c837571b059de659898bbf699f6c6d3fb/core/src/main/scala/org/apache/spark/scheduler/AsyncEventQueue.scala#L168]). A dropped event is never redelivered or resynced.
> {{ExecutorAllocationManager}}'s "executors needed" calculation is driven entirely by {{stageAttemptToNumTasks}}, which is populated *only* in {{onStageSubmitted}} ([ExecutorAllocationManager.scala#L687|https://github.com/apache/spark/blob/5abbd34c837571b059de659898bbf699f6c6d3fb/core/src/main/scala/org/apache/spark/ExecutorAllocationManager.scala#L687]) and consumed by {{maxNumExecutorsNeededPerResourceProfile}} ([ExecutorAllocationManager.scala#L297|https://github.com/apache/spark/blob/5abbd34c837571b059de659898bbf699f6c6d3fb/core/src/main/scala/org/apache/spark/ExecutorAllocationManager.scala#L297]). If the {{SparkListenerStageSubmitted}} event for a given stage is dropped specifically from the {{executorManagement}} queue:
> * The {{appStatus}} queue is unaffected, so the Spark UI/REST API continues to show the stage as {{RUNNING}} normally.
> * {{ExecutorAllocationManager}} never learns about that stage's tasks. {{maxNumExecutorsNeededPerResourceProfile}} permanently omits them from its calculation for the lifetime of the stage -- there is no periodic resync.
> * When the stage eventually completes, {{onStageCompleted}} removes a {{StageAttempt}} key that was never inserted; this is a silent no-op on the underlying {{mutable.HashMap}}, so there is no error, warning, or other signal that anything went wrong.
> The net effect: dynamic allocation can stop requesting new executors for an application that, from the UI/REST API and logs, looks completely healthy and busy.
> h3. How this was found
> Observed on a long-running (2+ day uptime) Spark Connect driver (Spark 4.0.1) in a Kubernetes deployment. {{ExecutorAllocationManager}}'s JMX metrics showed:
> {code}
> numberMaxNeededExecutors = 0   (with 2 pending tasks on an active stage)
> numberTargetExecutors    = 0
> {code}
> while the Spark UI's REST API ({{/api/v1/applications/<id>/jobs}}) simultaneously reported the job as {{RUNNING}}. A completely unrelated, concurrently-running session on the same driver process was independently affected in the same way, ruling out any per-job cause. The only workaround was restarting the driver process. Thread dumps showed the dynamic-allocation scheduling thread alive and running normally (not deadlocked) -- only the *calculation result* was wrong.
> h3. Reproduction
> Deterministically reproduced by constructing a {{LiveListenerBus}} with {{spark.scheduler.listenerbus.eventqueue.executorManagement.capacity=1}}, occupying the queue's single dispatch thread with a blocking listener, and posting a {{SparkListenerStageSubmitted}} event for a 2-task stage once the queue is full. The event is dropped (confirmed via {{queue.executorManagement.numDroppedEvents}}), and {{ExecutorAllocationManager.maxNumExecutorsNeededPerResourceProfile}} returns {{0}} for that stage, even though 2 tasks are pending.
> In a real cluster, this queue can fill up under a burst of many other events sharing the same queue ({{HeartbeatReceiver}} is also registered on {{executorManagement}}, [LiveListenerBus.scala#L70|https://github.com/apache/spark/blob/5abbd34c837571b059de659898bbf699f6c6d3fb/core/src/main/scala/org/apache/spark/scheduler/LiveListenerBus.scala#L70] area), or if a listener on that queue is briefly slow.
> h3. Expected behavior and proposed fix (this PR)
> 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.
> As a first, narrowly-scoped step, this PR makes the problem *observable* instead of silent: it exposes the executor-management queue's dropped-event count as a new {{ExecutorAllocationManager}} metric ({{numDroppedExecutorManagementEvents}}), so operators can alert on it and correlate "dynamic allocation stopped requesting executors" with "the executorManagement queue actually dropped an event" -- which is not currently possible (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 does not change any existing behavior; it only adds a new gauge metric, matching the existing pattern used by {{ExecutorAllocationManagerSource}}'s other gauges (e.g. {{numberMaxNeededExecutors}}).
> h3. Related work
> * SPARK-32597 (still open, unassigned) previously identified that event drops in the async listener bus can cause "inconsistent state for the Spark Application (sometimes application is hung state)" in general, and proposed a more invasive {{VariableLinkedBlockingQueue}} approach ([PR #29413|https://github.com/apache/spark/pull/29413], closed unmerged in 2020). This ticket documents a specific, reproducible manifestation of that general class of problem, with a much narrower first fix.
> * SPARK-26927 (Fixed) previously fixed a different specific mechanism -- event *ordering* (not dropping) across queues -- that similarly corrupted {{ExecutorAllocationManager}}'s bookkeeping and caused a production Spark Thrift Server hang.
> * SPARK-58446 (open, unassigned, [PR #57651|https://github.com/apache/spark/pull/57651]) reports a similarly-surfacing symptom -- dynamic allocation stuck at zero needed/target executors -- but from a distinct mechanism confirmed at the code level: a late {{TaskStart}}/{{SpeculativeTaskSubmitted}} event arriving *after* {{onStageCompleted}} has already removed the stage's bookkeeping re-creates stale state and drives the pending-task count negative. That fix guards {{onTaskStart}}/{{onSpeculativeTaskSubmitted}} with {{stageAttemptToNumTasks.contains(stageAttempt)}}; it 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.



--
This message was sent by Atlassian Jira
(v8.20.10#820010)

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