[SPARK-59418][CORE] Prefer idle executors and isolate repeated OOM retries

[sunchao]

JIRA: SPARK-59418

Why are the changes needed?

An OOM retry can encounter the same memory pressure as the original attempt. A free CPU slot does not imply ample executor memory, and moving the retry to another busy executor may reproduce the failure.

Consider an eight-core executor running eight one-CPU tasks. One task builds a large hash table while seven others compete for memory. That partition might fit with less competition, yet fail repeatedly at full concurrency. An executor-wide OOM also discards the other tasks' work. Reducing concurrency for the entire application can help, but sacrifices parallelism for tasks that do not need additional headroom.

The aim is to offer targeted recovery after an observed OOM. Requesting more CPUs is an indirect concurrency control, not a proportional memory reservation. This proposal keeps task CPU requests unchanged and makes the placement/isolation decision explicit.

Related work: SPARK-58187 / PR #57329 proposes increasing task CPUs on OOM retries. This PR is an alternative design based on bounded isolation. The earlier SPARK-21082 discussion raised memory estimation and locality concerns; this policy needs no per-task memory estimate and explicitly limits how much capacity and waiting it can consume.

What changes were proposed in this PR?

Start with idle placement, then escalate to isolation

With spark.scheduler.oomRetry.enabled=true, the first OOM retry prefers an eligible executor with no running tasks. It falls back to ordinary placement if no suitable idle executor is available. This initial preference does not reserve capacity: ordinary work can subsequently share that executor.

After two OOM failures of the same task, reserve an idle or least-busy compatible executor, let its existing tasks finish, and keep new work off it. The retry then runs alone until its attempt terminates. In the example above, the other seven slots remain unused during this attempt. Any extra headroom comes from reduced competition; neither the task's CPU request nor the executor's memory budget increases.

Bound the impact on other work

At most one executor per application is reserved. spark.scheduler.oomRetry.isolationTimeout defaults to 60 seconds from the task's latest OOM, including time waiting behind another reservation. Once that deadline expires, a pending retry can use ordinary placement. It does not limit a retry that has already started in isolation, and the normal task failure budget remains unchanged.

Other executors continue ordinary scheduling. Dynamic allocation counts the reservation separately so its unused slots do not hide demand from the remaining tasks; existing executor limits and allocation controls still apply. Recovery can bypass preferred locations but continues to respect exclusions, resource profiles, and exact CPU/custom-resource requirements, including fractional CPUs. Barrier and pipelined task sets are excluded, and OOM-affected tasks are not speculated.

Supporting changes carry typed OOM information through task exceptions, executor loss, and event logs, and keep resource accounting and cleanup consistent during cancellation, lost output, rejected launches, and barrier preparation failures. Recognized signals include JVM/Spark OOM exceptions, application-caused Spark OOM exits, and observed Kubernetes executor-container OOMKilled termination. An unqualified exit code 137, sidecar OOM, or pod deletion is not sufficient.

User-facing behavior and limits

The two settings are new in 5.0.0, and recovery is disabled by default. Enabling it trades some locality and executor parallelism for a chance to recover from repeated OOMs. Supporting accounting and failed-launch cleanup fixes also apply while the recovery policy is disabled. The existing Kubernetes allocator recovery mode remains independent and unchanged.

Isolation cannot make an intrinsically oversized partition fit, free retained cached/native memory, or recognize failures without an OOM signal. Kubernetes container restarts can hide attribution when the OOM is visible only in lastState. This PR does not claim a measured production recovery rate or cost reduction.

How was this PR tested?

518 tests passed across nine suites on Apache Spark master at 1b8ef5fa621e4959063626ecc1ad328206c0d61f, using Java 17.0.20.1, Scala 2.13.18, and Maven 3.9.16. All nine suites completed with zero failures, errors, ignored, or canceled tests. Added regression tests cover the new recovery behavior and public-master fractional-CPU/pipelined-task compatibility.

The focused suites cover idle placement, draining/isolation, fractional CPU restoration, timeout/cancellation liveness, dynamic allocation, task/executor failure attribution, serialization/event logs, barrier resource rollback, and oversized launch cleanup. Local-executor tests inject managed OOM failures; Kubernetes tests use container-status fixtures. These validate scheduler behavior, not live container OOM recovery or production memory savings.

git diff --check and Scalastyle passed for the Core and Kubernetes production and test sources. After adding the required lint annotations around the tests' intentional OOM throws, FailureSuite was rerun: all 18 tests passed.

Suites and reproduction command

Suite	Tests passed
JsonProtocolSuite	48
TaskSetManagerSuite	82
CoarseGrainedSchedulerBackendSuite	35
ExecutorResourcesAmountsSuite	16
ExecutorAllocationManagerSuite	71
TaskSchedulerImplSuite	148
SparkContextSuite	84
FailureSuite	18
ExecutorPodsLifecycleManagerSuite	16

With JAVA_HOME pointing to Java 17:

SPARK_LOCAL_IP=127.0.0.1 mvn -B -ntp -Pkubernetes \
  -pl resource-managers/kubernetes/core -am \
  -DwildcardSuites=org.apache.spark.util.JsonProtocolSuite,org.apache.spark.scheduler.TaskSetManagerSuite,org.apache.spark.scheduler.CoarseGrainedSchedulerBackendSuite,org.apache.spark.scheduler.ExecutorResourcesAmountsSuite,org.apache.spark.ExecutorAllocationManagerSuite,org.apache.spark.scheduler.TaskSchedulerImplSuite,org.apache.spark.SparkContextSuite,org.apache.spark.FailureSuite,org.apache.spark.scheduler.cluster.k8s.ExecutorPodsLifecycleManagerSuite \
  -Dtest=none -DfailIfNoTests=false test

Was this patch authored or co-authored using generative AI tooling?

Generated-by: OpenAI Codex desktop 26.810.11615 (Nightly)