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 1 commit
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
Prev Previous commit
SPARK-26734: Remove line break and stop tracker in test.
  • Loading branch information
@rlodge
rlodge committed Feb 5, 2019
commit 9cd6fb57b631316cd098dcba3337490d9582070c
3 changes: 1 addition & 2 deletions streaming/src/main/scala/org/apache/spark/streaming/scheduler/ReceivedBlockTracker.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -116,8 +116,7 @@ private[streaming] class ReceivedBlockTracker(
// 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,
mutable.ArrayBuffer(getReceivedBlockQueue(streamId).clone(): _*))
(streamId, mutable.ArrayBuffer(getReceivedBlockQueue(streamId).clone(): _*))
}.toMap
val allocatedBlocks = AllocatedBlocks(streamIdToBlocks)
if (writeToLog(BatchAllocationEvent(batchTime, allocatedBlocks))) {
Expand Down
2 changes: 2 additions & 0 deletions streaming/src/test/scala/org/apache/spark/streaming/ReceivedBlockTrackerSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -113,6 +113,8 @@ class ReceivedBlockTrackerSuite
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") {
Expand Down