zhouyejoe commented on a change in pull request #32007:
URL: https://github.com/apache/spark/pull/32007#discussion_r640996792
##########
File path:
common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/RemoteBlockPushResolver.java
##########
@@ -336,51 +436,26 @@ public StreamCallbackWithID
receiveBlockDataAsStream(PushBlockStream msg) {
final AppShufflePartitionInfo partitionInfo = partitionInfoBeforeCheck !=
null
&& partitionInfoBeforeCheck.mapTracker.contains(msg.mapIndex) ? null
: partitionInfoBeforeCheck;
- final String streamId = String.format("%s_%d_%d_%d",
- OneForOneBlockPusher.SHUFFLE_PUSH_BLOCK_PREFIX, appShuffleId.shuffleId,
msg.mapIndex,
- msg.reduceId);
if (partitionInfo != null) {
return new PushBlockStreamCallback(this, streamId, partitionInfo,
msg.mapIndex);
} else {
// For a duplicate block or a block which is late, respond back with a
callback that handles
// them differently.
- return new StreamCallbackWithID() {
- @Override
- public String getID() {
- return streamId;
- }
-
- @Override
- public void onData(String streamId, ByteBuffer buf) {
- // Ignore the requests. It reaches here either when a request is
received after the
- // shuffle file is finalized or when a request is for a duplicate
block.
- }
-
- @Override
- public void onComplete(String streamId) {
- if (isTooLate) {
- // Throw an exception here so the block data is drained from
channel and server
- // responds RpcFailure to the client.
- throw new RuntimeException(String.format("Block %s %s", streamId,
- ErrorHandler.BlockPushErrorHandler.TOO_LATE_MESSAGE_SUFFIX));
- }
- // For duplicate block that is received before the shuffle merge
finalizes, the
- // server should respond success to the client.
- }
-
- @Override
- public void onFailure(String streamId, Throwable cause) {
- }
- };
+ if (isTooLate) {
Review comment:
The implementation of the createCallbackForInvalidPushBlocks will have a
check for the error message. If the error message is null, it won't throw
RuntimeException. Will add a comment here. Refactoring this part is for
different types of error message when creating the callback.
--
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:
us...@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
