squito 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_r263600488
########## File path: core/src/main/scala/org/apache/spark/storage/ShuffleBlockFetcherIterator.scala ########## @@ -583,15 +595,28 @@ final class ShuffleBlockFetcherIterator( } /** - * Helper class that ensures a ManagedBuffer is released upon InputStream.close() + * Helper class that ensures a ManagedBuffer is released upon InputStream.close() and + * also detects stream corruption if streamCompressedOrEncrypted is true */ private class BufferReleasingInputStream( - private val delegate: InputStream, - private val iterator: ShuffleBlockFetcherIterator) + // This is visible for testing + private[storage] val delegate: InputStream, + private val iterator: ShuffleBlockFetcherIterator, + private val blockId: BlockId, + private val address: BlockManagerId, + private val streamCompressedOrEncrypted: Boolean) Review comment: I would rename `streamCompressedOrEncrypted` to `detectCorruption` or something like that, as the condition is a bit more complex now (you also check the `detectCorrupt` config when passing this in) ---------------------------------------------------------------- This is an automated message from the Apache Git Service. To respond to the message, please log on to 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]
