[rpc] Avoid retaining failed connections in the connection pool

fluss-rpc/src/main/java/org/apache/fluss/rpc/netty/client/NettyClient.java

@@ -191,12 +191,19 @@ private ServerConnection getOrCreateConnection(ServerNode node) {
                             node,
                             clientMetricGroup,
                             authenticatorSupplier.get(),
-                            (con, ignore) -> connections.remove(serverId, con));
+                            // Avoid recursive map updates when connecting fails synchronously.
+                            (con, ignore) ->
+                                    eventGroup.execute(() -> connections.remove(serverId, con)));
                 });
     }
 
     @VisibleForTesting
     Map<String, ServerConnection> connections() {
         return connections;
     }
+
+    @VisibleForTesting
+    Bootstrap bootstrap() {
+        return bootstrap;
+    }
 }

fluss-rpc/src/test/java/org/apache/fluss/rpc/netty/client/NettyClientTest.java

@@ -28,6 +28,7 @@
 import org.apache.fluss.rpc.TestingGatewayService;
 import org.apache.fluss.rpc.messages.ApiMessage;
 import org.apache.fluss.rpc.messages.ApiVersionsRequest;
+import org.apache.fluss.rpc.messages.ApiVersionsResponse;
 import org.apache.fluss.rpc.messages.GetTableInfoRequest;
 import org.apache.fluss.rpc.messages.LookupRequest;
 import org.apache.fluss.rpc.messages.PbLookupReqForBucket;
@@ -36,14 +37,25 @@
 import org.apache.fluss.rpc.netty.server.NettyServer;
 import org.apache.fluss.rpc.netty.server.RequestsMetrics;
 import org.apache.fluss.rpc.protocol.ApiKeys;
+import org.apache.fluss.shaded.netty4.io.netty.channel.EventLoop;
+import org.apache.fluss.shaded.netty4.io.netty.channel.EventLoopGroup;
+import org.apache.fluss.shaded.netty4.io.netty.resolver.AbstractAddressResolver;
+import org.apache.fluss.shaded.netty4.io.netty.resolver.AddressResolver;
+import org.apache.fluss.shaded.netty4.io.netty.resolver.AddressResolverGroup;
+import org.apache.fluss.shaded.netty4.io.netty.util.concurrent.EventExecutor;
+import org.apache.fluss.shaded.netty4.io.netty.util.concurrent.Promise;
 import org.apache.fluss.utils.NetUtils;
 import org.apache.fluss.utils.concurrent.FutureUtils;
 
 import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
 
 import java.net.ConnectException;
+import java.net.InetSocketAddress;
+import java.net.UnknownHostException;
 import java.time.Duration;
 import java.util.ArrayList;
 import java.util.Arrays;
@@ -73,6 +85,7 @@ public void setup() throws Exception {
         conf = new Configuration();
         // 3 worker threads is enough for this test
         conf.setInt(ConfigOptions.NETTY_SERVER_NUM_WORKER_THREADS, 3);
+        conf.setInt(ConfigOptions.NETTY_CLIENT_NUM_NETWORK_THREADS, 1);
         nettyClient = new NettyClient(conf, TestingClientMetricGroup.newInstance());
         buildNettyServer(1);
     }
@@ -255,7 +268,51 @@ void testExceptionWhenInitializeServerConnection() throws Exception {
                                         .sendRequest(serverNode, ApiKeys.API_VERSIONS, request)
                                         .get())
                 .hasMessageContaining("Disconnected from node");
-        assertThat(nettyClient.connections()).isEmpty();
+        retry(Duration.ofSeconds(10), () -> assertThat(nettyClient.connections()).isEmpty());
+    }
+
+    @ParameterizedTest
+    @ValueSource(ints = {1, 4})
+    void testReconnectAfterImmediateResolutionFailureOnEventLoop(int networkThreads)
+            throws Exception {
+        nettyClient.close();
+        conf.setInt(ConfigOptions.NETTY_CLIENT_NUM_NETWORK_THREADS, networkThreads);
+        nettyClient = new NettyClient(conf, TestingClientMetricGroup.newInstance());
+        InetSocketAddress resolvedAddress =
+                new InetSocketAddress(serverNode.host(), serverNode.port());
+        try (TestingAddressResolverGroup resolver =
+                new TestingAddressResolverGroup(resolvedAddress)) {
+            nettyClient.bootstrap().resolver(resolver);
+            EventLoopGroup eventLoopGroup = nettyClient.bootstrap().config().group();
+            // Preserve the chooser so connect() selects the caller's loop and fails synchronously.
+            EventLoop eventLoop = (EventLoop) eventLoopGroup.iterator().next();
+            ApiVersionsRequest request =
+                    new ApiVersionsRequest()
+                            .setClientSoftwareName("testing_client")
+                            .setClientSoftwareVersion("1.0");
+
+            CompletableFuture<ApiMessage> failedRequest =
+                    eventLoop
+                            .submit(
+                                    () ->
+                                            nettyClient.sendRequest(
+                                                    serverNode, ApiKeys.API_VERSIONS, request))
+                            .get();
+            assertThatThrownBy(failedRequest::get)
+                    .rootCause()
+                    .isInstanceOfAny(UnknownHostException.class, DisconnectException.class);
+            retry(Duration.ofSeconds(10), () -> assertThat(nettyClient.connections()).isEmpty());
+
+            resolver.failResolution = false;
+            CompletableFuture<ApiMessage> recoveredRequest =
+                    eventLoop
+                            .submit(
+                                    () ->
+                                            nettyClient.sendRequest(
+                                                    serverNode, ApiKeys.API_VERSIONS, request))
+                            .get();
+            assertThat(recoveredRequest.get()).isInstanceOf(ApiVersionsResponse.class);
+        }
     }
 
     private void buildNettyServer(int serverId) throws Exception {
@@ -276,4 +333,41 @@ private void buildNettyServer(int serverId) throws Exception {
             nettyServer.start();
         }
     }
+
+    private static final class TestingAddressResolverGroup
+            extends AddressResolverGroup<InetSocketAddress> {
+
+        private final InetSocketAddress resolvedAddress;
+        private volatile boolean failResolution = true;
+
+        private TestingAddressResolverGroup(InetSocketAddress resolvedAddress) {
+            this.resolvedAddress = resolvedAddress;
+        }
+
+        @Override
+        protected AddressResolver<InetSocketAddress> newResolver(EventExecutor executor) {
+            return new AbstractAddressResolver<InetSocketAddress>(executor) {
+                @Override
+                protected boolean doIsResolved(InetSocketAddress address) {
+                    return false;
+                }
+
+                @Override
+                protected void doResolve(
+                        InetSocketAddress address, Promise<InetSocketAddress> promise) {
+                    if (failResolution) {
+                        promise.setFailure(new UnknownHostException(address.getHostString()));
+                    } else {
+                        promise.setSuccess(resolvedAddress);
+                    }
+                }
+
+                @Override
+                protected void doResolveAll(
+                        InetSocketAddress address, Promise<List<InetSocketAddress>> promise) {
+                    throw new UnsupportedOperationException();
+                }
+            };
+        }
+    }
 }