This is an automated email from the ASF dual-hosted git repository.
szetszwo pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/ratis.git
The following commit(s) were added to refs/heads/master by this push:
new ac05d64f3 RATIS-2074. Intermittent fork timeout in
TestRaftWithNetty#testBasicAppendEntriesKillLeader. (#1080)
ac05d64f3 is described below
commit ac05d64f3f846d56658442896a39f4f5fd7495a2
Author: Tsz-Wo Nicholas Sze <[email protected]>
AuthorDate: Tue May 7 11:27:36 2024 -0700
RATIS-2074. Intermittent fork timeout in
TestRaftWithNetty#testBasicAppendEntriesKillLeader. (#1080)
---
.../java/org/apache/ratis/netty/NettyClient.java | 21 ++++++++++++++-------
.../java/org/apache/ratis/netty/NettyRpcProxy.java | 4 ++--
.../java/org/apache/ratis/netty/NettyUtils.java | 20 ++++++++++++++++++++
.../test/java/org/apache/ratis/RaftBasicTests.java | 2 ++
4 files changed, 38 insertions(+), 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
index 0cf4bd383..efea5fd0f 100644
--- 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.io.Closeable;
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 class NettyClient implements Closeable {
@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
index b7a04b050..b9788a8bb 100644
--- 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 class NettyRpcProxy implements Closeable {
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 @@ public class NettyRpcProxy implements Closeable {
}
};
- 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
index 8cce291af..37666bf18 100644
--- 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.CertificatesConf;
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.channel.socket.nio.NioSocketChannel;
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(5, TimeUnit.SECONDS);
class Print {
private static final AtomicBoolean PRINTED_EPOLL_UNAVAILABILITY_CAUSE =
new AtomicBoolean();
@@ -176,4 +181,19 @@ public interface NettyUtils {
return eventLoopGroup instanceof EpollEventLoopGroup ?
EpollServerSocketChannel.class : NioServerSocketChannel.class;
}
+
+ static void closeChannel(Channel channel, String name) {
+ final ChannelFuture f = channel.close();
+ final boolean completed;
+ try {
+ completed = f.await(CLOSE_TIMEOUT.getDuration(),
CLOSE_TIMEOUT.getUnit());
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ LOG.info("Interrupted closeChannel {} ", name, e);
+ return;
+ }
+ if (!completed) {
+ LOG.warn("closeChannel {} is not yet completed in {}", name,
CLOSE_TIMEOUT);
+ }
+ }
}
\ No newline at end of file
diff --git a/ratis-server/src/test/java/org/apache/ratis/RaftBasicTests.java
b/ratis-server/src/test/java/org/apache/ratis/RaftBasicTests.java
index 13ee08c83..9f360cb51 100644
--- a/ratis-server/src/test/java/org/apache/ratis/RaftBasicTests.java
+++ b/ratis-server/src/test/java/org/apache/ratis/RaftBasicTests.java
@@ -45,6 +45,7 @@ import org.apache.ratis.util.Timestamp;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Assumptions;
import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.Timeout;
import org.slf4j.Logger;
import org.slf4j.event.Level;
@@ -338,6 +339,7 @@ public abstract class RaftBasicTests<CLUSTER extends
MiniRaftCluster>
}
@Test
+ @Timeout(value = 300)
public void testWithLoad() throws Exception {
runWithNewCluster(NUM_SERVERS, cluster -> testWithLoad(10, 300, false,
cluster, LOG));
}
