diff --git a/.changes/next-release/feature-AWSSDKforJavav2-a91bebb.json b/.changes/next-release/feature-AWSSDKforJavav2-a91bebb.json new file mode 100644 index 000000000000..ce75baa4e048 --- /dev/null +++ b/.changes/next-release/feature-AWSSDKforJavav2-a91bebb.json @@ -0,0 +1,6 @@ +{ + "type": "feature", + "category": "AWS SDK for Java v2", + "contributor": "", + "description": "Adds timeouts to ResponsePublisher and ResponseInputStream to close connection if response not consumed" +} diff --git a/core/sdk-core/src/main/java/software/amazon/awssdk/core/ResponseInputStream.java b/core/sdk-core/src/main/java/software/amazon/awssdk/core/ResponseInputStream.java index 90df32b24113..8a7d9fea9b89 100644 --- a/core/sdk-core/src/main/java/software/amazon/awssdk/core/ResponseInputStream.java +++ b/core/sdk-core/src/main/java/software/amazon/awssdk/core/ResponseInputStream.java @@ -15,17 +15,33 @@ package software.amazon.awssdk.core; +import java.io.IOException; import java.io.InputStream; +import java.time.Duration; +import java.util.concurrent.Executors; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.ScheduledFuture; +import java.util.concurrent.TimeUnit; import software.amazon.awssdk.annotations.SdkPublicApi; +import software.amazon.awssdk.annotations.SdkTestInternalApi; import software.amazon.awssdk.core.io.SdkFilterInputStream; import software.amazon.awssdk.http.Abortable; import software.amazon.awssdk.http.AbortableInputStream; import software.amazon.awssdk.utils.IoUtils; +import software.amazon.awssdk.utils.Logger; import software.amazon.awssdk.utils.Validate; /** * Input stream that provides access to the unmarshalled POJO response returned by the service in addition to the streamed * contents. This input stream should be closed after all data has been read from the stream. + * + *

+ * NOTE: You must read this stream promptly to avoid automatic cancellation. The default timeout for reading is 60 + * seconds, which starts when the response stream is ready. If {@link #read()} is not invoked before the timeout, the stream will + * automatically abort to prevent resource leakage. + *

+ * The timeout can be customized by passing a {@link Duration} to the constructor, or disabled entirely by + * passing {@link Duration#ZERO} or a negative {@link Duration}. *

* Note about the Apache http client: This input stream can be used to leverage a feature of the Apache http client where * connections are released back to the connection pool to be reused. As such, calling {@link ResponseInputStream#close() close} @@ -43,19 +59,37 @@ @SdkPublicApi public final class ResponseInputStream extends SdkFilterInputStream implements Abortable { + private static final Logger log = Logger.loggerFor(ResponseInputStream.class); + private static final Duration DEFAULT_TIMEOUT = Duration.ofSeconds(60); private final ResponseT response; private final Abortable abortable; + private ScheduledFuture timeoutTask; + private volatile boolean hasRead = false; public ResponseInputStream(ResponseT resp, AbortableInputStream in) { + this(resp, in, null); + } + + public ResponseInputStream(ResponseT resp, AbortableInputStream in, Duration timeout) { super(in); this.response = Validate.paramNotNull(resp, "response"); this.abortable = Validate.paramNotNull(in, "abortableInputStream"); + + Duration resolvedTimeout = timeout != null ? timeout : DEFAULT_TIMEOUT; + scheduleTimeoutTask(resolvedTimeout); } public ResponseInputStream(ResponseT resp, InputStream in) { + this(resp, in, null); + } + + public ResponseInputStream(ResponseT resp, InputStream in, Duration timeout) { super(in); this.response = Validate.paramNotNull(resp, "response"); this.abortable = in instanceof Abortable ? (Abortable) in : null; + + Duration resolvedTimeout = timeout != null ? timeout : DEFAULT_TIMEOUT; + scheduleTimeoutTask(resolvedTimeout); } /** @@ -65,15 +99,77 @@ public ResponseT response() { return response; } + @Override + public int read() throws IOException { + cancelTimeoutTask(); + return super.read(); + } + + @Override + public int read(byte[] b) throws IOException { + cancelTimeoutTask(); + return super.read(b); + } + + @Override + public int read(byte[] b, int off, int len) throws IOException { + cancelTimeoutTask(); + return super.read(b, off, len); + } + + private void cancelTimeoutTask() { + if (!hasRead && timeoutTask != null) { + timeoutTask.cancel(false); + } + hasRead = true; + } + + private void scheduleTimeoutTask(Duration timeout) { + if (timeout.equals(Duration.ZERO) || timeout.isNegative()) { + return; + } + + long timeoutInMillis = timeout.toMillis(); + timeoutTask = TimeoutScheduler.INSTANCE.schedule(() -> { + if (!hasRead) { + log.debug(() -> String.format("InputStream was not read before timeout of [%d] milliseconds, aborting " + + "stream and closing connection.", timeoutInMillis)); + abort(); + } + }, timeoutInMillis, TimeUnit.MILLISECONDS); + } + + private static final class TimeoutScheduler { + static final ScheduledExecutorService INSTANCE = + Executors.newScheduledThreadPool(1, r -> { + Thread t = new Thread(r, "response-input-stream-timeout-scheduler"); + t.setDaemon(true); + return t; + }); + } + /** * Close the underlying connection, dropping all remaining data in the stream, and not leaving the * connection open to be used for future requests. */ @Override public void abort() { + if (timeoutTask != null) { + timeoutTask.cancel(false); + } if (abortable != null) { abortable.abort(); } - IoUtils.closeQuietly(in, null); + IoUtils.closeQuietlyV2(in, log); + } + + @SdkTestInternalApi + public boolean hasTimeoutTask() { + return timeoutTask != null; + } + + @SdkTestInternalApi + public boolean timeoutTaskDoneOrCancelled() { + return timeoutTask != null && timeoutTask.isDone(); } } diff --git a/core/sdk-core/src/main/java/software/amazon/awssdk/core/async/AsyncResponseTransformer.java b/core/sdk-core/src/main/java/software/amazon/awssdk/core/async/AsyncResponseTransformer.java index d7c872d89289..a7abf157a628 100644 --- a/core/sdk-core/src/main/java/software/amazon/awssdk/core/async/AsyncResponseTransformer.java +++ b/core/sdk-core/src/main/java/software/amazon/awssdk/core/async/AsyncResponseTransformer.java @@ -19,6 +19,7 @@ import java.io.InputStream; import java.nio.ByteBuffer; import java.nio.file.Path; +import java.time.Duration; import java.util.Map; import java.util.concurrent.CompletableFuture; import java.util.function.Consumer; @@ -271,6 +272,10 @@ static AsyncResponseTransformer> * other transformers, like {@link #toFile(Path)} and {@link #toBytes()}, which only have their {@link CompletableFuture} * completed after the entire response body has finished streaming. *

+ * The publisher has a default timeout of 60 seconds that starts when the response body begins streaming. If no subscriber is + * registered within this time, the subscription will be automatically cancelled. Use {@link #toPublisher(Duration)} to + * specify a custom timeout. + *

* You are responsible for subscribing to this publisher and managing the associated back-pressure. Therefore, this * transformer is only recommended for advanced use cases. *

@@ -293,6 +298,34 @@ static AsyncResponseTransformer(); } + /** + * Creates an {@link AsyncResponseTransformer} with a custom timeout that publishes the response body content through a + * {@link ResponsePublisher}, which is an {@link SdkPublisher} that also contains a reference to the {@link SdkResponse} + * returned by the service. + *

+ * When this transformer is used with an async client, the {@link CompletableFuture} that the client returns will be completed + * once the {@link SdkResponse} is available and the response body begins streaming. This behavior differs from some + * other transformers, like {@link #toFile(Path)} and {@link #toBytes()}, which only have their {@link CompletableFuture} + * completed after the entire response body has finished streaming. + *

+ * The timeout starts when the response body begins streaming. If no subscriber is registered within the specified timeout, + * the subscription will be automatically cancelled. To disable the timeout, pass {@link Duration#ZERO} or a negative + * {@link Duration}. + *

+ * You are responsible for subscribing to this publisher and managing the associated back-pressure. Therefore, this + * transformer is only recommended for advanced use cases. + * + * @param timeout Maximum time to wait for subscription before cancelling. Use {@link Duration#ZERO} or a negative + * {@link Duration} to disable timeout. + * @param Pojo response type. + * @return AsyncResponseTransformer instance. + * @see #toPublisher() + */ + static AsyncResponseTransformer> toPublisher(Duration timeout) { + return new PublisherAsyncResponseTransformer<>(timeout); + } + /** * Creates an {@link AsyncResponseTransformer} that allows reading the response body content as an {@link InputStream}. *

diff --git a/core/sdk-core/src/main/java/software/amazon/awssdk/core/async/ResponsePublisher.java b/core/sdk-core/src/main/java/software/amazon/awssdk/core/async/ResponsePublisher.java index d8f87899f785..bb44b4568f16 100644 --- a/core/sdk-core/src/main/java/software/amazon/awssdk/core/async/ResponsePublisher.java +++ b/core/sdk-core/src/main/java/software/amazon/awssdk/core/async/ResponsePublisher.java @@ -16,16 +16,31 @@ package software.amazon.awssdk.core.async; import java.nio.ByteBuffer; +import java.time.Duration; import java.util.Objects; +import java.util.concurrent.Executors; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.ScheduledFuture; +import java.util.concurrent.TimeUnit; import org.reactivestreams.Subscriber; +import org.reactivestreams.Subscription; import software.amazon.awssdk.annotations.SdkPublicApi; +import software.amazon.awssdk.annotations.SdkTestInternalApi; import software.amazon.awssdk.core.SdkResponse; +import software.amazon.awssdk.utils.Logger; import software.amazon.awssdk.utils.ToString; import software.amazon.awssdk.utils.Validate; /** * An {@link SdkPublisher} that publishes response body content and also contains a reference to the {@link SdkResponse} returned * by the service. + *

+ * NOTE: You must subscribe to this publisher promptly to avoid automatic cancellation. The default timeout for + * subscribing is 60 seconds, which starts when the response body begins streaming. If {@link #subscribe(Subscriber)} is not + * invoked before the timeout, the publisher will automatically cancel the underlying subscription to prevent resource leakage. + *

+ * The timeout can be customized by passing a {@link Duration} to the constructor, or disabled entirely by + * passing {@link Duration#ZERO} or a negative {@link Duration}. * * @param Pojo response type. * @see AsyncResponseTransformer#toPublisher() @@ -33,12 +48,23 @@ @SdkPublicApi public final class ResponsePublisher implements SdkPublisher { + private static final Logger log = Logger.loggerFor(ResponsePublisher.class); + private static final Duration DEFAULT_TIMEOUT = Duration.ofSeconds(60); private final ResponseT response; private final SdkPublisher publisher; + private ScheduledFuture timeoutTask; + private volatile boolean subscribed = false; public ResponsePublisher(ResponseT response, SdkPublisher publisher) { + this(response, publisher, null); + } + + public ResponsePublisher(ResponseT response, SdkPublisher publisher, Duration timeout) { this.response = Validate.paramNotNull(response, "response"); this.publisher = Validate.paramNotNull(publisher, "publisher"); + + Duration resolvedTimeout = timeout != null ? timeout : DEFAULT_TIMEOUT; + scheduleTimeoutTask(resolvedTimeout); } /** @@ -50,9 +76,59 @@ public ResponseT response() { @Override public void subscribe(Subscriber subscriber) { + subscribed = true; + if (timeoutTask != null) { + timeoutTask.cancel(false); + } + publisher.subscribe(subscriber); } + private void scheduleTimeoutTask(Duration timeout) { + if (timeout.equals(Duration.ZERO) || timeout.isNegative()) { + return; + } + + long timeoutInMillis = timeout.toMillis(); + timeoutTask = TimeoutScheduler.INSTANCE.schedule(() -> { + if (!subscribed) { + log.debug(() -> String.format("Publisher was not consumed before timeout of [%d] milliseconds, cancelling " + + "subscription and closing connection.", timeoutInMillis)); + + publisher.subscribe(new CancellingSubscriber()); + } + }, timeoutInMillis, TimeUnit.MILLISECONDS); + } + + private static final class TimeoutScheduler { + static final ScheduledExecutorService INSTANCE = + Executors.newScheduledThreadPool(1, r -> { + Thread t = new Thread(r, "response-publisher-timeout-scheduler"); + t.setDaemon(true); + return t; + }); + } + + private static class CancellingSubscriber implements Subscriber { + + @Override + public void onSubscribe(Subscription s) { + s.cancel(); + } + + @Override + public void onNext(ByteBuffer b) { + } + + @Override + public void onError(Throwable t) { + } + + @Override + public void onComplete() { + } + } + @Override public String toString() { return ToString.builder("ResponsePublisher") @@ -84,4 +160,14 @@ public int hashCode() { result = 31 * result + (publisher != null ? publisher.hashCode() : 0); return result; } + + @SdkTestInternalApi + public boolean hasTimeoutTask() { + return timeoutTask != null; + } + + @SdkTestInternalApi + public boolean timeoutTaskDoneOrCancelled() { + return timeoutTask != null && timeoutTask.isDone(); + } } diff --git a/core/sdk-core/src/main/java/software/amazon/awssdk/core/internal/async/PublisherAsyncResponseTransformer.java b/core/sdk-core/src/main/java/software/amazon/awssdk/core/internal/async/PublisherAsyncResponseTransformer.java index 3f0f9fa19fce..5acd685f7d99 100644 --- a/core/sdk-core/src/main/java/software/amazon/awssdk/core/internal/async/PublisherAsyncResponseTransformer.java +++ b/core/sdk-core/src/main/java/software/amazon/awssdk/core/internal/async/PublisherAsyncResponseTransformer.java @@ -16,6 +16,7 @@ package software.amazon.awssdk.core.internal.async; import java.nio.ByteBuffer; +import java.time.Duration; import java.util.concurrent.CompletableFuture; import software.amazon.awssdk.annotations.SdkInternalApi; import software.amazon.awssdk.core.SdkResponse; @@ -35,6 +36,14 @@ public final class PublisherAsyncResponseTransformer> future; private volatile ResponseT response; + private Duration timeout; + + public PublisherAsyncResponseTransformer() { + } + + public PublisherAsyncResponseTransformer(Duration timeout) { + this.timeout = timeout; + } @Override public CompletableFuture> prepare() { @@ -50,7 +59,7 @@ public void onResponse(ResponseT response) { @Override public void onStream(SdkPublisher publisher) { - future.complete(new ResponsePublisher<>(response, publisher)); + future.complete(new ResponsePublisher<>(response, publisher, timeout)); } @Override diff --git a/core/sdk-core/src/main/java/software/amazon/awssdk/core/sync/ResponseTransformer.java b/core/sdk-core/src/main/java/software/amazon/awssdk/core/sync/ResponseTransformer.java index f883b591829e..c4371f319ce2 100644 --- a/core/sdk-core/src/main/java/software/amazon/awssdk/core/sync/ResponseTransformer.java +++ b/core/sdk-core/src/main/java/software/amazon/awssdk/core/sync/ResponseTransformer.java @@ -26,6 +26,7 @@ import java.nio.file.Files; import java.nio.file.NoSuchFileException; import java.nio.file.Path; +import java.time.Duration; import java.util.Map; import software.amazon.awssdk.annotations.SdkProtectedApi; import software.amazon.awssdk.annotations.SdkPublicApi; @@ -233,11 +234,16 @@ public String name() { * be explicitly closed to release the connection. The unmarshalled response object can be obtained via the {@link * ResponseInputStream#response()} method. *

+ * The stream has a default timeout of 60 seconds that starts when the response stream is ready. If no read operation occurs + * within this time, the connection will be automatically aborted. Use {@link #toInputStream(Duration)} to specify a custom + * timeout. + *

* Note that the returned stream is not subject to the retry policy or timeout settings (except for socket timeout) * of the client. No retries will be performed in the event of a socket read failure or connection reset. * * @param Type of unmarshalled response POJO. * @return ResponseTransformer instance. + * @see #toInputStream(Duration) */ static ResponseTransformer> toInputStream() { return unmanaged(new ResponseTransformer>() { @@ -253,6 +259,34 @@ public String name() { }); } + /** + * Creates a response transformer that returns an unmanaged input stream with the response content and a custom timeout. + * This input stream must be explicitly closed to release the connection. + *

+ * The timeout starts when the response stream is ready. If no read operation occurs within the specified timeout, the + * connection will be automatically aborted. To disable the timeout, pass {@link Duration#ZERO} or a negative + * {@link Duration}. + * + * @param timeout Maximum time to wait for first read operation before aborting. Use {@link Duration#ZERO} or a negative + * {@link Duration} to disable timeout. + * @param Type of unmarshalled response POJO. + * @return ResponseTransformer instance. + * @see #toInputStream() + */ + static ResponseTransformer> toInputStream(Duration timeout) { + return unmanaged(new ResponseTransformer>() { + @Override + public ResponseInputStream transform(ResponseT response, AbortableInputStream inputStream) { + return new ResponseInputStream<>(response, inputStream, timeout); + } + + @Override + public String name() { + return TransformerType.STREAM.getName(); + } + }); + } + /** * Static helper method to create a response transformer that allows the connection to be left open. Useful for creating a * {@link ResponseTransformer} with a lambda or method reference rather than an anonymous inner class. diff --git a/core/sdk-core/src/test/java/software/amazon/awssdk/core/ResponseInputStreamTest.java b/core/sdk-core/src/test/java/software/amazon/awssdk/core/ResponseInputStreamTest.java index 79dbd44addf8..6710465a899d 100644 --- a/core/sdk-core/src/test/java/software/amazon/awssdk/core/ResponseInputStreamTest.java +++ b/core/sdk-core/src/test/java/software/amazon/awssdk/core/ResponseInputStreamTest.java @@ -15,64 +15,125 @@ package software.amazon.awssdk.core; +import static org.assertj.core.api.Assertions.assertThat; import static org.assertj.core.api.Assertions.assertThatCode; import static org.mockito.Mockito.never; +import static org.mockito.Mockito.verify; import java.io.IOException; import java.io.InputStream; +import java.time.Duration; +import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; import org.mockito.Mockito; import software.amazon.awssdk.http.Abortable; import software.amazon.awssdk.http.AbortableInputStream; class ResponseInputStreamTest { + + InputStream stream; + Abortable abortable; + AbortableInputStream abortableInputStream; + + @BeforeEach + public void setUp() throws Exception { + stream = Mockito.mock(InputStream.class); + abortable = Mockito.mock(Abortable.class); + abortableInputStream = AbortableInputStream.create(stream, abortable); + } + @Test void abort_withAbortable_closesUnderlyingStream() throws IOException { - InputStream stream = Mockito.mock(InputStream.class); - Abortable abortable = Mockito.mock(Abortable.class); - AbortableInputStream abortableInputStream = AbortableInputStream.create(stream, abortable); ResponseInputStream responseInputStream = new ResponseInputStream<>(new Object(), abortableInputStream); responseInputStream.abort(); - Mockito.verify(abortable).abort(); - Mockito.verify(stream).close(); + verify(abortable).abort(); + verify(stream).close(); + assertThat(responseInputStream.hasTimeoutTask()).isTrue(); + assertThat(responseInputStream.timeoutTaskDoneOrCancelled()).isTrue(); } @Test void failedClose_withinAbort_isIgnored() throws IOException { - InputStream stream = Mockito.mock(InputStream.class); - Abortable abortable = Mockito.mock(Abortable.class); - AbortableInputStream abortableInputStream = AbortableInputStream.create(stream, abortable); ResponseInputStream responseInputStream = new ResponseInputStream<>(new Object(), abortableInputStream); Mockito.doThrow(new IOException()).when(stream).close(); assertThatCode(responseInputStream::abort).doesNotThrowAnyException(); - Mockito.verify(abortable).abort(); - Mockito.verify(stream).close(); + verify(abortable).abort(); + verify(stream).close(); + assertThat(responseInputStream.hasTimeoutTask()).isTrue(); + assertThat(responseInputStream.timeoutTaskDoneOrCancelled()).isTrue(); } @Test void abort_withoutAbortable_closesUnderlyingStream() throws IOException { - InputStream stream = Mockito.mock(InputStream.class); ResponseInputStream responseInputStream = new ResponseInputStream<>(new Object(), stream); responseInputStream.abort(); - Mockito.verify(stream).close(); + verify(stream).close(); + assertThat(responseInputStream.hasTimeoutTask()).isTrue(); + assertThat(responseInputStream.timeoutTaskDoneOrCancelled()).isTrue(); } @Test void close_withAbortable_closesUnderlyingStream() throws IOException { - InputStream stream = Mockito.mock(InputStream.class); - Abortable abortable = Mockito.mock(Abortable.class); - AbortableInputStream abortableInputStream = AbortableInputStream.create(stream, abortable); ResponseInputStream responseInputStream = new ResponseInputStream<>(new Object(), abortableInputStream); responseInputStream.close(); - Mockito.verify(abortable, never()).abort(); - Mockito.verify(stream).close(); + verify(abortable, never()).abort(); + verify(stream).close(); + assertThat(responseInputStream.hasTimeoutTask()).isTrue(); + assertThat(responseInputStream.timeoutTaskDoneOrCancelled()).isFalse(); + } + + @Test + void customTimeout_noRead_abortsAfterTimeout() throws Exception { + ResponseInputStream responseInputStream = responseInputStream(Duration.ofSeconds(1)); + Thread.sleep(2000); + + verify(abortable).abort(); + verify(stream).close(); + assertThat(responseInputStream.hasTimeoutTask()).isTrue(); + assertThat(responseInputStream.timeoutTaskDoneOrCancelled()).isTrue(); + } + + @Test + void customTimouet_readBeforeTimeout_cancelsTimeout() throws Exception { + ResponseInputStream responseInputStream = responseInputStream(Duration.ofSeconds(1)); + responseInputStream.read(); + Thread.sleep(2000); + + verify(abortable, never()).abort(); + verify(stream).read(); + assertThat(responseInputStream.hasTimeoutTask()).isTrue(); + assertThat(responseInputStream.timeoutTaskDoneOrCancelled()).isTrue(); + } + + @Test + void zeroTimeout_disablesTimeout() throws Exception { + ResponseInputStream responseInputStream = responseInputStream(Duration.ZERO); + Thread.sleep(2000); + + verify(abortable, never()).abort(); + verify(stream, never()).close(); + assertThat(responseInputStream.hasTimeoutTask()).isFalse(); + } + + @Test + void negativeTimeout_disablesTimeout() throws Exception { + ResponseInputStream responseInputStream = responseInputStream(Duration.ofSeconds(-1)); + Thread.sleep(2000); + + verify(abortable, never()).abort(); + verify(stream, never()).close(); + assertThat(responseInputStream.hasTimeoutTask()).isFalse(); + } + + private ResponseInputStream responseInputStream(Duration timeout) { + return new ResponseInputStream<>(new Object(), abortableInputStream, timeout); } -} \ No newline at end of file +} diff --git a/core/sdk-core/src/test/java/software/amazon/awssdk/core/async/ResponsePublisherTest.java b/core/sdk-core/src/test/java/software/amazon/awssdk/core/async/ResponsePublisherTest.java index 0cb76b6dc78f..e1fe52cc4e45 100644 --- a/core/sdk-core/src/test/java/software/amazon/awssdk/core/async/ResponsePublisherTest.java +++ b/core/sdk-core/src/test/java/software/amazon/awssdk/core/async/ResponsePublisherTest.java @@ -15,15 +15,119 @@ package software.amazon.awssdk.core.async; +import static org.assertj.core.api.Assertions.assertThat; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; + +import java.nio.ByteBuffer; +import java.time.Duration; import nl.jqno.equalsverifier.EqualsVerifier; +import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; +import org.mockito.ArgumentCaptor; +import org.reactivestreams.Subscriber; +import org.reactivestreams.Subscription; +import software.amazon.awssdk.core.SdkResponse; class ResponsePublisherTest { + private SdkResponse response; + private SdkPublisher publisher; + private ResponsePublisher responsePublisher; + + @BeforeEach + public void setUp() throws Exception { + response = mock(SdkResponse.class); + publisher = mock(SdkPublisher.class); + } + + @Test + void defaultTimeout_noSubscriptionBeforeTimeout() throws Exception { + responsePublisher = new ResponsePublisher<>(response, publisher); + Thread.sleep(2000); + + verify(publisher, never()).subscribe(any(Subscriber.class)); + assertThat(responsePublisher.hasTimeoutTask()).isTrue(); + assertThat(responsePublisher.timeoutTaskDoneOrCancelled()).isFalse(); + } + + @Test + void customTimeout_noSubscription_cancelsSubscriptionAfterTimeout() throws Exception { + responsePublisher = responsePublisher(Duration.ofSeconds(1)); + Thread.sleep(2000); + + ArgumentCaptor> subscriberCaptor = ArgumentCaptor.forClass(Subscriber.class); + verify(publisher, times(1)).subscribe(subscriberCaptor.capture()); + + Subscriber cancellingSubscriber = subscriberCaptor.getValue(); + Subscription mockSubscription = mock(Subscription.class); + + cancellingSubscriber.onSubscribe(mockSubscription); + verify(mockSubscription, times(1)).cancel(); + + assertThat(responsePublisher.hasTimeoutTask()).isTrue(); + assertThat(responsePublisher.timeoutTaskDoneOrCancelled()).isTrue(); + } + + @Test + void customTimeout_subscribeBeforeTimeout_cancelsTimeoutTask() throws Exception { + responsePublisher = responsePublisher(Duration.ofSeconds(1)); + responsePublisher.subscribe(new TestSubscriber()); + Thread.sleep(2000); + + // Verify only one subscription occurred (test subscriber, not timeout cancellation) + verify(publisher, times(1)).subscribe(any(TestSubscriber.class)); + verify(publisher, times(1)).subscribe(any(Subscriber.class)); + assertThat(responsePublisher.hasTimeoutTask()).isTrue(); + assertThat(responsePublisher.timeoutTaskDoneOrCancelled()).isTrue(); + } + + @Test + void zeroTimeout_disablesTimeout() throws Exception { + responsePublisher = responsePublisher(Duration.ZERO); + Thread.sleep(2000); + + verify(publisher, never()).subscribe(any(Subscriber.class)); + assertThat(responsePublisher.hasTimeoutTask()).isFalse(); + } + + @Test + void negativeTimeout_disablesTimeout() throws Exception { + responsePublisher = responsePublisher(Duration.ofSeconds(-1)); + Thread.sleep(2000); + + verify(publisher, never()).subscribe(any(Subscriber.class)); + assertThat(responsePublisher.hasTimeoutTask()).isFalse(); + } + + private ResponsePublisher responsePublisher(Duration timeout) { + return new ResponsePublisher<>(response, publisher, timeout); + } + + private static class TestSubscriber implements Subscriber { + @Override + public void onSubscribe(Subscription s) { + s.request(Long.MAX_VALUE); + } + + @Override + public void onNext(ByteBuffer byteBuffer) {} + + @Override + public void onError(Throwable t) {} + + @Override + public void onComplete() {} + } + @Test void equalsAndHashcode() { EqualsVerifier.forClass(ResponsePublisher.class) .withNonnullFields("response", "publisher") + .withIgnoredFields("timeoutTask", "subscribed") .verify(); } } \ No newline at end of file diff --git a/services/s3/pom.xml b/services/s3/pom.xml index 2521f42936a3..425e6d8facc5 100644 --- a/services/s3/pom.xml +++ b/services/s3/pom.xml @@ -196,6 +196,11 @@ netty-transport test + + org.apache.httpcomponents + httpclient + test + software.amazon.eventstream eventstream diff --git a/services/s3/src/it/java/software/amazon/awssdk/services/s3/ResponseInputStreamTimeoutIntegrationTest.java b/services/s3/src/it/java/software/amazon/awssdk/services/s3/ResponseInputStreamTimeoutIntegrationTest.java new file mode 100644 index 000000000000..2e07849567ab --- /dev/null +++ b/services/s3/src/it/java/software/amazon/awssdk/services/s3/ResponseInputStreamTimeoutIntegrationTest.java @@ -0,0 +1,117 @@ +/* + * 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.services.s3; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; +import static software.amazon.awssdk.testutils.service.S3BucketUtils.temporaryBucketName; + +import java.io.IOException; +import java.time.Duration; +import org.apache.http.conn.ConnectionPoolTimeoutException; +import org.junit.AfterClass; +import org.junit.Before; +import org.junit.BeforeClass; +import org.junit.Test; +import software.amazon.awssdk.core.ResponseInputStream; +import software.amazon.awssdk.core.sync.RequestBody; +import software.amazon.awssdk.core.sync.ResponseTransformer; +import software.amazon.awssdk.http.apache.ApacheHttpClient; +import software.amazon.awssdk.services.s3.model.GetObjectRequest; +import software.amazon.awssdk.services.s3.model.GetObjectResponse; +import software.amazon.awssdk.services.s3.model.PutObjectRequest; + +public class ResponseInputStreamTimeoutIntegrationTest extends S3IntegrationTestBase { + + private static final String BUCKET = temporaryBucketName(GetObjectIntegrationTest.class); + private static final String KEY = "TestKey"; + private static final String CONTENT = "Hello"; + private static final GetObjectRequest getObjectRequest = GetObjectRequest.builder() + .bucket(BUCKET) + .key(KEY) + .build(); + private S3Client s3Client; + + @Before + public void init() { + s3Client = s3ClientBuilder() + .httpClientBuilder(ApacheHttpClient.builder().maxConnections(1)) + .overrideConfiguration(o -> o.retryStrategy(r -> r.maxAttempts(1))) + .build(); + } + + @BeforeClass + public static void setupFixture() throws IOException { + createBucket(BUCKET); + s3.putObject(PutObjectRequest.builder() + .bucket(BUCKET) + .key(KEY) + .build(), RequestBody.fromString(CONTENT)); + + + } + + @AfterClass + public static void tearDownFixture() { + deleteBucketAndAllContents(BUCKET); + } + + @Test + public void defaultTimeout_firstStreamNotConsumed_secondRequestTimesOut() { + s3Client.getObject(getObjectRequest); + + assertThatThrownBy(() -> s3Client.getObject(getObjectRequest)) + .hasRootCauseInstanceOf(ConnectionPoolTimeoutException.class) + .hasMessageContaining("Timeout waiting for connection from pool"); + } + + @Test + public void defaultTimeout_firstStreamConsumed_secondRequestSucceeds() throws IOException { + ResponseInputStream get1 = s3Client.getObject(getObjectRequest); + byte[] buf = new byte[CONTENT.length()]; + get1.read(buf); + + ResponseInputStream get2 = s3Client.getObject(getObjectRequest); + assertThat(get2.response().contentLength()).isEqualTo(CONTENT.length()); + } + + @Test + public void defaultTimeout_firstStreamAborted_secondRequestSucceeds() throws IOException { + ResponseInputStream get1 = s3Client.getObject(getObjectRequest); + get1.abort(); + + ResponseInputStream get2 = s3Client.getObject(getObjectRequest); + assertThat(get2.response().contentLength()).isEqualTo(CONTENT.length()); + } + + @Test + public void defaultTimeout_firstStreamClosed_secondRequestSucceeds() throws IOException { + ResponseInputStream get1 = s3Client.getObject(getObjectRequest); + get1.close(); + + ResponseInputStream get2 = s3Client.getObject(getObjectRequest); + assertThat(get2.response().contentLength()).isEqualTo(CONTENT.length()); + } + + @Test + public void customTimeout_waitForTimeout_secondRequestSucceeds() throws InterruptedException { + s3Client.getObject(getObjectRequest, ResponseTransformer.toInputStream(Duration.ofSeconds(2))); + Thread.sleep(3000); + + ResponseInputStream get2 = s3Client.getObject(getObjectRequest); + assertThat(get2.response().contentLength()).isEqualTo(CONTENT.length()); + } +} diff --git a/services/s3/src/it/java/software/amazon/awssdk/services/s3/ResponsePublisherTimeoutIntegrationTest.java b/services/s3/src/it/java/software/amazon/awssdk/services/s3/ResponsePublisherTimeoutIntegrationTest.java new file mode 100644 index 000000000000..cabd234977bd --- /dev/null +++ b/services/s3/src/it/java/software/amazon/awssdk/services/s3/ResponsePublisherTimeoutIntegrationTest.java @@ -0,0 +1,134 @@ +/* + * 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.services.s3; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; +import static software.amazon.awssdk.testutils.service.S3BucketUtils.temporaryBucketName; + +import java.io.IOException; +import java.nio.ByteBuffer; +import java.time.Duration; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.TimeoutException; +import org.junit.AfterClass; +import org.junit.Before; +import org.junit.BeforeClass; +import org.junit.Test; +import org.reactivestreams.Publisher; +import org.reactivestreams.Subscriber; +import org.reactivestreams.Subscription; +import software.amazon.awssdk.core.async.AsyncResponseTransformer; +import software.amazon.awssdk.core.async.ResponsePublisher; +import software.amazon.awssdk.core.sync.RequestBody; +import software.amazon.awssdk.http.nio.netty.NettyNioAsyncHttpClient; +import software.amazon.awssdk.services.s3.model.GetObjectRequest; +import software.amazon.awssdk.services.s3.model.GetObjectResponse; +import software.amazon.awssdk.services.s3.model.PutObjectRequest; + +public class ResponsePublisherTimeoutIntegrationTest extends S3IntegrationTestBase { + + private static final String BUCKET = temporaryBucketName(GetObjectIntegrationTest.class); + private static final String KEY = "TestKey"; + private static final String CONTENT = "Hello"; + private static final GetObjectRequest getObjectRequest = GetObjectRequest.builder() + .bucket(BUCKET) + .key(KEY) + .build(); + private S3AsyncClient s3AsyncClient; + + @Before + public void init() { + s3AsyncClient = s3AsyncClientBuilder().httpClientBuilder(NettyNioAsyncHttpClient.builder().maxConcurrency(1)).build(); + } + + @BeforeClass + public static void setupFixture() throws IOException { + createBucket(BUCKET); + s3.putObject(PutObjectRequest.builder() + .bucket(BUCKET) + .key(KEY) + .build(), RequestBody.fromString(CONTENT)); + } + + @AfterClass + public static void tearDownFixture() { + deleteBucketAndAllContents(BUCKET); + } + + @Test + public void defaultTimeout_firstPublisherNotConsumed_secondRequestTimesOut() { + getObjectWithDefaultTimeoutPublisher(); + CompletableFuture> get2 = getObjectWithDefaultTimeoutPublisher(); + + assertThatThrownBy(get2::join).hasRootCauseInstanceOf(TimeoutException.class) + .hasMessageContaining("Acquire operation took longer than the configured maximum time. " + + "This indicates that a request cannot get a connection from the " + + "pool within the specified maximum time."); + } + + @Test + public void defaultTimeout_firstPublisherConsumed_secondRequestSucceeds() { + CompletableFuture> get1 = getObjectWithDefaultTimeoutPublisher(); + consumeResponsePublisher(get1.join()); + CompletableFuture> get2 = getObjectWithDefaultTimeoutPublisher(); + + GetObjectResponse getObjectResponse = get2.join().response(); + assertThat(getObjectResponse.contentLength()).isEqualTo(CONTENT.length()); + } + + @Test + public void defaultTimeout_cancelFirstRequestFuture_secondRequestSucceeds() { + CompletableFuture> get1 = getObjectWithDefaultTimeoutPublisher(); + get1.cancel(true); + CompletableFuture> get2 = getObjectWithDefaultTimeoutPublisher(); + + GetObjectResponse getObjectResponse = get2.join().response(); + assertThat(getObjectResponse.contentLength()).isEqualTo(CONTENT.length()); + } + + @Test + public void customTimeout_waitForTimeout_secondRequestSucceeds() throws InterruptedException { + s3AsyncClient.getObject(getObjectRequest, AsyncResponseTransformer.toPublisher(Duration.ofSeconds(2))); + Thread.sleep(3000); + CompletableFuture> get2 = getObjectWithDefaultTimeoutPublisher(); + + GetObjectResponse getObjectResponse = get2.join().response(); + assertThat(getObjectResponse.contentLength()).isEqualTo(CONTENT.length()); + } + + private CompletableFuture> getObjectWithDefaultTimeoutPublisher() { + return s3AsyncClient.getObject(getObjectRequest, AsyncResponseTransformer.toPublisher()); + } + + private void consumeResponsePublisher(Publisher responsePublisher) { + responsePublisher.subscribe(new Subscriber() { + @Override + public void onSubscribe(Subscription subscription) { + subscription.request(Long.MAX_VALUE); + } + @Override + public void onNext(ByteBuffer byteBuffer) {} + + @Override + public void onError(Throwable throwable) {} + + @Override + public void onComplete() { + } + }); + } +}