This is an automated email from the ASF dual-hosted git repository. roryqi pushed a commit to branch branch-0.9 in repository https://gitbox.apache.org/repos/asf/incubator-uniffle.git
commit 54b1f596278cc8d5d1aef4a594672c10ab135c76 Author: RickyMa <[email protected]> AuthorDate: Fri Apr 12 21:34:38 2024 +0800 [#1596][FOLLOWUP] fix(netty): Send failed responses only when the channel is writable (#1641) ### What changes were proposed in this pull request? 1. Send failed responses only when the channel is writable. 2. Print debug logs when the data is successfully sent, reducing log output. 3. Reduce the duplicated error log. ### Why are the changes needed? A follow-up PR for: https://github.com/apache/incubator-uniffle/pull/1605. ### Does this PR introduce _any_ user-facing change? No. ### How was this patch tested? Existing UTs. --- .../server/netty/ShuffleServerNettyHandler.java | 76 ++++++++++++---------- 1 file changed, 40 insertions(+), 36 deletions(-) diff --git a/server/src/main/java/org/apache/uniffle/server/netty/ShuffleServerNettyHandler.java b/server/src/main/java/org/apache/uniffle/server/netty/ShuffleServerNettyHandler.java index e87f9aa4e..ed83c0b77 100644 --- a/server/src/main/java/org/apache/uniffle/server/netty/ShuffleServerNettyHandler.java +++ b/server/src/main/java/org/apache/uniffle/server/netty/ShuffleServerNettyHandler.java @@ -563,49 +563,53 @@ public class ShuffleServerNettyHandler implements BaseMessageHandler { + requestInfo + ", " + cause.getMessage(); - LOG.error(errorMsg, future.cause()); - RpcResponse errorResponse; - if (request instanceof GetLocalShuffleDataRequest) { - errorResponse = - new GetLocalShuffleDataResponse( - request.getRequestId(), - StatusCode.INTERNAL_ERROR, - errorMsg, - new NettyManagedBuffer(Unpooled.EMPTY_BUFFER)); - } else if (request instanceof GetLocalShuffleIndexRequest) { - errorResponse = - new GetLocalShuffleIndexResponse( - request.getRequestId(), - StatusCode.INTERNAL_ERROR, - errorMsg, - Unpooled.EMPTY_BUFFER, - 0L); - } else if (request instanceof GetMemoryShuffleDataRequest) { - errorResponse = - new GetMemoryShuffleDataResponse( - request.getRequestId(), - StatusCode.INTERNAL_ERROR, - errorMsg, - Lists.newArrayList(), - Unpooled.EMPTY_BUFFER); - } else { - LOG.error("Cannot handle request {}", request.type()); - return; + if (future.channel().isWritable()) { + RpcResponse errorResponse; + if (request instanceof GetLocalShuffleDataRequest) { + errorResponse = + new GetLocalShuffleDataResponse( + request.getRequestId(), + StatusCode.INTERNAL_ERROR, + errorMsg, + new NettyManagedBuffer(Unpooled.EMPTY_BUFFER)); + } else if (request instanceof GetLocalShuffleIndexRequest) { + errorResponse = + new GetLocalShuffleIndexResponse( + request.getRequestId(), + StatusCode.INTERNAL_ERROR, + errorMsg, + Unpooled.EMPTY_BUFFER, + 0L); + } else if (request instanceof GetMemoryShuffleDataRequest) { + errorResponse = + new GetMemoryShuffleDataResponse( + request.getRequestId(), + StatusCode.INTERNAL_ERROR, + errorMsg, + Lists.newArrayList(), + Unpooled.EMPTY_BUFFER); + } else { + LOG.error("Cannot handle request {}", request.type(), cause); + return; + } + client.getChannel().writeAndFlush(errorResponse); } - client.getChannel().writeAndFlush(errorResponse); LOG.error( "Failed to execute {} for {}. Took {} ms and could not retrieve {} bytes of data", request.getOperationType(), requestInfo, readTime, - dataSize); + dataSize, + cause); } else { - LOG.info( - "Successfully executed {} for {}. Took {} ms and retrieved {} bytes of data", - request.getOperationType(), - requestInfo, - readTime, - dataSize); + if (LOG.isDebugEnabled()) { + LOG.debug( + "Successfully executed {} for {}. Took {} ms and retrieved {} bytes of data", + request.getOperationType(), + requestInfo, + readTime, + dataSize); + } } } }
