diff --git a/fluss-rpc/src/main/java/org/apache/fluss/rpc/netty/client/NettyClient.java b/fluss-rpc/src/main/java/org/apache/fluss/rpc/netty/client/NettyClient.java index 50e714d0269..a244d5b4e9c 100644 --- a/fluss-rpc/src/main/java/org/apache/fluss/rpc/netty/client/NettyClient.java +++ b/fluss-rpc/src/main/java/org/apache/fluss/rpc/netty/client/NettyClient.java @@ -191,7 +191,9 @@ 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))); }); } @@ -199,4 +201,9 @@ private ServerConnection getOrCreateConnection(ServerNode node) { Map connections() { return connections; } + + @VisibleForTesting + Bootstrap bootstrap() { + return bootstrap; + } } diff --git a/fluss-rpc/src/test/java/org/apache/fluss/rpc/netty/client/NettyClientTest.java b/fluss-rpc/src/test/java/org/apache/fluss/rpc/netty/client/NettyClientTest.java index acacdbb45e1..bb6351de9e9 100644 --- a/fluss-rpc/src/test/java/org/apache/fluss/rpc/netty/client/NettyClientTest.java +++ b/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 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 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 { + + private final InetSocketAddress resolvedAddress; + private volatile boolean failResolution = true; + + private TestingAddressResolverGroup(InetSocketAddress resolvedAddress) { + this.resolvedAddress = resolvedAddress; + } + + @Override + protected AddressResolver newResolver(EventExecutor executor) { + return new AbstractAddressResolver(executor) { + @Override + protected boolean doIsResolved(InetSocketAddress address) { + return false; + } + + @Override + protected void doResolve( + InetSocketAddress address, Promise promise) { + if (failResolution) { + promise.setFailure(new UnknownHostException(address.getHostString())); + } else { + promise.setSuccess(resolvedAddress); + } + } + + @Override + protected void doResolveAll( + InetSocketAddress address, Promise> promise) { + throw new UnsupportedOperationException(); + } + }; + } + } }