Fixed the issue where invoking abort on AsyncResponseTransformer.toBl…

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

@@ -22,7 +22,7 @@
 import software.amazon.awssdk.core.SdkResponse;
 import software.amazon.awssdk.core.async.AsyncResponseTransformer;
 import software.amazon.awssdk.core.async.SdkPublisher;
-import software.amazon.awssdk.utils.async.InputStreamSubscriber;
+import software.amazon.awssdk.http.async.AbortableInputStreamSubscriber;
 
 /**
  * A {@link AsyncResponseTransformer} that allows performing blocking reads on the response data.
@@ -50,7 +50,7 @@ public void onResponse(ResponseT response) {
 
     @Override
     public void onStream(SdkPublisher<ByteBuffer> publisher) {
-        InputStreamSubscriber inputStreamSubscriber = new InputStreamSubscriber();
+        AbortableInputStreamSubscriber inputStreamSubscriber = AbortableInputStreamSubscriber.builder().build();
         publisher.subscribe(inputStreamSubscriber);
         future.complete(new ResponseInputStream<>(response, inputStreamSubscriber));
     }

http-client-spi/pom.xml

@@ -90,6 +90,16 @@
             <artifactId>byte-buddy</artifactId>
             <scope>test</scope>
         </dependency>
+        <dependency>
+            <groupId>org.mockito</groupId>
+            <artifactId>mockito-junit-jupiter</artifactId>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.mockito</groupId>
+            <artifactId>mockito-inline</artifactId>
+            <scope>test</scope>
+        </dependency>
     </dependencies>
 
     <build>

http-clients/aws-crt-client/src/main/java/software/amazon/awssdk/http/crt/internal/response/AbortableInputStreamSubscriber.java → http-client-spi/src/main/java/software/amazon/awssdk/http/async/AbortableInputStreamSubscriber.java

@@ -13,35 +13,49 @@
  * permissions and limitations under the License.
  */
 
-package software.amazon.awssdk.http.crt.internal.response;
+package software.amazon.awssdk.http.async;
 
 import java.io.IOException;
 import java.io.InputStream;
 import java.nio.ByteBuffer;
 import org.reactivestreams.Subscriber;
 import org.reactivestreams.Subscription;
-import software.amazon.awssdk.annotations.SdkInternalApi;
+import software.amazon.awssdk.annotations.SdkProtectedApi;
+import software.amazon.awssdk.annotations.SdkTestInternalApi;
 import software.amazon.awssdk.http.Abortable;
+import software.amazon.awssdk.utils.FunctionalUtils;
 import software.amazon.awssdk.utils.async.InputStreamSubscriber;
 
 /**
- * Wrapper of {@link InputStreamSubscriber} that also implements {@link Abortable} and closes the underlying connections when
- * {@link #close()} or {@link #abort()} is invoked.
+ * Wrapper of {@link InputStreamSubscriber} that also implements {@link Abortable}. By default, it will invoke {@link #close()}
+ * when {@link #abort()} is invoked. This behavior can be customized via {@link Builder#doAfterAbort(Runnable)}.
+ *
  */
-@SdkInternalApi
+@SdkProtectedApi
 public final class AbortableInputStreamSubscriber extends InputStream implements Subscriber<ByteBuffer>, Abortable {
-
     private final InputStreamSubscriber delegate;
-    private final Runnable closeConnection;
 
-    public AbortableInputStreamSubscriber(Runnable onClose, InputStreamSubscriber inputStreamSubscriber) {
-        this.delegate = inputStreamSubscriber;
-        this.closeConnection = onClose;
+    private final Runnable doAfterClose;
+    private final Runnable doAfterAbort;
+
+    private AbortableInputStreamSubscriber(Builder builder) {
+        this(builder, new InputStreamSubscriber());
+    }
+
+    @SdkTestInternalApi
+    AbortableInputStreamSubscriber(Builder builder, InputStreamSubscriber delegate) {
+        this.delegate = delegate;
+        this.doAfterClose = builder.doAfterClose == null ? FunctionalUtils.noOpRunnable() : builder.doAfterClose;
+        this.doAfterAbort = builder.doAfterAbort == null ? this::close : builder.doAfterAbort;
+    }
+
+    public static Builder builder() {
+        return new Builder();
     }
 
     @Override
     public void abort() {
-        close();
+        FunctionalUtils.invokeSafely(() -> doAfterAbort.run());
     }
 
     @Override
@@ -81,7 +95,27 @@ public void onComplete() {
 
     @Override
     public void close() {
-        closeConnection.run();
         delegate.close();
+        FunctionalUtils.invokeSafely(() -> doAfterClose.run());
+    }
+
+    public static final class Builder {
+        private Runnable doAfterClose;
+        private Runnable doAfterAbort;
+
+
+        public Builder doAfterClose(Runnable doAfterClose) {
+            this.doAfterClose = doAfterClose;
+            return this;
+        }
+
+        public Builder doAfterAbort(Runnable doAfterAbort) {
+            this.doAfterAbort = doAfterAbort;
+            return this;
+        }
+
+        public AbortableInputStreamSubscriber build() {
+            return new AbortableInputStreamSubscriber(this);
+        }
     }
 }

http-client-spi/src/test/java/software/amazon/awssdk/http/async/AbortableInputStreamSubscriberTest.java

@@ -0,0 +1,80 @@
+/*
+ * 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.http.async;
+
+import static org.mockito.Mockito.verify;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.Mock;
+import org.mockito.junit.jupiter.MockitoExtension;
+import software.amazon.awssdk.utils.async.InputStreamSubscriber;
+
+@ExtendWith(MockitoExtension.class)
+public class AbortableInputStreamSubscriberTest {
+
+    private AbortableInputStreamSubscriber abortableInputStreamSubscriber;
+
+    @Mock
+    private Runnable onClose;
+
+    @Mock
+    private Runnable onAbort;
+
+    @Mock
+    private InputStreamSubscriber inputStreamSubscriber;
+
+    @BeforeEach
+    void setUp() {
+        abortableInputStreamSubscriber = new AbortableInputStreamSubscriber(AbortableInputStreamSubscriber.builder()
+                                                                                                          .doAfterClose(onClose)
+                                                                                                          .doAfterAbort(onAbort),
+                                                                            inputStreamSubscriber);
+
+
+    }
+
+    @Test
+    void close_closeConfigured_shouldInvokeOnClose() {
+        abortableInputStreamSubscriber.close();
+        verify(inputStreamSubscriber).close();
+        verify(onClose).run();
+    }
+
+    @Test
+    void abort_abortConfigured_shouldInvokeDoAfterAbort() {
+        abortableInputStreamSubscriber.abort();
+        verify(onAbort).run();
+    }
+
+    @Test
+    void abort_abortNotConfigured_shouldInvokeOnClose() {
+        abortableInputStreamSubscriber = new AbortableInputStreamSubscriber(AbortableInputStreamSubscriber.builder()
+                                                                                                          .doAfterClose(onClose),
+                                                                            inputStreamSubscriber);
+        abortableInputStreamSubscriber.abort();
+        verify(onClose).run();
+    }
+
+    @Test
+    void close_closeNotConfigured_shouldCloseDelegate() {
+        abortableInputStreamSubscriber = new AbortableInputStreamSubscriber(AbortableInputStreamSubscriber.builder(),
+                                                                            inputStreamSubscriber);
+        abortableInputStreamSubscriber.close();
+        verify(inputStreamSubscriber).close();
+    }
+}

http-clients/aws-crt-client/src/main/java/software/amazon/awssdk/http/crt/internal/response/InputStreamAdaptingHttpStreamResponseHandler.java

@@ -31,9 +31,9 @@
 import software.amazon.awssdk.http.AbortableInputStream;
 import software.amazon.awssdk.http.SdkHttpFullResponse;
 import software.amazon.awssdk.http.SdkHttpResponse;
+import software.amazon.awssdk.http.async.AbortableInputStreamSubscriber;
 import software.amazon.awssdk.http.crt.AwsCrtHttpClient;
 import software.amazon.awssdk.utils.Logger;
-import software.amazon.awssdk.utils.async.InputStreamSubscriber;
 import software.amazon.awssdk.utils.async.SimplePublisher;
 
 /**
@@ -87,8 +87,10 @@ public void onResponseHeaders(HttpStream stream, int responseStatusCode, int blo
     @Override
     public int onResponseBody(HttpStream stream, byte[] bodyBytesIn) {
         if (inputStreamSubscriber == null) {
-            inputStreamSubscriber = new AbortableInputStreamSubscriber(() -> responseHandlerHelper.closeConnection(stream),
-                                                                       new InputStreamSubscriber());
+            inputStreamSubscriber =
+                AbortableInputStreamSubscriber.builder()
+                                              .doAfterClose(() -> responseHandlerHelper.closeConnection(stream))
+                                              .build();
             simplePublisher.subscribe(inputStreamSubscriber);
             // For response with a payload, we need to complete the future here to allow downstream to retrieve the data from
             // the stream directly.

http-clients/netty-nio-client/src/main/java/software/amazon/awssdk/http/nio/netty/internal/ResponseHandler.java

@@ -314,7 +314,7 @@ private void onCancel() {
                     try {
                         SdkCancellationException e = new SdkCancellationException(
                                 "Subscriber cancelled before all events were published");
-                        log.warn(channelContext.channel(), () -> "Subscriber cancelled before all events were published");
+                        log.debug(channelContext.channel(), () -> "Subscriber cancelled before all events were published");
                         executeFuture.completeExceptionally(e);
                     } finally {
                         runAndLogError(channelContext.channel(), () -> "Could not release channel back to the pool",

test/codegen-generated-classes-test/pom.xml

@@ -235,6 +235,12 @@
             <version>${awsjavasdk.version}</version>
             <scope>test</scope>
         </dependency>
+        <dependency>
+            <groupId>software.amazon.awssdk</groupId>
+            <artifactId>netty-nio-client</artifactId>
+            <version>${awsjavasdk.version}</version>
+            <scope>test</scope>
+        </dependency>
         <dependency>
             <groupId>io.reactivex.rxjava2</groupId>
             <artifactId>rxjava</artifactId>