diff --git a/httpclient5/src/main/java/org/apache/hc/client5/http/impl/classic/ResponseEntityProxy.java b/httpclient5/src/main/java/org/apache/hc/client5/http/impl/classic/ResponseEntityProxy.java index 5e6b91da88..0500368cda 100644 --- a/httpclient5/src/main/java/org/apache/hc/client5/http/impl/classic/ResponseEntityProxy.java +++ b/httpclient5/src/main/java/org/apache/hc/client5/http/impl/classic/ResponseEntityProxy.java @@ -62,10 +62,17 @@ public static void enhance(final ClassicHttpResponse response, final ExecRuntime private void cleanup() throws IOException { if (this.execRuntime != null) { - if (this.execRuntime.isEndpointConnected()) { - this.execRuntime.disconnectEndpoint(); + try { + if (this.execRuntime.isEndpointConnected()) { + this.execRuntime.disconnectEndpoint(); + } + } finally { + // discardEndpoint() is the only path that returns the lease to the + // pool (its manager.release(...) runs inside a finally). Guard it so + // an IOException from disconnectEndpoint() on a dead socket cannot + // strand the connection as permanently leased. + this.execRuntime.discardEndpoint(); } - this.execRuntime.discardEndpoint(); } } diff --git a/httpclient5/src/test/java/org/apache/hc/client5/http/impl/classic/TestResponseEntityProxy.java b/httpclient5/src/test/java/org/apache/hc/client5/http/impl/classic/TestResponseEntityProxy.java index a376fa3d68..409770fa66 100644 --- a/httpclient5/src/test/java/org/apache/hc/client5/http/impl/classic/TestResponseEntityProxy.java +++ b/httpclient5/src/test/java/org/apache/hc/client5/http/impl/classic/TestResponseEntityProxy.java @@ -28,19 +28,33 @@ package org.apache.hc.client5.http.impl.classic; import java.io.ByteArrayInputStream; +import java.io.IOException; import java.io.InputStream; import java.util.List; +import java.util.concurrent.atomic.AtomicInteger; +import org.apache.hc.client5.http.HttpRoute; import org.apache.hc.client5.http.classic.ExecRuntime; +import org.apache.hc.client5.http.io.ConnectionEndpoint; +import org.apache.hc.client5.http.io.HttpClientConnectionManager; +import org.apache.hc.client5.http.io.LeaseRequest; +import org.apache.hc.client5.http.protocol.HttpClientContext; import org.apache.hc.core5.function.Supplier; +import org.apache.hc.core5.http.ClassicHttpRequest; import org.apache.hc.core5.http.ClassicHttpResponse; import org.apache.hc.core5.http.Header; import org.apache.hc.core5.http.HttpEntity; +import org.apache.hc.core5.http.HttpHost; import org.apache.hc.core5.http.impl.io.ChunkedInputStream; +import org.apache.hc.core5.http.impl.io.HttpRequestExecutor; import org.apache.hc.core5.http.impl.io.SessionInputBufferImpl; import org.apache.hc.core5.http.io.SessionInputBuffer; import org.apache.hc.core5.http.io.entity.BasicHttpEntity; import org.apache.hc.core5.http.message.BasicClassicHttpResponse; +import org.apache.hc.core5.http.protocol.HttpContext; +import org.apache.hc.core5.io.CloseMode; +import org.apache.hc.core5.util.TimeValue; +import org.apache.hc.core5.util.Timeout; import org.junit.jupiter.api.Assertions; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; @@ -48,6 +62,7 @@ import org.mockito.Mock; import org.mockito.Mockito; import org.mockito.MockitoAnnotations; +import org.slf4j.LoggerFactory; class TestResponseEntityProxy { @@ -137,4 +152,117 @@ void testWriteToNullDrainsAndReleasesStream() throws Exception { } + + @Test + void testCleanupDiscardsEndpointWhenDisconnectEndpointThrows() throws Exception { + // Simulate a dead socket: disconnectEndpoint() throws (endpoint.close() on + // a broken connection). Without the fix, discardEndpoint() would be skipped + // and the connection would remain permanently leased in the pool. + Mockito.when(execRuntime.isEndpointConnected()).thenReturn(Boolean.TRUE); + Mockito.doThrow(new IOException("simulated dead socket")) + .when(execRuntime).disconnectEndpoint(); + + final ArgumentCaptor captor = ArgumentCaptor.forClass(HttpEntity.class); + ResponseEntityProxy.enhance(response, execRuntime); + Mockito.verify(response).setEntity(captor.capture()); + final HttpEntity wrappedEntity = captor.getValue(); + + Assertions.assertThrows(IOException.class, wrappedEntity::close); + + Mockito.verify(execRuntime).disconnectEndpoint(); + Mockito.verify(execRuntime).discardEndpoint(); + } + + /** + * Regression test for HTTPCLIENT-2432. + *

+ * On paths that reach {@code cleanup()} while the endpoint is still leased + * (e.g. {@code streamAbort()}, or {@code streamClosed()} swallowing a + * {@code SocketException}), {@code disconnectEndpoint()} is invoked on a + * connection whose socket may be in a broken state. Prior to the fix, an + * {@code IOException} from {@code endpoint.close()} propagated out and + * skipped {@code discardEndpoint()} - the only path that calls + * {@code manager.release(...)} and returns the lease to the pool. Under + * load this exhausted the pool. + *

+ * The test wires a real {@link InternalExecRuntime} to a fake connection + * manager whose leased endpoint throws from {@code close()}, then triggers + * {@code cleanup()} via {@code streamAbort()} and asserts the manager + * received exactly one {@code release()} call. + */ + @Test + void testPoolLeaseReturnedWhenDisconnectEndpointThrows() throws Exception { + final AtomicInteger releaseCount = new AtomicInteger(); + final ConnectionEndpoint brokenEndpoint = new ConnectionEndpoint() { + @Override + public ClassicHttpResponse execute(final String id, final ClassicHttpRequest request, + final HttpRequestExecutor executor, final HttpContext context) { + throw new UnsupportedOperationException(); + } + @Override + public boolean isConnected() { + return true; + } + @Override + public void setSocketTimeout(final Timeout timeout) { + } + @Override + public void close(final CloseMode closeMode) { + } + @Override + public void close() throws IOException { + throw new IOException("simulated dead socket"); + } + }; + + final HttpClientConnectionManager fakeManager = new HttpClientConnectionManager() { + @Override + public LeaseRequest lease(final String id, final HttpRoute route, + final Timeout requestTimeout, final Object state) { + return new LeaseRequest() { + @Override + public ConnectionEndpoint get(final Timeout timeout) { + return brokenEndpoint; + } + @Override + public boolean cancel() { + return false; + } + }; + } + @Override + public void release(final ConnectionEndpoint endpoint, final Object newState, + final TimeValue validDuration) { + releaseCount.incrementAndGet(); + } + @Override + public void connect(final ConnectionEndpoint endpoint, final TimeValue connectTimeout, + final HttpContext context) { + } + @Override + public void upgrade(final ConnectionEndpoint endpoint, final HttpContext context) { + } + @Override + public void close() { + } + @Override + public void close(final CloseMode closeMode) { + } + }; + + final InternalExecRuntime runtime = new InternalExecRuntime( + LoggerFactory.getLogger(TestResponseEntityProxy.class), + fakeManager, + new HttpRequestExecutor(), + null); + runtime.acquireEndpoint("id1", new HttpRoute(new HttpHost("localhost", 80)), + null, HttpClientContext.create()); + + final ResponseEntityProxy proxy = new ResponseEntityProxy(entity, runtime); + + Assertions.assertThrows(IOException.class, () -> proxy.streamAbort(null)); + + Assertions.assertEquals(1, releaseCount.get(), + "connection lease must be returned to the pool even when disconnectEndpoint() throws"); + } } \ No newline at end of file