Prevent conn leak on thread interrupt (#6835)

* Prevent conn leak on thread interrupt

If a thread blocked waiting on a connection pool is interrupted, it's
possible for the connection to become leaked if the pool then completes
the future from its side.

This commit prevents that from happening by intercepting the interrupt
and attempting to cancel the lease future. If that is unsuccessful, the
connection is returned to the upper, and the thread is reinterrupted.

Adresses #6786

* Review comments

* Checkstyle fix

Author: dagnir

.changes/next-release/bugfix-Apache5HTTPClient-9de458e.json

@@ -0,0 +1,6 @@
+{
+    "type": "bugfix",
+    "category": "Apache 5 HTTP Client",
+    "contributor": "",
+    "description": "Fixed a connection leak that could occur when the thread waiting to acquire a connection from the pool is interrupted. Fixes [#6786](https://github.com/aws/aws-sdk-java-v2/issues/6786)."
+}

http-clients/apache5-client/src/main/java/software/amazon/awssdk/http/apache5/Apache5HttpClient.java

@@ -50,7 +50,6 @@
 import org.apache.hc.client5.http.impl.classic.HttpClientBuilder;
 import org.apache.hc.client5.http.impl.classic.HttpClients;
 import org.apache.hc.client5.http.impl.io.PoolingHttpClientConnectionManager;
-import org.apache.hc.client5.http.impl.io.PoolingHttpClientConnectionManagerBuilder;
 import org.apache.hc.client5.http.impl.routing.DefaultRoutePlanner;
 import org.apache.hc.client5.http.io.HttpClientConnectionManager;
 import org.apache.hc.client5.http.protocol.HttpClientContext;
@@ -92,6 +91,7 @@
 import software.amazon.awssdk.http.apache5.internal.SdkProxyRoutePlanner;
 import software.amazon.awssdk.http.apache5.internal.conn.ClientConnectionManagerFactory;
 import software.amazon.awssdk.http.apache5.internal.conn.IdleConnectionReaper;
+import software.amazon.awssdk.http.apache5.internal.conn.SafePoolingHttpClientConnectionManagerBuilder;
 import software.amazon.awssdk.http.apache5.internal.conn.SdkConnectionKeepAliveStrategy;
 import software.amazon.awssdk.http.apache5.internal.conn.SdkTlsSocketFactory;
 import software.amazon.awssdk.http.apache5.internal.impl.Apache5HttpRequestFactory;
@@ -755,8 +755,8 @@ public PoolingHttpClientConnectionManager create(Apache5HttpClient.DefaultBuilde
 
             TlsSocketStrategy tlsStrategy = getPreferredTlsStrategy(configuration, standardOptions);
 
-            PoolingHttpClientConnectionManagerBuilder builder =
-                PoolingHttpClientConnectionManagerBuilder.create()
+            SafePoolingHttpClientConnectionManagerBuilder builder =
+                SafePoolingHttpClientConnectionManagerBuilder.create()
                                                          .setTlsSocketStrategy(tlsStrategy)
                                                          .setSchemePortResolver(DefaultSchemePortResolver.INSTANCE)
                                                          .setDnsResolver(configuration.dnsResolver);

http-clients/apache5-client/src/main/java/software/amazon/awssdk/http/apache5/internal/conn/SafePoolingHttpClientConnectionManager.java

@@ -0,0 +1,44 @@
+/*
+ * Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License").
+ * You may not use this file except in compliance with the License.
+ * A copy of the License is located at
+ *
+ *  http://aws.amazon.com/apache2.0
+ *
+ * or in the "license" file accompanying this file. This file is distributed
+ * on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either
+ * express or implied. See the License for the specific language governing
+ * permissions and limitations under the License.
+ */
+
+package software.amazon.awssdk.http.apache5.internal.conn;
+
+import org.apache.hc.client5.http.impl.io.ManagedHttpClientConnectionFactory;
+import org.apache.hc.client5.http.impl.io.PoolingHttpClientConnectionManager;
+import org.apache.hc.client5.http.io.HttpClientConnectionOperator;
+import org.apache.hc.core5.pool.DefaultDisposalCallback;
+import org.apache.hc.core5.pool.PoolReusePolicy;
+import software.amazon.awssdk.annotations.SdkInternalApi;
+
+/**
+ * Specialization of {@link PoolingHttpClientConnectionManager} to enable use of {@link SafeStrictConnPool} to prevent leaking
+ * connections when the thread waiting on the future is interrupted.
+ */
+@SdkInternalApi
+public final class SafePoolingHttpClientConnectionManager extends PoolingHttpClientConnectionManager {
+    public SafePoolingHttpClientConnectionManager(HttpClientConnectionOperator connectionOperator) {
+        super(connectionOperator,
+            new SafeStrictConnPool(
+                DEFAULT_MAX_CONNECTIONS_PER_ROUTE,
+                DEFAULT_MAX_TOTAL_CONNECTIONS,
+                null,
+                PoolReusePolicy.LIFO,
+                new DefaultDisposalCallback<>(),
+                null
+            ),
+            ManagedHttpClientConnectionFactory.INSTANCE
+        );
+    }
+}

http-clients/apache5-client/src/main/java/software/amazon/awssdk/http/apache5/internal/conn/SafePoolingHttpClientConnectionManagerBuilder.java

@@ -0,0 +1,235 @@
+/*
+ * Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License").
+ * You may not use this file except in compliance with the License.
+ * A copy of the License is located at
+ *
+ *  http://aws.amazon.com/apache2.0
+ *
+ * or in the "license" file accompanying this file. This file is distributed
+ * on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either
+ * express or implied. See the License for the specific language governing
+ * permissions and limitations under the License.
+ */
+
+/*
+ * ====================================================================
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ * ====================================================================
+ *
+ * This software consists of voluntary contributions made by many
+ * individuals on behalf of the Apache Software Foundation.  For more
+ * information on the Apache Software Foundation, please see
+ * <http://www.apache.org/>.
+ *
+ */
+
+package software.amazon.awssdk.http.apache5.internal.conn;
+
+import org.apache.hc.client5.http.DnsResolver;
+import org.apache.hc.client5.http.HttpRoute;
+import org.apache.hc.client5.http.SchemePortResolver;
+import org.apache.hc.client5.http.config.ConnectionConfig;
+import org.apache.hc.client5.http.config.TlsConfig;
+import org.apache.hc.client5.http.impl.io.DefaultHttpClientConnectionOperator;
+import org.apache.hc.client5.http.impl.io.PoolingHttpClientConnectionManagerBuilder;
+import org.apache.hc.client5.http.io.HttpClientConnectionOperator;
+import org.apache.hc.client5.http.ssl.DefaultClientTlsStrategy;
+import org.apache.hc.client5.http.ssl.TlsSocketStrategy;
+import org.apache.hc.core5.function.Resolver;
+import org.apache.hc.core5.http.HttpHost;
+import org.apache.hc.core5.http.URIScheme;
+import org.apache.hc.core5.http.config.RegistryBuilder;
+import org.apache.hc.core5.http.io.SocketConfig;
+import software.amazon.awssdk.annotations.SdkInternalApi;
+
+/**
+ * This is a fork of {@link PoolingHttpClientConnectionManagerBuilder} from Apache 5. The purpose of this forked class is to
+ * enable usage of the {@link SafePoolingHttpClientConnectionManager} to enable the workaround for
+ * https://github.com/aws/aws-sdk-java-v2/issues/6786.
+ */
+// This a direct copy of PoolingHttpClientConnectionManagerBuilder with minor changes to remove methods we don't use and
+// updates to follow our style guide.
+@SdkInternalApi
+public final class SafePoolingHttpClientConnectionManagerBuilder {
+
+    private TlsSocketStrategy tlsSocketStrategy;
+    private SchemePortResolver schemePortResolver;
+    private DnsResolver dnsResolver;
+    private Resolver<HttpRoute, SocketConfig> socketConfigResolver;
+    private Resolver<HttpRoute, ConnectionConfig> connectionConfigResolver;
+    private Resolver<HttpHost, TlsConfig> tlsConfigResolver;
+
+    private int maxConnTotal;
+    private int maxConnPerRoute;
+
+    public static SafePoolingHttpClientConnectionManagerBuilder create() {
+        return new SafePoolingHttpClientConnectionManagerBuilder();
+    }
+
+    /**
+     * Sets {@link TlsSocketStrategy} instance.
+     *
+     * @return this instance.
+     */
+    public SafePoolingHttpClientConnectionManagerBuilder setTlsSocketStrategy(TlsSocketStrategy tlsSocketStrategy) {
+        this.tlsSocketStrategy = tlsSocketStrategy;
+        return this;
+    }
+
+    /**
+     * Sets {@link DnsResolver} instance.
+     *
+     * @return this instance.
+     */
+    public SafePoolingHttpClientConnectionManagerBuilder setDnsResolver(DnsResolver dnsResolver) {
+        this.dnsResolver = dnsResolver;
+        return this;
+    }
+
+    /**
+     * Sets {@link SchemePortResolver} instance.
+     *
+     * @return this instance.
+     */
+    public SafePoolingHttpClientConnectionManagerBuilder setSchemePortResolver(SchemePortResolver schemePortResolver) {
+        this.schemePortResolver = schemePortResolver;
+        return this;
+    }
+
+    /**
+     * Sets maximum total connection value.
+     *
+     * @return this instance.
+     */
+    public SafePoolingHttpClientConnectionManagerBuilder setMaxConnTotal(int maxConnTotal) {
+        this.maxConnTotal = maxConnTotal;
+        return this;
+    }
+
+    /**
+     * Sets maximum connection per route value.
+     *
+     * @return this instance.
+     */
+    public SafePoolingHttpClientConnectionManagerBuilder setMaxConnPerRoute(int maxConnPerRoute) {
+        this.maxConnPerRoute = maxConnPerRoute;
+        return this;
+    }
+
+    /**
+     * Sets the same {@link SocketConfig} for all routes.
+     *
+     * @return this instance.
+     */
+    public SafePoolingHttpClientConnectionManagerBuilder setDefaultSocketConfig(SocketConfig config) {
+        this.socketConfigResolver = route -> config;
+        return this;
+    }
+
+    /**
+     * Sets {@link Resolver} of {@link SocketConfig} on a per route basis.
+     *
+     * @return this instance.
+     * @since 5.2
+     */
+    public SafePoolingHttpClientConnectionManagerBuilder setSocketConfigResolver(
+        Resolver<HttpRoute, SocketConfig> socketConfigResolver) {
+        this.socketConfigResolver = socketConfigResolver;
+        return this;
+    }
+
+    /**
+     * Sets the same {@link ConnectionConfig} for all routes.
+     *
+     * @return this instance.
+     * @since 5.2
+     */
+    public SafePoolingHttpClientConnectionManagerBuilder setDefaultConnectionConfig(ConnectionConfig config) {
+        this.connectionConfigResolver = route -> config;
+        return this;
+    }
+
+    /**
+     * Sets {@link Resolver} of {@link ConnectionConfig} on a per route basis.
+     *
+     * @return this instance.
+     * @since 5.2
+     */
+    public SafePoolingHttpClientConnectionManagerBuilder setConnectionConfigResolver(
+             Resolver<HttpRoute, ConnectionConfig> connectionConfigResolver) {
+        this.connectionConfigResolver = connectionConfigResolver;
+        return this;
+    }
+
+    /**
+     * Sets the same {@link TlsConfig} for all hosts.
+     *
+     * @return this instance.
+     * @since 5.2
+     */
+    public SafePoolingHttpClientConnectionManagerBuilder setDefaultTlsConfig(TlsConfig config) {
+        this.tlsConfigResolver = host -> config;
+        return this;
+    }
+
+    /**
+     * Sets {@link Resolver} of {@link TlsConfig} on a per host basis.
+     *
+     * @return this instance.
+     * @since 5.2
+     */
+    public SafePoolingHttpClientConnectionManagerBuilder setTlsConfigResolver(
+             Resolver<HttpHost, TlsConfig> tlsConfigResolver) {
+        this.tlsConfigResolver = tlsConfigResolver;
+        return this;
+    }
+
+    protected HttpClientConnectionOperator createConnectionOperator(SchemePortResolver schemePortResolver,
+                                                                    DnsResolver dnsResolver,
+                                                                    TlsSocketStrategy tlsSocketStrategy) {
+        return new DefaultHttpClientConnectionOperator(schemePortResolver, dnsResolver,
+                                                       RegistryBuilder.<TlsSocketStrategy>create()
+                        .register(URIScheme.HTTPS.id, tlsSocketStrategy)
+                        .build());
+    }
+
+    public SafePoolingHttpClientConnectionManager build() {
+        TlsSocketStrategy tlsSocketStrategyCopy;
+        if (tlsSocketStrategy != null) {
+            tlsSocketStrategyCopy = tlsSocketStrategy;
+        } else {
+            tlsSocketStrategyCopy = DefaultClientTlsStrategy.createDefault();
+        }
+
+        SafePoolingHttpClientConnectionManager poolingmgr = new SafePoolingHttpClientConnectionManager(
+                createConnectionOperator(schemePortResolver, dnsResolver, tlsSocketStrategyCopy));
+        poolingmgr.setSocketConfigResolver(socketConfigResolver);
+        poolingmgr.setConnectionConfigResolver(connectionConfigResolver);
+        poolingmgr.setTlsConfigResolver(tlsConfigResolver);
+        if (maxConnTotal > 0) {
+            poolingmgr.setMaxTotal(maxConnTotal);
+        }
+        if (maxConnPerRoute > 0) {
+            poolingmgr.setDefaultMaxPerRoute(maxConnPerRoute);
+        }
+        return poolingmgr;
+    }
+
+}

http-clients/apache5-client/src/main/java/software/amazon/awssdk/http/apache5/internal/conn/SafeStrictConnPool.java

@@ -0,0 +1,59 @@
+/*
+ * Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License").
+ * You may not use this file except in compliance with the License.
+ * A copy of the License is located at
+ *
+ *  http://aws.amazon.com/apache2.0
+ *
+ * or in the "license" file accompanying this file. This file is distributed
+ * on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either
+ * express or implied. See the License for the specific language governing
+ * permissions and limitations under the License.
+ */
+
+package software.amazon.awssdk.http.apache5.internal.conn;
+
+import java.util.concurrent.Future;
+import org.apache.hc.client5.http.HttpRoute;
+import org.apache.hc.client5.http.io.ManagedHttpClientConnection;
+import org.apache.hc.core5.concurrent.FutureCallback;
+import org.apache.hc.core5.pool.ConnPoolListener;
+import org.apache.hc.core5.pool.DisposalCallback;
+import org.apache.hc.core5.pool.PoolEntry;
+import org.apache.hc.core5.pool.PoolReusePolicy;
+import org.apache.hc.core5.pool.StrictConnPool;
+import org.apache.hc.core5.util.TimeValue;
+import org.apache.hc.core5.util.Timeout;
+import software.amazon.awssdk.annotations.SdkInternalApi;
+import software.amazon.awssdk.http.apache5.internal.utils.CancelOnInterruptWrapper;
+
+/**
+ * Specialization of {@link StrictConnPool} that prevents leaking the connection when thread waiting on the future is
+ * interrupted.
+ */
+@SdkInternalApi
+public final class SafeStrictConnPool extends StrictConnPool<HttpRoute, ManagedHttpClientConnection> {
+    public SafeStrictConnPool(int defaultMaxPerRoute,
+                              int maxTotal,
+                              TimeValue timeToLive,
+                              PoolReusePolicy policy,
+                              DisposalCallback<ManagedHttpClientConnection> disposalCallback,
+                              ConnPoolListener<HttpRoute> connPoolListener) {
+        super(defaultMaxPerRoute, maxTotal, timeToLive, policy, disposalCallback, connPoolListener);
+    }
+
+    public Future<PoolEntry<HttpRoute, ManagedHttpClientConnection>> lease(HttpRoute route,
+                                                                           Object state,
+                                                                           Timeout requestTimeout,
+                                                                           FutureCallback<PoolEntry<HttpRoute,
+                                                                               ManagedHttpClientConnection>> callback) {
+        return safeLease(super.lease(route, state, requestTimeout, callback));
+    }
+
+    private Future<PoolEntry<HttpRoute, ManagedHttpClientConnection>> safeLease(
+        Future<PoolEntry<HttpRoute, ManagedHttpClientConnection>> leaseFuture) {
+        return new CancelOnInterruptWrapper<>(leaseFuture);
+    }
+}