Fixed an issue where "Error thrown from TransformingAsyncResponseHandler#onError, ignoring" could be logged during an unrelated failure.

Fixes #1812

Author: millems

.changes/next-release/bugfix-AWSSDKforJavav2-ea88e81.json

@@ -0,0 +1,5 @@
+{
+    "type": "bugfix",
+    "category": "AWS SDK for Java v2",
+    "description": "Fixed an issue where \"Error thrown from TransformingAsyncResponseHandler#onError, ignoring\" could be logged during an unrelated failure."
+}

core/sdk-core/src/main/java/software/amazon/awssdk/core/internal/http/TransformingAsyncResponseHandler.java

@@ -16,7 +16,9 @@
 package software.amazon.awssdk.core.internal.http;
 
 import java.util.concurrent.CompletableFuture;
+import org.reactivestreams.Publisher;
 import software.amazon.awssdk.annotations.SdkInternalApi;
+import software.amazon.awssdk.http.SdkHttpResponse;
 import software.amazon.awssdk.http.async.SdkAsyncHttpResponseHandler;
 
 /**
@@ -29,8 +31,11 @@ public interface TransformingAsyncResponseHandler<ResultT> extends SdkAsyncHttpR
     /**
      * Return the future holding the transformed response.
      * <p>
-     * This method is guaranteed to be called before the request is executed, and before {@link
-     * SdkAsyncHttpResponseHandler#onHeaders(software.amazon.awssdk.http.SdkHttpResponse)} is signaled.
+     * This method is guaranteed to be called before the request is executed, before {@link
+     * #onHeaders(SdkHttpResponse)} and before {@link #onStream(Publisher)} is signaled.
+     * <p>
+     * Note: {@link #onError(Throwable)} *may* be invoked before this method is invoked, but it will
+     * never be invoked in parallel with this method.
      *
      * @return The future holding the transformed response.
      */

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

@@ -73,12 +73,26 @@ public void onStream(Publisher<ByteBuffer> publisher) {
 
     @Override
     public void onError(Throwable err) {
-        streamFuture.completeExceptionally(err);
+        // Method assumes onError will be never be executed concurrently with prepare.
+        if (streamFuture != null) {
+            streamFuture.completeExceptionally(err);
+        } else {
+            CompletableFuture<ByteArrayOutputStream> streamFuture = new CompletableFuture<>();
+            this.streamFuture = streamFuture;
+            streamFuture.completeExceptionally(err);
+        }
     }
 
     @Override
     public CompletableFuture<T> prepare() {
-        streamFuture = new CompletableFuture<>();
+        // Method assumes onError will be never be executed concurrently with prepare.
+        CompletableFuture<ByteArrayOutputStream> streamFuture = this.streamFuture;
+
+        if (streamFuture == null) {
+            streamFuture = new CompletableFuture<>();
+            this.streamFuture = streamFuture;
+        }
+
         return streamFuture.thenCompose(baos -> {
             ByteArrayInputStream content = new ByteArrayInputStream(baos.toByteArray());
             // Ignore aborts - we already have all of the content.

core/sdk-core/src/test/java/software/amazon/awssdk/core/internal/http/async/AsyncResponseHandlerTest.java

@@ -0,0 +1,31 @@
+package software.amazon.awssdk.core.internal.http.async;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import java.util.concurrent.CompletableFuture;
+import org.junit.Test;
+
+public class AsyncResponseHandlerTest {
+    @Test
+    public void onErrorAfterPrepareForwardsErrorToFuture() {
+        AsyncResponseHandler<Object> responseHandler = new AsyncResponseHandler<>(null, null, null);
+        CompletableFuture<Object> future = responseHandler.prepare();
+
+        assertThat(future).isNotCompleted();
+
+        Throwable t = new Throwable();
+        responseHandler.onError(t);
+
+        assertThat(future).hasFailedWithThrowableThat().isEqualTo(t);
+    }
+
+    @Test
+    public void onErrorBeforePrepareStoresExceptionForPrepare() {
+        AsyncResponseHandler<Object> responseHandler = new AsyncResponseHandler<>(null, null, null);
+
+        Throwable t = new Throwable();
+        responseHandler.onError(t);
+
+        assertThat(responseHandler.prepare()).hasFailedWithThrowableThat().isEqualTo(t);
+    }
+}