RATIS-2629. Handle Netty based request asynchronously and improve exception handling

ratis-netty/pom.xml

@@ -88,6 +88,11 @@
       <artifactId>junit-platform-launcher</artifactId>
       <scope>test</scope>
     </dependency>
+    <dependency>
+      <groupId>org.mockito</groupId>
+      <artifactId>mockito-core</artifactId>
+      <scope>test</scope>
+    </dependency>
 
   </dependencies>
 

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 {

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<ChannelFuture> channel;
   private final InetSocketAddress socketAddress;
 
+  private final ExecutorService requestExecutor;
+
   @ChannelHandler.Sharable
   class InboundHandler extends SimpleChannelInboundHandler<RaftNettyServerRequestProto> {
     @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);
     }
   }
 

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<String> 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();
+    }
+  }
+}