Fixed an issue that could cause exception to be thrown from CombinedResponseAsyncHttpResponseHandler#onStream when onError got invoked first (#3603)

Author: zoewangg

.changes/next-release/bugfix-AWSSDKforJavav2-6f2699a.json

@@ -0,0 +1,6 @@
+{
+    "type": "bugfix",
+    "category": "AWS SDK for Java v2",
+    "contributor": "",
+    "description": "Fixed an issue that could cause exception to be thrown from `CombinedResponseAsyncHttpResponseHandler#onStream` when `onError` got invoked first"
+}

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

@@ -75,6 +75,11 @@ public void onStream(Publisher<ByteBuffer> publisher) {
         Validate.isTrue(headersFuture != null, "onStream() invoked without prepare().");
         Validate.isTrue(headersFuture.isDone(), "headersFuture is still not completed when onStream() is "
                                                 + "invoked.");
+
+        if (headersFuture.isCompletedExceptionally()) {
+            return;
+        }
+
         SdkHttpResponse sdkHttpResponse = headersFuture.join();
         if (sdkHttpResponse.isSuccessful()) {
             successResponseHandler.onStream(publisher);

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

@@ -17,7 +17,10 @@
 
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.mockito.Mockito.times;
 import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.verifyNoInteractions;
+import static org.mockito.Mockito.verifyNoMoreInteractions;
 import static org.mockito.Mockito.when;
 
 import io.reactivex.Flowable;
@@ -72,6 +75,18 @@ void onStream_invokedWithoutOnHeaders_shouldThrowException() {
 
     }
 
+    @Test
+    void onStream_HeadersFutureCompleteSuccessfully_shouldNotThrowException() {
+        when(successResponseHandler.prepare()).thenReturn(CompletableFuture.completedFuture(null));
+
+        responseHandler.prepare();
+        responseHandler.onError(new RuntimeException("error"));
+        Flowable<ByteBuffer> publisher = publisher();
+        responseHandler.onStream(publisher);
+        verify(successResponseHandler, times(0)).onStream(publisher);
+        verify(errorResponseHandler, times(0)).onStream(publisher);
+    }
+
     @Test
     void successResponse_shouldCompleteHeaderFuture() {
         when(successResponseHandler.prepare()).thenReturn(CompletableFuture.completedFuture(null));