[SPARK-24294] Throw SparkException when OOM in BroadcastExchangeExec #21342

jinxing64 · 2018-05-16T07:51:35Z

What changes were proposed in this pull request?

When OutOfMemoryError thrown from BroadcastExchangeExec, scala.concurrent.Future will hit scala bug – scala/bug#9554, and hang until future timeout:

We could wrap the OOM inside SparkException to resolve this issue.

How was this patch tested?

Manually tested.

SparkQA · 2018-05-16T10:48:15Z

Test build #90675 has finished for PR 21342 at commit 9d83f1f.

This patch fails Spark unit tests.
This patch merges cleanly.
This patch adds no public classes.

SparkQA · 2018-05-16T11:40:00Z

Test build #90676 has finished for PR 21342 at commit 99e5b2a.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

jinxing64 · 2018-05-16T12:57:39Z

cc @sameeragarwal @hvanhovell @cloud-fan @jiangxb1987
Please take a look at this when you have time.

kiszk · 2018-05-16T18:05:20Z

The change looks good.
Is it better to add a comment that describes why we throw SparkException instead of OOM with a reference to scala/bug#9554. In the future, we would throw OOM if scala fixes this issue.

sameeragarwal · 2018-05-16T19:28:02Z

LGTM, Thanks! I've seen variations of this bug in the past as well and am not aware of a better way to fix this issue.

scala> Await.result(Future(throw new RuntimeException("foo")), 50.seconds)
java.lang.RuntimeException: foo // regular exceptions fail fast

scala> Await.result(Future(throw new OutOfMemoryError("foo")), 50.seconds)
java.util.concurrent.TimeoutException: Futures timed out after [50 seconds] // OOME times out

gatorsmile · 2018-05-17T00:15:28Z

Could we hold this PR? This change sounds risky. OutOfMemoryError is not issued by Spark. If we eat it, it could cause serious outcomes. cc @zsxwing @JoshRosen

JoshRosen · 2018-05-17T00:59:43Z

I'm also in favor of delaying for a couple of days for more detailed review because historically I think these types of changes have been high risk. The risk calculus might be a bit different if this was fixing a critical "wrong result" correctness bug, but it seems like this is a longstanding annoyance which causes either poor performance or visible job failures, so I don't see an extreme urgency to get this in immediately. Therefore let's take a bit of time to ponder things and sleep on it to be sure that we've thought through corner-cases (to be clear, I do think this patch is generally in a good direction).

Some specifics:

The old code had some places which purposely caught OutOfMemoryError thrown from layers of the spilling code. I do not know whether the expected sources of OOMs were only the throw sites modified here or whether the intent was also to catch OOMs from allocating too big arrays, etc. The latter would have been a dodgy pattern and bad idea in the first place, but I just wanted to note this as a potential risk for unintended / implicit behavior changes. If we want to be super conservative about that we could update throw sites but keep catch sites and extend them to catch both OOM cases.
Should we maybe throw an OutOfMemoryError subclass and then pattern-match on our subclass in a couple of specific places? That might help reduce change-surface.

jinxing64 · 2018-05-17T10:12:56Z

Thanks a lot for looking into this.

The issue is that, sometimes user would configure spark.sql.broadcastTimeout as bigger value, because the relationFuture in BroadcastExchangeExec could cost more time. When OOM
happens, it makes no sense user should wait until timeout.

Yes, there is implicit behavior changes in this pr. In current change we throw a SparkException and eat the OutOfMemoryError. But in existing code, when OOM happens, java.util.concurrent.TimeoutException is thrown but not OutOfMemoryError.

If we really want to throw a OutOfMemoryError here, can we do below steps?

When OOM, catch in future and throw SparkException
relationFuture onFailure {
case t: SparkException => throw OutOfMemoryError()
}

kiszk · 2018-05-17T12:43:27Z

Basically, I agree with this two-step direction. However, I think that it is too risky to simply throw and catch SparkException. This is because other SparkException may be caught in step 2. For example, this line throws SparkException.

Can we throw and catch another Exception or filter the caught SparkException with certain string?

cloud-fan · 2018-05-17T12:57:13Z

How about we create a special Exception class to wrap OOM and InterupttedException, and say it's only used to work around the scala bug?

SparkQA · 2018-05-20T05:03:41Z

Test build #90846 has finished for PR 21342 at commit bb5a74b.

This patch fails to build.
This patch merges cleanly.
This patch adds the following public classes (experimental):
public final class SparkOutOfMemoryException extends Exception

jinxing64 · 2018-05-20T05:32:16Z

@kiszk @cloud-fan
Thanks a lot for comments.
I tested manually and found that the bug exists in all fatal throwable.
In current change, I catch all fatal throwable and wrap in SparkFatalException，ThreadUtils will catch SparkFatalException and re-throw the wrapped throwable.

jinxing64 · 2018-05-20T05:32:50Z

I will update the pr title&description if the change is on the right direction.

SparkQA · 2018-05-20T05:33:38Z

Test build #90847 has finished for PR 21342 at commit bd24746.

This patch fails to build.
This patch merges cleanly.
This patch adds no public classes.

kiszk · 2018-05-20T08:31:29Z

retest this please

SparkQA · 2018-05-20T12:52:07Z

Test build #90853 has finished for PR 21342 at commit bd24746.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

felixcheung · 2018-05-20T21:41:31Z

core/src/main/java/org/apache/spark/memory/SparkOutOfMemoryException.java

+@Private
+public final class SparkOutOfMemoryException extends Exception {
+
+  private OutOfMemoryError oe;


so OutOfMemoryError subclasses java.lang.Error
where as this is subclassing from Exeception, does this matter here?

@felixcheung thanks for review.
In current change there is no SparkOutOfMemoryException. I wrap fatal Throwable in SparkFatalException

cloud-fan · 2018-05-21T14:20:35Z

sql/core/src/main/scala/org/apache/spark/sql/execution/exchange/BroadcastExchangeExec.scala

@@ -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 =>


not related to this PR, but I'm a little worried about catching OOM here. Spark has SparkOutOfMemoryError, and it seems more reasonable to catch SparkOutOfMemoryError. This can be fixed in another PR.

Agreed; let's fix that separately.

To be clear, @cloud-fan, do you mean that, ideally during relationFuture, all the OOM error thrown of type SparkOutOfMemoryError ? (SparkOutOfMemoryError subclass of OutOfMemoryError)

yea, I think Spark will only throw SparkOutOfMemoryError and we should only extend the error message for SparkOutOfMemoryError.

Sure, I'm glad take this and resolve in another pr.

cloud-fan · 2018-05-21T14:21:00Z

LGTM, which scala version has fixed this bug?

gatorsmile · 2018-05-21T15:37:13Z

core/src/main/scala/org/apache/spark/util/SparkFatalException.scala

+ * fatal throwable in {@link scala.concurrent.Future}'s body, and re-throw
+ * SparkFatalException, which wraps the fatal throwable inside.
+ */
+private[spark] final class SparkFatalException(val throwable: Throwable) extends Exception


Will this change impact https://github.com/apache/spark/blob/master/core/src/main/scala/org/apache/spark/util/SparkUncaughtExceptionHandler.scala#L42 ? cc @zsxwing @JoshRosen

I believe it will not. The SparkFatalException has a short life cycle: it is created inside Future {} and then caught and stripped by ThreadUtils.awaitResult.

Are there places where we fetch results from Futures without going through the ThreadUtils.awaitResult? In other words, is that a narrow waist? Would it make sense to add a second redundant layer of unwrapping at the top of SparkUncaughtExceptionHandler to handle that case? Not sure yet, but just thinking aloud here.

OTOH I guess we're actually only using this in one place right now, so I think things are correct as written, but I was just kind of abstractly worrying about potential future pitfalls in case people start using this pattern in new code without also noticing the ThreadUtils.awayResult requirement.

kiszk · 2018-05-21T16:03:02Z

sql/core/src/main/scala/org/apache/spark/sql/execution/exchange/BroadcastExchangeExec.scala

          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 " +


Just curious: Can we perform object operations (allocate OutOfMemoryError, allocate and concatenate Strings) when we caught OutOfMemoryError?
I think that we have space since we failed to allocate a large object.

I agree that we're likely to have reclaimable space at this point, so the chance of a second OOM / failure here seems small. I'm pretty sure that the OutOfMemoryError being caught here often originates from Spark itself where we explicitly throw another OutOfMemoryError at a lower layer of the system, in which case we still actually have heap to allocate strings. We should investigate and clean up that practice, but let's do that in a separate PR.

JoshRosen · 2018-05-22T01:18:56Z

Thanks for the updates. The net change / scope of changes have been significantly reduced here, so I feel that this change is a lot less risky now.

I left only one nitpicky comment at https://github.com/apache/spark/pull/21342/files#r189753488 worrying about potential future risks from people coming along and writing new code throwing SparkFatalException in a context where it can bubble up to the uncaught exception handler. If we want to be super defensive we could add some logic at

spark/core/src/main/scala/org/apache/spark/util/SparkUncaughtExceptionHandler.scala

Line 42 in 3244707

if (exception.isInstanceOf[OutOfMemoryError]) {

to also catch a SparkFatalException on top of an OOM and treat that as an OOM. Debatable if we want to do that, but it's a great way of addressing @gatorsmile's comment at #21342 (comment) and avoids future breakage.

Otherwise, LGTM.

jinxing64 · 2018-05-22T13:41:16Z

https://issues.scala-lang.org/browse/SI-9554?orig=1 is still "OPEN", not sure which scala version can fix this bug.

cloud-fan · 2018-05-22T13:44:43Z

core/src/main/scala/org/apache/spark/util/SparkFatalException.scala

+/**
+ * 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.


let's also mention that, we must use ThreadUtil.awaitResult to run the Future, catch this exception and re-throw the original exception.

jinxing64 · 2018-05-23T13:26:09Z

Thanks again for comments !
I updated this pr and added small defensive logic in SparkUncaughtExceptionHandler.scala.
Please take another look ~

SparkQA · 2018-05-23T17:37:47Z

Test build #91039 has finished for PR 21342 at commit f5a5a81.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

JoshRosen · 2018-05-23T19:36:31Z

Updated changes LGTM. Thanks for working on this!

gatorsmile · 2018-05-23T20:11:38Z

Thanks! Merged to master.

jinxing64 · 2018-05-24T02:01:39Z

Thanks for merging !

viktorklang · 2018-08-23T12:13:13Z

Scala 2.13.0-M5 and forward has improved handling of InterruptedExceptions and fatal errors should be propagated, I'd welcome feedback on the new logic when possible. Thank you!

scala/bug#9554

When OutOfMemoryError thrown from BroadcastExchangeExec, scala.concurrent.Future will hit scala bug – scala/bug#9554, and hang until future timeout: We could wrap the OOM inside SparkException to resolve this issue. Manually tested. Author: jinxing <jinxing6042@126.com> Closes apache#21342 from jinxing64/SPARK-24294. (cherry-picked from commit b7a036b) Ref: LIHADOOP-40177 RB=1414263 BUG=LIHADOOP-40177 R=fli,mshen,yezhou A=yezhou

[SPARK-24294] Throw SparkException when OOM in BroadcastExchangeExec

99e5b2a

jinxing64 force-pushed the SPARK-24294 branch from 9d83f1f to 99e5b2a Compare May 16, 2018 08:01

Catch SparkOutOfMemoryException and re-throw OOMError

bb5a74b

Catch all fatal throwable and wrap in SparkFatalException

bd24746

felixcheung reviewed May 20, 2018

View reviewed changes

cloud-fan reviewed May 21, 2018

View reviewed changes

gatorsmile reviewed May 21, 2018

View reviewed changes

kiszk reviewed May 21, 2018

View reviewed changes

cloud-fan reviewed May 22, 2018

View reviewed changes

refine doc and defensive code.

f5a5a81

asfgit closed this in b7a036b May 23, 2018

jinxing64 mentioned this pull request May 24, 2018

[SPARK-24379] BroadcastExchangeExec should catch SparkOutOfMemory and re-throw SparkFatalException, which wraps SparkOutOfMemory inside. #21424

Closed

[SPARK-24294] Throw SparkException when OOM in BroadcastExchangeExec #21342

[SPARK-24294] Throw SparkException when OOM in BroadcastExchangeExec #21342

Conversation

jinxing64 commented May 16, 2018

What changes were proposed in this pull request?

How was this patch tested?

SparkQA commented May 16, 2018

SparkQA commented May 16, 2018

jinxing64 commented May 16, 2018

kiszk commented May 16, 2018

sameeragarwal commented May 16, 2018

gatorsmile commented May 17, 2018

JoshRosen commented May 17, 2018

jinxing64 commented May 17, 2018

kiszk commented May 17, 2018

cloud-fan commented May 17, 2018

SparkQA commented May 20, 2018

jinxing64 commented May 20, 2018

jinxing64 commented May 20, 2018

SparkQA commented May 20, 2018

kiszk commented May 20, 2018

SparkQA commented May 20, 2018

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

cloud-fan commented May 21, 2018

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

JoshRosen commented May 22, 2018

jinxing64 commented May 22, 2018

Choose a reason for hiding this comment

jinxing64 commented May 23, 2018 • edited

SparkQA commented May 23, 2018

JoshRosen commented May 23, 2018

gatorsmile commented May 23, 2018

jinxing64 commented May 24, 2018

viktorklang commented Aug 23, 2018

jinxing64 commented May 23, 2018 •

edited