diff options
author | uncleGen <hustyugm@gmail.com> | 2016-12-17 13:19:30 +0000 |
---|---|---|
committer | Sean Owen <sowen@cloudera.com> | 2016-12-17 13:19:30 +0000 |
commit | 6d2379b3b762cdeff98db5ef4d963135c432580a (patch) | |
tree | 17a3fbe6cc727fa17b622ec93816ec58d8c182c4 /core/src/main | |
parent | 2bc1c95154d071d53c9ef2e9e404eaf50ceb4675 (diff) | |
download | spark-6d2379b3b762cdeff98db5ef4d963135c432580a.tar.gz spark-6d2379b3b762cdeff98db5ef4d963135c432580a.tar.bz2 spark-6d2379b3b762cdeff98db5ef4d963135c432580a.zip |
[SPARK-18485][CORE] Underlying integer overflow when create ChunkedByteBufferOutputStream in MemoryStore
## What changes were proposed in this pull request?
There is an underlying integer overflow when create ChunkedByteBufferOutputStream in MemoryStore. This PR provide a check before cast.
## How was this patch tested?
add new unit test
Author: uncleGen <hustyugm@gmail.com>
Closes #15915 from uncleGen/SPARK-18485.
Diffstat (limited to 'core/src/main')
-rw-r--r-- | core/src/main/scala/org/apache/spark/storage/memory/MemoryStore.scala | 10 |
1 files changed, 9 insertions, 1 deletions
diff --git a/core/src/main/scala/org/apache/spark/storage/memory/MemoryStore.scala b/core/src/main/scala/org/apache/spark/storage/memory/MemoryStore.scala index fff21218b1..c08275c7e0 100644 --- a/core/src/main/scala/org/apache/spark/storage/memory/MemoryStore.scala +++ b/core/src/main/scala/org/apache/spark/storage/memory/MemoryStore.scala @@ -331,7 +331,15 @@ private[spark] class MemoryStore( var unrollMemoryUsedByThisBlock = 0L // Underlying buffer for unrolling the block val redirectableStream = new RedirectableOutputStream - val bbos = new ChunkedByteBufferOutputStream(initialMemoryThreshold.toInt, allocator) + val chunkSize = if (initialMemoryThreshold > Int.MaxValue) { + logWarning(s"Initial memory threshold of ${Utils.bytesToString(initialMemoryThreshold)} " + + s"is too large to be set as chunk size. Chunk size has been capped to " + + s"${Utils.bytesToString(Int.MaxValue)}") + Int.MaxValue + } else { + initialMemoryThreshold.toInt + } + val bbos = new ChunkedByteBufferOutputStream(chunkSize, allocator) redirectableStream.setOutputStream(bbos) val serializationStream: SerializationStream = { val autoPick = !blockId.isInstanceOf[StreamBlockId] |