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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
