You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by we...@apache.org on 2024/10/24 05:33:48 UTC
(spark) branch master updated: [SPARK-50034][CORE] Fix Misreporting of Fatal Errors as Uncaught Exceptions in `SparkUncaughtExceptionHandler`
This is an automated email from the ASF dual-hosted git repository.
wenchen pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/master by this push:
new aeddc3f55e60 [SPARK-50034][CORE] Fix Misreporting of Fatal Errors as Uncaught Exceptions in `SparkUncaughtExceptionHandler`
aeddc3f55e60 is described below
commit aeddc3f55e604616d683bdcc3ce7c21c95b018f6
Author: Mingkang Li <mi...@databricks.com>
AuthorDate: Thu Oct 24 13:33:33 2024 +0800
[SPARK-50034][CORE] Fix Misreporting of Fatal Errors as Uncaught Exceptions in `SparkUncaughtExceptionHandler`
### What changes were proposed in this pull request?
#### Summary
This PR addresses the misreporting of fatal errors (such as `OutOfMemoryError`) as uncaught exceptions in `SparkUncaughtExceptionHandler`.
#### Context
In `Executor.scala`, an exception is considered fatal if any exception in the chain (or its causes) is fatal. This behavior is controlled by the `spark.executor.killOnFatalError.depth` config, which limits the depth to which the chain is inspected. If a fatal error is found, `SparkUncaughtExceptionHandler` is called.
Currently, `SparkUncaughtExceptionHandler` only considers the top-level exception when reporting the exit code, rather than traversing the full exception chain to identify the true fatal cause. As a result, some fatal errors, such as `OutOfMemoryError`, are mistakenly reported as uncaught exceptions.
For instance, if we have an OOM exception with the following structure:
```
RuntimeException
- Caused by: RuntimeException
- Caused by: java.lang.OutOfMemory
```
SparkUncaughtExceptionHandler would quit the executor with error code SparkExitCode.UNCAUGHT_EXCEPTION, when the true cause is an OOM error.
#### Fix
This PR modifies `SparkUncaughtExceptionHandler` to:
- Inspect the exception chain (up to the configured depth).
- Ensure that the actual fatal error is correctly identified and reflected in the exit code.
### Why are the changes needed?
This is a bug because some nonfatal exceptions are mistakenly reported as an uncaught exception when the true cause is some other fatal error down the exception chain.
### Does this PR introduce _any_ user-facing change?
No.
### How was this patch tested?
Unit tests are added to `SparkUncaughtExceptionHandlerSuite.scala`.
### Was this patch authored or co-authored using generative AI tooling?
No.
Closes #48551 from mingkangli-db/SPARK-50034.
Authored-by: Mingkang Li <mi...@databricks.com>
Signed-off-by: Wenchen Fan <we...@databricks.com>
---
.../scala/org/apache/spark/executor/Executor.scala | 2 +-
.../org/apache/spark/internal/config/package.scala | 6 +--
.../spark/util/SparkUncaughtExceptionHandler.scala | 51 ++++++++++++++++------
.../util/SparkUncaughtExceptionHandlerSuite.scala | 50 ++++++++++++++++++++-
4 files changed, 91 insertions(+), 18 deletions(-)
diff --git a/core/src/main/scala/org/apache/spark/executor/Executor.scala b/core/src/main/scala/org/apache/spark/executor/Executor.scala
index 0f0e7cb3d897..72d97bd78700 100644
--- a/core/src/main/scala/org/apache/spark/executor/Executor.scala
+++ b/core/src/main/scala/org/apache/spark/executor/Executor.scala
@@ -177,7 +177,7 @@ private[spark] class Executor(
// Whether to monitor killed / interrupted tasks
private val taskReaperEnabled = conf.get(TASK_REAPER_ENABLED)
- private val killOnFatalErrorDepth = conf.get(EXECUTOR_KILL_ON_FATAL_ERROR_DEPTH)
+ private val killOnFatalErrorDepth = conf.get(KILL_ON_FATAL_ERROR_DEPTH)
private val systemLoader = Utils.getContextOrSparkClassLoader
diff --git a/core/src/main/scala/org/apache/spark/internal/config/package.scala b/core/src/main/scala/org/apache/spark/internal/config/package.scala
index 134d1d6bdb88..6b164aecb2af 100644
--- a/core/src/main/scala/org/apache/spark/internal/config/package.scala
+++ b/core/src/main/scala/org/apache/spark/internal/config/package.scala
@@ -2454,11 +2454,11 @@ package object config {
.booleanConf
.createWithDefault(false)
- private[spark] val EXECUTOR_KILL_ON_FATAL_ERROR_DEPTH =
+ private[spark] val KILL_ON_FATAL_ERROR_DEPTH =
ConfigBuilder("spark.executor.killOnFatalError.depth")
.doc("The max depth of the exception chain in a failed task Spark will search for a fatal " +
- "error to check whether it should kill an executor. 0 means not checking any fatal " +
- "error, 1 means checking only the exception but not the cause, and so on.")
+ "error to check whether it should kill the JVM process. 0 means not checking any fatal" +
+ " error, 1 means checking only the exception but not the cause, and so on.")
.internal()
.version("3.1.0")
.intConf
diff --git a/core/src/main/scala/org/apache/spark/util/SparkUncaughtExceptionHandler.scala b/core/src/main/scala/org/apache/spark/util/SparkUncaughtExceptionHandler.scala
index c1ea4f929101..25a6322743ba 100644
--- a/core/src/main/scala/org/apache/spark/util/SparkUncaughtExceptionHandler.scala
+++ b/core/src/main/scala/org/apache/spark/util/SparkUncaughtExceptionHandler.scala
@@ -17,9 +17,11 @@
package org.apache.spark.util
+import org.apache.spark.SparkEnv
import org.apache.spark.executor.{ExecutorExitCode, KilledByTaskReaperException}
import org.apache.spark.internal.{Logging, MDC}
import org.apache.spark.internal.LogKeys.THREAD
+import org.apache.spark.internal.config.KILL_ON_FATAL_ERROR_DEPTH
/**
* The default uncaught exception handler for Spark daemons. It terminates the whole process for
@@ -36,6 +38,17 @@ private[spark] class SparkUncaughtExceptionHandler(val exitOnUncaughtException:
val _ = SparkExitCode.OOM
}
+ // The maximum depth to search in the exception cause chain for a fatal error,
+ // as defined by KILL_ON_FATAL_ERROR_DEPTH.
+ //
+ // SPARK-50034: When this handler is called, there is a fatal error in the cause chain within
+ // the specified depth. We should identify that fatal error and exit with the
+ // correct exit code.
+ private val killOnFatalErrorDepth: Int =
+ // At this point SparkEnv might be None
+ Option(SparkEnv.get).map(_.conf.get(KILL_ON_FATAL_ERROR_DEPTH)).getOrElse(5)
+
+
override def uncaughtException(thread: Thread, exception: Throwable): Unit = {
try {
val mdc = MDC(THREAD, thread)
@@ -50,19 +63,31 @@ private[spark] class SparkUncaughtExceptionHandler(val exitOnUncaughtException:
// We may have been called from a shutdown hook. If so, we must not call System.exit().
// (If we do, we will deadlock.)
if (!ShutdownHookManager.inShutdown()) {
- exception match {
- case _: OutOfMemoryError =>
- System.exit(SparkExitCode.OOM)
- case e: SparkFatalException if e.throwable.isInstanceOf[OutOfMemoryError] =>
- // SPARK-24294: This is defensive code, in case that SparkFatalException is
- // misused and uncaught.
- System.exit(SparkExitCode.OOM)
- case _: KilledByTaskReaperException if exitOnUncaughtException =>
- System.exit(ExecutorExitCode.KILLED_BY_TASK_REAPER)
- case _ if exitOnUncaughtException =>
- System.exit(SparkExitCode.UNCAUGHT_EXCEPTION)
- case _ =>
- // SPARK-30310: Don't System.exit() when exitOnUncaughtException is false
+ // Traverse the causes up to killOnFatalErrorDepth layers
+ var currentException: Throwable = exception
+ var depth = 0
+
+ while (currentException != null && depth < killOnFatalErrorDepth) {
+ currentException match {
+ case _: OutOfMemoryError =>
+ System.exit(SparkExitCode.OOM)
+ case e: SparkFatalException if e.throwable.isInstanceOf[OutOfMemoryError] =>
+ // SPARK-24294: This is defensive code, in case that SparkFatalException is
+ // misused and uncaught.
+ System.exit(SparkExitCode.OOM)
+ case _: KilledByTaskReaperException if exitOnUncaughtException =>
+ System.exit(ExecutorExitCode.KILLED_BY_TASK_REAPER)
+ // No match, continue traversing the cause chain
+ case _ =>
+ }
+ // Move to the next cause in the chain
+ currentException = currentException.getCause
+ depth += 1
+ }
+
+ // SPARK-30310: Don't System.exit() when exitOnUncaughtException is false
+ if (exitOnUncaughtException) {
+ System.exit(SparkExitCode.UNCAUGHT_EXCEPTION)
}
}
} catch {
diff --git a/core/src/test/scala/org/apache/spark/util/SparkUncaughtExceptionHandlerSuite.scala b/core/src/test/scala/org/apache/spark/util/SparkUncaughtExceptionHandlerSuite.scala
index 484340966155..20b3fa7e7d75 100644
--- a/core/src/test/scala/org/apache/spark/util/SparkUncaughtExceptionHandlerSuite.scala
+++ b/core/src/test/scala/org/apache/spark/util/SparkUncaughtExceptionHandlerSuite.scala
@@ -39,7 +39,15 @@ class SparkUncaughtExceptionHandlerSuite extends SparkFunSuite {
(ThrowableTypes.SparkFatalRuntimeException, true, SparkExitCode.UNCAUGHT_EXCEPTION),
(ThrowableTypes.SparkFatalRuntimeException, false, 0),
(ThrowableTypes.SparkFatalOutOfMemoryError, true, SparkExitCode.OOM),
- (ThrowableTypes.SparkFatalOutOfMemoryError, false, SparkExitCode.OOM)
+ (ThrowableTypes.SparkFatalOutOfMemoryError, false, SparkExitCode.OOM),
+ (ThrowableTypes.NestedOOMError, true, SparkExitCode.OOM),
+ (ThrowableTypes.NestedOOMError, false, SparkExitCode.OOM),
+ (ThrowableTypes.NestedSparkFatalException, true, SparkExitCode.OOM),
+ (ThrowableTypes.NestedSparkFatalException, false, SparkExitCode.OOM),
+ (ThrowableTypes.NonFatalNestedErrors, true, SparkExitCode.UNCAUGHT_EXCEPTION),
+ (ThrowableTypes.NonFatalNestedErrors, false, 0),
+ (ThrowableTypes.DeepNestedOOMError, true, SparkExitCode.UNCAUGHT_EXCEPTION),
+ (ThrowableTypes.DeepNestedOOMError, false, 0)
).foreach {
case (throwable: ThrowableTypes.ThrowableTypesVal,
exitOnUncaughtException: Boolean, expectedExitCode) =>
@@ -74,6 +82,46 @@ object ThrowableTypes extends Enumeration {
val SparkFatalOutOfMemoryError = ThrowableTypesVal("SparkFatalException(OutOfMemoryError)",
new SparkFatalException(new OutOfMemoryError))
+ // SPARK-50034: If there is a fatal error in the cause chain,
+ // we should also identify that fatal error and exit with the
+ // correct exit code.
+ val NestedOOMError = ThrowableTypesVal(
+ "NestedFatalError",
+ new RuntimeException("Nonfatal Level 1",
+ new RuntimeException("Nonfatal Level 2",
+ new RuntimeException("Nonfatal Level 3",
+ new OutOfMemoryError())))
+ )
+
+ val NestedSparkFatalException = ThrowableTypesVal(
+ "NestedSparkFatalException",
+ new RuntimeException("Nonfatal Level 1",
+ new RuntimeException("Nonfatal Level 2",
+ new SparkFatalException(new OutOfMemoryError())))
+ )
+
+ // Nested exception with non-fatal errors only
+ val NonFatalNestedErrors = ThrowableTypesVal(
+ "NonFatalNestedErrors",
+ new RuntimeException("Nonfatal Level 1",
+ new RuntimeException("Nonfatal Level 2",
+ new RuntimeException("Nonfatal Level 3",
+ new RuntimeException("Nonfatal Level 4")))
+ )
+ )
+
+ // Should not report as OOM when its depth is greater than killOnFatalErrorDepth
+ val DeepNestedOOMError = ThrowableTypesVal(
+ "DeepNestedOOMError",
+ new RuntimeException("Nonfatal Level 1",
+ new RuntimeException("Nonfatal Level 2",
+ new RuntimeException("Nonfatal Level 3",
+ new RuntimeException("Nonfatal Level 4",
+ new RuntimeException("Nonfatal Level 5",
+ new OutOfMemoryError()))))
+ )
+ )
+
// returns the actual Throwable by its name
def getThrowableByName(name: String): Throwable = {
super.withName(name).asInstanceOf[ThrowableTypesVal].t
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org