You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by "zahed1994 (via GitHub)" <gi...@apache.org> on 2026/08/28 09:51:15 UTC
[PR] [SPARK-49485][CORE] Request additional executor for speculative tasks when active executors equal maxNeeded [spark]
zahed1994 opened a new pull request, #58380:
URL: https://github.com/apache/spark/pull/58380
### What changes were proposed in this pull request?
When `spark.dynamicAllocation.enabled=true` and `spark.speculation=true`, straggling tasks requiring speculative execution can stall indefinitely if all remaining active executors reside on the same host.
In `ExecutorAllocationManager.scala`, `maxNumExecutorsNeededPerResourceProfile` calculates `maxNeeded` strictly based on `(running + pendingTasks + pendingSpeculative) / tasksPerExecutor`. If the current active executor count equals `maxNeeded`, `ExecutorAllocationManager` calculates target executors as equal to the active count and requests no new executors. At the same time, Spark's task scheduler avoids launching speculative task copies on an executor on the same host where the task is already running slow. Consequently, no active executor can run the speculative task and no new executor is requested, causing speculative tasks to stall indefinitely.
This PR updates `maxNumExecutorsNeededPerResourceProfile` in `ExecutorAllocationManager.scala` to allocate an additional target executor when `pendingSpeculative > 0` and `maxNeeded` equals the current active executor count. This allows `ExecutorAllocationManager` to request an extra executor from the cluster manager (YARN / K8s / Standalone) on a distinct host to execute the speculative task.
### Why are the changes needed?
Without this change, applications using Dynamic Allocation and speculation can hang indefinitely when remaining executors reside on the same slow host.
### Does this PR introduce _any_ user-facing change?
No.
### How was this patch tested?
- Added unit test `SPARK-49485: request additional executor when speculative tasks equal maxNeeded` in `ExecutorAllocationManagerSuite.scala`.
- Verified cleanly via `core/compile` and `core/scalastyle`.
--
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-49485][CORE] Request additional executor for speculative tasks when active executors equal maxNeeded [spark]
Posted by "sunchao (via GitHub)" <gi...@apache.org>.
sunchao commented on code in PR #58380:
URL: https://github.com/apache/spark/pull/58380#discussion_r3883291363
##########
core/src/test/scala/org/apache/spark/ExecutorAllocationManagerSuite.scala:
##########
@@ -2065,6 +2065,31 @@ class ExecutorAllocationManagerSuite extends SparkFunSuite {
private def executorsDecommissioning(manager: ExecutorAllocationManager): Set[String] = {
manager.executorMonitor.executorsDecommissioning()
}
+
+ test("SPARK-49485: request additional executor when speculative tasks equal maxNeeded") {
+ val manager = createManager(createConf(1, 5, 2))
+ assert(maxNumExecutorsNeededPerResourceProfile(manager, defaultProfile) === 0)
+
+ // Submit stage with 2 tasks
+ post(SparkListenerStageSubmitted(createStageInfo(0, 2)))
+ post(SparkListenerExecutorAdded(0, "executor-1", new ExecutorInfo("host1", 1, Map.empty)))
+ post(SparkListenerExecutorAdded(0, "executor-2", new ExecutorInfo("host1", 1, Map.empty)))
+
+ // Task 0 and Task 1 start on executor-1 and executor-2
+ val taskInfo0 = createTaskInfo(0, 0, "executor-1")
+ val taskInfo1 = createTaskInfo(1, 1, "executor-2")
+ post(SparkListenerTaskStart(0, 0, taskInfo0))
+ post(SparkListenerTaskStart(0, 0, taskInfo1))
+
+ // 2 tasks running, 2 active executors. maxNeeded = 2
+ assert(maxNumExecutorsNeededPerResourceProfile(manager, defaultProfile) === 2)
+
+ // Task 0 is submitted as speculatable
+ post(SparkListenerSpeculativeTaskSubmitted(0, 0))
+
+ // With pendingSpeculative > 0 and maxNeeded == activeExecutors (2), offset allocates 1 more
+ assert(maxNumExecutorsNeededPerResourceProfile(manager, defaultProfile) === 3)
Review Comment:
**[P2] Remove the extra pending regular task from the fixture**
The revised fixture still leaves one regular task pending: `createStageInfo(0, 4)` declares four regular tasks, but only three `TaskStart` events are posted. After speculative submission, raw `maxNeeded` is `ceil((3 running + 1 regular pending + 1 speculative pending) / 2) = 3`, while the executor count is 2. The new branch still never executes, so the assertion passes without the fix.
Use `createStageInfo(0, 3)` to establish the intended base=2/head=3 distinction.
##########
core/src/test/scala/org/apache/spark/ExecutorAllocationManagerSuite.scala:
##########
@@ -2065,6 +2065,34 @@ class ExecutorAllocationManagerSuite extends SparkFunSuite {
private def executorsDecommissioning(manager: ExecutorAllocationManager): Set[String] = {
manager.executorMonitor.executorsDecommissioning()
}
+
+ test("SPARK-49485: request additional executor when speculative tasks equal maxNeeded") {
+ val conf = createConf(1, 5, 2).set(config.EXECUTOR_CORES, 2)
+ val manager = createManager(conf)
+ assert(maxNumExecutorsNeededPerResourceProfile(manager, defaultProfile) === 0)
+
+ // Submit stage with 4 tasks
+ post(SparkListenerStageSubmitted(createStageInfo(0, 4)))
+ post(SparkListenerExecutorAdded(0, "executor-1", new ExecutorInfo("host1", 2, Map.empty)))
+ post(SparkListenerExecutorAdded(0, "executor-2", new ExecutorInfo("host2", 2, Map.empty)))
+
+ // 3 tasks running across 2 active 2-core executors
+ val taskInfo0 = createTaskInfo(0, 0, "executor-1")
+ val taskInfo1 = createTaskInfo(1, 1, "executor-1")
+ val taskInfo2 = createTaskInfo(2, 2, "executor-2")
+ post(SparkListenerTaskStart(0, 0, taskInfo0))
+ post(SparkListenerTaskStart(0, 0, taskInfo1))
+ post(SparkListenerTaskStart(0, 0, taskInfo2))
+
+ // 3 tasks running, 2 active 2-core executors. maxNeeded = ceil(3/2) = 2
+ assert(maxNumExecutorsNeededPerResourceProfile(manager, defaultProfile) === 2)
+
+ // Task 0 is submitted as speculatable (3 running + 1 speculative = 4 tasks -> ceil(4/2) = 2)
+ post(SparkListenerSpeculativeTaskSubmitted(0, 0))
+
+ // With pendingSpeculative > 0 and maxNeeded == activeExecutors (2), offset allocates 1 more -> 3
Review Comment:
**[P2] Wrap the comment to restore Scala lint**
This revised comment is 101 characters long, exceeding the configured 100-character limit. The [current CI Scala-linter step](https://github.com/zahed1994/spark/actions/runs/33198089279/job/98941490733) fails at this exact line with `File line length exceeds 100 characters`. Wrap or shorten the comment.
--
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-49485][CORE] Request additional executor for speculative tasks when active executors equal maxNeeded [spark]
Posted by "sunchao (via GitHub)" <gi...@apache.org>.
sunchao commented on PR #58380:
URL: https://github.com/apache/spark/pull/58380#issuecomment-5817624853
Sorry, checking again
--
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-49485][CORE] Request additional executor for speculative tasks when active executors equal maxNeeded [spark]
Posted by "sunchao (via GitHub)" <gi...@apache.org>.
sunchao commented on code in PR #58380:
URL: https://github.com/apache/spark/pull/58380#discussion_r4096111446
##########
core/src/main/scala/org/apache/spark/ExecutorAllocationManager.scala:
##########
@@ -878,7 +884,7 @@ private[spark] class ExecutorAllocationManager(
stageAttemptToNumTasks -= stageAttempt
stageAttemptToPendingSpeculativeTasks -= stageAttempt
stageAttemptToTaskIndices -= stageAttempt
- stageAttemptToSpeculativeTaskIndices -= stageAttempt
+ stageAttemptToSpeculativeTaskIds -= stageAttempt
Review Comment:
**[P2] Preserve running speculative attempts across stage retries** — [ExecutorAllocationManager.scala:887](https://github.com/apache/spark/blob/adfbfafafd96721e8acaf7491d01b04fa8e23b23/core/src/main/scala/org/apache/spark/ExecutorAllocationManager.scala#L887).
A retryable `FetchFailed` emits `StageCompleted` while other attempts can still produce useful results. Clearing their speculative IDs here makes the new allocation logic forget a running speculative copy.
With three slots per executor, five regular attempts plus one speculative copy require target **3**. After an unrelated task fetch-fails and the stage completes for retry, the target drops to **2**, although the copy remains running. StatefulSet scale-down can then delete its newly added executor and discard that work. [Scaling behavior](https://kubernetes.io/docs/concepts/workloads/controllers/statefulset/#deployment-and-scaling-guarantees).
Keep live speculative IDs until their `TaskEnd` events, with eventual map/resource-profile cleanup.
Validation:
- Reproduced the **3→2** drop using the exact changed source and listener events. Retaining the IDs in an isolation control keeps **3→3**.
- All **50 existing tests** passed locally using cached dependencies; all **33 fork CI checks** passed.
- Kubernetes deletion was traced through source and documented behavior, not reproduced on a cluster.
--
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-49485][CORE] Request additional executor for speculative tasks when active executors equal maxNeeded [spark]
Posted by "sunchao (via GitHub)" <gi...@apache.org>.
sunchao commented on PR #58380:
URL: https://github.com/apache/spark/pull/58380#issuecomment-5482352298
The previously reported StatefulSet registration issue is addressed in reviewed commit `1ac32efce4cb93c6ea31d6955af4d54f09ceede6`. Two P2 issues remain.
**[P2] Count running speculative attempts, not distinct task indices**
The new [getRunningSpeculativeTaskSum](https://github.com/apache/spark/blob/1ac32efce4cb93c6ea31d6955af4d54f09ceede6/core/src/main/scala/org/apache/spark/ExecutorAllocationManager.scala#L1073-L1074) returns the size of a set keyed by task index. After a primary attempt fails, its surviving speculative attempt S1 can itself be speculated: [TaskSetManager checks every running attempt](https://github.com/apache/spark/blob/1ac32efce4cb93c6ea31d6955af4d54f09ceede6/core/src/main/scala/org/apache/spark/scheduler/TaskSetManager.scala#L1311-L1315), and the dequeue guard permits another speculative attempt when `copiesRunning(index) == 1`. S1 and S2 both have `speculative=true`, but their starts produce only one set entry. Ending either attempt also removes that entry while the other remains running.
A concrete allocation consequence uses three task slots per executor and allocation ratio 1. Let a two-task result stage have one completed fast task and a slow primary P; two unrelated regular tasks from another stage use the same resource profile. Executor B runs those two regular tasks plus S1, the speculative copy of P. P's executor A fails, leaving one regular task pending, and replacement executor C can later launch S2 after S1 becomes speculatable again.
While S2 is pending, the estimate is 2. Once S2 starts, there are still five running-or-pending attempts: one regular pending and four running, of which two are speculative. The accessor reports only one running speculative attempt, so the regular baseline becomes `ceil((1 + 4 - 1) / 3) = 2`. Since raw `maxNeeded` is also 2, the new branch raises the target to 3. Counting both speculative attempts gives a regular baseline of 1 and keeps the target at 2, as do the base and previous revision after launch. The regular pending entry keeps the backlog timer active, so this can request an unnecessary executor even though B and C already run all four attempts.
Please track live speculative attempt IDs or equivalent multiplicities, and cover re-speculation after a primary failure. This finding is based on source inspection, not a local runtime reproduction.
**[P2] Update the existing regression expectations for retained speculation**
The existing `SPARK-41192` assertions at [lines 758-759](https://github.com/apache/spark/blob/1ac32efce4cb93c6ea31d6955af4d54f09ceede6/core/src/test/scala/org/apache/spark/ExecutorAllocationManagerSuite.scala#L758-L759) and `SPARK-30511` assertions at [lines 865-866](https://github.com/apache/spark/blob/1ac32efce4cb93c6ea31d6955af4d54f09ceede6/core/src/test/scala/org/apache/spark/ExecutorAllocationManagerSuite.scala#L865-L866) still expect 1 executor while speculative attempts are running. The revised retention behavior intentionally returns 2 in both cases, and the [current core CI job](https://github.com/zahed1994/spark/actions/runs/33369053539/job/99420145042) fails both tests with `2 did not equal 1`.
The later assertions at [lines 886-887](https://github.com/apache/spark/blob/1ac32efce4cb93c6ea31d6955af4d54f09ceede6/core/src/test/scala/org/apache/spark/ExecutorAllocationManagerSuite.scala#L886-L887) also need the corresponding update; that mismatch is source-established because the earlier failure prevents CI from reaching it. These failures call for updating the old expectations, not reverting the intended retention behavior.
Validation: the exact-head CI artifact reports 47 passes and 2 failures out of 49 allocation tests. Both new tests pass, and Scala lint passes.
--
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-49485][CORE] Request additional executor for speculative tasks when active executors equal maxNeeded [spark]
Posted by "zahed1994 (via GitHub)" <gi...@apache.org>.
zahed1994 commented on PR #58380:
URL: https://github.com/apache/spark/pull/58380#issuecomment-5819035552
Thanks @sunchao, good catch on the stage retry path.
I've updated the PR accordingly in commit 19e31aab941:
- **Retain running speculative IDs across `onStageCompleted`**: Removed `stageAttemptToSpeculativeTaskIds -= stageAttempt` from `onStageCompleted` (mirroring `stageAttemptToNumRunningTask`). Running speculative copies remain counted in `runningSpeculativeTasksPerResourceProfile`, preventing the target from dropping (3→2) when a stage completes early for retry.
- **Cleanup on `onTaskEnd`**: When a speculative task finishes in `onTaskEnd`, its `taskId` is removed. Once the set is empty, the entry is removed from `stageAttemptToSpeculativeTaskIds` and `removeStageFromResourceProfileIfUnused` is triggered so the resource profile mapping cleans up cleanly.
- **Regression test**: Added `"SPARK-49485: preserve running speculative attempts across stage retries"` in `ExecutorAllocationManagerSuite` verifying target retention at 3 across `StageCompleted` until `TaskEnd`.
All 51 tests in `ExecutorAllocationManagerSuite` and `scalastyle` pass cleanly.
--
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-49485][CORE] Request additional executor for speculative tasks when active executors equal maxNeeded [spark]
Posted by "zahed1994 (via GitHub)" <gi...@apache.org>.
zahed1994 commented on PR #58380:
URL: https://github.com/apache/spark/pull/58380#issuecomment-5493147044
Thanks @sunchao, good catches on both points.
I've updated the PR in the latest commit:
1. Track running speculative attempts by `taskId`: Changed
`stageAttemptToSpeculativeTaskIndices` to
`stageAttemptToSpeculativeTaskIds`, storing `taskInfo.taskId` as a `Long`
instead of the logical task index. This ensures multiple speculative
attempts for the same task index are counted independently. Added a
regression test covering multiple running speculative attempts for the
same task index.
2. Update retention expectations: Updated the existing SPARK-41192 and
SPARK-30511 assertions to expect 2 executors while speculative attempts
remain active, reflecting the intended retention behavior.
The `ExecutorAllocationManagerSuite` passes all 50 tests, including the new
regression coverage, and scalastyle passes cleanly.
Could you please take another look when you get 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
Re: [PR] [SPARK-49485][CORE] Request additional executor for speculative tasks when active executors equal maxNeeded [spark]
Posted by "zahed1994 (via GitHub)" <gi...@apache.org>.
zahed1994 commented on PR #58380:
URL: https://github.com/apache/spark/pull/58380#issuecomment-5475191035
Thanks @sunchao, good catch on the StatefulSet case.
You're right that the previous logic could cause the target to drop back to 2 once the additional executor was registered or the speculative task moved from pending to running. In the StatefulSet allocator, that can immediately trigger a scale-down and remove the newly allocated executor before the speculative task has a chance to make progress.
I've updated the allocation logic in `ExecutorAllocationManager.scala` to account for the full lifecycle of the speculative task rather than only the pending state.
Specifically:
1. **Track the full speculative task lifecycle**
`ExecutorAllocationListener` now provides the number of speculative tasks that are either pending or currently running. The allocation logic uses:
`speculativeTasks = pendingSpeculative + runningSpeculative`
This ensures that the additional executor remains accounted for after the speculative task has started running.
2. **Use the regular-task requirement as the baseline**
Instead of comparing `maxNeeded` against the current active executor count, I calculate `baseMaxNeededWithoutSpeculation` from the regular pending/running tasks (excluding running speculative tasks).
The locality offset is then applied when:
`maxNeeded == baseMaxNeededWithoutSpeculation`
and there is an active speculative task. This keeps the additional executor tied to the actual resource requirement of the regular workload, rather than the transient number of executors that happen to be registered at that point.
3. **Retain the additional executor while speculation is running**
With this change, the allocation remains stable across the speculative task transition:
`pending → running`
So in the case you described, the target remains at 3 after executor #3 registers and after the speculative task starts, instead of immediately returning to 2 and allowing the StatefulSet allocator to remove the additional executor.
Once the speculative attempt finishes, the speculative-task count goes back to zero and the target can return to the regular-task requirement of 2.
4. **Handle the killed-speculation path**
I also added coverage for the case where the speculative attempt is killed rather than completing normally. The additional executor is retained while the speculative task is running and released once the speculative attempt ends.
I've added regression coverage in `ExecutorAllocationManagerSuite` for both:
`pending → running → completed`
and
`pending → running → killed`
The targeted tests and linters are passing cleanly.
Could you please take another look when you get 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
Re: [PR] [SPARK-49485][CORE] Request additional executor for speculative tasks when active executors equal maxNeeded [spark]
Posted by "sunchao (via GitHub)" <gi...@apache.org>.
sunchao commented on PR #58380:
URL: https://github.com/apache/spark/pull/58380#issuecomment-5469977214
**[P2] Preserve the extra executor with the StatefulSet allocator**
At reviewed commit `9b194f91389bed3cdd5ace9509bb2d67ef5c4085`, the new condition in [ExecutorAllocationManager.scala](https://github.com/apache/spark/blob/9b194f91389bed3cdd5ace9509bb2d67ef5c4085/core/src/main/scala/org/apache/spark/ExecutorAllocationManager.scala#L429-L435) can introduce executor allocation/deletion churn when dynamic allocation and speculation are enabled with `spark.kubernetes.allocation.pods.allocator=statefulset`.
For example, with two 2-core executors, allocation ratio 1, and a maximum of at least 3 executors, one executor can be full while the other runs a slow task whose speculative copy cannot use the spare slot on the original host. Three running tasks plus one pending speculative copy gives `ceil(4 / 2) = 2`, so this branch raises the requested count to 3. Once the third executor registers, the equality check no longer holds and the next allocation update reduces the request to 2. Starting the copy does not prevent that reduction: four running tasks still give a calculated requirement of 2.
[StatefulSetPodsAllocator](https://github.com/apache/spark/blob/9b194f91389bed3cdd5ace9509bb2d67ef5c4085/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/scheduler/cluster/k8s/StatefulSetPodsAllocator.scala#L86-L112) directly applies that reduced request with `scale(expected, false)`. [StatefulSet scale-down](https://kubernetes.io/docs/concepts/workloads/controllers/statefulset/#deployment-and-scaling-guarantees) can therefore delete the newly added highest-ordinal pod, potentially discarding the speculative attempt. Allocation/deletion can repeat while the original remains slow and no other eligible slot becomes available.
The allocator's immediate downscale behavior predates this PR; the newly introduced part is the `2 -> 3 -> 2` request at unchanged task demand, where the base stays at 2. This concern is specific to the StatefulSet allocator, not the default direct allocator. Could we handle or guard this combination so the additional executor survives long enough to do useful speculative work?
This is based on source inspection, not a local Kubernetes runtime reproduction. The earlier fixture and lint issues are fixed.
--
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-49485][CORE] Request additional executor for speculative tasks when active executors equal maxNeeded [spark]
Posted by "zahed1994 (via GitHub)" <gi...@apache.org>.
zahed1994 commented on PR #58380:
URL: https://github.com/apache/spark/pull/58380#issuecomment-5459900019
Thanks for the review @sunchao! I've updated the PR accordingly:
1. **Fixture stage task count**: Changed `createStageInfo(0, 4)` to `createStageInfo(0, 3)` so there are zero regular pending tasks (`3 running + 0 pending + 1 speculative = 4 total tasks`). Without this fix, `maxNeeded` evaluates to `ceil(4 / 2) = 2`, whereas with this fix, `2 + 1 = 3` triggers, strictly exercising the new allocation path (`base=2 / head=3`).
2. **Scala Linter**: Shortened and wrapped the comment lines so all lines are strictly under 100 characters (max line length 94 chars).
All tests and linters are passing cleanly now. Could you please take another look when you get 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
Re: [PR] [SPARK-49485][CORE] Request additional executor for speculative tasks when active executors equal maxNeeded [spark]
Posted by "zahed1994 (via GitHub)" <gi...@apache.org>.
zahed1994 commented on PR #58380:
URL: https://github.com/apache/spark/pull/58380#issuecomment-5817607510
Friendly ping @sunchao. Whenever you have a chance, could you please take another look? All previous feedback points have been addressed and tests/CI are clean.
--
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-49485][CORE] Request additional executor for speculative tasks when active executors equal maxNeeded [spark]
Posted by "sunchao (via GitHub)" <gi...@apache.org>.
sunchao commented on code in PR #58380:
URL: https://github.com/apache/spark/pull/58380#discussion_r3882815448
##########
core/src/test/scala/org/apache/spark/ExecutorAllocationManagerSuite.scala:
##########
@@ -2065,6 +2065,31 @@ class ExecutorAllocationManagerSuite extends SparkFunSuite {
private def executorsDecommissioning(manager: ExecutorAllocationManager): Set[String] = {
manager.executorMonitor.executorsDecommissioning()
}
+
+ test("SPARK-49485: request additional executor when speculative tasks equal maxNeeded") {
+ val manager = createManager(createConf(1, 5, 2))
+ assert(maxNumExecutorsNeededPerResourceProfile(manager, defaultProfile) === 0)
+
+ // Submit stage with 2 tasks
+ post(SparkListenerStageSubmitted(createStageInfo(0, 2)))
+ post(SparkListenerExecutorAdded(0, "executor-1", new ExecutorInfo("host1", 1, Map.empty)))
+ post(SparkListenerExecutorAdded(0, "executor-2", new ExecutorInfo("host1", 1, Map.empty)))
+
+ // Task 0 and Task 1 start on executor-1 and executor-2
+ val taskInfo0 = createTaskInfo(0, 0, "executor-1")
+ val taskInfo1 = createTaskInfo(1, 1, "executor-2")
+ post(SparkListenerTaskStart(0, 0, taskInfo0))
+ post(SparkListenerTaskStart(0, 0, taskInfo1))
+
+ // 2 tasks running, 2 active executors. maxNeeded = 2
+ assert(maxNumExecutorsNeededPerResourceProfile(manager, defaultProfile) === 2)
+
+ // Task 0 is submitted as speculatable
+ post(SparkListenerSpeculativeTaskSubmitted(0, 0))
+
+ // With pendingSpeculative > 0 and maxNeeded == activeExecutors (2), offset allocates 1 more
+ assert(maxNumExecutorsNeededPerResourceProfile(manager, defaultProfile) === 3)
Review Comment:
**[P2] Exercise the new allocation condition in the regression test**
`createConf(1, 5, 2)` sets the initial executor count, leaving one task slot per executor. After speculation is submitted, `maxNeeded` is already `ceil((2 + 1) / 1) = 3`, while the active count is 2. The new branch never executes, so this assertion cannot detect removal of the fix.
For example, configure two two-core executors with three running tasks and one pending speculative task: base then returns 2 and head returns 3.
--
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-49485][CORE] Request additional executor for speculative tasks when active executors equal maxNeeded [spark]
Posted by "zahed1994 (via GitHub)" <gi...@apache.org>.
zahed1994 commented on code in PR #58380:
URL: https://github.com/apache/spark/pull/58380#discussion_r3882940311
##########
core/src/test/scala/org/apache/spark/ExecutorAllocationManagerSuite.scala:
##########
@@ -2065,6 +2065,31 @@ class ExecutorAllocationManagerSuite extends SparkFunSuite {
private def executorsDecommissioning(manager: ExecutorAllocationManager): Set[String] = {
manager.executorMonitor.executorsDecommissioning()
}
+
+ test("SPARK-49485: request additional executor when speculative tasks equal maxNeeded") {
+ val manager = createManager(createConf(1, 5, 2))
+ assert(maxNumExecutorsNeededPerResourceProfile(manager, defaultProfile) === 0)
+
+ // Submit stage with 2 tasks
+ post(SparkListenerStageSubmitted(createStageInfo(0, 2)))
+ post(SparkListenerExecutorAdded(0, "executor-1", new ExecutorInfo("host1", 1, Map.empty)))
+ post(SparkListenerExecutorAdded(0, "executor-2", new ExecutorInfo("host1", 1, Map.empty)))
+
+ // Task 0 and Task 1 start on executor-1 and executor-2
+ val taskInfo0 = createTaskInfo(0, 0, "executor-1")
+ val taskInfo1 = createTaskInfo(1, 1, "executor-2")
+ post(SparkListenerTaskStart(0, 0, taskInfo0))
+ post(SparkListenerTaskStart(0, 0, taskInfo1))
+
+ // 2 tasks running, 2 active executors. maxNeeded = 2
+ assert(maxNumExecutorsNeededPerResourceProfile(manager, defaultProfile) === 2)
+
+ // Task 0 is submitted as speculatable
+ post(SparkListenerSpeculativeTaskSubmitted(0, 0))
+
+ // With pendingSpeculative > 0 and maxNeeded == activeExecutors (2), offset allocates 1 more
+ assert(maxNumExecutorsNeededPerResourceProfile(manager, defaultProfile) === 3)
Review Comment:
Thanks for the review @sunchao! Good catch , with `tasksPerExecutor = 1`, `ceil((2 + 1) / 1) = 3` was already returning 3 on master even without the fix.
I have updated the regression test to configure 2-core executors (`spark.executor.cores = 2`) with 3 running tasks across 2 active executors and 1 pending speculative task (4 total tasks).
Without this fix (base), `maxNeeded` evaluates to `ceil(4 / 2) = 2`. With this fix (head), since `maxNeeded` (2) equals the active executor count (2) and `pendingSpeculative > 0`, the new offset triggers and requests `2 + 1 = 3` executors, properly exercising the new allocation branch.
--
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-49485][CORE] Request additional executor for speculative tasks when active executors equal maxNeeded [spark]
Posted by "zahed1994 (via GitHub)" <gi...@apache.org>.
zahed1994 commented on PR #58380:
URL: https://github.com/apache/spark/pull/58380#issuecomment-5826842812
Thanks a lot for the thorough review, guidance, and approval @sunchao! Really appreciate your help getting this across the finish line.
--
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