dongjunhwang created SPARK-58935:
------------------------------------
Summary: 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
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: [email protected]
For additional commands, e-mail: [email protected]
