RATIS-2074. Intermittent fork timeout in TestRaftWithNetty#testBasicA…

…ppendEntriesKillLeader.
szetszwo · May 6, 2024 · 5d2ec79 · 5d2ec79
1 parent 839f8fb
commit 5d2ec79
Show file tree

Hide file tree

Showing 3 changed files with 34 additions and 9 deletions.
diff --git a/ratis-netty/src/main/java/org/apache/ratis/netty/NettyClient.java b/ratis-netty/src/main/java/org/apache/ratis/netty/NettyClient.java
@@ -33,13 +33,17 @@
 import java.net.InetSocketAddress;
 
 public class NettyClient implements Closeable {
-  private final LifeCycle lifeCycle = new LifeCycle(JavaUtils.getClassSimpleName(getClass()));
-
+  private final LifeCycle lifeCycle;
+  private final String serverAddress;
   private Channel channel;
 
+  NettyClient(String serverAddress) {
+    this.lifeCycle = new LifeCycle(JavaUtils.getClassSimpleName(getClass()) + "-" + serverAddress);
+    this.serverAddress = serverAddress;
+  }
+
   /** Connects to the given server address. */
-  public void connect(String serverAddress, EventLoopGroup group,
-                      ChannelInitializer<SocketChannel> initializer)
+  public void connect(EventLoopGroup group, ChannelInitializer<SocketChannel> initializer)
       throws InterruptedException {
     final InetSocketAddress address = NetUtils.createSocketAddr(serverAddress);
 
@@ -57,13 +61,16 @@ public void connect(String serverAddress, EventLoopGroup group,
 
   @Override
   public void close() {
-    lifeCycle.checkStateAndClose(() -> {
-      channel.close().syncUninterruptibly();
-    });
+    lifeCycle.checkStateAndClose(() -> NettyUtils.closeChannel(channel, serverAddress));
   }
 
   public ChannelFuture writeAndFlush(Object msg) {
     lifeCycle.assertCurrentState(LifeCycle.States.RUNNING);
     return channel.writeAndFlush(msg);
   }
+
+  @Override
+  public String toString() {
+    return lifeCycle.toString();
+  }
 }
diff --git a/ratis-netty/src/main/java/org/apache/ratis/netty/NettyRpcProxy.java b/ratis-netty/src/main/java/org/apache/ratis/netty/NettyRpcProxy.java
@@ -99,7 +99,7 @@ public static long getCallId(RaftNettyServerReplyProto proto) {
 
 
   class Connection implements Closeable {
-    private final NettyClient client = new NettyClient();
+    private final NettyClient client = new NettyClient(peer.getAddress());
     private final Queue<CompletableFuture<RaftNettyServerReplyProto>> replies
         = new LinkedList<>();
 
@@ -137,7 +137,7 @@ protected void initChannel(SocketChannel ch) throws Exception {
         }
       };
 
-      client.connect(peer.getAddress(), group, initializer);
+      client.connect(group, initializer);
     }
 
     synchronized ChannelFuture offer(RaftNettyServerRequestProto request,

diff --git a/ratis-netty/src/main/java/org/apache/ratis/netty/NettyUtils.java b/ratis-netty/src/main/java/org/apache/ratis/netty/NettyUtils.java
@@ -22,6 +22,8 @@
 import org.apache.ratis.security.TlsConf.KeyManagerConf;
 import org.apache.ratis.security.TlsConf.PrivateKeyConf;
 import org.apache.ratis.security.TlsConf.TrustManagerConf;
+import org.apache.ratis.thirdparty.io.netty.channel.Channel;
+import org.apache.ratis.thirdparty.io.netty.channel.ChannelFuture;
 import org.apache.ratis.thirdparty.io.netty.channel.EventLoopGroup;
 import org.apache.ratis.thirdparty.io.netty.channel.ServerChannel;
 import org.apache.ratis.thirdparty.io.netty.channel.epoll.Epoll;
@@ -35,16 +37,19 @@
 import org.apache.ratis.thirdparty.io.netty.handler.ssl.SslContext;
 import org.apache.ratis.thirdparty.io.netty.handler.ssl.SslContextBuilder;
 import org.apache.ratis.util.ConcurrentUtils;
+import org.apache.ratis.util.TimeDuration;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import javax.net.ssl.KeyManager;
 import javax.net.ssl.TrustManager;
+import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.function.Function;
 
 public interface NettyUtils {
   Logger LOG = LoggerFactory.getLogger(NettyUtils.class);
+  TimeDuration CLOSE_TIMEOUT = TimeDuration.valueOf(10, TimeUnit.SECONDS);
 
   class Print {
     private static final AtomicBoolean PRINTED_EPOLL_UNAVAILABILITY_CAUSE = new AtomicBoolean();
@@ -176,4 +181,17 @@ static Class<? extends ServerChannel> getServerChannelClass(EventLoopGroup event
     return eventLoopGroup instanceof EpollEventLoopGroup ?
         EpollServerSocketChannel.class : NioServerSocketChannel.class;
   }
+
+  static void closeChannel(Channel channel, String name) {
+    final ChannelFuture f = channel.close();
+    try {
+      f.await(CLOSE_TIMEOUT.getDuration(), CLOSE_TIMEOUT.getUnit());
+    } catch (InterruptedException e) {
+      Thread.currentThread().interrupt();
+      LOG.info("Interrupted closeChannel {} ", name, e);
+    }
+    if (!f.isSuccess()) {
+      LOG.warn("Failed to close channel {} in {}", name, CLOSE_TIMEOUT);
+    }
+  }
 }