shutdownAndWait.

szetszwo · Dec 1, 2023 · d5019c1 · d5019c1
1 parent ae21773
commit d5019c1
Show file tree

Hide file tree

Showing 3 changed files with 7 additions and 5 deletions.
diff --git a/ratis-netty/src/main/java/org/apache/ratis/netty/client/NettyClientStreamRpc.java b/ratis-netty/src/main/java/org/apache/ratis/netty/client/NettyClientStreamRpc.java
@@ -288,7 +288,7 @@ synchronized boolean shouldFlush(boolean force, int countMin, SizeInBytes bytesM
   private final OutstandingRequests outstandingRequests = new OutstandingRequests();
 
   public NettyClientStreamRpc(RaftPeer server, TlsConf tlsConf, RaftProperties properties) {
-    this.name = JavaUtils.getClassSimpleName(getClass()) + "->" + server;
+    this.name = JavaUtils.getClassSimpleName(getClass()) + "->" + server.getId();
     this.requestTimeout = RaftClientConfigKeys.DataStream.requestTimeout(properties);
     this.closeTimeout = requestTimeout.multiply(2);
 

diff --git a/ratis-netty/src/main/java/org/apache/ratis/netty/server/DataStreamManagement.java b/ratis-netty/src/main/java/org/apache/ratis/netty/server/DataStreamManagement.java
@@ -279,8 +279,10 @@ public Set<ClientInvocationId> remove(ChannelId channelId) {
   }
 
   void shutdown() {
-    requestExecutor.shutdown();
-    writeExecutor.shutdown();
+    ConcurrentUtils.shutdownAndWait(TimeDuration.ONE_SECOND, requestExecutor,
+        timeout -> LOG.warn("{}: requestExecutor shutdown timeout in {}", this, timeout));
+    ConcurrentUtils.shutdownAndWait(TimeDuration.ONE_SECOND, writeExecutor,
+        timeout -> LOG.warn("{}: writeExecutor shutdown timeout in {}", this, timeout));
   }
 
   private CompletableFuture<DataStream> stream(RaftClientRequest request, StateMachine stateMachine) {

diff --git a/ratis-netty/src/main/java/org/apache/ratis/netty/server/NettyServerStreamRpc.java b/ratis-netty/src/main/java/org/apache/ratis/netty/server/NettyServerStreamRpc.java
@@ -318,9 +318,9 @@ public void close() {
       bossGroup.shutdownGracefully(0, 100, TimeUnit.MILLISECONDS);
       workerGroup.shutdownGracefully(0, 100, TimeUnit.MILLISECONDS);
       ConcurrentUtils.shutdownAndWait(TimeDuration.ONE_SECOND, bossGroup,
-          timeout -> LOG.warn("{}: bossGroup shutdown timeout in " + timeout, this));
+          timeout -> LOG.warn("{}: bossGroup shutdown timeout in {}", this, timeout));
       ConcurrentUtils.shutdownAndWait(TimeDuration.ONE_SECOND, workerGroup,
-          timeout -> LOG.warn("{}: workerGroup shutdown timeout in " + timeout, this));
+          timeout -> LOG.warn("{}: workerGroup shutdown timeout in {}", this, timeout));
     } catch (InterruptedException e) {
       Thread.currentThread().interrupt();
       LOG.error(this + ": Interrupted close()", e);