From c2945273d572489e1ce692f3ffbbfc4b35f1e955 Mon Sep 17 00:00:00 2001 From: goutamadwant Date: Tue, 30 Jun 2026 23:52:40 -0700 Subject: [PATCH 1/2] Enforce Java 11 response body timeouts --- .../java/feign/http2client/Http2Client.java | 129 +++++++++++++++++- .../test/Http2ClientAsyncTest.java | 23 ++++ .../http2client/test/Http2ClientTest.java | 15 ++ 3 files changed, 164 insertions(+), 3 deletions(-) diff --git a/java11/src/main/java/feign/http2client/Http2Client.java b/java11/src/main/java/feign/http2client/Http2Client.java index 9eb718df9..8148edccc 100644 --- a/java11/src/main/java/feign/http2client/Http2Client.java +++ b/java11/src/main/java/feign/http2client/Http2Client.java @@ -38,6 +38,7 @@ import java.net.http.HttpRequest.Builder; import java.net.http.HttpResponse; import java.net.http.HttpResponse.BodyHandlers; +import java.net.http.HttpTimeoutException; import java.time.Duration; import java.util.Arrays; import java.util.Collection; @@ -52,6 +53,11 @@ import java.util.TreeSet; import java.util.concurrent.CompletableFuture; import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.Executors; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.ScheduledFuture; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; import java.util.function.Function; import java.util.stream.Collectors; import java.util.zip.GZIPInputStream; @@ -59,6 +65,14 @@ public class Http2Client implements Client, AsyncClient { + private static final ScheduledExecutorService BODY_READ_TIMEOUT_EXECUTOR = + Executors.newSingleThreadScheduledExecutor( + runnable -> { + Thread thread = new Thread(runnable, "feign-http2client-body-timeout"); + thread.setDaemon(true); + return thread; + }); + private final HttpClient client; private final Map> clients = new ConcurrentHashMap<>(); @@ -109,7 +123,7 @@ public Response execute(Request request, Options options) throws IOException { throw new IOException(e); } - return toFeignResponse(request, httpResponse); + return toFeignResponse(request, httpResponse, options); } @Override @@ -125,17 +139,22 @@ public CompletableFuture execute( HttpClient clientForRequest = getOrCreateClient(options); CompletableFuture> future = clientForRequest.sendAsync(httpRequest, HttpResponse.BodyHandlers.ofInputStream()); - return future.thenApply(httpResponse -> toFeignResponse(request, httpResponse)); + return future.thenApply(httpResponse -> toFeignResponse(request, httpResponse, options)); } protected Response toFeignResponse(Request request, HttpResponse httpResponse) { + return toFeignResponse(request, httpResponse, null); + } + + private Response toFeignResponse( + Request request, HttpResponse httpResponse, Options options) { final OptionalLong length = httpResponse.headers().firstValueAsLong("Content-Length"); final Integer contentLength = length.isPresent() && length.getAsLong() >= 0 && length.getAsLong() <= Integer.MAX_VALUE ? (int) length.getAsLong() : null; - InputStream body = httpResponse.body(); + InputStream body = withReadTimeout(httpResponse.body(), options); if (httpResponse.headers().allValues(CONTENT_ENCODING).contains(ENCODING_GZIP)) { try { @@ -156,6 +175,110 @@ protected Response toFeignResponse(Request request, HttpResponse ht .build(); } + private static InputStream withReadTimeout(InputStream body, Options options) { + if (body == null || options == null || options.readTimeout() <= 0) { + return body; + } + return new TimeoutInputStream(body, options.readTimeout(), options.readTimeoutUnit()); + } + + private static final class TimeoutInputStream extends InputStream { + + private final InputStream delegate; + private final long timeout; + private final TimeUnit timeoutUnit; + + private TimeoutInputStream(InputStream delegate, long timeout, TimeUnit timeoutUnit) { + this.delegate = delegate; + this.timeout = timeout; + this.timeoutUnit = timeoutUnit; + } + + @Override + public int read() throws IOException { + return readWithTimeout(delegate::read); + } + + @Override + public int read(byte[] b, int off, int len) throws IOException { + return readWithTimeout(() -> delegate.read(b, off, len)); + } + + @Override + public int available() throws IOException { + return delegate.available(); + } + + @Override + public void close() throws IOException { + delegate.close(); + } + + private int readWithTimeout(BodyRead read) throws IOException { + final AtomicBoolean completed = new AtomicBoolean(false); + final AtomicBoolean timedOut = new AtomicBoolean(false); + final ScheduledFuture timeoutFuture = + BODY_READ_TIMEOUT_EXECUTOR.schedule( + () -> { + if (completed.compareAndSet(false, true)) { + timedOut.set(true); + try { + delegate.close(); + } catch (IOException ignored) { + } + } + }, + timeout, + timeoutUnit); + + try { + final int result = read.read(); + if (completed.compareAndSet(false, true)) { + timeoutFuture.cancel(false); + return result; + } + throw timeoutException(null); + } catch (IOException e) { + if (completed.compareAndSet(false, true)) { + timeoutFuture.cancel(false); + } + final HttpTimeoutException timeoutException = findTimeoutException(e); + if (timedOut.get() || timeoutException != null) { + throw timeoutException == null ? timeoutException(e) : timeoutException; + } + throw e; + } catch (RuntimeException e) { + if (completed.compareAndSet(false, true)) { + timeoutFuture.cancel(false); + } + throw e; + } + } + + private static HttpTimeoutException timeoutException(IOException cause) { + final HttpTimeoutException exception = new HttpTimeoutException("response timed out"); + if (cause != null) { + exception.initCause(cause); + } + return exception; + } + + private static HttpTimeoutException findTimeoutException(Throwable throwable) { + Throwable current = throwable; + while (current != null) { + if (current instanceof HttpTimeoutException) { + return (HttpTimeoutException) current; + } + current = current.getCause(); + } + return null; + } + } + + private interface BodyRead { + int read() throws IOException; + } + private HttpClient getOrCreateClient(Options options) { if (doesClientConfigurationDiffer(options)) { // create a new client from the existing one - but with connectTimeout and followRedirect diff --git a/java11/src/test/java/feign/http2client/test/Http2ClientAsyncTest.java b/java11/src/test/java/feign/http2client/test/Http2ClientAsyncTest.java index c5a29c43e..4147d2f35 100644 --- a/java11/src/test/java/feign/http2client/test/Http2ClientAsyncTest.java +++ b/java11/src/test/java/feign/http2client/test/Http2ClientAsyncTest.java @@ -63,6 +63,7 @@ import java.io.IOException; import java.lang.reflect.Type; import java.net.URI; +import java.net.http.HttpTimeoutException; import java.nio.charset.StandardCharsets; import java.time.Clock; import java.time.Instant; @@ -510,6 +511,23 @@ void doesntRetryAfterResponseIsSent() throws Throwable { assertThat(exception.getMessage()).contains("timeout reading POST http://"); } + @Test + void timeoutReadingResponseBody() throws Throwable { + server.enqueue(new MockResponse().setBody("foo").setBodyDelay(1, TimeUnit.SECONDS)); + + final TestInterfaceAsync api = + newAsyncBuilder() + .options( + new Request.Options(500, TimeUnit.MILLISECONDS, 500, TimeUnit.MILLISECONDS, true)) + .target("http://localhost:" + server.getPort()); + + final CompletableFuture cf = api.post(); + server.takeRequest(); + + Throwable exception = assertThrows(FeignException.class, () -> unwrap(cf)); + assertThat(exception).hasCauseInstanceOf(HttpTimeoutException.class); + } + @Test void throwsFeignExceptionIncludingBody() throws Throwable { server.enqueue(new MockResponse().setBody("success!")); @@ -1060,6 +1078,11 @@ TestInterfaceAsyncBuilder dismiss404() { return this; } + TestInterfaceAsyncBuilder options(Request.Options options) { + delegate.options(options); + return this; + } + TestInterfaceAsyncBuilder queryMapEndcoder(QueryMapEncoder queryMapEncoder) { delegate.queryMapEncoder(queryMapEncoder); return this; diff --git a/java11/src/test/java/feign/http2client/test/Http2ClientTest.java b/java11/src/test/java/feign/http2client/test/Http2ClientTest.java index 4f3602322..af8996e14 100644 --- a/java11/src/test/java/feign/http2client/test/Http2ClientTest.java +++ b/java11/src/test/java/feign/http2client/test/Http2ClientTest.java @@ -175,6 +175,21 @@ void timeoutTest() { assertThat(exception).hasCauseInstanceOf(HttpTimeoutException.class); } + @Test + void timeoutReadingResponseBody() { + server.enqueue(new MockResponse().setBody("foo").setBodyDelay(1, TimeUnit.SECONDS)); + + final TestInterface api = + newBuilder() + .retryer(Retryer.NEVER_RETRY) + .options( + new Request.Options(500, TimeUnit.MILLISECONDS, 500, TimeUnit.MILLISECONDS, true)) + .target(TestInterface.class, server.url("/").toString()); + + FeignException exception = assertThrows(FeignException.class, () -> api.timeout()); + assertThat(exception).hasCauseInstanceOf(HttpTimeoutException.class); + } + @Test void getWithRequestBody() throws Exception { // MockWebServer rejects GET requests carrying a body ("Request must not have a body"), From 77694542cfc4de0f4b7bd9b7cebd1367362fce2f Mon Sep 17 00:00:00 2001 From: goutamadwant Date: Wed, 12 Aug 2026 22:00:38 -0700 Subject: [PATCH 2/2] Preserve response hooks and use one body timeout Keep both response conversion extension points on the production path. Schedule one deadline per body stream and retain delegated stream behavior. --- .../java/feign/http2client/Http2Client.java | 183 +++++++++++++----- .../Http2ClientContentLengthTest.java | 54 +++++- .../test/Http2ClientAsyncTest.java | 7 + .../http2client/test/Http2ClientTest.java | 52 +++++ 4 files changed, 244 insertions(+), 52 deletions(-) diff --git a/java11/src/main/java/feign/http2client/Http2Client.java b/java11/src/main/java/feign/http2client/Http2Client.java index 2504d6682..0574ebc1f 100644 --- a/java11/src/main/java/feign/http2client/Http2Client.java +++ b/java11/src/main/java/feign/http2client/Http2Client.java @@ -32,6 +32,7 @@ import java.net.http.HttpClient; import java.net.http.HttpClient.Redirect; import java.net.http.HttpClient.Version; +import java.net.http.HttpHeaders; import java.net.http.HttpRequest; import java.net.http.HttpRequest.BodyPublisher; import java.net.http.HttpRequest.BodyPublishers; @@ -57,14 +58,15 @@ import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.ScheduledFuture; import java.util.concurrent.TimeUnit; -import java.util.concurrent.atomic.AtomicBoolean; import java.util.function.Function; import java.util.stream.Collectors; import java.util.zip.GZIPInputStream; import java.util.zip.InflaterInputStream; +import javax.net.ssl.SSLSession; public class Http2Client implements Client, AsyncClient { + // Shared by all clients so response-body deadlines do not create one thread per client. private static final ScheduledExecutorService BODY_READ_TIMEOUT_EXECUTOR = Executors.newSingleThreadScheduledExecutor( runnable -> { @@ -143,18 +145,13 @@ public CompletableFuture execute( } protected Response toFeignResponse(Request request, HttpResponse httpResponse) { - return toFeignResponse(request, httpResponse, null); - } - - private Response toFeignResponse( - Request request, HttpResponse httpResponse, Options options) { final OptionalLong length = httpResponse.headers().firstValueAsLong("Content-Length"); Integer contentLength = length.isPresent() && length.getAsLong() >= 0 && length.getAsLong() <= Integer.MAX_VALUE ? (int) length.getAsLong() : null; - InputStream body = withReadTimeout(httpResponse.body(), options); + InputStream body = httpResponse.body(); if (httpResponse.headers().allValues(CONTENT_ENCODING).contains(ENCODING_GZIP)) { try { @@ -178,6 +175,15 @@ private Response toFeignResponse( .build(); } + protected Response toFeignResponse( + Request request, HttpResponse httpResponse, Options options) { + final InputStream body = httpResponse.body(); + final InputStream timedBody = withReadTimeout(body, options); + return toFeignResponse( + request, + body == timedBody ? httpResponse : new TimeoutHttpResponse(httpResponse, timedBody)); + } + private static InputStream withReadTimeout(InputStream body, Options options) { if (body == null || options == null || options.readTimeout() <= 0) { return body; @@ -188,23 +194,51 @@ private static InputStream withReadTimeout(InputStream body, Options options) { private static final class TimeoutInputStream extends InputStream { private final InputStream delegate; - private final long timeout; - private final TimeUnit timeoutUnit; + private final ScheduledFuture timeoutFuture; + private volatile boolean timedOut; private TimeoutInputStream(InputStream delegate, long timeout, TimeUnit timeoutUnit) { this.delegate = delegate; - this.timeout = timeout; - this.timeoutUnit = timeoutUnit; + this.timeoutFuture = + BODY_READ_TIMEOUT_EXECUTOR.schedule( + () -> { + timedOut = true; + try { + delegate.close(); + } catch (IOException ignored) { + } + }, + timeout, + timeoutUnit); } @Override public int read() throws IOException { - return readWithTimeout(delegate::read); + try { + return afterRead(delegate.read()); + } catch (IOException e) { + throw translateException(e); + } } @Override public int read(byte[] b, int off, int len) throws IOException { - return readWithTimeout(() -> delegate.read(b, off, len)); + try { + return afterRead(delegate.read(b, off, len)); + } catch (IOException e) { + throw translateException(e); + } + } + + @Override + public long skip(long n) throws IOException { + try { + final long skipped = delegate.skip(n); + checkTimedOut(); + return skipped; + } catch (IOException e) { + throw translateException(e); + } } @Override @@ -212,52 +246,52 @@ public int available() throws IOException { return delegate.available(); } + @Override + public synchronized void mark(int readLimit) { + delegate.mark(readLimit); + } + + @Override + public synchronized void reset() throws IOException { + delegate.reset(); + } + + @Override + public boolean markSupported() { + return delegate.markSupported(); + } + @Override public void close() throws IOException { + timeoutFuture.cancel(false); delegate.close(); } - private int readWithTimeout(BodyRead read) throws IOException { - final AtomicBoolean completed = new AtomicBoolean(false); - final AtomicBoolean timedOut = new AtomicBoolean(false); - final ScheduledFuture timeoutFuture = - BODY_READ_TIMEOUT_EXECUTOR.schedule( - () -> { - if (completed.compareAndSet(false, true)) { - timedOut.set(true); - try { - delegate.close(); - } catch (IOException ignored) { - } - } - }, - timeout, - timeoutUnit); + private int afterRead(int result) throws HttpTimeoutException { + checkTimedOut(); + if (result == -1) { + timeoutFuture.cancel(false); + } + return result; + } - try { - final int result = read.read(); - if (completed.compareAndSet(false, true)) { - timeoutFuture.cancel(false); - return result; - } + private void checkTimedOut() throws HttpTimeoutException { + if (timedOut) { throw timeoutException(null); - } catch (IOException e) { - if (completed.compareAndSet(false, true)) { - timeoutFuture.cancel(false); - } - final HttpTimeoutException timeoutException = findTimeoutException(e); - if (timedOut.get() || timeoutException != null) { - throw timeoutException == null ? timeoutException(e) : timeoutException; - } - throw e; - } catch (RuntimeException e) { - if (completed.compareAndSet(false, true)) { - timeoutFuture.cancel(false); - } - throw e; } } + private IOException translateException(IOException exception) { + final HttpTimeoutException timeoutException = findTimeoutException(exception); + if (timeoutException != null) { + return timeoutException; + } + if (timedOut) { + return timeoutException(exception); + } + return exception; + } + private static HttpTimeoutException timeoutException(IOException cause) { final HttpTimeoutException exception = new HttpTimeoutException("response timed out"); if (cause != null) { @@ -278,8 +312,55 @@ private static HttpTimeoutException findTimeoutException(Throwable throwable) { } } - private interface BodyRead { - int read() throws IOException; + private static final class TimeoutHttpResponse implements HttpResponse { + + private final HttpResponse delegate; + private final InputStream body; + + private TimeoutHttpResponse(HttpResponse delegate, InputStream body) { + this.delegate = delegate; + this.body = body; + } + + @Override + public int statusCode() { + return delegate.statusCode(); + } + + @Override + public HttpRequest request() { + return delegate.request(); + } + + @Override + public Optional> previousResponse() { + return delegate.previousResponse(); + } + + @Override + public HttpHeaders headers() { + return delegate.headers(); + } + + @Override + public InputStream body() { + return body; + } + + @Override + public Optional sslSession() { + return delegate.sslSession(); + } + + @Override + public URI uri() { + return delegate.uri(); + } + + @Override + public Version version() { + return delegate.version(); + } } private HttpClient getOrCreateClient(Options options) { diff --git a/java11/src/test/java/feign/http2client/Http2ClientContentLengthTest.java b/java11/src/test/java/feign/http2client/Http2ClientContentLengthTest.java index bb3970595..cd4488428 100644 --- a/java11/src/test/java/feign/http2client/Http2ClientContentLengthTest.java +++ b/java11/src/test/java/feign/http2client/Http2ClientContentLengthTest.java @@ -16,6 +16,7 @@ package feign.http2client; import static org.assertj.core.api.Assertions.assertThat; +import static org.junit.jupiter.api.Assertions.assertThrows; import feign.Request; import feign.Request.HttpMethod; @@ -23,17 +24,21 @@ import feign.Util; import java.io.ByteArrayInputStream; import java.io.ByteArrayOutputStream; +import java.io.IOException; import java.io.InputStream; import java.net.URI; import java.net.http.HttpClient.Version; import java.net.http.HttpHeaders; import java.net.http.HttpRequest; import java.net.http.HttpResponse; +import java.net.http.HttpTimeoutException; import java.nio.charset.StandardCharsets; import java.util.Collections; import java.util.List; import java.util.Map; import java.util.Optional; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; import java.util.zip.GZIPOutputStream; import javax.net.ssl.SSLSession; import org.junit.jupiter.api.Test; @@ -41,6 +46,11 @@ class Http2ClientContentLengthTest { private static HttpResponse responseWithContentLength(String contentLength) { + return responseWithContentLength(contentLength, new ByteArrayInputStream(new byte[0])); + } + + private static HttpResponse responseWithContentLength( + String contentLength, InputStream body) { final HttpHeaders headers = HttpHeaders.of(Map.of("Content-Length", List.of(contentLength)), (name, value) -> true); return new HttpResponse<>() { @@ -66,7 +76,7 @@ public HttpHeaders headers() { @Override public InputStream body() { - return new ByteArrayInputStream(new byte[0]); + return body; } @Override @@ -185,4 +195,46 @@ void gzipDecodedBodyReportsUnknownLength() throws Exception { assertThat(Util.toString(response.body().asReader(StandardCharsets.UTF_8))) .isEqualTo("Compressed Data"); } + + @Test + void responseBodyTimeoutUsesOneStreamDeadline() throws Exception { + final CountDownLatch closed = new CountDownLatch(1); + final InputStream body = + new ByteArrayInputStream(new byte[] {1}) { + @Override + public void close() throws IOException { + closed.countDown(); + super.close(); + } + }; + final Request.Options options = + new Request.Options(1, TimeUnit.SECONDS, 1, TimeUnit.MILLISECONDS, true); + final Response response = + new Http2Client().toFeignResponse(request(), responseWithContentLength("1", body), options); + + assertThat(closed.await(5, TimeUnit.SECONDS)).isTrue(); + assertThrows(HttpTimeoutException.class, response.body().asInputStream()::read); + } + + @Test + void responseBodyTimeoutPreservesStreamOperations() throws Exception { + final Request.Options options = + new Request.Options(1, TimeUnit.SECONDS, 1, TimeUnit.MINUTES, true); + final Response response = + new Http2Client() + .toFeignResponse( + request(), + responseWithContentLength("3", new ByteArrayInputStream(new byte[] {1, 2, 3})), + options); + final InputStream body = response.body().asInputStream(); + + assertThat(body.markSupported()).isTrue(); + body.mark(3); + assertThat(body.read()).isEqualTo(1); + assertThat(body.skip(1)).isEqualTo(1); + assertThat(body.read()).isEqualTo(3); + body.reset(); + assertThat(body.read()).isEqualTo(1); + body.close(); + } } diff --git a/java11/src/test/java/feign/http2client/test/Http2ClientAsyncTest.java b/java11/src/test/java/feign/http2client/test/Http2ClientAsyncTest.java index 4147d2f35..3f7a00a1c 100644 --- a/java11/src/test/java/feign/http2client/test/Http2ClientAsyncTest.java +++ b/java11/src/test/java/feign/http2client/test/Http2ClientAsyncTest.java @@ -45,6 +45,7 @@ import feign.RequestTemplate; import feign.Response; import feign.ResponseMapper; +import feign.Retryer; import feign.Target; import feign.Target.HardCodedTarget; import feign.Util; @@ -517,6 +518,7 @@ void timeoutReadingResponseBody() throws Throwable { final TestInterfaceAsync api = newAsyncBuilder() + .retryer(Retryer.NEVER_RETRY) .options( new Request.Options(500, TimeUnit.MILLISECONDS, 500, TimeUnit.MILLISECONDS, true)) .target("http://localhost:" + server.getPort()); @@ -1083,6 +1085,11 @@ TestInterfaceAsyncBuilder options(Request.Options options) { return this; } + TestInterfaceAsyncBuilder retryer(Retryer retryer) { + delegate.retryer(retryer); + return this; + } + TestInterfaceAsyncBuilder queryMapEndcoder(QueryMapEncoder queryMapEncoder) { delegate.queryMapEncoder(queryMapEncoder); return this; diff --git a/java11/src/test/java/feign/http2client/test/Http2ClientTest.java b/java11/src/test/java/feign/http2client/test/Http2ClientTest.java index af8996e14..963bdbc51 100644 --- a/java11/src/test/java/feign/http2client/test/Http2ClientTest.java +++ b/java11/src/test/java/feign/http2client/test/Http2ClientTest.java @@ -190,6 +190,32 @@ void timeoutReadingResponseBody() { assertThat(exception).hasCauseInstanceOf(HttpTimeoutException.class); } + @Test + void invokesTwoArgumentResponseOverride() { + server.enqueue(new MockResponse().setBody("foo")); + final TwoArgumentOverrideClient client = new TwoArgumentOverrideClient(); + final TestInterface api = + Feign.builder() + .client(client) + .target(TestInterface.class, "http://localhost:" + server.getPort()); + + assertThat(api.get()).isEqualTo("foo"); + assertThat(client.invoked).isTrue(); + } + + @Test + void invokesThreeArgumentResponseOverride() { + server.enqueue(new MockResponse().setBody("foo")); + final ThreeArgumentOverrideClient client = new ThreeArgumentOverrideClient(); + final TestInterface api = + Feign.builder() + .client(client) + .target(TestInterface.class, "http://localhost:" + server.getPort()); + + assertThat(api.get()).isEqualTo("foo"); + assertThat(client.invoked).isTrue(); + } + @Test void getWithRequestBody() throws Exception { // MockWebServer rejects GET requests carrying a body ("Request must not have a body"), @@ -394,4 +420,30 @@ public void parsesRequestAndResponse() throws IOException, InterruptedException public Feign.Builder newBuilder() { return Feign.builder().client(new Http2Client()); } + + private static final class TwoArgumentOverrideClient extends Http2Client { + + private boolean invoked; + + @Override + protected Response toFeignResponse( + Request request, java.net.http.HttpResponse response) { + invoked = true; + return super.toFeignResponse(request, response); + } + } + + private static final class ThreeArgumentOverrideClient extends Http2Client { + + private boolean invoked; + + @Override + protected Response toFeignResponse( + Request request, + java.net.http.HttpResponse response, + Request.Options options) { + invoked = true; + return super.toFeignResponse(request, response, options); + } + } }