[SPARK-26734] [STREAMING] Fix StackOverflowError with large block queue #23716

rlodge · 2019-01-31T22:15:51Z

What changes were proposed in this pull request?

SPARK-23991 introduced a bug in ReceivedBlockTracker#allocateBlocksToBatch: when a queue with more than a few thousand blocks are in the queue, serializing the queue throws a StackOverflowError. This change just adds dequeueAll to the new clone operation on the queue so that the fix in 23991 is preserved but the serialized data comes from an ArrayBuffer which doesn't have the serialization problems that mutable.Queue has.

How was this patch tested?

A unit test was added.

assembly/pom.xml

srowen · 2019-02-01T00:24:31Z

streaming/src/main/scala/org/apache/spark/streaming/scheduler/ReceivedBlockTracker.scala

    if (lastAllocatedBatchTime == null || batchTime > lastAllocatedBatchTime) {
      val streamIdToBlocks = streamIds.map { streamId =>
-        (streamId, getReceivedBlockQueue(streamId).clone())
+        (streamId, getReceivedBlockQueue(streamId).clone().dequeueAll(x => true))


This just basically copies the content right? would it be more robust to just construct a new mutable.Seq from the queue to be sure about what the type is? I know it won't be a Queue here but sounds like the representation does kind of matter.

Was https://issues.apache.org/jira/browse/SPARK-23358 / https://issues.apache.org/jira/browse/SPARK-23391 really the cause or did it mostly uncover this?

Sure, I could do:

mutable.ArrayBuffer.empty[ReceivedBlockInfo] ++= getReceivedBlockQueue(streamId).clone()

which ensures that we know exactly the sequence type being serialized; would you prefer I did that?

https://issues.apache.org/jira/browse/SPARK-23991 caused it in the sense that it moved from using dequeueAll to just serializing a clone of the queue, and the queue just doesn't serialize correctly after some number of entries (a scala bug, IMO). You could say it "uncovered" a scala bug that was sitting there, but prior to that checking the code wouldn't error with large numbers of entries because dequeueAll constructs an ArrayBuffer and that's what was being serialized.

Oh I was looking at the wrong JIRA, number typo. Yeah I see it. @gaborgsomogyi that makes sense?

How about ArrayBuffer(...:_*)? might be better as it could exactly allocate the size.

That's not a scala but java limitation. Serialization is vulnerable to stack overflow for certain kind of structures; for example, a long linked list with no special writeObject() methods will be serialized by recursively writing each link. If you've got a 100k links, you're going to try to use 100k stack frames, and quite likely fail with a StackOverflowError. The main thing here is to use something which is not linked.

I would also choose a solution which allocates the buffer in one step and we could avoid amortized O(1) issue.

Additionally it will be a weird construct which can be tried to simplified by later developers. I would add a comment here which describes the issue not to enforce people to analyze this again.

Yes I think that's the point here indeed, but a Queue apparently has a linked-list-like representation. ArrayBuffer won't. Are you saying that's the right change or no? my point was we're not actually guaranteed what dequeueAll returns.

A comment would be good, sure.

I think .dequeueAll(x => true) works but not the right change and I would propose similar things what you've mentioned, for example ArrayBuffer.... The main point is to allocate the buffer in one step which is not the case with dequeueAll.

Sure, I can add a comment. Is

ArrayBuffer(queue.clone(): _*)

sufficient, or do we need

val clonedQueue = getReceivedBlockQueue(streamId).clone() val bufferToWrite = new mutable.ArrayBuffer[ReceivedBlockInfo](clonedQueue.size) bufferToWrite ++= clonedQueue (streamId, bufferToWrite)

The first one looks good to me. It definitely makes an ArrayBuffer! and I can only imagine the ordering that is produced is the order of dequeueing the elements. It was so when I tried it locally.

Ok, I pushed those changes.

gaborgsomogyi

Thanks for the efforts! I would write down the root cause of this issue in the description as well. It's not obvious why it solves the problem.

gaborgsomogyi · 2019-02-01T11:18:58Z

streaming/src/main/scala/org/apache/spark/streaming/scheduler/ReceivedBlockTracker.scala

    if (lastAllocatedBatchTime == null || batchTime > lastAllocatedBatchTime) {
      val streamIdToBlocks = streamIds.map { streamId =>
-        (streamId, getReceivedBlockQueue(streamId).clone())
+        (streamId, getReceivedBlockQueue(streamId).clone().dequeueAll(x => true))


I would also choose a solution which allocates the buffer in one step and we could avoid amortized O(1) issue.

Additionally it will be a weird construct which can be tried to simplified by later developers. I would add a comment here which describes the issue not to enforce people to analyze this again.

gaborgsomogyi · 2019-02-01T11:20:11Z

streaming/src/test/scala/org/apache/spark/streaming/ReceivedBlockTrackerSuite.scala

+    val receivedBlockTracker = createTracker(setCheckpointDir = true)
+    val blockInfos = generateBlockInfos(100000)
+    blockInfos.map(receivedBlockTracker.addBlock)
+    receivedBlockTracker.allocateBlocksToBatch(1)


I would add assertions here as well. If it's not throwing exception doesn't mean the same blocks deserialized.

gaborgsomogyi · 2019-02-01T11:33:24Z

streaming/src/test/scala/org/apache/spark/streaming/ReceivedBlockTrackerSuite.scala

  }

+  test("block addition, and block to batch allocation with many blocks") {
+    val receivedBlockTracker = createTracker(setCheckpointDir = true)


setCheckpointDir is already true.

…ArrayBuffer.

gaborgsomogyi

Apart from the minor things looks good.

gaborgsomogyi · 2019-02-05T08:23:05Z

streaming/src/main/scala/org/apache/spark/streaming/scheduler/ReceivedBlockTracker.scala

      val streamIdToBlocks = streamIds.map { streamId =>
-        (streamId, getReceivedBlockQueue(streamId).clone())
+        (streamId,
+          mutable.ArrayBuffer(getReceivedBlockQueue(streamId).clone(): _*))


Nit: no line break required.

gaborgsomogyi · 2019-02-05T08:25:41Z

streaming/src/test/scala/org/apache/spark/streaming/ReceivedBlockTrackerSuite.scala


+  test("block addition, and block to batch allocation with many blocks") {
+    val receivedBlockTracker = createTracker()
+    receivedBlockTracker.isWriteAheadLogEnabled should be (true)


Would be good to close the tracker. I know not all tests does this (which is a bug) but it would be good to make it clean here.

Sure, done.

gaborgsomogyi

LGTM.

SparkQA · 2019-02-06T16:00:19Z

Test build #4550 has finished for PR 23716 at commit 9cd6fb5.

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

## What changes were proposed in this pull request? SPARK-23991 introduced a bug in `ReceivedBlockTracker#allocateBlocksToBatch`: when a queue with more than a few thousand blocks are in the queue, serializing the queue throws a StackOverflowError. This change just adds `dequeueAll` to the new `clone` operation on the queue so that the fix in 23991 is preserved but the serialized data comes from an ArrayBuffer which doesn't have the serialization problems that mutable.Queue has. ## How was this patch tested? A unit test was added. Closes #23716 from rlodge/SPARK-26734. Authored-by: Ross Lodge <[email protected]> Signed-off-by: Sean Owen <[email protected]> (cherry picked from commit 8427e9b) Signed-off-by: Sean Owen <[email protected]>

srowen · 2019-02-06T16:48:17Z

Merged to master/2.4/2.3

## What changes were proposed in this pull request? SPARK-23991 introduced a bug in `ReceivedBlockTracker#allocateBlocksToBatch`: when a queue with more than a few thousand blocks are in the queue, serializing the queue throws a StackOverflowError. This change just adds `dequeueAll` to the new `clone` operation on the queue so that the fix in 23991 is preserved but the serialized data comes from an ArrayBuffer which doesn't have the serialization problems that mutable.Queue has. ## How was this patch tested? A unit test was added. Closes apache#23716 from rlodge/SPARK-26734. Authored-by: Ross Lodge <[email protected]> Signed-off-by: Sean Owen <[email protected]>

## What changes were proposed in this pull request? SPARK-23991 introduced a bug in `ReceivedBlockTracker#allocateBlocksToBatch`: when a queue with more than a few thousand blocks are in the queue, serializing the queue throws a StackOverflowError. This change just adds `dequeueAll` to the new `clone` operation on the queue so that the fix in 23991 is preserved but the serialized data comes from an ArrayBuffer which doesn't have the serialization problems that mutable.Queue has. ## How was this patch tested? A unit test was added. Closes apache#23716 from rlodge/SPARK-26734. Authored-by: Ross Lodge <[email protected]> Signed-off-by: Sean Owen <[email protected]> (cherry picked from commit 8427e9b) Signed-off-by: Sean Owen <[email protected]>

rlodge added 2 commits January 31, 2019 09:03

Fix for issue SPARK-26734

e45ab3d

Add a test

c6d967f

srowen requested changes Jan 31, 2019

View reviewed changes

assembly/pom.xml Outdated Show resolved Hide resolved

Revert unintentionally changed POMs

7062df3

srowen reviewed Feb 1, 2019

View reviewed changes

gaborgsomogyi requested changes Feb 1, 2019

View reviewed changes

SPARK-26734: Add comment. Make test more robust. Explicitly allocate …

485ef84

…ArrayBuffer.

gaborgsomogyi reviewed Feb 5, 2019

View reviewed changes

SPARK-26734: Remove line break and stop tracker in test.

9cd6fb5

gaborgsomogyi approved these changes Feb 6, 2019

View reviewed changes

srowen closed this in 8427e9b Feb 6, 2019

[SPARK-26734] [STREAMING] Fix StackOverflowError with large block queue #23716

[SPARK-26734] [STREAMING] Fix StackOverflowError with large block queue #23716

Uh oh!

Conversation

rlodge commented Jan 31, 2019

What changes were proposed in this pull request?

How was this patch tested?

Uh oh!

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!

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!

rlodge Feb 4, 2019 • 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!

gaborgsomogyi left a comment

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!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

gaborgsomogyi left a comment

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!

Choose a reason for hiding this comment

Uh oh!

gaborgsomogyi left a comment

Choose a reason for hiding this comment

Uh oh!

SparkQA commented Feb 6, 2019

Uh oh!

srowen commented Feb 6, 2019

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

4 participants

rlodge Feb 4, 2019 •

edited

Loading