Ngone51 commented on a change in pull request #33451:
URL: https://github.com/apache/spark/pull/33451#discussion_r677449980
##########
File path:
core/src/main/scala/org/apache/spark/storage/ShuffleBlockFetcherIterator.scala
##########
@@ -822,8 +836,15 @@ final class ShuffleBlockFetcherIterator(
}
} catch {
case e: IOException =>
- buf.release()
+ // When shuffle checksum is enabled, for a block that is
corrupted twice,
+ // we'd calculate the checksum of the block by consuming the
remaining data
+ // in the buf. So, we should release the buf later.
+ if (!(checksumEnabled && corruptedBlocks.contains(blockId))) {
+ buf.release()
+ }
Review comment:
> To clarify, with this change, for all fetches after the first failure,
we will diagnose (except if cause == disk) ?
No. We will only diagnose if a block corrupted at second time.
(Previously, we'd diagnose when the block corrupted at first time. And we
decide whether retry the block depends on the diagnosis result. But this way
has a problem that it blocks the fetcher's thread so may introduce regression.
Now, for the block corrupted at first time, we still always retry it (this
remains the same behavior as now). If it corrupts again, then we'd diagnose it
and throw fetch failure with the cause (if any).)
--
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.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
