HTTPCLIENT-2416 - Fix pool entry leak on late lease completion

Author: arturobernalg

httpclient5-testing/src/test/java/org/apache/hc/client5/testing/sync/TestConnectionManagement.java

@@ -29,7 +29,12 @@
 
 import java.io.IOException;
 import java.net.InetSocketAddress;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
 import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicLong;
 import java.util.function.Consumer;
 
 import org.apache.hc.client5.http.HttpRoute;
@@ -63,6 +68,7 @@
 import org.apache.hc.core5.io.CloseMode;
 import org.apache.hc.core5.pool.PoolConcurrencyPolicy;
 import org.apache.hc.core5.pool.PoolReusePolicy;
+import org.apache.hc.core5.pool.PoolStats;
 import org.apache.hc.core5.util.TimeValue;
 import org.apache.hc.core5.util.Timeout;
 import org.junit.jupiter.api.AfterEach;
@@ -391,4 +397,49 @@ void testConnectionTimeoutSetting() throws Exception {
         connManager.close();
     }
 
+    @Test
+    void testConnectionRequestTimeout() throws Exception {
+        configureServer(bootstrap -> bootstrap
+                .register("/random/*", new RandomHandler()));
+        final HttpHost target = startServer();
+
+        connManager.setMaxTotal(1);
+
+        final HttpRoute route = new HttpRoute(target, null, false);
+        final Timeout connRequestTimeout = Timeout.ofMicroseconds(1);
+
+        final int concurrentThreads = 10;
+        final CountDownLatch countDownLatch = new CountDownLatch(concurrentThreads);
+        final AtomicLong n = new AtomicLong(concurrentThreads * 100);
+
+        final ExecutorService executorService = Executors.newFixedThreadPool(concurrentThreads);
+        for (int i = 0; i < concurrentThreads; i++) {
+            executorService.execute(() -> {
+                try {
+                    while (n.decrementAndGet() > 0) {
+                        try {
+                            final LeaseRequest request = connManager.lease("id1", route, connRequestTimeout, null);
+                            final ConnectionEndpoint connectionEndpoint = request.get(connRequestTimeout);
+                            connManager.release(connectionEndpoint, null, null);
+                        } catch (final InterruptedException ex) {
+                            Thread.currentThread().interrupt();
+                            Assertions.fail("Unexpected exception", ex);
+                        } catch (final TimeoutException | ExecutionException ignored) {
+                        }
+                    }
+                } finally {
+                    countDownLatch.countDown();
+                }
+            });
+        }
+
+        Assertions.assertTrue(countDownLatch.await(TIMEOUT.getDuration(), TIMEOUT.getTimeUnit()));
+        Assertions.assertTrue(n.get() <= 0);
+
+        final PoolStats stats = connManager.getStats(route);
+        Assertions.assertEquals(0, stats.getLeased());
+
+        connManager.close();
+    }
+
 }

httpclient5/src/main/java/org/apache/hc/client5/http/impl/io/PoolingHttpClientConnectionManager.java

@@ -29,6 +29,7 @@
 import java.io.IOException;
 import java.nio.file.Path;
 import java.util.Set;
+import java.util.concurrent.CancellationException;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.Future;
 import java.util.concurrent.TimeoutException;
@@ -388,8 +389,20 @@ public ConnectionEndpoint get(
                     final PoolEntry<HttpRoute, ManagedHttpClientConnection> poolEntry;
                     try {
                         poolEntry = leaseFuture.get(timeout.getDuration(), timeout.getTimeUnit());
-                    } catch (final TimeoutException ex) {
-                        leaseFuture.cancel(true);
+                    } catch (final TimeoutException | InterruptedException ex) {
+                        if (!leaseFuture.cancel(true)) {
+                            try {
+                                final PoolEntry<HttpRoute, ManagedHttpClientConnection> latePoolEntry = leaseFuture.get(
+                                        Timeout.ZERO_MILLISECONDS.getDuration(),
+                                        Timeout.ZERO_MILLISECONDS.getTimeUnit());
+                                if (latePoolEntry != null) {
+                                    pool.release(latePoolEntry, false);
+                                }
+                            } catch (final TimeoutException | ExecutionException | CancellationException ignore) {
+                            } catch (final InterruptedException interrupted) {
+                                    Thread.currentThread().interrupt();
+                            }
+                        }
                         throw ex;
                     }
                     if (LOG.isDebugEnabled()) {

httpclient5/src/test/java/org/apache/hc/client5/http/impl/io/TestPoolingHttpClientConnectionManager.java

@@ -171,6 +171,56 @@ void testLeaseFutureTimeout() throws Exception {
                 connRequest1.get(Timeout.ofSeconds(1)));
     }
 
+    @Test
+    void testLeaseFutureTimeoutLateLeaseReleased() throws Exception {
+        final HttpHost target = new HttpHost("localhost", 80);
+        final HttpRoute route = new HttpRoute(target);
+        final PoolEntry<HttpRoute, ManagedHttpClientConnection> entry = new PoolEntry<>(route, TimeValue.NEG_ONE_MILLISECOND);
+
+        Mockito.when(future.get(1, TimeUnit.SECONDS)).thenThrow(new TimeoutException());
+        Mockito.when(future.cancel(true)).thenReturn(false);
+        Mockito.when(future.get(0L, TimeUnit.MILLISECONDS)).thenReturn(entry);
+        Mockito.when(pool.lease(
+                Mockito.eq(route),
+                Mockito.eq(null),
+                Mockito.any(),
+                Mockito.eq(null)))
+                .thenReturn(future);
+
+        final LeaseRequest connRequest1 = mgr.lease("some-id", route, null);
+        Assertions.assertThrows(TimeoutException.class, () ->
+                connRequest1.get(Timeout.ofSeconds(1)));
+
+        Mockito.verify(future).cancel(true);
+        Mockito.verify(future).get(0L, TimeUnit.MILLISECONDS);
+        Mockito.verify(pool).release(entry, false);
+    }
+
+    @Test
+    void testLeaseFutureInterruptedLateLeaseReleased() throws Exception {
+        final HttpHost target = new HttpHost("localhost", 80);
+        final HttpRoute route = new HttpRoute(target);
+        final PoolEntry<HttpRoute, ManagedHttpClientConnection> entry = new PoolEntry<>(route, TimeValue.NEG_ONE_MILLISECOND);
+
+        Mockito.when(future.get(1, TimeUnit.SECONDS)).thenThrow(new InterruptedException());
+        Mockito.when(future.cancel(true)).thenReturn(false);
+        Mockito.when(future.get(0L, TimeUnit.MILLISECONDS)).thenReturn(entry);
+        Mockito.when(pool.lease(
+                Mockito.eq(route),
+                Mockito.eq(null),
+                Mockito.any(),
+                Mockito.eq(null)))
+                .thenReturn(future);
+
+        final LeaseRequest connRequest1 = mgr.lease("some-id", route, null);
+        Assertions.assertThrows(InterruptedException.class, () ->
+                connRequest1.get(Timeout.ofSeconds(1)));
+
+        Mockito.verify(future).cancel(true);
+        Mockito.verify(future).get(0L, TimeUnit.MILLISECONDS);
+        Mockito.verify(pool).release(entry, false);
+    }
+
     @Test
     void testReleaseReusable() throws Exception {
         final HttpHost target = new HttpHost("localhost", 80);