From 4776b4c3e9c29c19abf3477c4cbffc960cb52b14 Mon Sep 17 00:00:00 2001 From: Abhishek Pal Date: Sat, 1 Aug 2026 22:18:22 +0530 Subject: [PATCH 1/5] RATIS-2629. Handle Netty based request asynchronously and improve exception handling --- ratis-netty/pom.xml | 5 + .../apache/ratis/netty/NettyConfigKeys.java | 21 ++++ .../ratis/netty/server/NettyRpcService.java | 44 +++++++- .../netty/server/TestNettyRpcService.java | 105 ++++++++++++++++++ 4 files changed, 169 insertions(+), 6 deletions(-) create mode 100644 ratis-netty/src/test/java/org/apache/ratis/netty/server/TestNettyRpcService.java diff --git a/ratis-netty/pom.xml b/ratis-netty/pom.xml index 5688a1fa4a..6b720332c9 100644 --- a/ratis-netty/pom.xml +++ b/ratis-netty/pom.xml @@ -88,6 +88,11 @@ junit-platform-launcher test + + org.mockito + mockito-core + test + diff --git a/ratis-netty/src/main/java/org/apache/ratis/netty/NettyConfigKeys.java b/ratis-netty/src/main/java/org/apache/ratis/netty/NettyConfigKeys.java index fd0906ebfe..5984fff11f 100644 --- a/ratis-netty/src/main/java/org/apache/ratis/netty/NettyConfigKeys.java +++ b/ratis-netty/src/main/java/org/apache/ratis/netty/NettyConfigKeys.java @@ -74,6 +74,27 @@ static boolean useEpoll(RaftProperties properties) { static void setUseEpoll(RaftProperties properties, boolean enable) { setBoolean(properties::setBoolean, USE_EPOLL_KEY, enable); } + + String ASYNC_REQUEST_THREAD_POOL_CACHED_KEY = PREFIX + ".async.request.thread.pool.cached"; + boolean ASYNC_REQUEST_THREAD_POOL_CACHED_DEFAULT = true; + static boolean asyncRequestThreadPoolCached(RaftProperties properties) { + return getBoolean(properties::getBoolean, ASYNC_REQUEST_THREAD_POOL_CACHED_KEY, + ASYNC_REQUEST_THREAD_POOL_CACHED_DEFAULT, getDefaultLog()); + } + static void setAsyncRequestThreadPoolCached(RaftProperties properties, boolean useCached) { + setBoolean(properties::setBoolean, ASYNC_REQUEST_THREAD_POOL_CACHED_KEY, useCached); + } + + String ASYNC_REQUEST_THREAD_POOL_SIZE_KEY = PREFIX + ".async.request.thread.pool.size"; + int ASYNC_REQUEST_THREAD_POOL_SIZE_DEFAULT = 32; + static int asyncRequestThreadPoolSize(RaftProperties properties) { + return getInt(properties::getInt, ASYNC_REQUEST_THREAD_POOL_SIZE_KEY, + ASYNC_REQUEST_THREAD_POOL_SIZE_DEFAULT, getDefaultLog(), + requireMin(0), requireMax(65536)); + } + static void setAsyncRequestThreadPoolSize(RaftProperties properties, int size) { + setInt(properties::setInt, ASYNC_REQUEST_THREAD_POOL_SIZE_KEY, size); + } } interface Client { diff --git a/ratis-netty/src/main/java/org/apache/ratis/netty/server/NettyRpcService.java b/ratis-netty/src/main/java/org/apache/ratis/netty/server/NettyRpcService.java index f7d2805e8a..e8d353cec7 100644 --- a/ratis-netty/src/main/java/org/apache/ratis/netty/server/NettyRpcService.java +++ b/ratis-netty/src/main/java/org/apache/ratis/netty/server/NettyRpcService.java @@ -42,6 +42,7 @@ import org.apache.ratis.proto.netty.NettyProtos.RaftNettyServerReplyProto; import org.apache.ratis.proto.netty.NettyProtos.RaftNettyServerRequestProto; import org.apache.ratis.util.CodeInjectionForTesting; +import org.apache.ratis.util.ConcurrentUtils; import org.apache.ratis.util.JavaUtils; import org.apache.ratis.util.MemoizedSupplier; import org.apache.ratis.util.ProtoUtils; @@ -50,8 +51,8 @@ import java.io.IOException; import java.net.InetSocketAddress; +import java.util.concurrent.ExecutorService; import java.util.concurrent.TimeUnit; -import java.util.Objects; /** * A netty server endpoint that acts as the communication layer. @@ -87,12 +88,31 @@ public static Builder newBuilder() { private final MemoizedSupplier channel; private final InetSocketAddress socketAddress; + private final ExecutorService requestExecutor; + @ChannelHandler.Sharable class InboundHandler extends SimpleChannelInboundHandler { @Override protected void channelRead0(ChannelHandlerContext ctx, RaftNettyServerRequestProto proto) { - final RaftNettyServerReplyProto reply = handle(proto); - ctx.writeAndFlush(reply); + requestExecutor.execute(() -> { + final RaftNettyServerReplyProto reply; + try { + // handle() already builds an error reply whenever it has a request context + reply = handle(proto); + } catch (Exception e) { + // Close the channel so the client fails fast instead of blocking until timeout. + LOG.warn("{}: Failed to handle request; closing the channel.", getId(), e); + ctx.close(); + return; + } + ctx.writeAndFlush(reply); + }); + } + + @Override + public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) { + LOG.warn("{}: exceptionCaught on channel {}; closing it.", getId(), ctx.channel(), cause); + ctx.close(); } } @@ -116,6 +136,11 @@ protected void initChannel(SocketChannel ch) { } }; + this.requestExecutor = ConcurrentUtils.newThreadPoolWithMax( + NettyConfigKeys.Server.asyncRequestThreadPoolCached(server.getProperties()), + NettyConfigKeys.Server.asyncRequestThreadPoolSize(server.getProperties()), + server.getId() + "-request-"); + final boolean useEpoll = NettyConfigKeys.Server.useEpoll(server.getProperties()); this.bossGroup = NettyUtils.newEventLoopGroup(CLASS_NAME + "-bossGroup", 0, useEpoll); this.workerGroup = NettyUtils.newEventLoopGroup(CLASS_NAME + "-workerGroup",0, useEpoll); @@ -155,6 +180,7 @@ public void startImpl() throws IOException { @Override public void closeImpl() throws IOException { + ConcurrentUtils.shutdownAndWait(requestExecutor); final ChannelFuture f = getChannel().close(); f.syncUninterruptibly(); bossGroup.shutdownGracefully(0, 100, TimeUnit.MILLISECONDS); @@ -296,9 +322,15 @@ RaftNettyServerReplyProto handle(RaftNettyServerRequestProto proto) { throw new UnsupportedOperationException("Request case not supported: " + proto.getRaftNettyServerRequestCase()); } - } catch (IOException ioe) { - return toRaftNettyServerReplyProto( - Objects.requireNonNull(rpcRequest, "rpcRequest = null"), ioe); + } catch (Exception e) { + if (rpcRequest == null) { + // let InboundHandler close the channel so the client fails fast. + throw new IllegalStateException(getId() + ": Failed to handle request " + proto, e); + } + // The client deserializes the reply and casts it to IOException, so always send an + // IOException regardless of the actual failure type. + final IOException ioe = e instanceof IOException ? (IOException) e : new IOException(e); + return toRaftNettyServerReplyProto(rpcRequest, ioe); } } diff --git a/ratis-netty/src/test/java/org/apache/ratis/netty/server/TestNettyRpcService.java b/ratis-netty/src/test/java/org/apache/ratis/netty/server/TestNettyRpcService.java new file mode 100644 index 0000000000..ad20fb2ec4 --- /dev/null +++ b/ratis-netty/src/test/java/org/apache/ratis/netty/server/TestNettyRpcService.java @@ -0,0 +1,105 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.ratis.netty.server; + +import org.apache.ratis.conf.RaftProperties; +import org.apache.ratis.proto.RaftProtos.RaftRpcRequestProto; +import org.apache.ratis.proto.RaftProtos.RequestVoteRequestProto; +import org.apache.ratis.proto.netty.NettyProtos.RaftNettyServerReplyProto; +import org.apache.ratis.proto.netty.NettyProtos.RaftNettyServerReplyProto.RaftNettyServerReplyCase; +import org.apache.ratis.proto.netty.NettyProtos.RaftNettyServerRequestProto; +import org.apache.ratis.protocol.RaftPeerId; +import org.apache.ratis.server.RaftServer; +import org.apache.ratis.thirdparty.io.netty.channel.ChannelHandlerContext; +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Test; +import org.mockito.Mockito; + +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.TimeUnit; + +/** Tests for {@link NettyRpcService} request handling. */ +public class TestNettyRpcService { + private static final RaftPeerId ID = RaftPeerId.valueOf("s0"); + + private static RaftServer newMockServer() { + final RaftServer server = Mockito.mock(RaftServer.class); + Mockito.when(server.getId()).thenReturn(ID); + Mockito.when(server.getProperties()).thenReturn(new RaftProperties()); + return server; + } + + private static RaftNettyServerRequestProto newRequestVoteProto() { + final RaftRpcRequestProto rpc = RaftRpcRequestProto.newBuilder() + .setRequestorId(ID.toByteString()) + .setReplyId(ID.toByteString()) + .setCallId(1) + .build(); + final RequestVoteRequestProto request = RequestVoteRequestProto.newBuilder() + .setServerRequest(rpc) + .build(); + return RaftNettyServerRequestProto.newBuilder() + .setRequestVoteRequest(request) + .build(); + } + + /** + * A non-{@link java.io.IOException} thrown by the server must be turned into an error reply + * instead of escaping the handler and leaving the client to block until its request timeout. + */ + @Test + public void testHandleReturnsErrorReplyOnRuntimeException() throws Exception { + final RaftServer server = newMockServer(); + Mockito.when(server.requestVote(Mockito.any())).thenThrow(new RuntimeException("injected")); + + final NettyRpcService service = NettyRpcService.newBuilder().setServer(server).build(); + service.start(); + try { + final RaftNettyServerReplyProto reply = service.handle(newRequestVoteProto()); + Assertions.assertEquals(RaftNettyServerReplyCase.EXCEPTIONREPLY, reply.getRaftNettyServerReplyCase()); + } finally { + service.close(); + } + } + + /** Requests must be handled off the Netty I/O event loop, on the request executor thread. */ + @Test + public void testRequestHandledOffEventLoop() throws Exception { + final RaftServer server = newMockServer(); + final CompletableFuture handlingThreadName = new CompletableFuture<>(); + Mockito.when(server.requestVote(Mockito.any())).thenAnswer(invocation -> { + handlingThreadName.complete(Thread.currentThread().getName()); + throw new RuntimeException("injected"); + }); + + final NettyRpcService service = NettyRpcService.newBuilder().setServer(server).build(); + service.start(); + try { + final ChannelHandlerContext ctx = Mockito.mock(ChannelHandlerContext.class); + service.new InboundHandler().channelRead0(ctx, newRequestVoteProto()); + + final String threadName = handlingThreadName.get(5, TimeUnit.SECONDS); + Assertions.assertTrue(threadName.startsWith(ID + "-request-"), + "Request was handled on an unexpected thread: " + threadName); + Assertions.assertNotEquals(Thread.currentThread().getName(), threadName, + "Request was handled on the calling thread, not offloaded"); + } finally { + service.close(); + } + } +} From 269a5702b1d145a975a5cedeffc7ea1382561188 Mon Sep 17 00:00:00 2001 From: Abhishek Pal Date: Sun, 2 Aug 2026 21:27:06 +0530 Subject: [PATCH 2/5] Mirror gRPC async handling --- .../ratis/netty/server/NettyRpcService.java | 257 ++++++++++-------- .../netty/server/TestNettyRpcService.java | 2 +- 2 files changed, 139 insertions(+), 120 deletions(-) diff --git a/ratis-netty/src/main/java/org/apache/ratis/netty/server/NettyRpcService.java b/ratis-netty/src/main/java/org/apache/ratis/netty/server/NettyRpcService.java index e8d353cec7..36bbbf6218 100644 --- a/ratis-netty/src/main/java/org/apache/ratis/netty/server/NettyRpcService.java +++ b/ratis-netty/src/main/java/org/apache/ratis/netty/server/NettyRpcService.java @@ -21,8 +21,6 @@ import org.apache.ratis.netty.NettyConfigKeys; import org.apache.ratis.netty.NettyRpcProxy; import org.apache.ratis.util.NettyUtils; -import org.apache.ratis.protocol.GroupInfoReply; -import org.apache.ratis.protocol.GroupListReply; import org.apache.ratis.protocol.RaftClientReply; import org.apache.ratis.protocol.RaftPeerId; import org.apache.ratis.rpc.SupportedRpcType; @@ -51,6 +49,8 @@ import java.io.IOException; import java.net.InetSocketAddress; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CompletionException; import java.util.concurrent.ExecutorService; import java.util.concurrent.TimeUnit; @@ -90,23 +90,35 @@ public static Builder newBuilder() { private final ExecutorService requestExecutor; - @ChannelHandler.Sharable class InboundHandler extends SimpleChannelInboundHandler { + /** + * Tail of this channel's chain of request-handling tasks. + * Requests on a channel must be handled in arrival order. + */ + private CompletableFuture tail = CompletableFuture.completedFuture(null); + @Override protected void channelRead0(ChannelHandlerContext ctx, RaftNettyServerRequestProto proto) { - requestExecutor.execute(() -> { - final RaftNettyServerReplyProto reply; + tail = tail.handleAsync((prev, prevError) -> { + final CompletableFuture replyFuture; try { - // handle() already builds an error reply whenever it has a request context - reply = handle(proto); + replyFuture = handleAsync(proto); } catch (Exception e) { - // Close the channel so the client fails fast instead of blocking until timeout. + // No request context to build a reply; fail fast by closing the channel. LOG.warn("{}: Failed to handle request; closing the channel.", getId(), e); ctx.close(); - return; + return null; } - ctx.writeAndFlush(reply); - }); + replyFuture.whenComplete((reply, e) -> { + if (e != null) { + LOG.warn("{}: Failed to handle request; closing the channel.", getId(), e); + ctx.close(); + } else { + ctx.writeAndFlush(reply); + } + }); + return null; + }, requestExecutor); } @Override @@ -207,114 +219,108 @@ public InetSocketAddress getInetSocketAddress() { } } - RaftNettyServerReplyProto handle(RaftNettyServerRequestProto proto) { + CompletableFuture handleAsync(RaftNettyServerRequestProto proto) { RaftRpcRequestProto rpcRequest = null; try { + final CompletableFuture replyFuture; switch (proto.getRaftNettyServerRequestCase()) { - case REQUESTVOTEREQUEST: + case REQUESTVOTEREQUEST: { + // requestVote has no async variant; it is fast and does not block on commit. final RequestVoteRequestProto request = proto.getRequestVoteRequest(); rpcRequest = request.getServerRequest(); - final RequestVoteReplyProto reply = server.requestVote(request); - return RaftNettyServerReplyProto.newBuilder() - .setRequestVoteReply(reply) - .build(); - - case TRANSFERLEADERSHIPREQUEST: - final TransferLeadershipRequestProto transferLeadershipRequest = proto.getTransferLeadershipRequest(); - rpcRequest = transferLeadershipRequest.getRpcRequest(); - final RaftClientReply transferLeadershipReply = server.transferLeadership( - ClientProtoUtils.toTransferLeadershipRequest(transferLeadershipRequest)); - return RaftNettyServerReplyProto.newBuilder() - .setRaftClientReply(ClientProtoUtils.toRaftClientReplyProto(transferLeadershipReply)) - .build(); - - case STARTLEADERELECTIONREQUEST: - final StartLeaderElectionRequestProto startLeaderElectionRequest = proto.getStartLeaderElectionRequest(); - rpcRequest = startLeaderElectionRequest.getServerRequest(); - final StartLeaderElectionReplyProto startLeaderElectionReply = - server.startLeaderElection(startLeaderElectionRequest); - return RaftNettyServerReplyProto.newBuilder().setStartLeaderElectionReply(startLeaderElectionReply).build(); - - case SNAPSHOTMANAGEMENTREQUEST: - final SnapshotManagementRequestProto snapshotManagementRequest = proto.getSnapshotManagementRequest(); - rpcRequest = snapshotManagementRequest.getRpcRequest(); - final RaftClientReply snapshotManagementReply = server.snapshotManagement( - ClientProtoUtils.toSnapshotManagementRequest(snapshotManagementRequest)); - return RaftNettyServerReplyProto.newBuilder() - .setRaftClientReply(ClientProtoUtils.toRaftClientReplyProto(snapshotManagementReply)) - .build(); - - case LEADERELECTIONMANAGEMENTREQUEST: - final LeaderElectionManagementRequestProto leaderElectionManagementRequest = - proto.getLeaderElectionManagementRequest(); - rpcRequest = leaderElectionManagementRequest.getRpcRequest(); - final RaftClientReply leaderElectionManagementReply = server.leaderElectionManagement( - ClientProtoUtils.toLeaderElectionManagementRequest(leaderElectionManagementRequest)); - return RaftNettyServerReplyProto.newBuilder() - .setRaftClientReply(ClientProtoUtils.toRaftClientReplyProto(leaderElectionManagementReply)) - .build(); - - case APPENDENTRIESREQUEST: - final AppendEntriesRequestProto appendEntriesRequest = proto.getAppendEntriesRequest(); - rpcRequest = appendEntriesRequest.getServerRequest(); - final AppendEntriesReplyProto appendEntriesReply = server.appendEntries(appendEntriesRequest); - return RaftNettyServerReplyProto.newBuilder() - .setAppendEntriesReply(appendEntriesReply) - .build(); - - case INSTALLSNAPSHOTREQUEST: - final InstallSnapshotRequestProto installSnapshotRequest = proto.getInstallSnapshotRequest(); - rpcRequest = installSnapshotRequest.getServerRequest(); - final InstallSnapshotReplyProto installSnapshotReply = server.installSnapshot(installSnapshotRequest); - return RaftNettyServerReplyProto.newBuilder() - .setInstallSnapshotReply(installSnapshotReply) - .build(); - - case RAFTCLIENTREQUEST: - final RaftClientRequestProto raftClientRequest = proto.getRaftClientRequest(); - rpcRequest = raftClientRequest.getRpcRequest(); - final RaftClientReply raftClientReply = server.submitClientRequest( - ClientProtoUtils.toRaftClientRequest(raftClientRequest)); - return RaftNettyServerReplyProto.newBuilder() - .setRaftClientReply(ClientProtoUtils.toRaftClientReplyProto(raftClientReply)) - .build(); - - case SETCONFIGURATIONREQUEST: - final SetConfigurationRequestProto configurationRequest = proto.getSetConfigurationRequest(); - rpcRequest = configurationRequest.getRpcRequest(); - final RaftClientReply configurationReply = server.setConfiguration( - ClientProtoUtils.toSetConfigurationRequest(configurationRequest)); - return RaftNettyServerReplyProto.newBuilder() - .setRaftClientReply(ClientProtoUtils.toRaftClientReplyProto(configurationReply)) - .build(); - - case GROUPMANAGEMENTREQUEST: - final GroupManagementRequestProto groupManagementRequest = proto.getGroupManagementRequest(); - rpcRequest = groupManagementRequest.getRpcRequest(); - final RaftClientReply groupManagementReply = server.groupManagement( - ClientProtoUtils.toGroupManagementRequest(groupManagementRequest)); - return RaftNettyServerReplyProto.newBuilder() - .setRaftClientReply(ClientProtoUtils.toRaftClientReplyProto(groupManagementReply)) - .build(); - - case GROUPLISTREQUEST: - final GroupListRequestProto groupListRequest = proto.getGroupListRequest(); - rpcRequest = groupListRequest.getRpcRequest(); - final GroupListReply groupListReply = server.getGroupList( - ClientProtoUtils.toGroupListRequest(groupListRequest)); - return RaftNettyServerReplyProto.newBuilder() - .setGroupListReply(ClientProtoUtils.toGroupListReplyProto(groupListReply)) - .build(); - - case GROUPINFOREQUEST: - final GroupInfoRequestProto groupInfoRequest = proto.getGroupInfoRequest(); - rpcRequest = groupInfoRequest.getRpcRequest(); - final GroupInfoReply groupInfoReply = server.getGroupInfo( - ClientProtoUtils.toGroupInfoRequest(groupInfoRequest)); - return RaftNettyServerReplyProto.newBuilder() - .setGroupInfoReply(ClientProtoUtils.toGroupInfoReplyProto(groupInfoReply)) - .build(); - + replyFuture = CompletableFuture.completedFuture(RaftNettyServerReplyProto.newBuilder() + .setRequestVoteReply(server.requestVote(request)) + .build()); + break; + } + case TRANSFERLEADERSHIPREQUEST: { + final TransferLeadershipRequestProto request = proto.getTransferLeadershipRequest(); + rpcRequest = request.getRpcRequest(); + replyFuture = server.transferLeadershipAsync(ClientProtoUtils.toTransferLeadershipRequest(request)) + .thenApply(NettyRpcService::toRaftClientReply); + break; + } + case STARTLEADERELECTIONREQUEST: { + // startLeaderElection has no async variant; it is fast and does not block on commit. + final StartLeaderElectionRequestProto request = proto.getStartLeaderElectionRequest(); + rpcRequest = request.getServerRequest(); + replyFuture = CompletableFuture.completedFuture(RaftNettyServerReplyProto.newBuilder() + .setStartLeaderElectionReply(server.startLeaderElection(request)) + .build()); + break; + } + case SNAPSHOTMANAGEMENTREQUEST: { + final SnapshotManagementRequestProto request = proto.getSnapshotManagementRequest(); + rpcRequest = request.getRpcRequest(); + replyFuture = server.snapshotManagementAsync(ClientProtoUtils.toSnapshotManagementRequest(request)) + .thenApply(NettyRpcService::toRaftClientReply); + break; + } + case LEADERELECTIONMANAGEMENTREQUEST: { + final LeaderElectionManagementRequestProto request = proto.getLeaderElectionManagementRequest(); + rpcRequest = request.getRpcRequest(); + replyFuture = server.leaderElectionManagementAsync( + ClientProtoUtils.toLeaderElectionManagementRequest(request)) + .thenApply(NettyRpcService::toRaftClientReply); + break; + } + case APPENDENTRIESREQUEST: { + final AppendEntriesRequestProto request = proto.getAppendEntriesRequest(); + rpcRequest = request.getServerRequest(); + replyFuture = server.appendEntriesAsync(request) + .thenApply(reply -> RaftNettyServerReplyProto.newBuilder() + .setAppendEntriesReply(reply) + .build()); + break; + } + case INSTALLSNAPSHOTREQUEST: { + // installSnapshot has no async variant; it runs on this per-channel serialized path. + final InstallSnapshotRequestProto request = proto.getInstallSnapshotRequest(); + rpcRequest = request.getServerRequest(); + replyFuture = CompletableFuture.completedFuture(RaftNettyServerReplyProto.newBuilder() + .setInstallSnapshotReply(server.installSnapshot(request)) + .build()); + break; + } + case RAFTCLIENTREQUEST: { + final RaftClientRequestProto request = proto.getRaftClientRequest(); + rpcRequest = request.getRpcRequest(); + replyFuture = server.submitClientRequestAsync(ClientProtoUtils.toRaftClientRequest(request)) + .thenApply(NettyRpcService::toRaftClientReply); + break; + } + case SETCONFIGURATIONREQUEST: { + final SetConfigurationRequestProto request = proto.getSetConfigurationRequest(); + rpcRequest = request.getRpcRequest(); + replyFuture = server.setConfigurationAsync(ClientProtoUtils.toSetConfigurationRequest(request)) + .thenApply(NettyRpcService::toRaftClientReply); + break; + } + case GROUPMANAGEMENTREQUEST: { + final GroupManagementRequestProto request = proto.getGroupManagementRequest(); + rpcRequest = request.getRpcRequest(); + replyFuture = server.groupManagementAsync(ClientProtoUtils.toGroupManagementRequest(request)) + .thenApply(NettyRpcService::toRaftClientReply); + break; + } + case GROUPLISTREQUEST: { + final GroupListRequestProto request = proto.getGroupListRequest(); + rpcRequest = request.getRpcRequest(); + replyFuture = server.getGroupListAsync(ClientProtoUtils.toGroupListRequest(request)) + .thenApply(reply -> RaftNettyServerReplyProto.newBuilder() + .setGroupListReply(ClientProtoUtils.toGroupListReplyProto(reply)) + .build()); + break; + } + case GROUPINFOREQUEST: { + final GroupInfoRequestProto request = proto.getGroupInfoRequest(); + rpcRequest = request.getRpcRequest(); + replyFuture = server.getGroupInfoAsync(ClientProtoUtils.toGroupInfoRequest(request)) + .thenApply(reply -> RaftNettyServerReplyProto.newBuilder() + .setGroupInfoReply(ClientProtoUtils.toGroupInfoReplyProto(reply)) + .build()); + break; + } case RAFTNETTYSERVERREQUEST_NOT_SET: throw new IllegalArgumentException("Request case not set in proto: " + proto.getRaftNettyServerRequestCase()); @@ -322,18 +328,31 @@ RaftNettyServerReplyProto handle(RaftNettyServerRequestProto proto) { throw new UnsupportedOperationException("Request case not supported: " + proto.getRaftNettyServerRequestCase()); } + + final RaftRpcRequestProto request = rpcRequest; + // Convert an asynchronous failure into an error reply (the client casts it to IOException). + return replyFuture.exceptionally(e -> toRaftNettyServerReplyProto(request, toIOException(e))); } catch (Exception e) { + // A synchronous failure before the reply future was created. if (rpcRequest == null) { - // let InboundHandler close the channel so the client fails fast. + // No request context to build a targeted reply; let InboundHandler close the channel. throw new IllegalStateException(getId() + ": Failed to handle request " + proto, e); } - // The client deserializes the reply and casts it to IOException, so always send an - // IOException regardless of the actual failure type. - final IOException ioe = e instanceof IOException ? (IOException) e : new IOException(e); - return toRaftNettyServerReplyProto(rpcRequest, ioe); + return CompletableFuture.completedFuture(toRaftNettyServerReplyProto(rpcRequest, toIOException(e))); } } + private static RaftNettyServerReplyProto toRaftClientReply(RaftClientReply reply) { + return RaftNettyServerReplyProto.newBuilder() + .setRaftClientReply(ClientProtoUtils.toRaftClientReplyProto(reply)) + .build(); + } + + private static IOException toIOException(Throwable t) { + final Throwable cause = t instanceof CompletionException && t.getCause() != null ? t.getCause() : t; + return cause instanceof IOException ? (IOException) cause : new IOException(cause); + } + private static RaftNettyServerReplyProto toRaftNettyServerReplyProto( RaftRpcRequestProto request, IOException e) { final RaftRpcReplyProto.Builder rpcReply = RaftRpcReplyProto.newBuilder() diff --git a/ratis-netty/src/test/java/org/apache/ratis/netty/server/TestNettyRpcService.java b/ratis-netty/src/test/java/org/apache/ratis/netty/server/TestNettyRpcService.java index ad20fb2ec4..8579909135 100644 --- a/ratis-netty/src/test/java/org/apache/ratis/netty/server/TestNettyRpcService.java +++ b/ratis-netty/src/test/java/org/apache/ratis/netty/server/TestNettyRpcService.java @@ -70,7 +70,7 @@ public void testHandleReturnsErrorReplyOnRuntimeException() throws Exception { final NettyRpcService service = NettyRpcService.newBuilder().setServer(server).build(); service.start(); try { - final RaftNettyServerReplyProto reply = service.handle(newRequestVoteProto()); + final RaftNettyServerReplyProto reply = service.handleAsync(newRequestVoteProto()).join(); Assertions.assertEquals(RaftNettyServerReplyCase.EXCEPTIONREPLY, reply.getRaftNettyServerReplyCase()); } finally { service.close(); From 944136923892b7955d4949e25481d61582c99c6e Mon Sep 17 00:00:00 2001 From: Abhishek Pal Date: Sun, 2 Aug 2026 21:55:42 +0530 Subject: [PATCH 3/5] Address CI issues --- .../ratis/netty/server/NettyRpcService.java | 126 +++++++++--------- 1 file changed, 65 insertions(+), 61 deletions(-) diff --git a/ratis-netty/src/main/java/org/apache/ratis/netty/server/NettyRpcService.java b/ratis-netty/src/main/java/org/apache/ratis/netty/server/NettyRpcService.java index 36bbbf6218..81dd657f8f 100644 --- a/ratis-netty/src/main/java/org/apache/ratis/netty/server/NettyRpcService.java +++ b/ratis-netty/src/main/java/org/apache/ratis/netty/server/NettyRpcService.java @@ -224,103 +224,107 @@ CompletableFuture handleAsync(RaftNettyServerRequestP try { final CompletableFuture replyFuture; switch (proto.getRaftNettyServerRequestCase()) { - case REQUESTVOTEREQUEST: { + case REQUESTVOTEREQUEST: // requestVote has no async variant; it is fast and does not block on commit. - final RequestVoteRequestProto request = proto.getRequestVoteRequest(); - rpcRequest = request.getServerRequest(); + final RequestVoteRequestProto requestVoteRequest = proto.getRequestVoteRequest(); + rpcRequest = requestVoteRequest.getServerRequest(); replyFuture = CompletableFuture.completedFuture(RaftNettyServerReplyProto.newBuilder() - .setRequestVoteReply(server.requestVote(request)) + .setRequestVoteReply(server.requestVote(requestVoteRequest)) .build()); break; - } - case TRANSFERLEADERSHIPREQUEST: { - final TransferLeadershipRequestProto request = proto.getTransferLeadershipRequest(); - rpcRequest = request.getRpcRequest(); - replyFuture = server.transferLeadershipAsync(ClientProtoUtils.toTransferLeadershipRequest(request)) + + case TRANSFERLEADERSHIPREQUEST: + final TransferLeadershipRequestProto transferLeadershipRequest = proto.getTransferLeadershipRequest(); + rpcRequest = transferLeadershipRequest.getRpcRequest(); + replyFuture = server.transferLeadershipAsync( + ClientProtoUtils.toTransferLeadershipRequest(transferLeadershipRequest)) .thenApply(NettyRpcService::toRaftClientReply); break; - } - case STARTLEADERELECTIONREQUEST: { + + case STARTLEADERELECTIONREQUEST: // startLeaderElection has no async variant; it is fast and does not block on commit. - final StartLeaderElectionRequestProto request = proto.getStartLeaderElectionRequest(); - rpcRequest = request.getServerRequest(); + final StartLeaderElectionRequestProto startLeaderElectionRequest = proto.getStartLeaderElectionRequest(); + rpcRequest = startLeaderElectionRequest.getServerRequest(); replyFuture = CompletableFuture.completedFuture(RaftNettyServerReplyProto.newBuilder() - .setStartLeaderElectionReply(server.startLeaderElection(request)) + .setStartLeaderElectionReply(server.startLeaderElection(startLeaderElectionRequest)) .build()); break; - } - case SNAPSHOTMANAGEMENTREQUEST: { - final SnapshotManagementRequestProto request = proto.getSnapshotManagementRequest(); - rpcRequest = request.getRpcRequest(); - replyFuture = server.snapshotManagementAsync(ClientProtoUtils.toSnapshotManagementRequest(request)) + + case SNAPSHOTMANAGEMENTREQUEST: + final SnapshotManagementRequestProto snapshotManagementRequest = proto.getSnapshotManagementRequest(); + rpcRequest = snapshotManagementRequest.getRpcRequest(); + replyFuture = server.snapshotManagementAsync( + ClientProtoUtils.toSnapshotManagementRequest(snapshotManagementRequest)) .thenApply(NettyRpcService::toRaftClientReply); break; - } - case LEADERELECTIONMANAGEMENTREQUEST: { - final LeaderElectionManagementRequestProto request = proto.getLeaderElectionManagementRequest(); - rpcRequest = request.getRpcRequest(); + + case LEADERELECTIONMANAGEMENTREQUEST: + final LeaderElectionManagementRequestProto leaderElectionManagementRequest = + proto.getLeaderElectionManagementRequest(); + rpcRequest = leaderElectionManagementRequest.getRpcRequest(); replyFuture = server.leaderElectionManagementAsync( - ClientProtoUtils.toLeaderElectionManagementRequest(request)) + ClientProtoUtils.toLeaderElectionManagementRequest(leaderElectionManagementRequest)) .thenApply(NettyRpcService::toRaftClientReply); break; - } - case APPENDENTRIESREQUEST: { - final AppendEntriesRequestProto request = proto.getAppendEntriesRequest(); - rpcRequest = request.getServerRequest(); - replyFuture = server.appendEntriesAsync(request) + + case APPENDENTRIESREQUEST: + final AppendEntriesRequestProto appendEntriesRequest = proto.getAppendEntriesRequest(); + rpcRequest = appendEntriesRequest.getServerRequest(); + replyFuture = server.appendEntriesAsync(appendEntriesRequest) .thenApply(reply -> RaftNettyServerReplyProto.newBuilder() .setAppendEntriesReply(reply) .build()); break; - } - case INSTALLSNAPSHOTREQUEST: { + + case INSTALLSNAPSHOTREQUEST: // installSnapshot has no async variant; it runs on this per-channel serialized path. - final InstallSnapshotRequestProto request = proto.getInstallSnapshotRequest(); - rpcRequest = request.getServerRequest(); + final InstallSnapshotRequestProto installSnapshotRequest = proto.getInstallSnapshotRequest(); + rpcRequest = installSnapshotRequest.getServerRequest(); replyFuture = CompletableFuture.completedFuture(RaftNettyServerReplyProto.newBuilder() - .setInstallSnapshotReply(server.installSnapshot(request)) + .setInstallSnapshotReply(server.installSnapshot(installSnapshotRequest)) .build()); break; - } - case RAFTCLIENTREQUEST: { - final RaftClientRequestProto request = proto.getRaftClientRequest(); - rpcRequest = request.getRpcRequest(); - replyFuture = server.submitClientRequestAsync(ClientProtoUtils.toRaftClientRequest(request)) + + case RAFTCLIENTREQUEST: + final RaftClientRequestProto raftClientRequest = proto.getRaftClientRequest(); + rpcRequest = raftClientRequest.getRpcRequest(); + replyFuture = server.submitClientRequestAsync(ClientProtoUtils.toRaftClientRequest(raftClientRequest)) .thenApply(NettyRpcService::toRaftClientReply); break; - } - case SETCONFIGURATIONREQUEST: { - final SetConfigurationRequestProto request = proto.getSetConfigurationRequest(); - rpcRequest = request.getRpcRequest(); - replyFuture = server.setConfigurationAsync(ClientProtoUtils.toSetConfigurationRequest(request)) + + case SETCONFIGURATIONREQUEST: + final SetConfigurationRequestProto setConfigurationRequest = proto.getSetConfigurationRequest(); + rpcRequest = setConfigurationRequest.getRpcRequest(); + replyFuture = server.setConfigurationAsync( + ClientProtoUtils.toSetConfigurationRequest(setConfigurationRequest)) .thenApply(NettyRpcService::toRaftClientReply); break; - } - case GROUPMANAGEMENTREQUEST: { - final GroupManagementRequestProto request = proto.getGroupManagementRequest(); - rpcRequest = request.getRpcRequest(); - replyFuture = server.groupManagementAsync(ClientProtoUtils.toGroupManagementRequest(request)) + + case GROUPMANAGEMENTREQUEST: + final GroupManagementRequestProto groupManagementRequest = proto.getGroupManagementRequest(); + rpcRequest = groupManagementRequest.getRpcRequest(); + replyFuture = server.groupManagementAsync(ClientProtoUtils.toGroupManagementRequest(groupManagementRequest)) .thenApply(NettyRpcService::toRaftClientReply); break; - } - case GROUPLISTREQUEST: { - final GroupListRequestProto request = proto.getGroupListRequest(); - rpcRequest = request.getRpcRequest(); - replyFuture = server.getGroupListAsync(ClientProtoUtils.toGroupListRequest(request)) + + case GROUPLISTREQUEST: + final GroupListRequestProto groupListRequest = proto.getGroupListRequest(); + rpcRequest = groupListRequest.getRpcRequest(); + replyFuture = server.getGroupListAsync(ClientProtoUtils.toGroupListRequest(groupListRequest)) .thenApply(reply -> RaftNettyServerReplyProto.newBuilder() .setGroupListReply(ClientProtoUtils.toGroupListReplyProto(reply)) .build()); break; - } - case GROUPINFOREQUEST: { - final GroupInfoRequestProto request = proto.getGroupInfoRequest(); - rpcRequest = request.getRpcRequest(); - replyFuture = server.getGroupInfoAsync(ClientProtoUtils.toGroupInfoRequest(request)) + + case GROUPINFOREQUEST: + final GroupInfoRequestProto groupInfoRequest = proto.getGroupInfoRequest(); + rpcRequest = groupInfoRequest.getRpcRequest(); + replyFuture = server.getGroupInfoAsync(ClientProtoUtils.toGroupInfoRequest(groupInfoRequest)) .thenApply(reply -> RaftNettyServerReplyProto.newBuilder() .setGroupInfoReply(ClientProtoUtils.toGroupInfoReplyProto(reply)) .build()); break; - } + case RAFTNETTYSERVERREQUEST_NOT_SET: throw new IllegalArgumentException("Request case not set in proto: " + proto.getRaftNettyServerRequestCase()); @@ -332,7 +336,7 @@ CompletableFuture handleAsync(RaftNettyServerRequestP final RaftRpcRequestProto request = rpcRequest; // Convert an asynchronous failure into an error reply (the client casts it to IOException). return replyFuture.exceptionally(e -> toRaftNettyServerReplyProto(request, toIOException(e))); - } catch (Exception e) { + } catch (IOException | RuntimeException e) { // A synchronous failure before the reply future was created. if (rpcRequest == null) { // No request context to build a targeted reply; let InboundHandler close the channel. From 1282dcea521215a22b2ce0c711203cb80f0b5a93 Mon Sep 17 00:00:00 2001 From: Abhishek Pal Date: Sun, 2 Aug 2026 22:21:49 +0530 Subject: [PATCH 4/5] Switch to a fixed pool --- .../src/main/java/org/apache/ratis/netty/NettyConfigKeys.java | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/ratis-netty/src/main/java/org/apache/ratis/netty/NettyConfigKeys.java b/ratis-netty/src/main/java/org/apache/ratis/netty/NettyConfigKeys.java index 5984fff11f..e7ba5d8101 100644 --- a/ratis-netty/src/main/java/org/apache/ratis/netty/NettyConfigKeys.java +++ b/ratis-netty/src/main/java/org/apache/ratis/netty/NettyConfigKeys.java @@ -76,7 +76,9 @@ static void setUseEpoll(RaftProperties properties, boolean enable) { } String ASYNC_REQUEST_THREAD_POOL_CACHED_KEY = PREFIX + ".async.request.thread.pool.cached"; - boolean ASYNC_REQUEST_THREAD_POOL_CACHED_DEFAULT = true; + // Default to a fixed pool. + // TODO: Refer to https://issues.apache.org/jira/browse/RATIS-2637 + boolean ASYNC_REQUEST_THREAD_POOL_CACHED_DEFAULT = false; static boolean asyncRequestThreadPoolCached(RaftProperties properties) { return getBoolean(properties::getBoolean, ASYNC_REQUEST_THREAD_POOL_CACHED_KEY, ASYNC_REQUEST_THREAD_POOL_CACHED_DEFAULT, getDefaultLog()); From 9a6720bec3566d899a0b30988af6bb142cea0d91 Mon Sep 17 00:00:00 2001 From: Abhishek Pal Date: Mon, 3 Aug 2026 07:38:47 +0530 Subject: [PATCH 5/5] Address timeout duration, fix minor issue which can cause leak --- .../apache/ratis/netty/server/NettyRpcService.java | 14 ++++++++------ 1 file changed, 8 insertions(+), 6 deletions(-) diff --git a/ratis-netty/src/main/java/org/apache/ratis/netty/server/NettyRpcService.java b/ratis-netty/src/main/java/org/apache/ratis/netty/server/NettyRpcService.java index 81dd657f8f..b93137d196 100644 --- a/ratis-netty/src/main/java/org/apache/ratis/netty/server/NettyRpcService.java +++ b/ratis-netty/src/main/java/org/apache/ratis/netty/server/NettyRpcService.java @@ -44,6 +44,7 @@ import org.apache.ratis.util.JavaUtils; import org.apache.ratis.util.MemoizedSupplier; import org.apache.ratis.util.ProtoUtils; +import org.apache.ratis.util.TimeDuration; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -148,11 +149,6 @@ protected void initChannel(SocketChannel ch) { } }; - this.requestExecutor = ConcurrentUtils.newThreadPoolWithMax( - NettyConfigKeys.Server.asyncRequestThreadPoolCached(server.getProperties()), - NettyConfigKeys.Server.asyncRequestThreadPoolSize(server.getProperties()), - server.getId() + "-request-"); - final boolean useEpoll = NettyConfigKeys.Server.useEpoll(server.getProperties()); this.bossGroup = NettyUtils.newEventLoopGroup(CLASS_NAME + "-bossGroup", 0, useEpoll); this.workerGroup = NettyUtils.newEventLoopGroup(CLASS_NAME + "-workerGroup",0, useEpoll); @@ -167,6 +163,11 @@ protected void initChannel(SocketChannel ch) { .handler(new LoggingHandler(LogLevel.INFO)) .childHandler(initializer) .bind(socketAddress)); + + this.requestExecutor = ConcurrentUtils.newThreadPoolWithMax( + NettyConfigKeys.Server.asyncRequestThreadPoolCached(server.getProperties()), + NettyConfigKeys.Server.asyncRequestThreadPoolSize(server.getProperties()), + server.getId() + "-request-"); } @Override @@ -192,7 +193,8 @@ public void startImpl() throws IOException { @Override public void closeImpl() throws IOException { - ConcurrentUtils.shutdownAndWait(requestExecutor); + ConcurrentUtils.shutdownAndWait(TimeDuration.ONE_SECOND, requestExecutor, + timeout -> LOG.warn("{}: requestExecutor shutdown timeout in {}", this, timeout)); final ChannelFuture f = getChannel().close(); f.syncUninterruptibly(); bossGroup.shutdownGracefully(0, 100, TimeUnit.MILLISECONDS);