Double-check before closing

trpc-core/src/main/java/com/tencent/trpc/core/cluster/RpcClusterClientManager.java

@@ -128,19 +128,57 @@ 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<String, RpcClientProxy> 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);
             }
         }));
     }
 
     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;
     }
 
     /**
@@ -306,6 +344,11 @@ public ProtocolConfig getProtocolConfig() {
             return delegate.getProtocolConfig();
         }
 
+        @Override
+        public int getPendingRequestCount() {
+            return delegate.getPendingRequestCount();
+        }
+
         @Override
         public int hashCode() {
             return Objects.hash(delegate);
@@ -327,4 +370,4 @@ public boolean equals(Object obj) {
         }
     }
 
-}
+}

trpc-core/src/main/java/com/tencent/trpc/core/cluster/def/DefClusterInvoker.java

@@ -72,6 +72,9 @@ protected CompletionStage<Response> doInvoke(Request request, CompletionStage<Se
     protected ConsumerInvokerProxy<T> getInvoker(ServiceInstance instance) {
         String key = toUniqKey(instance);
         ConsumerInvokerProxy<T> 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;
         }

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.
+     *
+     * <p>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}.</p>
+     *
+     * <p>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.</p>
+     *
+     * @return the number of in-flight requests, 0 if unknown
+     */
+    default int getPendingRequestCount() {
+        return 0;
+    }
+
 }

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.
+     *
+     * <p>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}.</p>
+     *
+     * @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.

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}
      *