[SPARK-34939][CORE] Throw fetch failure exception when unable to deserialize broadcasted map statuses #32033

viirya · 2021-04-02T04:29:29Z

What changes were proposed in this pull request?

This patch catches IOException, which is possibly thrown due to unable to deserialize map statuses (e.g., broadcasted value is destroyed), when deserilizing map statuses. Once IOException is caught, MetadataFetchFailedException is thrown to let Spark handle it.

Why are the changes needed?

One customer encountered application error. From the log, it is caused by accessing non-existing broadcasted value. The broadcasted value is map statuses. E.g.,

[info]   Cause: java.io.IOException: org.apache.spark.SparkException: Failed to get broadcast_0_piece0 of broadcast_0                                                                                                                  
[info]   at org.apache.spark.util.Utils$.tryOrIOException(Utils.scala:1410)                                                                                                                                                            
[info]   at org.apache.spark.broadcast.TorrentBroadcast.readBroadcastBlock(TorrentBroadcast.scala:226)                                                                                                                                 
[info]   at org.apache.spark.broadcast.TorrentBroadcast.getValue(TorrentBroadcast.scala:103)                                                                                                                                           
[info]   at org.apache.spark.broadcast.Broadcast.value(Broadcast.scala:70)                                                                                                                                                             
[info]   at org.apache.spark.MapOutputTracker$.$anonfun$deserializeMapStatuses$3(MapOutputTracker.scala:967)                                                                                                                           
[info]   at org.apache.spark.internal.Logging.logInfo(Logging.scala:57)                                                                                                                                                                
[info]   at org.apache.spark.internal.Logging.logInfo$(Logging.scala:56)                                                                                                                                                               
[info]   at org.apache.spark.MapOutputTracker$.logInfo(MapOutputTracker.scala:887)                                                                                                                                                     
[info]   at org.apache.spark.MapOutputTracker$.deserializeMapStatuses(MapOutputTracker.scala:967)

There is a race-condition. After map statuses are broadcasted and the executors obtain serialized broadcasted map statuses. If any fetch failure happens after, Spark scheduler invalidates cached map statuses and destroy broadcasted value of the map statuses. Then any executor trying to deserialize serialized broadcasted map statuses and access broadcasted value, IOException will be thrown. Currently we don't catch it in MapOutputTrackerWorker and above exception will fail the application.

Normally we should throw a fetch failure exception for such case. Spark scheduler will handle this.

Does this PR introduce any user-facing change?

No

How was this patch tested?

Unit test.

viirya · 2021-04-02T04:39:36Z

core/src/main/scala/org/apache/spark/MapOutputTracker.scala

   * explicitly destroyed later on when the ShuffleMapStage is garbage-collected.
   */
-  private[this] var cachedSerializedBroadcast: Broadcast[Array[Byte]] = _
+  private[spark] var cachedSerializedBroadcast: Broadcast[Array[Byte]] = _


Expose this for test.

SparkQA · 2021-04-02T05:37:08Z

Kubernetes integration test unable to build dist.

exiting with code: 1
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/41417/

HyukjinKwon

Looks fine. cc @mridulm @Ngone51 FYI

SparkQA · 2021-04-02T07:36:44Z

Test build #136839 has finished for PR 32033 at commit 12e93fa.

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

dbtsai · 2021-04-02T07:55:41Z

core/src/main/scala/org/apache/spark/MapOutputTracker.scala

+          try {
+            fetchedStatuses = MapOutputTracker.deserializeMapStatuses(fetchedBytes, conf)
+          } catch {
+            case e: SparkException =>


The failure could be DIRECT, how can you ensure it's only catching exception from broadcast?

Oh, never mind. I saw the code in the following.

dbtsai · 2021-04-02T07:58:43Z

core/src/main/scala/org/apache/spark/MapOutputTracker.scala

+          val bcast = deserializeObject(bytes, 1, bytes.length - 1).
+            asInstanceOf[Broadcast[Array[Byte]]]
+          logInfo("Broadcast mapstatuses size = " + bytes.length +
+            ", actual size = " + bcast.value.length)


Maybe we should move line 964 to 967 out of the try block like in DIRECT case.

This is for the need of writing the test case. In the test case, if we call getStatuses, the mapoutput tracker worker will ask tracker master for new broadcasted value. So we cannot test the situation we need.

Ngone51 · 2021-04-02T08:11:17Z

This seems to be the same issue with #27604. cc @liupc @cloud-fan

Ngone51 · 2021-04-02T08:26:56Z

core/src/main/scala/org/apache/spark/shuffle/FetchFailedException.scala

-    message: String)
-  extends FetchFailedException(null, shuffleId, -1L, -1, reduceId, message)
+    message: String,
+    cause: Throwable = null)


I don't see cause is used anywhere. Shall we covert the cause to stack string and append to the message?

Sure, I just want to keep original stack trace.

SparkQA · 2021-04-02T17:34:23Z

Kubernetes integration test starting
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/41441/

SparkQA · 2021-04-02T17:34:25Z

Kubernetes integration test status failure
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/41441/

SparkQA · 2021-04-02T19:10:14Z

Test build #136863 has finished for PR 32033 at commit 0ff8919.

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

dongjoon-hyun · 2021-04-03T06:04:14Z

core/src/main/scala/org/apache/spark/MapOutputTracker.scala

+          deserializeObject(bcast.value, 1, bcast.value.length - 1).asInstanceOf[Array[MapStatus]]
+        } catch {
+          case e: IOException =>
+            logError("Exception encountered during deserializing broadcasted map statuses: ", e)


If this is recoverable, maybe shall we lower the level to Warn?

Okay, makes sense.

SparkQA · 2021-04-03T08:57:39Z

Kubernetes integration test starting
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/41457/

SparkQA · 2021-04-03T08:57:40Z

Kubernetes integration test status failure
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/41457/

SparkQA · 2021-04-03T10:51:52Z

Test build #136881 has finished for PR 32033 at commit 7719c3a.

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

viirya · 2021-04-03T17:28:23Z

@dbtsai @Ngone51 Any more comments? Otherwise I will merge this in days. I think there is conflict in 2.4, I will make a backport after.

viirya · 2021-04-03T17:28:33Z

Thanks @dongjoon-hyun

… deserialize broadcasted map statuses ### What changes were proposed in this pull request? This patch catches `IOException`, which is possibly thrown due to unable to deserialize map statuses (e.g., broadcasted value is destroyed), when deserilizing map statuses. Once `IOException` is caught, `MetadataFetchFailedException` is thrown to let Spark handle it. This is a backport of #32033 to branch-2.4. ### Why are the changes needed? One customer encountered application error. From the log, it is caused by accessing non-existing broadcasted value. The broadcasted value is map statuses. E.g., ``` [info] Cause: java.io.IOException: org.apache.spark.SparkException: Failed to get broadcast_0_piece0 of broadcast_0 [info] at org.apache.spark.util.Utils$.tryOrIOException(Utils.scala:1410) [info] at org.apache.spark.broadcast.TorrentBroadcast.readBroadcastBlock(TorrentBroadcast.scala:226) [info] at org.apache.spark.broadcast.TorrentBroadcast.getValue(TorrentBroadcast.scala:103) [info] at org.apache.spark.broadcast.Broadcast.value(Broadcast.scala:70) [info] at org.apache.spark.MapOutputTracker$.$anonfun$deserializeMapStatuses$3(MapOutputTracker.scala:967) [info] at org.apache.spark.internal.Logging.logInfo(Logging.scala:57) [info] at org.apache.spark.internal.Logging.logInfo$(Logging.scala:56) [info] at org.apache.spark.MapOutputTracker$.logInfo(MapOutputTracker.scala:887) [info] at org.apache.spark.MapOutputTracker$.deserializeMapStatuses(MapOutputTracker.scala:967) ``` There is a race-condition. After map statuses are broadcasted and the executors obtain serialized broadcasted map statuses. If any fetch failure happens after, Spark scheduler invalidates cached map statuses and destroy broadcasted value of the map statuses. Then any executor trying to deserialize serialized broadcasted map statuses and access broadcasted value, `IOException` will be thrown. Currently we don't catch it in `MapOutputTrackerWorker` and above exception will fail the application. Normally we should throw a fetch failure exception for such case. Spark scheduler will handle this. ### Does this PR introduce _any_ user-facing change? No ### How was this patch tested? Unit test. Closes #32045 from viirya/fix-broadcast. Authored-by: Liang-Chi Hsieh <[email protected]> Signed-off-by: Dongjoon Hyun <[email protected]>

…rialize broadcasted map statuses ### What changes were proposed in this pull request? This patch catches `IOException`, which is possibly thrown due to unable to deserialize map statuses (e.g., broadcasted value is destroyed), when deserilizing map statuses. Once `IOException` is caught, `MetadataFetchFailedException` is thrown to let Spark handle it. ### Why are the changes needed? One customer encountered application error. From the log, it is caused by accessing non-existing broadcasted value. The broadcasted value is map statuses. E.g., ``` [info] Cause: java.io.IOException: org.apache.spark.SparkException: Failed to get broadcast_0_piece0 of broadcast_0 [info] at org.apache.spark.util.Utils$.tryOrIOException(Utils.scala:1410) [info] at org.apache.spark.broadcast.TorrentBroadcast.readBroadcastBlock(TorrentBroadcast.scala:226) [info] at org.apache.spark.broadcast.TorrentBroadcast.getValue(TorrentBroadcast.scala:103) [info] at org.apache.spark.broadcast.Broadcast.value(Broadcast.scala:70) [info] at org.apache.spark.MapOutputTracker$.$anonfun$deserializeMapStatuses$3(MapOutputTracker.scala:967) [info] at org.apache.spark.internal.Logging.logInfo(Logging.scala:57) [info] at org.apache.spark.internal.Logging.logInfo$(Logging.scala:56) [info] at org.apache.spark.MapOutputTracker$.logInfo(MapOutputTracker.scala:887) [info] at org.apache.spark.MapOutputTracker$.deserializeMapStatuses(MapOutputTracker.scala:967) ``` There is a race-condition. After map statuses are broadcasted and the executors obtain serialized broadcasted map statuses. If any fetch failure happens after, Spark scheduler invalidates cached map statuses and destroy broadcasted value of the map statuses. Then any executor trying to deserialize serialized broadcasted map statuses and access broadcasted value, `IOException` will be thrown. Currently we don't catch it in `MapOutputTrackerWorker` and above exception will fail the application. Normally we should throw a fetch failure exception for such case. Spark scheduler will handle this. ### Does this PR introduce _any_ user-facing change? No ### How was this patch tested? Unit test. Closes #32033 from viirya/fix-broadcast-master. Authored-by: Liang-Chi Hsieh <[email protected]> Signed-off-by: Dongjoon Hyun <[email protected]> (cherry picked from commit 571acc8) Signed-off-by: Dongjoon Hyun <[email protected]>

dongjoon-hyun · 2021-04-04T01:39:22Z

Merged to master/3.1/3.0.

viirya · 2021-04-04T03:30:57Z

Thank all!

cloud-fan · 2021-04-05T07:21:32Z

core/src/main/scala/org/apache/spark/MapOutputTracker.scala

+        } catch {
+          case e: IOException =>
+            logWarning("Exception encountered during deserializing broadcasted map statuses: ", e)
+            throw new SparkException("Unable to deserialize broadcasted map statuses", e)


shall we throw MetadataFetchFailedException directly here?

~~Throw MetadataFetchFailedException here and catch it and rethrow?~~

Oh, recall why I did this. To construct MetadataFetchFailedException needs shuffleId.

I choose to not throw MetadataFetchFailedException as deserializeMapStatuses doesn't have shuffleId and doesn't need it at all.

ah I see, thanks for the explanation!

mridulm · 2021-04-05T18:19:25Z

Got to this a bit late, looks good to me. Nice catch @viirya !

viirya · 2021-04-05T18:32:13Z

Thanks @cloud-fan @mridulm

Ngone51 · 2021-04-06T07:08:31Z

Late LGTM.

…rialize broadcasted map statuses ### What changes were proposed in this pull request? This patch catches `IOException`, which is possibly thrown due to unable to deserialize map statuses (e.g., broadcasted value is destroyed), when deserilizing map statuses. Once `IOException` is caught, `MetadataFetchFailedException` is thrown to let Spark handle it. ### Why are the changes needed? One customer encountered application error. From the log, it is caused by accessing non-existing broadcasted value. The broadcasted value is map statuses. E.g., ``` [info] Cause: java.io.IOException: org.apache.spark.SparkException: Failed to get broadcast_0_piece0 of broadcast_0 [info] at org.apache.spark.util.Utils$.tryOrIOException(Utils.scala:1410) [info] at org.apache.spark.broadcast.TorrentBroadcast.readBroadcastBlock(TorrentBroadcast.scala:226) [info] at org.apache.spark.broadcast.TorrentBroadcast.getValue(TorrentBroadcast.scala:103) [info] at org.apache.spark.broadcast.Broadcast.value(Broadcast.scala:70) [info] at org.apache.spark.MapOutputTracker$.$anonfun$deserializeMapStatuses$3(MapOutputTracker.scala:967) [info] at org.apache.spark.internal.Logging.logInfo(Logging.scala:57) [info] at org.apache.spark.internal.Logging.logInfo$(Logging.scala:56) [info] at org.apache.spark.MapOutputTracker$.logInfo(MapOutputTracker.scala:887) [info] at org.apache.spark.MapOutputTracker$.deserializeMapStatuses(MapOutputTracker.scala:967) ``` There is a race-condition. After map statuses are broadcasted and the executors obtain serialized broadcasted map statuses. If any fetch failure happens after, Spark scheduler invalidates cached map statuses and destroy broadcasted value of the map statuses. Then any executor trying to deserialize serialized broadcasted map statuses and access broadcasted value, `IOException` will be thrown. Currently we don't catch it in `MapOutputTrackerWorker` and above exception will fail the application. Normally we should throw a fetch failure exception for such case. Spark scheduler will handle this. ### Does this PR introduce _any_ user-facing change? No ### How was this patch tested? Unit test. Closes apache#32033 from viirya/fix-broadcast-master. Authored-by: Liang-Chi Hsieh <[email protected]> Signed-off-by: Dongjoon Hyun <[email protected]> (cherry picked from commit 571acc8) Signed-off-by: Dongjoon Hyun <[email protected]>

… deserialize broadcasted map statuses This patch catches `IOException`, which is possibly thrown due to unable to deserialize map statuses (e.g., broadcasted value is destroyed), when deserilizing map statuses. Once `IOException` is caught, `MetadataFetchFailedException` is thrown to let Spark handle it. This is a backport of apache#32033 to branch-2.4. One customer encountered application error. From the log, it is caused by accessing non-existing broadcasted value. The broadcasted value is map statuses. E.g., ``` [info] Cause: java.io.IOException: org.apache.spark.SparkException: Failed to get broadcast_0_piece0 of broadcast_0 [info] at org.apache.spark.util.Utils$.tryOrIOException(Utils.scala:1410) [info] at org.apache.spark.broadcast.TorrentBroadcast.readBroadcastBlock(TorrentBroadcast.scala:226) [info] at org.apache.spark.broadcast.TorrentBroadcast.getValue(TorrentBroadcast.scala:103) [info] at org.apache.spark.broadcast.Broadcast.value(Broadcast.scala:70) [info] at org.apache.spark.MapOutputTracker$.$anonfun$deserializeMapStatuses$3(MapOutputTracker.scala:967) [info] at org.apache.spark.internal.Logging.logInfo(Logging.scala:57) [info] at org.apache.spark.internal.Logging.logInfo$(Logging.scala:56) [info] at org.apache.spark.MapOutputTracker$.logInfo(MapOutputTracker.scala:887) [info] at org.apache.spark.MapOutputTracker$.deserializeMapStatuses(MapOutputTracker.scala:967) ``` There is a race-condition. After map statuses are broadcasted and the executors obtain serialized broadcasted map statuses. If any fetch failure happens after, Spark scheduler invalidates cached map statuses and destroy broadcasted value of the map statuses. Then any executor trying to deserialize serialized broadcasted map statuses and access broadcasted value, `IOException` will be thrown. Currently we don't catch it in `MapOutputTrackerWorker` and above exception will fail the application. Normally we should throw a fetch failure exception for such case. Spark scheduler will handle this. No Unit test. Closes apache#32045 from viirya/fix-broadcast. Authored-by: Liang-Chi Hsieh <[email protected]> Signed-off-by: Dongjoon Hyun <[email protected]> (cherry picked from commit 30436b5) RB=2855790 BUG=LIHADOOP-61824 G=spark-reviewers R=yezhou,mmuralid,vsowrira A=mmuralid,vsowrira

viirya added 2 commits April 1, 2021 19:02

throw fetch failure exception when unable to deserialize map statuses.

cf197ec

Add test case.

12e93fa

github-actions bot added the CORE label Apr 2, 2021

viirya commented Apr 2, 2021

View reviewed changes

viirya changed the title ~~[SPARK-34939][CORE] Throw fetch failure exception when unable to deserialize map statuses~~ [SPARK-34939][CORE] Throw fetch failure exception when unable to deserialize broadcasted map statuses Apr 2, 2021

HyukjinKwon reviewed Apr 2, 2021

View reviewed changes

dbtsai reviewed Apr 2, 2021

View reviewed changes

Ngone51 reviewed Apr 2, 2021

View reviewed changes

Inline cause to fetch failure exception message.

0ff8919

dongjoon-hyun reviewed Apr 3, 2021

View reviewed changes

Use warn log level.

7719c3a

dongjoon-hyun approved these changes Apr 3, 2021

View reviewed changes

viirya mentioned this pull request Apr 3, 2021

[SPARK-34939][CORE][2.4] Throw fetch failure exception when unable to deserialize broadcasted map statuses #32045

Closed

dongjoon-hyun closed this in 571acc8 Apr 4, 2021

cloud-fan reviewed Apr 5, 2021

View reviewed changes

viirya deleted the fix-broadcast-master branch December 27, 2023 18:27

[SPARK-34939][CORE] Throw fetch failure exception when unable to deserialize broadcasted map statuses #32033

[SPARK-34939][CORE] Throw fetch failure exception when unable to deserialize broadcasted map statuses #32033

Uh oh!

Conversation

viirya commented Apr 2, 2021 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

What changes were proposed in this pull request?

Why are the changes needed?

Does this PR introduce any user-facing change?

How was this patch tested?

Uh oh!

Choose a reason for hiding this comment

Uh oh!

SparkQA commented Apr 2, 2021

Uh oh!

HyukjinKwon left a comment • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

SparkQA commented Apr 2, 2021

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Ngone51 commented Apr 2, 2021

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

SparkQA commented Apr 2, 2021

Uh oh!

SparkQA commented Apr 2, 2021

Uh oh!

SparkQA commented Apr 2, 2021

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

SparkQA commented Apr 3, 2021

Uh oh!

SparkQA commented Apr 3, 2021

Uh oh!

SparkQA commented Apr 3, 2021

Uh oh!

viirya commented Apr 3, 2021

Uh oh!

viirya commented Apr 3, 2021

Uh oh!

dongjoon-hyun commented Apr 4, 2021

Uh oh!

viirya commented Apr 4, 2021

Uh oh!

Choose a reason for hiding this comment

Uh oh!

viirya Apr 5, 2021 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

mridulm commented Apr 5, 2021

Uh oh!

viirya commented Apr 5, 2021

Uh oh!

Ngone51 commented Apr 6, 2021

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

viirya commented Apr 2, 2021 •

edited

Loading

HyukjinKwon left a comment •

edited

Loading

viirya Apr 5, 2021 •

edited

Loading