You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by li...@apache.org on 2018/05/23 20:12:08 UTC
spark git commit: [SPARK-24294] Throw SparkException when OOM in
BroadcastExchangeExec
Repository: spark
Updated Branches:
refs/heads/master 84557bc9f -> b7a036b75
[SPARK-24294] Throw SparkException when OOM in BroadcastExchangeExec
## What changes were proposed in this pull request?
When OutOfMemoryError thrown from BroadcastExchangeExec, scala.concurrent.Future will hit scala bug – https://github.com/scala/bug/issues/9554, and hang until future timeout:
We could wrap the OOM inside SparkException to resolve this issue.
## How was this patch tested?
Manually tested.
Author: jinxing <ji...@126.com>
Closes #21342 from jinxing64/SPARK-24294.
Project: http://git-wip-us.apache.org/repos/asf/spark/repo
Commit: http://git-wip-us.apache.org/repos/asf/spark/commit/b7a036b7
Tree: http://git-wip-us.apache.org/repos/asf/spark/tree/b7a036b7
Diff: http://git-wip-us.apache.org/repos/asf/spark/diff/b7a036b7
Branch: refs/heads/master
Commit: b7a036b75b8a1d287ac014b85e90d555753064c9
Parents: 84557bc
Author: jinxing <ji...@126.com>
Authored: Wed May 23 13:12:05 2018 -0700
Committer: Xiao Li <ga...@gmail.com>
Committed: Wed May 23 13:12:05 2018 -0700
----------------------------------------------------------------------
.../apache/spark/util/SparkFatalException.scala | 27 ++++++++++++++++++++
.../util/SparkUncaughtExceptionHandler.scala | 13 +++++++---
.../org/apache/spark/util/ThreadUtils.scala | 2 ++
.../exchange/BroadcastExchangeExec.scala | 13 +++++++---
4 files changed, 48 insertions(+), 7 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/spark/blob/b7a036b7/core/src/main/scala/org/apache/spark/util/SparkFatalException.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/org/apache/spark/util/SparkFatalException.scala b/core/src/main/scala/org/apache/spark/util/SparkFatalException.scala
new file mode 100644
index 0000000..1aa2009
--- /dev/null
+++ b/core/src/main/scala/org/apache/spark/util/SparkFatalException.scala
@@ -0,0 +1,27 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.spark.util
+
+/**
+ * SPARK-24294: To bypass scala bug: https://github.com/scala/bug/issues/9554, we catch
+ * fatal throwable in {@link scala.concurrent.Future}'s body, and re-throw
+ * SparkFatalException, which wraps the fatal throwable inside.
+ * Note that SparkFatalException should only be thrown from a {@link scala.concurrent.Future},
+ * which is run by using ThreadUtils.awaitResult. ThreadUtils.awaitResult will catch
+ * it and re-throw the original exception/error.
+ */
+private[spark] final class SparkFatalException(val throwable: Throwable) extends Exception
http://git-wip-us.apache.org/repos/asf/spark/blob/b7a036b7/core/src/main/scala/org/apache/spark/util/SparkUncaughtExceptionHandler.scala
----------------------------------------------------------------------
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 e0f5af5..1b34fbd 100644
--- a/core/src/main/scala/org/apache/spark/util/SparkUncaughtExceptionHandler.scala
+++ b/core/src/main/scala/org/apache/spark/util/SparkUncaughtExceptionHandler.scala
@@ -39,10 +39,15 @@ 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()) {
- if (exception.isInstanceOf[OutOfMemoryError]) {
- System.exit(SparkExitCode.OOM)
- } else if (exitOnUncaughtException) {
- System.exit(SparkExitCode.UNCAUGHT_EXCEPTION)
+ 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 _ if exitOnUncaughtException =>
+ System.exit(SparkExitCode.UNCAUGHT_EXCEPTION)
}
}
} catch {
http://git-wip-us.apache.org/repos/asf/spark/blob/b7a036b7/core/src/main/scala/org/apache/spark/util/ThreadUtils.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/org/apache/spark/util/ThreadUtils.scala b/core/src/main/scala/org/apache/spark/util/ThreadUtils.scala
index 81aaf79..165a15c 100644
--- a/core/src/main/scala/org/apache/spark/util/ThreadUtils.scala
+++ b/core/src/main/scala/org/apache/spark/util/ThreadUtils.scala
@@ -200,6 +200,8 @@ private[spark] object ThreadUtils {
val awaitPermission = null.asInstanceOf[scala.concurrent.CanAwait]
awaitable.result(atMost)(awaitPermission)
} catch {
+ case e: SparkFatalException =>
+ throw e.throwable
// TimeoutException is thrown in the current thread, so not need to warp the exception.
case NonFatal(t) if !t.isInstanceOf[TimeoutException] =>
throw new SparkException("Exception thrown in awaitResult: ", t)
http://git-wip-us.apache.org/repos/asf/spark/blob/b7a036b7/sql/core/src/main/scala/org/apache/spark/sql/execution/exchange/BroadcastExchangeExec.scala
----------------------------------------------------------------------
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/exchange/BroadcastExchangeExec.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/exchange/BroadcastExchangeExec.scala
index 9e0ec94..c55f9b8 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/exchange/BroadcastExchangeExec.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/exchange/BroadcastExchangeExec.scala
@@ -19,6 +19,7 @@ package org.apache.spark.sql.execution.exchange
import scala.concurrent.{ExecutionContext, Future}
import scala.concurrent.duration._
+import scala.util.control.NonFatal
import org.apache.spark.{broadcast, SparkException}
import org.apache.spark.launcher.SparkLauncher
@@ -30,7 +31,7 @@ import org.apache.spark.sql.execution.{SparkPlan, SQLExecution}
import org.apache.spark.sql.execution.joins.HashedRelation
import org.apache.spark.sql.execution.metric.SQLMetrics
import org.apache.spark.sql.internal.SQLConf
-import org.apache.spark.util.ThreadUtils
+import org.apache.spark.util.{SparkFatalException, ThreadUtils}
/**
* A [[BroadcastExchangeExec]] collects, transforms and finally broadcasts the result of
@@ -111,12 +112,18 @@ case class BroadcastExchangeExec(
SQLMetrics.postDriverMetricUpdates(sparkContext, executionId, metrics.values.toSeq)
broadcasted
} catch {
+ // SPARK-24294: To bypass scala bug: https://github.com/scala/bug/issues/9554, we throw
+ // SparkFatalException, which is a subclass of Exception. ThreadUtils.awaitResult
+ // will catch this exception and re-throw the wrapped fatal throwable.
case oe: OutOfMemoryError =>
- throw new OutOfMemoryError(s"Not enough memory to build and broadcast the table to " +
+ throw new SparkFatalException(
+ new OutOfMemoryError(s"Not enough memory to build and broadcast the table to " +
s"all worker nodes. As a workaround, you can either disable broadcast by setting " +
s"${SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key} to -1 or increase the spark driver " +
s"memory by setting ${SparkLauncher.DRIVER_MEMORY} to a higher value")
- .initCause(oe.getCause)
+ .initCause(oe.getCause))
+ case e if !NonFatal(e) =>
+ throw new SparkFatalException(e)
}
}
}(BroadcastExchangeExec.executionContext)
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org