feat: optimize long link test

Author: wardseptember

trpc-proto/trpc-proto-http/src/main/java/com/tencent/trpc/proto/http/client/Http2ConsumerInvoker.java

@@ -15,7 +15,6 @@
 import static com.tencent.trpc.core.common.Constants.DEFAULT_CLIENT_REQUEST_TIMEOUT_MS;
 import static com.tencent.trpc.proto.http.common.HttpConstants.CONNECTION_REQUEST_TIMEOUT;
 
-import autovalue.shaded.com.google.common.common.base.Objects;
 import com.tencent.trpc.core.common.config.BackendConfig;
 import com.tencent.trpc.core.common.config.ConsumerConfig;
 import com.tencent.trpc.core.common.config.ProtocolConfig;
@@ -29,6 +28,7 @@
 import java.nio.charset.StandardCharsets;
 import java.util.HashMap;
 import java.util.Map;
+import java.util.Objects;
 import java.util.concurrent.Future;
 import java.util.concurrent.TimeUnit;
 import org.apache.hc.client5.http.async.methods.SimpleHttpRequest;
@@ -44,6 +44,11 @@
 
 /**
  * HTTP/2 protocol client invoker, supporting both h2 and http2c.
+ *
+ * <p>Each {@link #send(Request)} entry signals the underlying {@link Http2cRpcClient} that
+ * it is being used (drives the idle-eviction heuristic) and reports success / failure to
+ * drive the consecutive-failure counter that flips the client to unavailable on sustained
+ * backend outages.</p>
  */
 public class Http2ConsumerInvoker<T> extends AbstractConsumerInvoker<T> {
 
@@ -60,19 +65,38 @@ public Http2ConsumerInvoker(Http2cRpcClient client, ConsumerConfig<T> config,
      *
      * @param request client request
      * @return Response
-     * @throws Exception if send request failed
+     * @throws Exception declared to honour the abstract contract; the implementation never
+     *         lets exceptions escape — every failure path is wrapped into a Response with
+     *         {@code exception != null}.
      */
     @Override
     public Response send(Request request) throws Exception {
+        Http2cRpcClient http2cRpcClient = (Http2cRpcClient) client;
+        // Mark "used" before any work so even a failed request keeps the idle-eviction timer
+        // accurate (a failing client is still actively used and must not be reaped as orphan).
+        http2cRpcClient.markUsed();
+
         int requestTimeout = config.getBackendConfig().getRequestTimeout();
-        SimpleHttpRequest simpleHttpRequest = buildRequest(request, requestTimeout);
+        SimpleHttpRequest simpleHttpRequest;
+        try {
+            simpleHttpRequest = buildRequest(request, requestTimeout);
+        } catch (Exception ex) {
+            http2cRpcClient.markFailure();
+            return RpcUtils.newResponse(request, null, ex);
+        }
 
         try {
             SimpleHttpResponse simpleHttpResponse = execute(request, requestTimeout,
-                    simpleHttpRequest);
-
-            return handleResponse(request, simpleHttpResponse);
+                    simpleHttpRequest, http2cRpcClient);
+            Response response = handleResponse(request, simpleHttpResponse);
+            if (response.getException() == null) {
+                http2cRpcClient.markSuccess();
+            } else {
+                http2cRpcClient.markFailure();
+            }
+            return response;
         } catch (Exception e) {
+            http2cRpcClient.markFailure();
             return RpcUtils.newResponse(request, null, e);
         }
 
@@ -132,15 +156,12 @@ private Response handleResponse(Request request, SimpleHttpResponse simpleHttpRe
      * @param request TRPC request
      * @param requestTimeout request timeout
      * @param simpleHttpRequest HTTP request
+     * @param http2cRpcClient already-resolved owning client (avoids a redundant cast)
      * @return HTTP response
      * @throws Exception if do HTTP request failed
      */
     private SimpleHttpResponse execute(Request request, int requestTimeout,
-            SimpleHttpRequest simpleHttpRequest) throws Exception {
-        Http2cRpcClient http2cRpcClient = (Http2cRpcClient) client;
-        // Refresh idle counter so the cluster manager's reconnect-check timer keeps treating
-        // this client as healthy while it's actively serving requests.
-        http2cRpcClient.markUsed();
+            SimpleHttpRequest simpleHttpRequest, Http2cRpcClient http2cRpcClient) throws Exception {
         CloseableHttpAsyncClient httpAsyncClient = http2cRpcClient.getHttpAsyncClient();
         Future<SimpleHttpResponse> httpResponseFuture = httpAsyncClient.execute(simpleHttpRequest,
                 new FutureCallback<SimpleHttpResponse>() {
@@ -210,7 +231,7 @@ private SimpleHttpRequest buildRequest(Request request, int requestTimeout) thro
         }
         // Set custom business headers, consistent with the TRPC protocol, only process String and byte[]
         request.getAttachments().forEach((k, v) -> {
-            if (Objects.equal(k, HttpHeaders.TRANSFER_ENCODING) || Objects.equal(k, HttpHeaders.CONTENT_LENGTH)) {
+            if (Objects.equals(k, HttpHeaders.TRANSFER_ENCODING) || Objects.equals(k, HttpHeaders.CONTENT_LENGTH)) {
                 return;
             }
             if (v instanceof String) {

trpc-proto/trpc-proto-http/src/main/java/com/tencent/trpc/proto/http/client/Http2RpcClient.java

@@ -15,6 +15,8 @@
 import static com.tencent.trpc.transport.http.common.Constants.KEYSTORE_PATH;
 
 import com.tencent.trpc.core.common.config.ProtocolConfig;
+import com.tencent.trpc.core.exception.ErrorCode;
+import com.tencent.trpc.core.exception.TRpcException;
 import com.tencent.trpc.core.logger.Logger;
 import com.tencent.trpc.core.logger.LoggerFactory;
 import java.io.File;
@@ -24,28 +26,41 @@
 import org.apache.hc.client5.http.impl.nio.PoolingAsyncClientConnectionManagerBuilder;
 import org.apache.hc.core5.http2.HttpVersionPolicy;
 import org.apache.hc.core5.http2.ssl.ConscryptClientTlsStrategy;
+import org.apache.hc.core5.pool.PoolReusePolicy;
+import org.apache.hc.core5.reactor.IOReactorConfig;
 import org.apache.hc.core5.ssl.SSLContexts;
+import org.apache.hc.core5.util.TimeValue;
 
 /**
- * HTTP 2 protocol client.
+ * HTTP/2 (TLS) protocol client. Inherits long-connection state ({@code lastUsedNanos},
+ * {@code consecutiveFailures}, {@link #markUsed}, {@link #markSuccess}, {@link #markFailure}
+ * and the overridden {@link #isAvailable()}) from {@link Http2cRpcClient}; the differences
+ * are the TLS handshake and the explicit {@link HttpVersionPolicy} negotiation.
+ *
+ * <p>The connection manager is sized and tuned identically to {@link Http2cRpcClient}: pool
+ * limits derived from {@code maxConns}, idle / expired eviction, SO_KEEPALIVE and a hard
+ * connection TTL.</p>
  */
 public class Http2RpcClient extends Http2cRpcClient {
 
-    private static final Logger logger = LoggerFactory.getLogger(HttpRpcClient.class);
+    private static final Logger logger = LoggerFactory.getLogger(Http2RpcClient.class);
+
+    private static final int VALIDATE_AFTER_INACTIVITY_MS = 2000;
+    private static final long EVICT_IDLE_CONNECTIONS_SECONDS = 60L;
+    private static final int CONNECTION_TTL_MINUTES = 10;
 
     /**
      * The protocol type used for interaction with the server, such as HTTP1, H2, or protocol negotiation.
      * In trpc, the interaction is forced to use H2 or HTTP1 protocol based on the configuration.
      */
     protected HttpVersionPolicy clientVersionPolicy;
-
     public Http2RpcClient(ProtocolConfig config) {
         super(config);
         this.clientVersionPolicy = HttpVersionPolicy.FORCE_HTTP_2;
     }
 
     @Override
-    protected void doOpen() {
+    protected void doOpen() throws TRpcException {
         try {
             String keyStorePath = String
                     .valueOf(getProtocolConfig().getExtMap().get(KEYSTORE_PATH));
@@ -61,20 +76,46 @@ protected void doOpen() {
                     .build();
 
             // 2. Configure connection pool.
+            int maxConns = protocolConfig.getMaxConns();
             final PoolingAsyncClientConnectionManager cm = PoolingAsyncClientConnectionManagerBuilder
                     .create().useSystemProperties()
                     .setTlsStrategy(new ConscryptClientTlsStrategy(sslContext))
+                    .setMaxConnTotal(maxConns)
+                    .setMaxConnPerRoute(maxConns)
+                    .setConnPoolPolicy(PoolReusePolicy.LIFO)
+                    .setValidateAfterInactivity(TimeValue.ofMilliseconds(VALIDATE_AFTER_INACTIVITY_MS))
+                    .setConnectionTimeToLive(TimeValue.ofMinutes(CONNECTION_TTL_MINUTES))
                     .build();
 
             // 3. Configure the client to force HTTPS protocol to use HTTP1 communication and H2 protocol
             // to use H2 communication.
             httpAsyncClient = HttpAsyncClients.custom()
-                    .setVersionPolicy(this.clientVersionPolicy).setConnectionManager(cm)
+                    .setVersionPolicy(this.clientVersionPolicy)
+                    .setConnectionManager(cm)
+                    .setIOReactorConfig(IOReactorConfig.custom()
+                            .setSoKeepAlive(true)
+                            .build())
+                    .evictExpiredConnections()
+                    .evictIdleConnections(TimeValue.ofSeconds(EVICT_IDLE_CONNECTIONS_SECONDS))
                     .build();
             // 4. Start the client.
             httpAsyncClient.start();
         } catch (Exception e) {
-            logger.error("httpAsyncClient error: ", e);
+            logger.error("open https/h2 client ({}) failed",
+                    getProtocolConfig().toSimpleString(), e);
+            throw TRpcException.newFrameException(ErrorCode.TRPC_CLIENT_CONNECT_ERR,
+                    "open https/h2 client (" + getProtocolConfig().toSimpleString() + ") failed",
+                    e);
         }
     }
+
+    /**
+     * Defensive close ensuring the inherited handle is released. We let the parent's
+     * {@code doClose} do the actual cleanup; this method exists in case future TLS-only
+     * resources need explicit release.
+     */
+    @Override
+    protected void doClose() {
+        super.doClose();
+    }
 }