[#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.
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 e87f9aa..ed83c0b 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 @@
+ 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);
+ }
}
}
}
