From 8a878a00758ec4b72cbf5ea3be6e079ff2601307 Mon Sep 17 00:00:00 2001 From: eric-zc1 Date: Fri, 18 Sep 2026 15:48:25 +0800 Subject: [PATCH 1/2] fix: double-check idle state before closing a RpcClient --- .../core/cluster/RpcClusterClientManager.java | 25 +++++++++++++++++-- 1 file changed, 23 insertions(+), 2 deletions(-) diff --git a/trpc-core/src/main/java/com/tencent/trpc/core/cluster/RpcClusterClientManager.java b/trpc-core/src/main/java/com/tencent/trpc/core/cluster/RpcClusterClientManager.java index 320b8b0f5..9027a784f 100644 --- a/trpc-core/src/main/java/com/tencent/trpc/core/cluster/RpcClusterClientManager.java +++ b/trpc-core/src/main/java/com/tencent/trpc/core/cluster/RpcClusterClientManager.java @@ -128,11 +128,32 @@ public static void scanUnusedClient() { }); unusedClientMap.forEach((bConfig, value) -> value.forEach(e -> { try { + RpcClientProxy proxy = (RpcClientProxy) e; + // Double-check before closing: ensure the client is still idle to avoid closing a client in use. + // This prevents race condition where getOrCreateClient() gets a client right before it's closed. + if (!isIdleTimeout(bConfig, proxy)) { + // lastUsedNanos was updated, meaning a business thread is using it, try to put it back + Map clientMap = CLUSTER_MAP.get(bConfig); + if (clientMap != null) { + RpcClientProxy existing = clientMap.putIfAbsent( + proxy.getProtocolConfig().toUniqId(), proxy); + if (existing == null) { + // Successfully put back, do not close + logger.info("RpcClient {} rescued from closing due to recent usage", + proxy.getProtocolConfig().toSimpleString()); + return; + } + } + // Failed to put back (a new client already exists), still need to close the old one + } e.close(); - } finally { logger.warn("RpcClient in clusterName={}, naming={}, remove rpc client{}, due to unused time > {} ms", bConfig.getName(), bConfig.getNamingOptions().getServiceNaming(), e.getProtocolConfig().toSimpleString(), bConfig.getIdleTimeout()); + } catch (Exception ex) { + logger.error("Failed to close RpcClient in clusterName={}, naming={}, client={}", + bConfig.getName(), bConfig.getNamingOptions().getServiceNaming(), + e.getProtocolConfig().toSimpleString(), ex); } })); } @@ -327,4 +348,4 @@ public boolean equals(Object obj) { } } -} \ No newline at end of file +} From aac715b11882fd58af8ed977739aef30b0eb41d4 Mon Sep 17 00:00:00 2001 From: eric-zc1 Date: Fri, 18 Sep 2026 15:52:58 +0800 Subject: [PATCH 2/2] fix: skip cleaning an idle RpcClient which still has in-flight requests RpcClusterClientManager#scanUnusedClient removes a client from CLUSTER_MAP and then closes it, while DefClusterInvoker#getInvoker only checks whether the invoker is already cached. A request issued in that window is therefore sent through a client which is being torn down, and DefResponseFutureManager #closeClient fails all its in-flight requests with "Client(...) stop" (error code 999). That code is not in the circuit breaker list and the failure is even reported as a success, so the exception is simply propagated to the caller. - RpcClient#getPendingRequestCount(): new method, with a default implementation returning 0 to keep binary compatibility for existing implementations. It is implemented by DefRpcClient on top of DefResponseFutureManager#getPendingCount() (i.e. the size of the in-flight future map) and delegated by RpcClientProxy; - RpcClusterClientManager#isIdleTimeout: although the client is idle for too long, skip cleaning it when there are still requests in flight, so that it is re-checked in the next round instead of being closed; - DefClusterInvoker#getInvoker: document why the availability check is required on the fast path. --- .../core/cluster/RpcClusterClientManager.java | 24 +- .../core/cluster/def/DefClusterInvoker.java | 3 + .../com/tencent/trpc/core/rpc/RpcClient.java | 16 ++ .../cluster/RpcClusterClientManagerTest.java | 211 +++++++++++++++--- .../support/DefResponseFutureManager.java | 13 ++ .../trpc/proto/support/DefRpcClient.java | 8 + .../proto/support/DefResponseFutureTest.java | 36 +++ 7 files changed, 284 insertions(+), 27 deletions(-) diff --git a/trpc-core/src/main/java/com/tencent/trpc/core/cluster/RpcClusterClientManager.java b/trpc-core/src/main/java/com/tencent/trpc/core/cluster/RpcClusterClientManager.java index 9027a784f..f6f3453b2 100644 --- a/trpc-core/src/main/java/com/tencent/trpc/core/cluster/RpcClusterClientManager.java +++ b/trpc-core/src/main/java/com/tencent/trpc/core/cluster/RpcClusterClientManager.java @@ -161,7 +161,24 @@ public static void scanUnusedClient() { private static boolean isIdleTimeout(BackendConfig bConfig, RpcClientProxy clientProxy) { long unusedNanosLimit = TimeUnit.MILLISECONDS.toNanos(bConfig.getIdleTimeout()); long lastUsedNanos = clientProxy.getLastUsedNanos(); - return lastUsedNanos > 0 && unusedNanosLimit > 0 && (System.nanoTime() - lastUsedNanos) > unusedNanosLimit; + boolean idleTooLong = lastUsedNanos > 0 && unusedNanosLimit > 0 + && (System.nanoTime() - lastUsedNanos) > unusedNanosLimit; + if (!idleTooLong) { + return false; + } + // The client is idle for a long time, but there are still requests in flight on it. + // Skip cleaning in this round and re-check in the next one, otherwise closeClient() would + // forcibly fail all those in-flight requests with "Client(...) stop". + int pending = clientProxy.getPendingRequestCount(); + if (pending > 0) { + logger.warn("RpcClient in clusterName={}, naming={}, client={} idle > {} ms, " + + "but {} request(s) still in flight, skip cleaning this round", + bConfig.getName(), bConfig.getNamingOptions().getServiceNaming(), + clientProxy.getProtocolConfig().toSimpleString(), + bConfig.getIdleTimeout(), pending); + return false; + } + return true; } /** @@ -327,6 +344,11 @@ public ProtocolConfig getProtocolConfig() { return delegate.getProtocolConfig(); } + @Override + public int getPendingRequestCount() { + return delegate.getPendingRequestCount(); + } + @Override public int hashCode() { return Objects.hash(delegate); diff --git a/trpc-core/src/main/java/com/tencent/trpc/core/cluster/def/DefClusterInvoker.java b/trpc-core/src/main/java/com/tencent/trpc/core/cluster/def/DefClusterInvoker.java index bf3155f18..8fc2ca9bb 100644 --- a/trpc-core/src/main/java/com/tencent/trpc/core/cluster/def/DefClusterInvoker.java +++ b/trpc-core/src/main/java/com/tencent/trpc/core/cluster/def/DefClusterInvoker.java @@ -72,6 +72,9 @@ protected CompletionStage doInvoke(Request request, CompletionStage getInvoker(ServiceInstance instance) { String key = toUniqKey(instance); ConsumerInvokerProxy result = invokerCache.get(key); + // Keep consistent with createInvoker: the invoker must be rebuilt once its client is closed + // (or is being closed), otherwise the request would be sent to a client which is being torn + // down by RpcClusterClientManager#scanUnusedClient. if (result != null && result.isAvailable()) { return result; } diff --git a/trpc-core/src/main/java/com/tencent/trpc/core/rpc/RpcClient.java b/trpc-core/src/main/java/com/tencent/trpc/core/rpc/RpcClient.java index 04f3b6edf..c37189bcf 100644 --- a/trpc-core/src/main/java/com/tencent/trpc/core/rpc/RpcClient.java +++ b/trpc-core/src/main/java/com/tencent/trpc/core/rpc/RpcClient.java @@ -61,4 +61,20 @@ public interface RpcClient { */ ProtocolConfig getProtocolConfig(); + /** + * Get the number of requests which are still in flight on this client. + * + *

It is used by the idle client cleaner to skip a client which still has in-flight requests, + * otherwise closing the client would forcibly fail those requests with {@code Client(...) stop}.

+ * + *

A {@code default} implementation is provided to keep binary compatibility with the existing + * third-party implementations, which simply reports "no in-flight request" and thus keeps the old + * cleaning behavior.

+ * + * @return the number of in-flight requests, 0 if unknown + */ + default int getPendingRequestCount() { + return 0; + } + } diff --git a/trpc-core/src/test/java/com/tencent/trpc/core/cluster/RpcClusterClientManagerTest.java b/trpc-core/src/test/java/com/tencent/trpc/core/cluster/RpcClusterClientManagerTest.java index 7ee1943da..f2c025a72 100644 --- a/trpc-core/src/test/java/com/tencent/trpc/core/cluster/RpcClusterClientManagerTest.java +++ b/trpc-core/src/test/java/com/tencent/trpc/core/cluster/RpcClusterClientManagerTest.java @@ -21,6 +21,7 @@ import com.tencent.trpc.core.rpc.ConsumerInvoker; import com.tencent.trpc.core.rpc.RpcClient; import java.lang.reflect.Field; +import java.lang.reflect.Method; import java.util.Map; import org.junit.Assert; import org.junit.Test; @@ -98,45 +99,203 @@ public void testScanWithEmptyCluster() { RpcClusterClientManager.scanUnusedClient(); } + /** + * An idle client which still has in-flight requests must be kept, otherwise those requests would be + * forcibly failed with "Client(...) stop" by {@code DefResponseFutureManager#closeClient}. + * Once the requests are done, the client is expected to be cleaned in the next round. + */ + @Test + public void testScanSkipsClientWithInFlightRequest() throws Exception { + BackendConfig backendConfig = new BackendConfig(); + backendConfig.setIdleTimeout(1); + backendConfig.setNamingUrl("ip://127.0.0.1:8085"); + InFlightProtocolConfigTest config = new InFlightProtocolConfigTest("127.0.0.1", 8085); + RpcClusterClientManager.getOrCreateClient(backendConfig, config); + Map innerMap = innerClientsOf(backendConfig); + assertEquals(1, innerMap.size()); + + Thread.sleep(10); + config.setPendingRequestCount(1); + RpcClusterClientManager.scanUnusedClient(); + assertEquals("idle client with in-flight request should not be cleaned", 1, innerMap.size()); + + config.setPendingRequestCount(0); + RpcClusterClientManager.scanUnusedClient(); + assertEquals("idle client without in-flight request should be cleaned", 0, innerMap.size()); + RpcClusterClientManager.shutdownBackendConfig(backendConfig); + } + + /** + * A client may be used by a business thread after it has been picked as unused but before it is really + * closed. In that case it must be put back instead of being closed. + */ + @Test + public void testScanRescuesClientUsedAgainBeforeClosing() throws Exception { + BackendConfig backendConfig = new BackendConfig(); + backendConfig.setIdleTimeout(1); + backendConfig.setNamingUrl("ip://127.0.0.1:8086"); + + ProtocolConfigTest configA = new ProtocolConfigTest("127.0.0.1", 8087); + ProtocolConfigTest configB = new ProtocolConfigTest("127.0.0.1", 8088); + RpcClusterClientManager.getOrCreateClient(backendConfig, configA); + RpcClusterClientManager.getOrCreateClient(backendConfig, configB); + + Map innerMap = innerClientsOf(backendConfig); + assertEquals(2, innerMap.size()); + Object clientA = innerMap.get(configA.toUniqId()); + Object clientB = innerMap.get(configB.toUniqId()); + + // Whichever client is closed first refreshes the other one, which simulates a business thread + // using it during the cleaning process. The refreshed one must be rescued. + configA.setOnClose(() -> updateLastUsedNanos(clientB)); + configB.setOnClose(() -> updateLastUsedNanos(clientA)); + + Thread.sleep(10); + RpcClusterClientManager.scanUnusedClient(); + assertEquals("the client used again before closing should be rescued", 1, innerMap.size()); + RpcClusterClientManager.shutdownBackendConfig(backendConfig); + } + + /** + * A failure while closing one client must not break the cleaning of the others. + */ + @Test + public void testScanContinuesWhenClosingFails() throws Exception { + BackendConfig backendConfig = new BackendConfig(); + backendConfig.setIdleTimeout(1); + backendConfig.setNamingUrl("ip://127.0.0.1:8089"); + ProtocolConfigTest config = new ProtocolConfigTest("127.0.0.1", 8090); + config.setOnClose(() -> { + throw new IllegalStateException("close failed"); + }); + RpcClusterClientManager.getOrCreateClient(backendConfig, config); + Map innerMap = innerClientsOf(backendConfig); + assertEquals(1, innerMap.size()); + + Thread.sleep(10); + RpcClusterClientManager.scanUnusedClient(); + assertEquals("client should be removed even if closing it throws", 0, innerMap.size()); + RpcClusterClientManager.shutdownBackendConfig(backendConfig); + } + + @SuppressWarnings("unchecked") + private static Map innerClientsOf(BackendConfig backendConfig) throws Exception { + Field field = RpcClusterClientManager.class.getDeclaredField("CLUSTER_MAP"); + field.setAccessible(true); + Map> clusterMap = + (Map>) field.get(null); + Map innerMap = clusterMap.get(backendConfig); + Assert.assertNotNull(innerMap); + return innerMap; + } + + private static void updateLastUsedNanos(Object clientProxy) { + try { + Method method = clientProxy.getClass().getMethod("updateLastUsedNanos"); + method.setAccessible(true); + method.invoke(clientProxy); + } catch (Exception ex) { + throw new IllegalStateException("failed to update lastUsedNanos", ex); + } + } + private static class ProtocolConfigTest extends ProtocolConfig { + private Runnable onClose = () -> { + }; + + ProtocolConfigTest() { + } + + ProtocolConfigTest(String ip, int port) { + setIp(ip); + setPort(port); + setNetwork("tcp"); + } + + void setOnClose(Runnable onClose) { + this.onClose = onClose; + } + @Override public RpcClient createClient() { - return new RpcClient() { + return new TestRpcClient(this, () -> onClose.run()); + } + } - @Override - public void open() throws TRpcException { - } + private static class InFlightProtocolConfigTest extends ProtocolConfig { - @Override - public boolean isClosed() { - return false; - } + private volatile int pendingRequestCount; - @Override - public boolean isAvailable() { - return true; - } + InFlightProtocolConfigTest(String ip, int port) { + setIp(ip); + setPort(port); + setNetwork("tcp"); + } - @Override - public ProtocolConfig getProtocolConfig() { - return ProtocolConfigTest.this; - } + void setPendingRequestCount(int pendingRequestCount) { + this.pendingRequestCount = pendingRequestCount; + } + @Override + public RpcClient createClient() { + return new TestRpcClient(this, () -> { + }) { @Override - public void close() { + public int getPendingRequestCount() { + return pendingRequestCount; } + }; + } + } - @Override - public ConsumerInvoker createInvoker(ConsumerConfig consumerConfig) { - return null; - } + /** + * Test client which does not override {@link RpcClient#getPendingRequestCount()}, so the default + * implementation (no in-flight request) is exercised as well. + */ + private static class TestRpcClient implements RpcClient { - @Override - public CloseFuture closeFuture() { - return new CloseFuture(); - } - }; + private final ProtocolConfig protocolConfig; + + private final Runnable onClose; + + TestRpcClient(ProtocolConfig protocolConfig, Runnable onClose) { + this.protocolConfig = protocolConfig; + this.onClose = onClose; + } + + @Override + public void open() throws TRpcException { + } + + @Override + public boolean isClosed() { + return false; + } + + @Override + public boolean isAvailable() { + return true; + } + + @Override + public ProtocolConfig getProtocolConfig() { + return protocolConfig; + } + + @Override + public void close() { + onClose.run(); + } + + @Override + public ConsumerInvoker createInvoker(ConsumerConfig consumerConfig) { + return null; + } + + @Override + public CloseFuture closeFuture() { + return new CloseFuture(); } } } diff --git a/trpc-proto/trpc-rpc-support/src/main/java/com/tencent/trpc/proto/support/DefResponseFutureManager.java b/trpc-proto/trpc-rpc-support/src/main/java/com/tencent/trpc/proto/support/DefResponseFutureManager.java index df1af9260..815e7be2e 100644 --- a/trpc-proto/trpc-rpc-support/src/main/java/com/tencent/trpc/proto/support/DefResponseFutureManager.java +++ b/trpc-proto/trpc-rpc-support/src/main/java/com/tencent/trpc/proto/support/DefResponseFutureManager.java @@ -79,6 +79,19 @@ public DefResponseFuture newFuture(RpcClientContext context, ConsumerInvoker return future; } + /** + * Get the number of requests which are still in flight, i.e. the {@link DefResponseFuture}s that have + * not been completed (by a response or by the timeout watcher) yet. + * + *

It is used by the idle client cleaner to avoid closing a client which still has in-flight + * requests, otherwise those requests would be forcibly failed with {@code Client(...) stop}.

+ * + * @return the number of in-flight requests + */ + public int getPendingCount() { + return futureMap.size(); + } + /** * Removes and force stops all {@link DefResponseFuture}s related to a tRPC client. * Should be called when a tRPC client closes. diff --git a/trpc-proto/trpc-rpc-support/src/main/java/com/tencent/trpc/proto/support/DefRpcClient.java b/trpc-proto/trpc-rpc-support/src/main/java/com/tencent/trpc/proto/support/DefRpcClient.java index 0a488da22..f483ba29f 100644 --- a/trpc-proto/trpc-rpc-support/src/main/java/com/tencent/trpc/proto/support/DefRpcClient.java +++ b/trpc-proto/trpc-rpc-support/src/main/java/com/tencent/trpc/proto/support/DefRpcClient.java @@ -85,6 +85,14 @@ public boolean isAvailable() { return super.isAvailable() && (transport != null && transport.isConnected()); } + /** + * {@inheritDoc} + */ + @Override + public int getPendingRequestCount() { + return futureManager.getPendingCount(); + } + /** * {@inheritDoc} * diff --git a/trpc-proto/trpc-rpc-support/src/test/java/com/tencent/trpc/proto/support/DefResponseFutureTest.java b/trpc-proto/trpc-rpc-support/src/test/java/com/tencent/trpc/proto/support/DefResponseFutureTest.java index 14ff7affb..5567b0ddb 100644 --- a/trpc-proto/trpc-rpc-support/src/test/java/com/tencent/trpc/proto/support/DefResponseFutureTest.java +++ b/trpc-proto/trpc-rpc-support/src/test/java/com/tencent/trpc/proto/support/DefResponseFutureTest.java @@ -110,6 +110,42 @@ public void test() throws Exception { shutdownListener.onShutdown(); } + @Test + public void testPendingRequestCount() throws Exception { + // The timeout manager is a static field shared by all DefResponseFutureManager instances, and it is + // closed by DefResponseFutureTest#test(). Reset it so that this test is order independent. + DefResponseFutureManager.reset(); + + ProtocolConfig config = ProtocolConfig.newInstance(); + config.setIp("127.0.0.1"); + config.setPort(8889); + DefRpcClient rpcClient = new DefRpcClient(config, new TestClientCodec()); + ConsumerInvoker invoker = new DefConsumerInvoker(rpcClient, new ConsumerConfig<>()); + + // No request in flight yet + assertEquals(0, rpcClient.getPendingRequestCount()); + assertEquals(0, rpcClient.getFutureManager().getPendingCount()); + + ClientTransport client = new NettyClientTransportFactory().create(config, + new ChannelHandlerAdapter() { + }, new TestClientCodec()); + DefRequest request = new DefRequest(); + request.setRequestId(2000); + request.getMeta().setTimeout(1000); + RpcClientContext context = new RpcClientContext(); + rpcClient.getFutureManager().newFuture(context, invoker, client, request); + + // One request in flight: the idle client cleaner must be able to see it + assertEquals(1, rpcClient.getFutureManager().getPendingCount()); + assertEquals(1, rpcClient.getPendingRequestCount()); + + rpcClient.getFutureManager().remove(request.getRequestId()); + assertEquals(0, rpcClient.getPendingRequestCount()); + + rpcClient.close(); + client.close(); + } + private class TestClientCodec extends ClientCodec { @Override