sunchao commented on code in PR #57329:
URL: https://github.com/apache/spark/pull/57329#discussion_r3648371107
##########
core/src/main/scala/org/apache/spark/scheduler/TaskSetManager.scala:
##########
@@ -1498,6 +1600,37 @@ private[spark] object TaskSetManager {
// Shared empty set used as default value for executorIdToTaskIds lookups
// to avoid allocating a new empty set on each executorLost call.
private val EMPTY_LONG_SET = new OpenHashSet[Long](0)
+
+ // Whether an ExceptionFailure represents an OutOfMemoryError: either a
fatal JVM heap
+ // OutOfMemoryError or the non-fatal SparkOutOfMemoryError (which subclasses
OutOfMemoryError),
+ // possibly wrapped in another exception (e.g.
FileFormatDataWriter.enrichWriteError wraps the
+ // cause in a SparkException). When the throwable is preserved, walk a
bounded cause chain the
+ // way Executor.isFatalError does (depthToCheck bounds the walk and guards
against a cause
+ // cycle). Otherwise fall back to the top-level serialized class name, which
is always present
+ // even when the throwable could not be preserved or is not loadable in the
driver; a wrapped
+ // OOM whose throwable was dropped cannot be recognized from the class name
alone, which is an
+ // accepted limitation of the fallback.
+ private def isOom(ef: ExceptionFailure, depthToCheck: Int): Boolean = {
+ ef.exception match {
+ case Some(t) => causedByOom(t, depthToCheck)
+ case None =>
+ ef.className == classOf[OutOfMemoryError].getName ||
+ ef.className == classOf[SparkOutOfMemoryError].getName
+ }
+ }
+
+ @scala.annotation.tailrec
+ private def causedByOom(t: Throwable, depthToCheck: Int): Boolean = {
+ if (depthToCheck <= 0) {
Review Comment:
[P2] Keep OOM classification independent of fatal-error search depth
`spark.executor.killOnFatalError.depth=0` is explicitly supported and means
only that the executor should not inspect a failed task for a fatal error.
Reusing it here makes `causedByOom` return `false` before checking even a
direct, preserved `SparkOutOfMemoryError`; at depth `1`, it also misses the
ordinary `SparkException`-wrapped SQL-write OOM. Since `SparkOutOfMemoryError`
does not terminate the executor, no exit-code fallback restores the
classification, so `spark.task.oomRetryCpusIncrement > 0` silently retries with
the original CPU allocation until the stage exhausts its failures. Please
classify a top-level OOM independently of the fatal-kill setting, use a
separately bounded cause traversal, and add regression coverage for depths `0`
and `1`.
##########
docs/configuration.md:
##########
@@ -3181,6 +3181,40 @@ Apart from these, the following properties are also
available, and may be useful
</td>
<td>0.5.0</td>
</tr>
+<tr>
+ <td><code>spark.task.oomRetryCpusIncrement</code></td>
+ <td>0</td>
+ <td>
+ Number of additional CPUs to allocate for each retry of a task that failed
due to
+ out-of-memory. Each OOM retry of a task gets
+ <code>spark.task.cpus + spark.task.oomRetryCpusIncrement * N</code> CPUs
(N = number of
Review Comment:
[P2] Document retry growth from the task ResourceProfile
This formula is incorrect for stage-level task resource profiles.
`TaskSchedulerImpl` passes
`ResourceProfile.getTaskCpusOrDefaultForProfile(prof, conf)` into
`resourceOffer`, so with `spark.task.cpus=1`, a stage profile requiring `4`
task CPUs, an `8`-core executor, and `spark.task.oomRetryCpusIncrement=1`, the
first OOM retry actually requires `5` CPUs, not the documented `2`. That
changes task schedulability and executor concurrency. Please describe the
formula as `profile task CPUs + increment * N`, falling back to
`spark.task.cpus` only when the profile supplies no override; update the
matching `OOM_RETRY_CPUS_INCREMENT` configuration documentation and add a
nondefault task-profile retry test.
--
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]
