Skip to content

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

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
rlodge wants to merge 5 commits into from
Closed

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

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
e45ab3d
Fix for issue SPARK-26734
rlodge Jan 31, 2019
c6d967f
Add a test
rlodge Jan 31, 2019
7062df3
Revert unintentionally changed POMs
rlodge Jan 31, 2019
485ef84
SPARK-26734: Add comment. Make test more robust. Explicitly allocat…
rlodge Feb 4, 2019
9cd6fb5
SPARK-26734: Remove line break and stop tracker in test.
rlodge Feb 5, 2019
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 5 additions & 1 deletion streaming/src/main/scala/org/apache/spark/streaming/scheduler/ReceivedBlockTracker.scala
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -111,8 +111,12 @@ private[streaming] class ReceivedBlockTracker(
*/
def allocateBlocksToBatch(batchTime: Time): Unit = synchronized {
if (lastAllocatedBatchTime == null || batchTime > lastAllocatedBatchTime) {
// We explicitly create an ArrayBuffer here because at least as of Scala 2.11 and 2.12
// a mutable.Queue fails serialization with a StackOverflow error if it has more than
// a few thousand elements. So we explicitly allocate a collection for serialization which
// we know doesn't have this issue. (See SPARK-26734).
val streamIdToBlocks = streamIds.map { streamId =>
(streamId, getReceivedBlockQueue(streamId).clone())
(streamId, mutable.ArrayBuffer(getReceivedBlockQueue(streamId).clone(): _*))
}.toMap
val allocatedBlocks = AllocatedBlocks(streamIdToBlocks)
if (writeToLog(BatchAllocationEvent(batchTime, allocatedBlocks))) {
Expand Down
25 changes: 23 additions & 2 deletions streaming/src/test/scala/org/apache/spark/streaming/ReceivedBlockTrackerSuite.scala
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -96,6 +96,27 @@ class ReceivedBlockTrackerSuite
receivedBlockTracker.getUnallocatedBlocks(streamId) shouldEqual blockInfos
}

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

@gaborgsomogyigaborgsomogyiFeb 5, 2019

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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.

@rlodgerlodgeFeb 5, 2019

Copy link
Copy Markdown
ContributorAuthor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Sure, done.


val blockInfos = generateBlockInfos(100000)
blockInfos.map(receivedBlockTracker.addBlock)
receivedBlockTracker.allocateBlocksToBatch(1)

@gaborgsomogyigaborgsomogyiFeb 1, 2019

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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

@rlodgerlodgeFeb 4, 2019

Copy link
Copy Markdown
ContributorAuthor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

sure


receivedBlockTracker.getUnallocatedBlocks(streamId) shouldEqual Seq.empty
receivedBlockTracker.hasUnallocatedReceivedBlocks should be (false)
receivedBlockTracker.getBlocksOfBatch(1) shouldEqual Map(streamId -> blockInfos)
receivedBlockTracker.getBlocksOfBatchAndStream(1, streamId) shouldEqual blockInfos

val expectedWrittenData1 = blockInfos.map(BlockAdditionEvent) :+
BatchAllocationEvent(1, AllocatedBlocks(Map(streamId -> blockInfos)))
getWrittenLogData() shouldEqual expectedWrittenData1
getWriteAheadLogFiles() should have size 1

receivedBlockTracker.stop()
}

test("recovery with write ahead logs should remove only allocated blocks from received queue") {
val manualClock = new ManualClock
val batchTime = manualClock.getTimeMillis()
Expand DownExpand Up@@ -362,8 +383,8 @@ class ReceivedBlockTrackerSuite
}

/** Generate blocks infos using random ids */
def generateBlockInfos(): Seq[ReceivedBlockInfo] = {
List.fill(5)(ReceivedBlockInfo(streamId, Some(0L), None,
def generateBlockInfos(blockCount: Int = 5): Seq[ReceivedBlockInfo] = {
List.fill(blockCount)(ReceivedBlockInfo(streamId, Some(0L), None,
BlockManagerBasedStoreResult(StreamBlockId(streamId, math.abs(Random.nextInt)), Some(0L))))
}

Expand Down