[
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: [email protected]
For additional commands, e-mail: [email protected]
