RATIS-2329. NettyRpcProxy should support handling netty channel exception to prevent replication stuck #1285

leixm · 2025-09-16T08:01:09Z

What changes were proposed in this pull request?

NettyRpcProxy should support handling netty channel exception

What is the link to the Apache JIRA

https://issues.apache.org/jira/browse/RATIS-2329

How was this patch tested?

Manual tests

leixm · 2025-09-16T08:03:09Z

leixm · 2025-09-16T08:09:09Z

Network anomaly, persistent TimeoutIOException

25/09/15 15:39:46,745 WARN [1@group-47BEDE733167->3-LogAppenderDefault-LogAppenderDaemon] LogAppender: 1@group-47BEDE733167->3-LogAppenderDefault: Failed to appendEntries (retry=1): org.apache.ratis.protocol.exceptions.TimeoutIOException
.......
25/09/15 17:14:31,450 WARN [1@group-47BEDE733167->3-LogAppenderDefault-LogAppenderDaemon] LogAppender: 1@group-47BEDE733167->3-LogAppenderDefault: Failed to appendEntries (retry=631): org.apache.ratis.protocol.exceptions.TimeoutIOException

Network recovery, append entries stuck like above stack, I think the connection has been closed, but the SimpleChannelInboundHandler implemented by NettyRpcProxy does not implement exceptionCaught.

Env: we use celeborn 0.4.1 version, which depends on ratis 2.5.1 version and configure netty mode.

leixm · 2025-09-16T08:09:25Z

@szetszwo Can you help review?

…tions.

szetszwo

@leixm , thanks a lot for working on this! The change looks good. Just have some minor comments inlined.

szetszwo · 2025-09-18T03:09:48Z

ratis-netty/src/main/java/org/apache/ratis/netty/NettyRpcProxy.java

+        private void failOutstandingRequests(Throwable cause) {
+          replies.forEach(f -> f.completeExceptionally(cause));
+          replies.clear();
+        }


Let's move it out. Then, it can also be used in close().

BTW, could you also change the IOException to AlreadyClosedException?

@@ -153,9 +169,14 @@ public class NettyRpcProxy implements Closeable { @Override public synchronized void close() { client.close(); + failOutstandingRequests(new AlreadyClosedException("Closing connection to " + peer)); + } + + private void failOutstandingRequests(Throwable cause) { if (!replies.isEmpty()) { - final IOException e = new IOException("Connection to " + peer + " is closed."); - replies.stream().forEach(f -> f.completeExceptionally(e)); + LOG.warn("Still have {} requests outstanding from {} connection: {}", + replies.size(), peer, cause.toString()); + replies.forEach(f -> f.completeExceptionally(cause)); replies.clear(); } }

szetszwo · 2025-09-18T03:23:21Z

ratis-netty/src/main/java/org/apache/ratis/netty/NettyRpcProxy.java

+        @Override
+        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
+          if (!replies.isEmpty()) {
+            LOG.error(
+              "Still have {} requests outstanding when caught exception from {} connection",
+              replies.size(),
+              peer,
+              cause);
+            failOutstandingRequests(cause);
+          }
+          client.close();
+        }
+
+        @Override
+        public void channelInactive(ChannelHandlerContext ctx) {
+          if (!replies.isEmpty()) {
+            LOG.error(
+              "Still have {} requests outstanding when connection from {} is closed",
+              replies.size(),
+              peer);
+            failOutstandingRequests(new IOException("Connection to " + peer + " is closed."));
+          }
+        }


Then, these two methods become very short. BTW, let's

Wrap the cause with IOException in the first method so it could include the peer.

Use AlreadyClosedException for the second method and use a different message.

@Override public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) { failOutstandingRequests(new IOException("Caught an exception for the connection to " + peer, cause)); client.close(); } @Override public void channelInactive(ChannelHandlerContext ctx) { failOutstandingRequests(new AlreadyClosedException("Channel to " + peer + " is inactive.")); }

leixm · 2025-09-25T07:23:29Z

@szetszwo Thank you very much for your guidance. I have two questions to discuss:

ExceptionCaught is set to IOException, which doesn't cause a reconnection. Refer to IOUtils#shouldReconnect, so this may still cause issues.
https://github.com/apache/ratis/blob/master/ratis-netty/src/main/java/org/apache/ratis/netty/NettyRpcProxy.java#L201, when a TimeoutException is caught, should we attempt to reconnect to avoid persistent timeouts in some exceptional cases?

Can we add a new exception type, such as NettyRpcException, to wrap both of the above cases into a NettyRpcException, and then modify IOUtils#shouldReconnect to support reconnecting when a NettyRpcException is encountered?

szetszwo · 2025-09-25T17:22:02Z

Can we add a new exception type, such as NettyRpcException, ...

That's a good idea. Let's add it.

szetszwo · 2025-09-25T21:43:47Z

ratis-common/src/main/java/org/apache/ratis/util/IOUtils.java

    return ReflectionUtils.isInstance(e,
        SocketException.class, SocketTimeoutException.class, ClosedChannelException.class, EOFException.class,
-        AlreadyClosedException.class);
+        AlreadyClosedException.class, NettyRpcException.class);


@leixm , Actually, NettyRpcException is not needed. Could we just use TimeoutIOException here?

I have removed NettyRpcException, but we should call failOutstandingRequests(cause); in exceptionCaught, and we should not wrap it to IOException, because there may have been a SocketTimeoutException that required reconnection, but because we wrapped it into an IOException, it ultimately did not reconnect.

Let's check also the causes?

static boolean shouldReconnect(Throwable e) { for (; e != null; e = e.getCause()) { if (ReflectionUtils.isInstance(e, SocketException.class, SocketTimeoutException.class, ClosedChannelException.class, EOFException.class, AlreadyClosedException.class, TimeoutIOException.class)) { return true; } } return false; }

szetszwo · 2025-09-26T18:39:47Z

@leixm , thanks for the update. There is a test failure. Please take a look. We probably have to update the test.

[INFO] Running org.apache.ratis.netty.TestRaftAsyncWithNetty
Error:  Tests run: 14, Failures: 1, Errors: 0, Skipped: 0, Time elapsed: 83.67 s <<< FAILURE! -- in org.apache.ratis.netty.TestRaftAsyncWithNetty
Error:  org.apache.ratis.netty.TestRaftAsyncWithNetty.testRequestAsyncWithRetryFailureAfterInitialMessages -- Time elapsed: 5.590 s <<< FAILURE!
java.lang.AssertionError: The test "retry-failure-0" does not throw anything.
	at org.apache.ratis.BaseTest.testFailureCase(BaseTest.java:161)
	at org.apache.ratis.BaseTest.testFailureCase(BaseTest.java:169)
	at org.apache.ratis.RaftAsyncTests.runTestRequestAsyncWithRetryFailure(RaftAsyncTests.java:195)
	at org.apache.ratis.RaftAsyncTests.lambda$runTestRequestAsyncWithRetryFailure$0(RaftAsyncTests.java:134)
	at org.apache.ratis.server.impl.MiniRaftCluster$Factory$Get.runWithNewCluster(MiniRaftCluster.java:142)
	at org.apache.ratis.server.impl.MiniRaftCluster$Factory$Get.runWithNewCluster(MiniRaftCluster.java:125)
	at org.apache.ratis.RaftAsyncTests.runTestRequestAsyncWithRetryFailure(RaftAsyncTests.java:134)
	at org.apache.ratis.RaftAsyncTests.testRequestAsyncWithRetryFailureAfterInitialMessages(RaftAsyncTests.java:129)
	at java.lang.reflect.Method.invoke(Method.java:498)
	at java.util.ArrayList.forEach(ArrayList.java:1259)
	at java.util.ArrayList.forEach(ArrayList.java:1259)

OneSizeFitsQuorum

Just left minor issues! I have change the title to make this bug's impact more clear

ratis-netty/src/main/java/org/apache/ratis/netty/NettyRpcProxy.java

Copilot

Pull Request Overview

This PR enhances NettyRpcProxy to handle Netty channel exceptions to prevent replication stuck scenarios. The changes improve error handling by adding proper exception handling for channel failures and modifying the reconnection logic to traverse exception causes.

Added exception handling for Netty channel failures in NettyRpcProxy
Enhanced reconnection logic to check exception cause chain
Improved logging and error reporting for outstanding requests

Reviewed Changes

Copilot reviewed 2 out of 2 changed files in this pull request and generated 2 comments.

File	Description
NettyRpcProxy.java	Added exception handlers for channel failures, improved thread-safe queue usage, and enhanced error handling for outstanding requests
IOUtils.java	Modified shouldReconnect method to traverse exception cause chain and added TimeoutIOException to reconnectable exceptions

_{Tip: Customize your code reviews with copilot-instructions.md. Create the file or learn how to get started.}

Copilot · 2025-09-27T15:56:46Z

ratis-netty/src/main/java/org/apache/ratis/netty/NettyRpcProxy.java

+        @Override
+        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
+          failOutstandingRequests(new IOException("Caught an exception for the connection to " + peer, cause));
+          client.close();


The exceptionCaught method should call super.exceptionCaught(ctx, cause) to ensure proper exception propagation in the Netty pipeline, or at minimum call ctx.close() to properly close the channel context.

Suggested change

client.close();

client.close();

ctx.close();

ctx.close() and client.close() are equivalent and do not need to be called repeatedly.

Copilot · 2025-09-27T15:56:47Z

ratis-netty/src/main/java/org/apache/ratis/netty/NettyRpcProxy.java

+
+        @Override
+        public void channelInactive(ChannelHandlerContext ctx) {
+          failOutstandingRequests(new AlreadyClosedException("Channel to " + peer + " is inactive."));


The channelInactive method should call super.channelInactive(ctx) to ensure proper cleanup in the Netty pipeline.

Suggested change

failOutstandingRequests(new AlreadyClosedException("Channel to " + peer + " is inactive."));

failOutstandingRequests(new AlreadyClosedException("Channel to " + peer + " is inactive."));

super.channelInactive(ctx);

Copilot seems to have a legitimate suggestion.

szetszwo

@leixm , thanks for the update! Please see the comment inlined.

szetszwo · 2025-09-29T16:15:15Z

ratis-netty/src/main/java/org/apache/ratis/netty/NettyRpcProxy.java

+
+        @Override
+        public void channelInactive(ChannelHandlerContext ctx) {
+          failOutstandingRequests(new AlreadyClosedException("Channel to " + peer + " is inactive."));


Copilot seems to have a legitimate suggestion.

leixm force-pushed the RATIS-2329 branch from 0b9917f to 822142f Compare September 16, 2025 08:21

RATIS-2329. NettyRpcProxy should support handling netty.channel excep…

6d22303

…tions.

szetszwo reviewed Sep 18, 2025

View reviewed changes

fix

1f53647

leixm force-pushed the RATIS-2329 branch from a0bcef9 to 1f53647 Compare September 25, 2025 07:15

fix

f708e63

szetszwo reviewed Sep 25, 2025

View reviewed changes

Xianming Lei added 6 commits September 26, 2025 10:06

fix

46d52fc

fix

441f366

fix

68aed9d

fix

7b87283

fix

a8c60c7

fix

20629e4

OneSizeFitsQuorum changed the title ~~RATIS-2329. NettyRpcProxy should support handling netty channel exception~~ RATIS-2329. NettyRpcProxy should support handling netty channel exception to prevent replication stuck Sep 27, 2025

OneSizeFitsQuorum reviewed Sep 27, 2025

View reviewed changes

OneSizeFitsQuorum requested a review from Copilot September 27, 2025 15:56

Copilot AI reviewed Sep 27, 2025

View reviewed changes

Xianming Lei added 3 commits September 29, 2025 15:12

fix

6995881

fix

b360ca2

fix

e80f08b

leixm requested review from OneSizeFitsQuorum and szetszwo September 29, 2025 08:31

szetszwo reviewed Sep 29, 2025

View reviewed changes

	failOutstandingRequests(new AlreadyClosedException("Channel to " + peer + " is inactive."));
	failOutstandingRequests(new AlreadyClosedException("Channel to " + peer + " is inactive."));
	super.channelInactive(ctx);

RATIS-2329. NettyRpcProxy should support handling netty channel exception to prevent replication stuck #1285

Are you sure you want to change the base?

RATIS-2329. NettyRpcProxy should support handling netty channel exception to prevent replication stuck #1285

Conversation

leixm commented Sep 16, 2025

What changes were proposed in this pull request?

What is the link to the Apache JIRA

How was this patch tested?

Uh oh!

leixm commented Sep 16, 2025

Uh oh!

leixm commented Sep 16, 2025

Uh oh!

leixm commented Sep 16, 2025

Uh oh!

szetszwo left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

leixm commented Sep 25, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

szetszwo commented Sep 25, 2025

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

szetszwo Sep 26, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

szetszwo commented Sep 26, 2025

Uh oh!

OneSizeFitsQuorum left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

Pull Request Overview

Reviewed Changes

Uh oh!

Copilot AI Sep 27, 2025

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Copilot AI Sep 27, 2025

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

szetszwo left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

leixm commented Sep 25, 2025 •

edited

Loading

szetszwo Sep 26, 2025 •

edited

Loading