ulysses-you commented on code in PR #57329:
URL: https://github.com/apache/spark/pull/57329#discussion_r3655476454
##########
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:
Fixed. `isOom` no longer reuses `spark.executor.killOnFatalError.depth`. A
top-level `OutOfMemoryError` is now classified unconditionally (from the
preserved throwable, or the serialized class name when it was not preserved),
and only the *wrapped* case walks a cause chain, bounded by a separate fixed
constant `OOM_CAUSE_SEARCH_DEPTH = 5`:
```scala
private def isOom(ef: ExceptionFailure): Boolean = {
ef.className == classOf[OutOfMemoryError].getName ||
ef.className == classOf[SparkOutOfMemoryError].getName ||
ef.exception.exists(causedByOom(_, OOM_CAUSE_SEARCH_DEPTH))
}
```
So `killOnFatalError.depth = 0` (or `1`) no longer suppresses OOM detection
-- neither for a direct `SparkOutOfMemoryError` nor for a
`SparkException`-wrapped OOM. Added a parameterized regression covering depth
`0` and `1` that asserts both a direct and a wrapped OOM still increment the
retry count.
--
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]
