[SPARK-49485][CORE] Request additional executor for speculative tasks when active executors equal maxNeeded

[zahed1994]

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.

[sunchao]

[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.

[zahed1994]

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.

[sunchao]

[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.

[sunchao]

[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 fails at this exact line with File line length exceeds 100 characters. Wrap or shorten the comment.