RATIS-2349. NettyClient#writeAndFlush should support throwing AlreadyClosedException (#1303)

Author: leixm

ratis-netty/src/main/java/org/apache/ratis/netty/NettyClient.java

@@ -17,6 +17,7 @@
  */
 package org.apache.ratis.netty;
 
+import org.apache.ratis.protocol.exceptions.AlreadyClosedException;
 import org.apache.ratis.thirdparty.io.netty.bootstrap.Bootstrap;
 import org.apache.ratis.thirdparty.io.netty.channel.Channel;
 import org.apache.ratis.thirdparty.io.netty.channel.ChannelFuture;
@@ -64,9 +65,19 @@ public void close() {
     lifeCycle.checkStateAndClose(() -> NettyUtils.closeChannel(channel, serverAddress));
   }
 
-  public ChannelFuture writeAndFlush(Object msg) {
-    lifeCycle.assertCurrentState(LifeCycle.States.RUNNING);
-    return channel.writeAndFlush(msg);
+  public ChannelFuture writeAndFlush(Object msg) throws AlreadyClosedException {
+    final LifeCycle.State state = lifeCycle.getCurrentState();
+    if (state.isRunning()) {
+      return channel.writeAndFlush(msg);
+    }
+    // For CLOSING, CLOSED, and EXCEPTION states, throw AlreadyClosedException to trigger reconnection
+    if (state.isClosingOrClosed() || state == LifeCycle.State.EXCEPTION) {
+      throw new AlreadyClosedException(
+        "Client is closed or failed: state=" + state + ", channel=" + channel);
+    }
+    // For other states (NEW, STARTING, PAUSING, PAUSED), this is a programming error
+    throw new IllegalStateException("Client is in unexpected state for writeAndFlush: " +
+      "state=" + state + ", channel=" + channel);
   }
 
   @Override

ratis-netty/src/main/java/org/apache/ratis/netty/NettyRpcProxy.java

@@ -157,7 +157,7 @@ protected void initChannel(SocketChannel ch) throws Exception {
     }
 
     synchronized ChannelFuture offer(RaftNettyServerRequestProto request,
-        CompletableFuture<RaftNettyServerReplyProto> reply) {
+        CompletableFuture<RaftNettyServerReplyProto> reply) throws AlreadyClosedException {
       replies.offer(reply);
       return client.writeAndFlush(request);
     }
@@ -199,7 +199,11 @@ public void close() {
 
   public CompletableFuture<RaftNettyServerReplyProto> sendAsync(RaftNettyServerRequestProto proto) {
     final CompletableFuture<RaftNettyServerReplyProto> reply = new CompletableFuture<>();
-    connection.offer(proto, reply);
+    try {
+      connection.offer(proto, reply);
+    } catch (AlreadyClosedException e) {
+      reply.completeExceptionally(e);
+    }
     return reply;
   }