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_r253976053
########## File path: core/src/main/scala/org/apache/spark/storage/ShuffleBlockFetcherIterator.scala ########## @@ -583,15 +595,27 @@ 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: for completeness, you should change `skip()` as well to do the same error handling (though we never really use skip currently, we should be consistent) ---------------------------------------------------------------- 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]
