From 21b5cc25d0bea08aeafff10061fb7bd517f358a8 Mon Sep 17 00:00:00 2001 From: eric-zc1 Date: Thu, 18 Dec 2025 10:54:53 +0800 Subject: [PATCH 1/2] Double-check before closing --- .../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 8fbdbf2f4..3e3ad865a 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 @@ -123,11 +123,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); } })); } @@ -322,4 +343,4 @@ public boolean equals(Object obj) { } } -} \ No newline at end of file +} From 67c3ea2ddc2840431e2a004784c411c16a14695b Mon Sep 17 00:00:00 2001 From: eric-zc1 Date: Fri, 18 Sep 2026 15:31:53 +0800 Subject: [PATCH 2/2] =?UTF-8?q?fix(cluster):=20=E8=B7=B3=E8=BF=87=E6=B8=85?= =?UTF-8?q?=E7=90=86=E4=BB=8D=E6=9C=89=E5=9C=A8=E9=80=94=E8=AF=B7=E6=B1=82?= =?UTF-8?q?=E7=9A=84=E7=A9=BA=E9=97=B2=E5=AE=A2=E6=88=B7=E7=AB=AF?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../core/cluster/RpcClusterClientManager.java | 24 ++++++++++++++++++- .../core/cluster/def/DefClusterInvoker.java | 3 +++ .../com/tencent/trpc/core/rpc/RpcClient.java | 16 +++++++++++++ .../support/DefResponseFutureManager.java | 13 ++++++++++ .../trpc/proto/support/DefRpcClient.java | 8 +++++++ 5 files changed, 63 insertions(+), 1 deletion(-) 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-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} *