Fixed the issue where AWS CRT-based S3 client was eagerly buffering data before the underlying CRT component was able to handle it. (#3800)

Author: zoewangg

.changes/next-release/bugfix-AWSCRTbasedS3client-7316932.json

@@ -0,0 +1,6 @@
+{
+    "type": "bugfix",
+    "category": "AWS CRT-based S3 client",
+    "contributor": "",
+    "description": "Fixed the issue where AWS CRT-based S3 client was eagerly buffering data before the underlying CRT component was able to handle it. See [#3726](https://github.com/aws/aws-sdk-java-v2/issues/3726)"
+}

.changes/next-release/bugfix-AWSCRTbasedS3client-89a9c8e.json

@@ -0,0 +1,6 @@
+{
+    "type": "bugfix",
+    "category": "AWS CRT-based S3 client",
+    "contributor": "",
+    "description": "Reduced the buffer used to upload object from 16MB to 1MB."
+}

services/s3/src/main/java/software/amazon/awssdk/services/s3/internal/crt/S3CrtRequestBodyStreamAdapter.java

@@ -16,6 +16,7 @@
 package software.amazon.awssdk.services.s3.internal.crt;
 
 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,19 +27,23 @@
  */
 @SdkInternalApi
 public final class S3CrtRequestBodyStreamAdapter implements HttpRequestBodyStream {
-    private static final long MINIMUM_BYTES_BUFFERED = 16 * 1024 * 1024L;
+    private static final long MINIMUM_BYTES_BUFFERED = 1024 * 1024L;
     private final SdkHttpContentPublisher bodyPublisher;
     private final ByteBufferStoringSubscriber requestBodySubscriber;
 
+    private final AtomicBoolean subscribed = new AtomicBoolean(false);
 
     public S3CrtRequestBodyStreamAdapter(SdkHttpContentPublisher bodyPublisher) {
         this.bodyPublisher = bodyPublisher;
         this.requestBodySubscriber = new ByteBufferStoringSubscriber(MINIMUM_BYTES_BUFFERED);
-        bodyPublisher.subscribe(requestBodySubscriber);
     }
 
     @Override
     public boolean sendRequestBody(ByteBuffer outBuffer) {
+        if (subscribed.compareAndSet(false, true)) {
+            bodyPublisher.subscribe(requestBodySubscriber);
+        }
+
         // blocking here because CRT S3 requires the buffer to be completely filled
         return requestBodySubscriber.blockingTransferTo(outBuffer) == ByteBufferStoringSubscriber.TransferResult.END_OF_STREAM;
     }