[apache#1596][FOLLOWUP] fix(netty): Send failed responses only when t…

…he channel is writable
rickyma · Apr 11, 2024 · 88b2564 · 88b2564
1 parent 1431c8a
commit 88b2564
Showing 1 changed file with 31 additions and 29 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
@@ -564,43 +564,45 @@ public void operationComplete(ChannelFuture future) {
                 + ", "
                 + 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());
+            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);
       } else {
-        LOG.info(
+        LOG.debug(
             "Successfully executed {} for {}. Took {} ms and retrieved {} bytes of data",
             request.getOperationType(),
             requestInfo,