Address comments

Author: davidh44

core/sdk-core/src/main/java/software/amazon/awssdk/core/ResponseInputStream.java

@@ -37,10 +37,11 @@
  *
  * <p>
  * <b>NOTE:</b> You must read this stream promptly to avoid automatic cancellation. The default timeout for reading is 60
- * seconds. If {@link #read()} is not invoked before the timeout, the stream will automatically abort to prevent resource leakage.
+ * 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.
  * <p>
  * The timeout can be customized by passing a {@link Duration} to the constructor, or disabled entirely by
- * passing {@link Duration#ZERO}.
+ * passing {@link Duration#ZERO} or a negative {@link Duration}.
  * <p>
  * 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}
@@ -79,10 +80,16 @@ public ResponseInputStream(ResponseT resp, AbortableInputStream in, Duration tim
     }
 
     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;
-        scheduleTimeoutTask(DEFAULT_TIMEOUT);
+
+        Duration resolvedTimeout = timeout != null ? timeout : DEFAULT_TIMEOUT;
+        scheduleTimeoutTask(resolvedTimeout);
     }
 
     /**
@@ -111,14 +118,14 @@ public int read(byte[] b, int off, int len) throws IOException {
     }
 
     private void cancelTimeoutTask() {
-        hasRead = true;
-        if (timeoutTask != null) {
+        if (!hasRead && timeoutTask != null) {
             timeoutTask.cancel(false);
         }
+        hasRead = true;
     }
 
     private void scheduleTimeoutTask(Duration timeout) {
-        if (timeout.equals(Duration.ZERO)) {
+        if (timeout.equals(Duration.ZERO) || timeout.isNegative()) {
             return;
         }
 

core/sdk-core/src/main/java/software/amazon/awssdk/core/async/AsyncResponseTransformer.java

@@ -272,8 +272,9 @@ static <ResponseT> AsyncResponseTransformer<ResponseT, ResponseBytes<ResponseT>>
      * other transformers, like {@link #toFile(Path)} and {@link #toBytes()}, which only have their {@link CompletableFuture}
      * completed after the entire response body has finished streaming.
      * <p>
-     * The publisher has a default timeout of 60 seconds. If no subscriber is registered within this time, the subscription
-     * will be automatically cancelled. Use {@link #toPublisher(Duration)} to specify a custom timeout.
+     * 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.
      * <p>
      * You are responsible for subscribing to this publisher and managing the associated back-pressure. Therefore, this
      * transformer is only recommended for advanced use cases.
@@ -307,13 +308,15 @@ static <ResponseT extends SdkResponse> AsyncResponseTransformer<ResponseT, Respo
      * other transformers, like {@link #toFile(Path)} and {@link #toBytes()}, which only have their {@link CompletableFuture}
      * completed after the entire response body has finished streaming.
      * <p>
-     * If no subscriber is registered within the specified timeout, the subscription will be automatically cancelled. To
-     * disable the timeout, pass {@link Duration#ZERO}.
+     * 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}.
      * <p>
      * 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} to disable timeout.
+     * @param timeout Maximum time to wait for subscription before cancelling. Use {@link Duration#ZERO} or a negative
+     * {@link Duration} to disable timeout.
      * @param <ResponseT> Pojo response type.
      * @return AsyncResponseTransformer instance.
      * @see #toPublisher()

core/sdk-core/src/main/java/software/amazon/awssdk/core/async/ResponsePublisher.java

@@ -35,12 +35,12 @@
  * An {@link SdkPublisher} that publishes response body content and also contains a reference to the {@link SdkResponse} returned
  * by the service.
  * <p>
- * <b>NOTE:</b> You must subscribe to this publisher promptly to avoid automatic cancellation. The default timeout
- * for subscribing is 60 seconds. If {@link #subscribe(Subscriber)} is not invoked before the timeout, the publisher
- * will automatically cancel the underlying subscription to prevent resource leakage.
+ * <b>NOTE:</b> 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.
  * <p>
  * The timeout can be customized by passing a {@link Duration} to the constructor, or disabled entirely by
- * passing {@link Duration#ZERO}.
+ * passing {@link Duration#ZERO} or a negative {@link Duration}.
  *
  * @param <ResponseT> Pojo response type.
  * @see AsyncResponseTransformer#toPublisher()
@@ -85,7 +85,7 @@ public void subscribe(Subscriber<? super ByteBuffer> subscriber) {
     }
 
     private void scheduleTimeoutTask(Duration timeout) {
-        if (timeout.equals(Duration.ZERO)) {
+        if (timeout.equals(Duration.ZERO) ||  timeout.isNegative()) {
             return;
         }
 

core/sdk-core/src/main/java/software/amazon/awssdk/core/sync/ResponseTransformer.java

@@ -234,8 +234,9 @@ public String name() {
      * be explicitly closed to release the connection. The unmarshalled response object can be obtained via the {@link
      * ResponseInputStream#response()} method.
      * <p>
-     * The stream has a default timeout of 60 seconds. If no read operation occurs within this time, the connection
-     * will be automatically aborted. Use {@link #toInputStream(Duration)} to specify a custom timeout.
+     * 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.
      * <p>
      * 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.
@@ -262,10 +263,12 @@ 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.
      * <p>
-     * If no read operation occurs within the specified timeout, the connection will be automatically aborted.
-     * Pass {@link Duration#ZERO} to disable the timeout.
+     * 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} to disable timeout.
+     * @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 <ResponseT> Type of unmarshalled response POJO.
      * @return ResponseTransformer instance.
      * @see #toInputStream()

core/sdk-core/src/test/java/software/amazon/awssdk/core/ResponseInputStreamTest.java

@@ -123,6 +123,16 @@ void zeroTimeout_disablesTimeout() throws Exception {
         assertThat(responseInputStream.hasTimeoutTask()).isFalse();
     }
 
+    @Test
+    void negativeTimeout_disablesTimeout() throws Exception {
+        ResponseInputStream<Object> responseInputStream = responseInputStream(Duration.ofSeconds(-1));
+        Thread.sleep(2000);
+
+        verify(abortable, never()).abort();
+        verify(stream, never()).close();
+        assertThat(responseInputStream.hasTimeoutTask()).isFalse();
+    }
+
     private ResponseInputStream<Object> responseInputStream(Duration timeout) {
         return new ResponseInputStream<>(new Object(), abortableInputStream, timeout);
     }

core/sdk-core/src/test/java/software/amazon/awssdk/core/async/ResponsePublisherTest.java

@@ -94,6 +94,15 @@ void zeroTimeout_disablesTimeout() throws Exception {
         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<SdkResponse> responsePublisher(Duration timeout) {
         return new ResponsePublisher<>(response, publisher, timeout);
     }