Check if stream is closed before trying to increment window. (#4833)

Author: zoewangg

.changes/next-release/bugfix-AWSCRTHTTPClient-7e448e2.json

@@ -0,0 +1,6 @@
+{
+    "type": "bugfix",
+    "category": "AWS CRT HTTP Client",
+    "contributor": "",
+    "description": "Fixed the issue in the AWS CRT HTTP client where the application could crash if stream.incrementWindow was invoked on a closed stream"
+}

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

@@ -95,7 +95,9 @@ public int onResponseBody(HttpStream stream, byte[] bodyBytesIn) {
                 return;
             }
 
-            stream.incrementWindow(bodyBytesIn.length);
+            if (!responseHandlerHelper.connectionClosed().get()) {
+                stream.incrementWindow(bodyBytesIn.length);
+            }
         });
 
         return 0;

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

@@ -92,8 +92,10 @@ public int onResponseBody(HttpStream stream, byte[] bodyBytesIn) {
                 return;
             }
 
-            // increment the window upon buffer consumption.
-            stream.incrementWindow(bodyBytesIn.length);
+            if (!responseHandlerHelper.connectionClosed().get()) {
+                // increment the window upon buffer consumption.
+                stream.incrementWindow(bodyBytesIn.length);
+            }
         });
 
         // Window will be incremented after the subscriber consumes the data, returning 0 here to disable it.

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

@@ -82,4 +82,8 @@ public void cleanUpConnectionBasedOnStatusCode(HttpStream stream) {
             releaseConnection(stream);
         }
     }
+
+    public AtomicBoolean connectionClosed() {
+        return connectionClosed;
+    }
 }

http-clients/aws-crt-client/src/test/java/software/amazon/awssdk/http/crt/internal/BaseHttpStreamResponseHandlerTest.java

@@ -16,9 +16,11 @@
 package software.amazon.awssdk.http.crt.internal;
 
 import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.mockito.ArgumentMatchers.anyInt;
 import static org.mockito.Mockito.never;
 import static org.mockito.Mockito.verify;
 
+import java.nio.charset.StandardCharsets;
 import java.util.concurrent.CompletableFuture;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
@@ -95,6 +97,22 @@ void failedToGetResponse_shouldShutdownConnection() {
         verify(httpStream).close();
     }
 
+    @Test
+    void streamClosed_shouldNotIncreaseStreamWindow() throws InterruptedException {
+        HttpHeader[] httpHeaders = getHttpHeaders();
+        responseHandler.onResponseHeaders(httpStream, 500, HttpHeaderBlock.MAIN.getValue(),
+                                          httpHeaders);
+        responseHandler.onResponseHeadersDone(httpStream, 0);
+        responseHandler.onResponseBody(httpStream, "{}".getBytes(StandardCharsets.UTF_8));
+
+        responseHandler.onResponseComplete(httpStream, 0);
+        requestFuture.join();
+        verify(crtConn).shutdown();
+        verify(crtConn).close();
+        verify(httpStream).close();
+        verify(httpStream, never()).incrementWindow(anyInt());
+    }
+
     private static HttpHeader[] getHttpHeaders() {
         HttpHeader[] httpHeaders = new HttpHeader[1];
         httpHeaders[0] = new HttpHeader("Content-Length", "1");

pom.xml

@@ -119,7 +119,7 @@
         <rxjava.version>2.2.21</rxjava.version>
         <commons-codec.verion>1.15</commons-codec.verion>
         <jmh.version>1.29</jmh.version>
-        <awscrt.version>0.29.2</awscrt.version>
+        <awscrt.version>0.29.7</awscrt.version>
 
         <!--Test dependencies -->
         <junit5.version>5.10.0</junit5.version>