Fixed a thread safety issue that could cause application to crash in …

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

@@ -0,0 +1,6 @@
+{
+    "type": "bugfix",
+    "category": "AWS CRT HTTP Client",
+    "contributor": "",
+    "description": "Fixed a thread safety issue that could cause application to crash in the edge case where the SDK attempted to invoke `incrementWindow` after the stream is closed in AWS CRT HTTP Client."
+}

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

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

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

@@ -101,11 +101,8 @@ public int onResponseBody(HttpStream stream, byte[] bodyBytesIn) {
                 failFutureAndCloseConnection(stream, failure);
                 return;
             }
-
-            if (!responseHandlerHelper.connectionClosed().get()) {
-                // increment the window upon buffer consumption.
-                stream.incrementWindow(bodyBytesIn.length);
-            }
+            // increment the window upon buffer consumption.
+            responseHandlerHelper.incrementWindow(stream, 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

@@ -30,14 +30,16 @@
  *
  * CRT connection will only be closed, i.e., not reused, in one of the following conditions:
  * 1. 5xx server error OR
- * 2. It fails to read the response.
+ * 2. It fails to read the response OR
+ * 3. the response stream is closed/aborted by the caller.
  */
 @SdkInternalApi
 public class ResponseHandlerHelper {
 
     private final SdkHttpResponse.Builder responseBuilder;
     private final HttpClientConnection connection;
-    private AtomicBoolean connectionClosed = new AtomicBoolean(false);
+    private boolean connectionClosed;
+    private final Object lock = new Object();
 
     public ResponseHandlerHelper(SdkHttpResponse.Builder responseBuilder, HttpClientConnection connection) {
         this.responseBuilder = responseBuilder;
@@ -57,20 +59,34 @@ public void onResponseHeaders(HttpStream stream, int responseStatusCode, int hea
      * Release the connection back to the pool so that it can be reused.
      */
     public void releaseConnection(HttpStream stream) {
-        if (connectionClosed.compareAndSet(false, true)) {
-            connection.close();
-            stream.close();
+        synchronized (lock) {
+            if (!connectionClosed) {
+                connectionClosed = true;
+                connection.close();
+                stream.close();
+            }
+        }
+    }
+
+    public void incrementWindow(HttpStream stream, int windowSize) {
+        synchronized (lock) {
+            if (!connectionClosed) {
+                stream.incrementWindow(windowSize);
+            }
         }
     }
 
     /**
      * Close the connection completely
      */
     public void closeConnection(HttpStream stream) {
-        if (connectionClosed.compareAndSet(false, true)) {
-            connection.shutdown();
-            connection.close();
-            stream.close();
+        synchronized (lock) {
+            if (!connectionClosed) {
+                connectionClosed = true;
+                connection.shutdown();
+                connection.close();
+                stream.close();
+            }
         }
     }
 
@@ -82,8 +98,4 @@ public void cleanUpConnectionBasedOnStatusCode(HttpStream stream) {
             releaseConnection(stream);
         }
     }
-
-    public AtomicBoolean connectionClosed() {
-        return connectionClosed;
-    }
 }