Fixed the issue where AWS CRT HTTP client was eagerly buffering data before the underlying CRT component was able to handle it and added error handling for the case where it failed to convert SDK request to CRT request (#6260)

Author: zoewangg

.changes/next-release/bugfix-AWSCRTAsyncHTTPClient-a045823.json

@@ -0,0 +1,6 @@
+{
+    "type": "bugfix",
+    "category": "AWS CRT Async HTTP Client",
+    "contributor": "",
+    "description": "Fixed potential connection leak issue when SDK failed to convert the SDK request to CRT request"
+}

.changes/next-release/bugfix-AWSCRTHTTPAsyncClient-88e9d07.json

@@ -0,0 +1,6 @@
+{
+    "type": "bugfix",
+    "category": "AWS CRT Async HTTP Client",
+    "contributor": "",
+    "description": "Fixed the issue where AWS CRT HTTP client was eagerly buffering data before the underlying CRT component was able to handle it"
+}

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

@@ -88,12 +88,12 @@ private void executeRequest(CrtAsyncRequestContext executionContext,
                                 CompletableFuture<Void> requestFuture,
                                 HttpClientConnection crtConn,
                                 AsyncExecuteRequest asyncRequest) {
-        HttpRequest crtRequest = CrtRequestAdapter.toAsyncCrtRequest(executionContext);
-        HttpStreamResponseHandler crtResponseHandler =
-            CrtResponseAdapter.toCrtResponseHandler(crtConn, requestFuture, asyncRequest.responseHandler());
-
         // Submit the request on the connection
         try {
+            HttpRequest crtRequest = CrtRequestAdapter.toAsyncCrtRequest(executionContext);
+            HttpStreamResponseHandler crtResponseHandler =
+                CrtResponseAdapter.toCrtResponseHandler(crtConn, requestFuture, asyncRequest.responseHandler());
+
             crtConn.makeRequest(crtRequest, crtResponseHandler).activate();
         } catch (HttpException e) {
             Throwable toThrow = wrapWithIoExceptionIfRetryable(e);

http-clients/aws-crt-client/src/main/java/software/amazon/awssdk/http/crt/internal/request/CrtRequestBodyAdapter.java

@@ -16,6 +16,7 @@
 package software.amazon.awssdk.http.crt.internal.request;
 
 import java.nio.ByteBuffer;
+import java.util.concurrent.atomic.AtomicBoolean;
 import software.amazon.awssdk.annotations.SdkInternalApi;
 import software.amazon.awssdk.crt.http.HttpRequestBodyStream;
 import software.amazon.awssdk.http.async.SdkHttpContentPublisher;
@@ -26,15 +27,19 @@
 final class CrtRequestBodyAdapter implements HttpRequestBodyStream {
     private final SdkHttpContentPublisher requestPublisher;
     private final ByteBufferStoringSubscriber requestBodySubscriber;
+    private final AtomicBoolean subscribed = new AtomicBoolean(false);
 
     CrtRequestBodyAdapter(SdkHttpContentPublisher requestPublisher, long readLimit) {
         this.requestPublisher = requestPublisher;
         this.requestBodySubscriber = new ByteBufferStoringSubscriber(readLimit);
-        requestPublisher.subscribe(requestBodySubscriber);
     }
 
     @Override
     public boolean sendRequestBody(ByteBuffer bodyBytesOut) {
+        if (subscribed.compareAndSet(false, true)) {
+            requestPublisher.subscribe(requestBodySubscriber);
+        }
+
         return requestBodySubscriber.transferTo(bodyBytesOut) == TransferResult.END_OF_STREAM;
     }
 

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

@@ -105,6 +105,29 @@ public void acquireConnectionThrowException_shouldInvokeOnError() {
         assertThat(executeFuture).hasFailedWithThrowableThat().hasCause(exception).isInstanceOf(IOException.class);
     }
 
+    @Test
+    public void invalidRequest_requestConversionThrowError_shouldInvokeOnError() {
+        CrtAsyncRequestContext context = CrtAsyncRequestContext.builder()
+                                                               .crtConnPool(connectionManager)
+                                                               .request(AsyncExecuteRequest.builder()
+                                                                                           .responseHandler(responseHandler)
+                                                                                           .build())
+                                                               .build();
+        CompletableFuture<HttpClientConnection> completableFuture = new CompletableFuture<>();
+
+        Mockito.when(connectionManager.acquireConnection()).thenReturn(completableFuture);
+        completableFuture.complete(httpClientConnection);
+
+        CompletableFuture<Void> executeFuture = requestExecutor.execute(context);
+
+        ArgumentCaptor<Exception> argumentCaptor = ArgumentCaptor.forClass(Exception.class);
+        Mockito.verify(responseHandler).onError(argumentCaptor.capture());
+
+        Exception actualException = argumentCaptor.getValue();
+        assertThat(actualException).isInstanceOf(NullPointerException.class);
+        assertThat(executeFuture).hasFailedWithThrowableThat().isInstanceOf(NullPointerException.class);
+    }
+
     @Test
     public void executeAsyncRequest_CrtRuntimeException_shouldInvokeOnError() {
         CrtRuntimeException exception = new CrtRuntimeException("");