ankuriitg commented on a change in pull request #23453: [SPARK-26089][CORE]
Handle corruption in large shuffle blocks
URL: https://github.com/apache/spark/pull/23453#discussion_r246931993
##########
File path:
core/src/main/scala/org/apache/spark/storage/ShuffleBlockFetcherIterator.scala
##########
@@ -451,16 +452,24 @@ final class ShuffleBlockFetcherIterator(
var isStreamCopied: Boolean = false
try {
input = streamWrapper(blockId, in)
- // Only copy the stream if it's wrapped by compression or
encryption, also the size of
- // block is small (the decompressed block is smaller than
maxBytesInFlight)
- if (detectCorrupt && !input.eq(in) && size < maxBytesInFlight / 3)
{
+ // Only copy the stream if it's wrapped by compression or
encryption upto a size of
+ // maxBytesInFlight/3. If stream is longer, then corruption will
be caught while reading
+ // the stream.
+ streamCompressedOrEncrypted = !input.eq(in)
+ if (detectCorrupt && streamCompressedOrEncrypted) {
isStreamCopied = true
val out = new ChunkedByteBufferOutputStream(64 * 1024,
ByteBuffer.allocate)
- // Decompress the whole block at once to detect any corruption,
which could increase
- // the memory usage tne potential increase the chance of OOM.
+ // Decompress the block upto maxBytesInFlight/3 at once to
detect any corruption which
+ // could increase the memory usage and potentially increase the
chance of OOM.
// TODO: manage the memory used here, and spill it into disk in
case of OOM.
- Utils.copyStream(input, out, closeStreams = true)
- input = out.toChunkedByteBuffer.toInputStream(dispose = true)
+ isStreamCopied = Utils.copyStreamUpTo(
+ input, out, maxBytesInFlight / 3, closeStreams = true)
+ if (isStreamCopied) {
+ input = out.toChunkedByteBuffer.toInputStream(dispose = true)
+ } else {
Review comment:
I am guessing you did not look into Utils.copyStreamUpTo(). isStreamCopied
is true when the size of the stream is less than maxBytesInFlight/3 and that
stream was copied. If it is longer, then it will go to the else block. If
stream wasn't copied because of corruption, it will be handled in catch block.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
[email protected]
With regards,
Apache Git Services
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
