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 50e714d026..e3d2488c71 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 @@ -76,6 +76,8 @@ public final class NettyClient implements RpcClient { private final Supplier authenticatorSupplier; + private final long requestTimeoutMs; + private volatile boolean isClosed = false; public NettyClient(Configuration conf, ClientMetricGroup clientMetricGroup) { @@ -87,6 +89,7 @@ public NettyClient(Configuration conf, ClientMetricGroup clientMetricGroup) { conf.getInt(ConfigOptions.NETTY_CLIENT_NUM_NETWORK_THREADS), "fluss-netty-client"); int connectTimeoutMs = (int) conf.get(ConfigOptions.CLIENT_CONNECT_TIMEOUT).toMillis(); + this.requestTimeoutMs = conf.get(ConfigOptions.CLIENT_REQUEST_TIMEOUT).toMillis(); int connectionMaxIdle = (int) conf.get(ConfigOptions.NETTY_CONNECTION_MAX_IDLE_TIME).getSeconds(); boolean preferHeap = @@ -191,6 +194,7 @@ private ServerConnection getOrCreateConnection(ServerNode node) { node, clientMetricGroup, authenticatorSupplier.get(), + requestTimeoutMs, (con, ignore) -> connections.remove(serverId, con)); }); } diff --git a/fluss-rpc/src/main/java/org/apache/fluss/rpc/netty/client/ServerConnection.java b/fluss-rpc/src/main/java/org/apache/fluss/rpc/netty/client/ServerConnection.java index 8eb9d805c2..9d8c7be26a 100644 --- a/fluss-rpc/src/main/java/org/apache/fluss/rpc/netty/client/ServerConnection.java +++ b/fluss-rpc/src/main/java/org/apache/fluss/rpc/netty/client/ServerConnection.java @@ -48,6 +48,7 @@ import org.apache.fluss.shaded.netty4.io.netty.channel.ChannelFuture; import org.apache.fluss.shaded.netty4.io.netty.channel.ChannelFutureListener; import org.apache.fluss.utils.ExponentialBackoff; +import org.apache.fluss.utils.concurrent.FutureUtils; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -85,6 +86,7 @@ final class ServerConnection { private final CompletableFuture closeFuture = new CompletableFuture<>(); private final ConnectionMetrics connectionMetrics; private final ClientAuthenticator authenticator; + private final long requestTimeoutMs; private final ExponentialBackoff backoff; private final Object lock = new Object(); @@ -114,11 +116,13 @@ final class ServerConnection { ServerNode node, ClientMetricGroup clientMetricGroup, ClientAuthenticator authenticator, + long requestTimeoutMs, BiConsumer closeCallback) { this.node = node; this.state = ConnectionState.CONNECTING; this.connectionMetrics = clientMetricGroup.createConnectionMetricGroup(node.uid()); this.authenticator = authenticator; + this.requestTimeoutMs = requestTimeoutMs; this.backoff = new ExponentialBackoff(100L, 2, 5000L, 0.2); whenClose(closeCallback); @@ -360,23 +364,41 @@ private CompletableFuture doSend( return responseFuture; } + FutureUtils.orTimeout( + inflight.responseFuture, + requestTimeoutMs, + TimeUnit.MILLISECONDS, + String.format( + "Timed out waiting for response from node %s after %d ms.", + node, requestTimeoutMs)); + inflight.responseFuture.whenComplete( + (ignored, throwable) -> { + if (inflightRequests.remove(inflight.requestId, inflight)) { + connectionMetrics.updateMetricsAfterGetResponse( + apiKey, inflight.requestStartTime, 0); + } + }); + channel.writeAndFlush(byteBuf) .addListener( (ChannelFutureListener) future -> { if (!future.isSuccess()) { - connectionMetrics.updateMetricsAfterGetResponse( - apiKey, inflight.requestStartTime, 0); - Throwable cause = future.cause(); - if (cause instanceof IOException) { - // when server close the channel, the cause will be - // IOException, if the cause is IOException, we wrap - // it as retryable NetworkException to retry to - // connect - cause = new NetworkException(cause); + if (inflightRequests.remove( + inflight.requestId, inflight)) { + connectionMetrics.updateMetricsAfterGetResponse( + apiKey, inflight.requestStartTime, 0); + Throwable cause = future.cause(); + if (cause instanceof IOException) { + // when server close the channel, the cause will + // be IOException, if the cause is IOException, + // we wrap it as retryable NetworkException to + // retry to connect + cause = new NetworkException(cause); + } + inflight.responseFuture.completeExceptionally( + cause); } - inflight.responseFuture.completeExceptionally(cause); - inflightRequests.remove(inflight.requestId); } }); return inflight.responseFuture; 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 329f5d2f28..49890a310f 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 @@ -27,6 +27,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; @@ -43,6 +44,7 @@ import org.junit.jupiter.api.Test; import java.net.ConnectException; +import java.time.Duration; import java.util.ArrayList; import java.util.Arrays; import java.util.Collections; @@ -51,6 +53,9 @@ import java.util.Set; import java.util.concurrent.CompletableFuture; import java.util.concurrent.ExecutionException; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; +import java.util.concurrent.atomic.AtomicInteger; import static org.apache.fluss.utils.NetUtils.getAvailablePort; import static org.assertj.core.api.Assertions.assertThat; @@ -99,6 +104,44 @@ void testSendIncompleteRequest() { .hasRootCauseMessage("Some required fields are missing"); } + @Test + void testClientRequestTimeout() throws Exception { + nettyClient.close(); + nettyServer.close(); + + conf.set(ConfigOptions.CLIENT_REQUEST_TIMEOUT, Duration.ofMillis(100)); + conf.set(ConfigOptions.NETTY_CONNECTION_MAX_IDLE_TIME, Duration.ofSeconds(2)); + nettyClient = new NettyClient(conf, TestingClientMetricGroup.newInstance()); + + AtomicInteger apiVersionsRequests = new AtomicInteger(); + buildNettyServer( + 1, + new TestingGatewayService() { + @Override + public CompletableFuture apiVersions( + ApiVersionsRequest request) { + if (apiVersionsRequests.incrementAndGet() == 1) { + return super.apiVersions(request); + } + return new CompletableFuture<>(); + } + }); + + ApiVersionsRequest request = + new ApiVersionsRequest() + .setClientSoftwareName("testing_client") + .setClientSoftwareVersion("1.0"); + CompletableFuture response = + nettyClient.sendRequest(serverNode, ApiKeys.API_VERSIONS, request); + + assertThatThrownBy(() -> response.get(1, TimeUnit.SECONDS)) + .isInstanceOf(ExecutionException.class) + .hasRootCauseInstanceOf(TimeoutException.class) + .rootCause() + .hasMessageContaining("Timed out waiting for response from node"); + assertThat(nettyClient.connections().get(serverNode.uid()).numInflightRequests()).isZero(); + } + @Test void testSendRequestToWrongServerType() { LookupRequest lookupRequest = new LookupRequest().setTableId(1); @@ -254,11 +297,16 @@ void testExceptionWhenInitializeServerConnection() throws Exception { } private void buildNettyServer(int serverId) throws Exception { + buildNettyServer(serverId, new TestingGatewayService()); + } + + private void buildNettyServer(int serverId, TestingGatewayService gatewayService) + throws Exception { try (NetUtils.Port availablePort = getAvailablePort()) { serverNode = new ServerNode( serverId, "localhost", availablePort.getPort(), ServerType.COORDINATOR); - service = new TestingGatewayService(); + service = gatewayService; MetricGroup metricGroup = NOPMetricsGroup.newInstance(); nettyServer = new NettyServer( diff --git a/fluss-rpc/src/test/java/org/apache/fluss/rpc/netty/client/ServerConnectionTest.java b/fluss-rpc/src/test/java/org/apache/fluss/rpc/netty/client/ServerConnectionTest.java index c5e10f3970..686660552f 100644 --- a/fluss-rpc/src/test/java/org/apache/fluss/rpc/netty/client/ServerConnectionTest.java +++ b/fluss-rpc/src/test/java/org/apache/fluss/rpc/netty/client/ServerConnectionTest.java @@ -20,6 +20,7 @@ import org.apache.fluss.cluster.Endpoint; import org.apache.fluss.cluster.ServerNode; import org.apache.fluss.cluster.ServerType; +import org.apache.fluss.config.ConfigOptions; import org.apache.fluss.config.Configuration; import org.apache.fluss.exception.DisconnectException; import org.apache.fluss.exception.InvalidServerTypeException; @@ -96,6 +97,9 @@ /** Test for {@link ServerConnection}. */ public class ServerConnectionTest { + private static final long REQUEST_TIMEOUT_MS = + ConfigOptions.CLIENT_REQUEST_TIMEOUT.defaultValue().toMillis(); + private EventLoopGroup eventLoopGroup; private Bootstrap bootstrap; private ClientAuthenticator clientAuthenticator; @@ -139,6 +143,7 @@ void testConnectionClose() { serverNode, TestingClientMetricGroup.newInstance(), clientAuthenticator, + REQUEST_TIMEOUT_MS, (con, ignore) -> {}); ConnectionState connectionState = connection.getConnectionState(); assertThat(connectionState).isEqualTo(ConnectionState.CONNECTING); @@ -169,10 +174,20 @@ void testConnectionMetrics() throws ExecutionException, InterruptedException { ClientMetricGroup client = new ClientMetricGroup(metricRegistry, "client"); ServerConnection connection = new ServerConnection( - bootstrap, serverNode, client, clientAuthenticator, (con, ignore) -> {}); + bootstrap, + serverNode, + client, + clientAuthenticator, + REQUEST_TIMEOUT_MS, + (con, ignore) -> {}); ServerConnection connection2 = new ServerConnection( - bootstrap, serverNode2, client, clientAuthenticator, (con, ignore) -> {}); + bootstrap, + serverNode2, + client, + clientAuthenticator, + REQUEST_TIMEOUT_MS, + (con, ignore) -> {}); LookupRequest request = new LookupRequest().setTableId(1); PbLookupReqForBucket pbLookupReqForBucket = request.addBucketsReq(); pbLookupReqForBucket.setBucketId(1); @@ -222,6 +237,7 @@ public ChannelFuture connect(String host, int port) { serverNode, TestingClientMetricGroup.newInstance(), clientAuthenticator, + REQUEST_TIMEOUT_MS, (con, ignore) -> {}); try { OutOfMemoryError error = new OutOfMemoryError("Direct buffer memory"); @@ -272,6 +288,7 @@ public ChannelFuture connect(String host, int port) { wrongServerTypeNode, TestingClientMetricGroup.newInstance(), clientAuthenticator, + REQUEST_TIMEOUT_MS, (con, ignore) -> {}); // Pending request will be rejected with InvalidServerTypeException which is @@ -304,6 +321,7 @@ void testRejectBucketCountChangeForOldServer() throws Exception { serverNode, TestingClientMetricGroup.newInstance(), clientAuthenticator, + REQUEST_TIMEOUT_MS, (con, ignore) -> {}); try { connection.validateVersionCompatibility( @@ -334,6 +352,7 @@ void testRejectHistoricalWritesForOldServer() throws Exception { serverNode, TestingClientMetricGroup.newInstance(), clientAuthenticator, + REQUEST_TIMEOUT_MS, (con, ignore) -> {}); try { assertThat(connection.send(ApiKeys.PUT_KV, putKvRequest(null)).get())