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

[jinxing64]

What changes were proposed in this pull request?

After #20014, Spark won't fails the entire executor but only fails the task suffering SparkOutOfMemoryError. After #21342, BroadcastExchangeExec try-catch OutOfMemoryError. Think about below scenario:

1. SparkOutOfMemoryError(subclass of OutOfMemoryError) is thrown in scala.concurrent.Future;
2. SparkOutOfMemoryError is caught and an OutOfMemoryError is wrapped in SparkFatalException and re-thrown;
3. ThreadUtils.awaitResult catches SparkFatalException and a OutOfMemoryError is thrown;
4. The OutOfMemoryError will go to SparkUncaughtExceptionHandler.uncaughtException and Executor fails.

So it makes more sense to catch SparkOutOfMemory and re-throw SparkFatalException, which wraps SparkOutOfMemory inside.

[jinxing64]

cc @cloud-fan @JoshRosen
Would you please help take a look at this when you have time ?

[cloud-fan]

since we fully control the creation of SparkOutOfMemoryError, can we move the error message to where we throw SparkOutOfMemoryError when building hash relation?

[SparkQA]

Test build #91105 has finished for PR 21424 at commit aa10470.

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

[SparkQA]

Test build #91204 has finished for PR 21424 at commit 3a9669c.

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

[SparkQA]

Test build #91203 has finished for PR 21424 at commit 8f224fb.

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

[cloud-fan]

ah i see. So the SparkOutOfMemoryError is thrown by BytesToBytesMap, we need to catch and rethrow it to attach the error message anyway.

I also found that we may throw OOM when calling child.executeCollectIterator which calls RDD#collect, seems the previous code is corrected.

[jinxing64]

So, I change back ?

[cloud-fan]

it seems we don't need to change anything, maybe just add some comments to say where OOM can occur, i.e. RDD#collect and BroadcastMode#transform

[jinxing64]

@cloud-fan

I also found that we may throw OOM

My previous understanding is that Spark throw SparkOutOfMemoryError when expect there's no memory -- such expectation is from Spark scope of memory management. So that it's safe to catch SparkOutOfMemoryError and mark the corresponding task as failed rather than the executor.

If OutOfMemoryError is thrown from JVM internal, e.g. OOM when new Array[bigSize], is it ok to catch it and continue running the executor as if nothing happened ? If it's ok, does it mean that an executor should never exit when OutOfMemoryError?

[cloud-fan]

That's a good point!

According to the document of SparkOutOfMemoryError, Spark should not kill the executor if SparkOutOfMemoryError is thrown. Broadcast is special because it's run on driver side, so SparkOutOfMemoryError is same as OutOfMemoryError, we need to kill the driver anyway.

That said, I think it's fine to catch OutOfMemoryError and enhance the error message here, to give users some details about why the job failed.

One thing we can do for future safety: do not change the exception type when enhancing the error message.

[jinxing64]

Broadcast is special because it's run on driver side, so SparkOutOfMemoryError is same as OutOfMemoryError, we need to kill the driver anyway.

Thanks a lot for deep explanation, I will close this pr and leave it as unchanged :)

[HyukjinKwon]

I left the JIRA resolved as "Won't fix" given the discussion above.