sunchao commented on code in PR #57329:
URL: https://github.com/apache/spark/pull/57329#discussion_r3648372078
##########
core/src/main/scala/org/apache/spark/scheduler/TaskSetManager.scala:
##########
@@ -125,6 +126,30 @@ private[spark] class TaskSetManager(
val successful = new Array[Boolean](numTasks)
private val numFailures = new Array[Int](numTasks)
+ // For each task, tracks the number of times it has failed due to
out-of-memory. Used to grow
+ // the number of CPUs allocated to a retry (see
spark.task.oomRetryCpusIncrement), which lowers
+ // the executor's concurrent task count and thus increases the retry's
execution-memory share.
+ // Reset to 0 when the task succeeds, mirroring numFailures.
+ private[scheduler] val numOomRetries = new Array[Int](numTasks)
+ private val oomRetryCpusIncrement = conf.get(config.OOM_RETRY_CPUS_INCREMENT)
+ // Executor exit codes treated as OOM (see
spark.task.oomRetryExecutorExitCodes); when an
+ // executor exits with one of these, the tasks it was running have their OOM
retry count bumped.
+ private val oomRetryExecutorExitCodes =
conf.get(config.OOM_RETRY_EXECUTOR_EXIT_CODES).toSet
+ // The executor total cores of this TaskSet's ResourceProfile. A
TaskSetManager is tied to a
+ // single ResourceProfile, so this is constant for its lifetime and caps the
OOM retry cpus.
+ private val executorCoresLimit: Int = {
+ val rp =
sched.sc.resourceProfileManager.resourceProfileFromId(taskSet.resourceProfileId)
+ rp.getExecutorCores.getOrElse(conf.get(EXECUTOR_CORES))
Review Comment:
[P1] The known ResourceProfile cap can still permanently strand retries on
the current head.
The `None` fix addresses unknown executor capacity, but a known profile or
configured capacity can still exceed the executor's *actual registered total*.
Spark's existing Kubernetes OOM recovery demonstrates this without synthetic
offers: exit `52` is described as `(JVM OOM)`,
`KubernetesClusterSchedulerBackend.doRemoveExecutor` enables recovery,
`BasicExecutorFeatureStep` sets `SPARK_EXECUTOR_CORES` to `spark.task.cpus`,
and the Kubernetes entrypoint passes that reduced value as the executor's
registered `--cores`. With a four-core executor profile, `spark.task.cpus=1`,
and `spark.task.oomRetryCpusIncrement=1`, the replacement registers **one**
core while `executorCoresLimit` remains **four**. The pending OOM retry
therefore requires **two** cores, rejects every offer forever, and never gets
another attempt or reaches `maxTaskFailures`.
The same failure is reproducible locally with `local[2,2]`,
`spark.executor.cores=4`, and increment `2`: the retry requests three CPUs from
an executor whose actual total is two. The added `FailureSuite` only tests
matching configured and actual capacities.
Please cap against each eligible executor's actual **registered total**
rather than profile/configured cores or currently free cores, and add coverage
for Kubernetes OOM recovery and unequal configured/actual local capacity.
##########
core/src/main/scala/org/apache/spark/scheduler/TaskSetManager.scala:
##########
@@ -1247,6 +1318,11 @@ private[spark] class TaskSetManager(
// that the task is not running, and it is NetworkFailure rather
than TaskFailure.
case _ => !info.launching
}
+ // Grow the CPUs of the retry when a running task's executor died of a
JVM heap OOM. See
+ // the SparkOutOfMemoryError branch in handleFailedTask for the
non-fatal counterpart.
+ if (oomRetryCpusIncrement > 0 && !isBarrier && isOomExit &&
exitCausedByApp) {
Review Comment:
[P2] The dropped-cause limitation is reachable from Spark's own
execution-memory spill path.
This is not limited to an unusual application-defined exception.
`TaskMemoryManager.spillConsumer` constructs
`SparkOutOfMemoryError("SPILL_OUT_OF_MEMORY", new HashMap<String, String>() {{
... }})` using a double-brace anonymous map. Inspecting the compiled
`TaskMemoryManager$1` confirms that the map captures both its nonserializable
`TaskMemoryManager` and `MemoryConsumer`. When
`FileFormatDataWriter.enrichWriteError` wraps that OOM in a `SparkException`,
`Executor.TaskRunner` hits `NotSerializableException` while serializing
`ExceptionFailure` and explicitly retries with `preserveCause=false`.
The driver consequently receives `exception=None` and the top-level class
name `org.apache.spark.SparkException`; neither OOM fallback matches. Since
`SparkOutOfMemoryError` is deliberately nonfatal, there is no executor-exit
fallback either, and the retry retains the same CPU allocation until the stage
fails. The existing wrapped-OOM test bypasses executor serialization, so it
cannot detect this real in-tree case.
Please preserve an explicit OOM classification before discarding the
throwable, or otherwise transport a serializable marker, and add a regression
that exercises the actual executor failure-serialization fallback.
--
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]
