otterc commented on a change in pull request #32140:
URL: https://github.com/apache/spark/pull/32140#discussion_r648682648
##########
File path:
core/src/main/scala/org/apache/spark/storage/ShuffleBlockFetcherIterator.scala
##########
@@ -767,6 +908,43 @@ final class ShuffleBlockFetcherIterator(
deferredFetchRequests.getOrElseUpdate(address, new
Queue[FetchRequest]())
defReqQueue.enqueue(request)
result = null
+
+ case IgnoreFetchResult(blockId, address, size, isNetworkReqDone) =>
+ if (pushBasedFetchHelper.isNotExecutorOrMergedLocal(address)) {
+ numBlocksInFlightPerAddress(address) =
numBlocksInFlightPerAddress(address) - 1
+ bytesInFlight -= size
+ }
+ if (isNetworkReqDone) {
+ reqsInFlight -= 1
+ logDebug("Number of requests in flight " + reqsInFlight)
+ }
+ numBlocksProcessed +=
pushBasedFetchHelper.initiateFallbackBlockFetchForMergedBlock(
+ blockId, address)
+ // Set result to null to trigger another iteration of the while loop
to get either
+ // a SuccessFetchResult or a FailureFetchResult.
+ result = null
+
+ case MergedBlocksMetaFetchResult(shuffleId, reduceId, blockSize,
numChunks, bitmaps,
+ address, _) =>
+ // The original meta request is processed so we decrease
numBlocksToFetch by 1. We will
+ // collect new chunks request and the count of this is added to
numBlocksToFetch in
+ // collectFetchReqsFromMergedBlocks.
+ numBlocksToFetch -= 1
+ val blocksToRequest =
pushBasedFetchHelper.createChunkBlockInfosFromMetaResponse(
+ shuffleId, reduceId, blockSize, numChunks, bitmaps)
+ val additionalRemoteReqs = new ArrayBuffer[FetchRequest]
+ collectFetchRequests(address, blocksToRequest.toSeq,
additionalRemoteReqs)
+ fetchRequests ++= additionalRemoteReqs
+ // Set result to null to force another iteration.
+ result = null
Review comment:
I have added a UT as well to verify this `iterator has just 1 merged
block and fails to fetch the meta`. PTAL
--
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]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
