Fix per async body buffer usage (#5939)

* Fix per async body buffer usage

`bufferPerAsyncRequestBody` represents the read buffer used by each
inflight `FileAsyncRequestBody`.

This commit fixes the calculation so that we account for the user
configured `bufferSizeInBytes` from the input `SplitConfiguration`. The
issue is that `asyncRequestBody.chunkSizeInBytes()` may be greater than
the desired `bufferSizeInBytes`.

This also fixes the creation of the `FileAsyncRequestBody` for each
chunk so that we configure `bufferPerAsyncRequestBody` for the
`chunkSizeInBytes`.

* Add changelog

Author: dagnir

.changes/next-release/bugfix-S3TransferManager-54510f5.json

@@ -0,0 +1,6 @@
+{
+    "type": "bugfix",
+    "category": "S3 Transfer Manager",
+    "contributor": "",
+    "description": "Fix an issue where `uploadFile` never completes for the Java based S3 TransferManager if the `apiCallBufferSizeInBytes` configured on the `MultipartConfiguration` is too small."
+}

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

@@ -28,6 +28,7 @@
 import software.amazon.awssdk.core.async.AsyncRequestBodySplitConfiguration;
 import software.amazon.awssdk.core.async.SdkPublisher;
 import software.amazon.awssdk.core.exception.SdkClientException;
+import software.amazon.awssdk.utils.NumericUtils;
 import software.amazon.awssdk.utils.Validate;
 import software.amazon.awssdk.utils.async.SimplePublisher;
 
@@ -67,7 +68,8 @@ public FileAsyncRequestBodySplitHelper(FileAsyncRequestBody asyncRequestBody,
         this.totalBufferSize = splitConfiguration.bufferSizeInBytes() == null ?
                                AsyncRequestBodySplitConfiguration.defaultConfiguration().bufferSizeInBytes() :
                                splitConfiguration.bufferSizeInBytes();
-        this.bufferPerAsyncRequestBody = asyncRequestBody.chunkSizeInBytes();
+        this.bufferPerAsyncRequestBody = Math.min(asyncRequestBody.chunkSizeInBytes(),
+                                                  NumericUtils.saturatedCast(totalBufferSize));
     }
 
     public SdkPublisher<AsyncRequestBody> split() {
@@ -131,6 +133,7 @@ private AsyncRequestBody newFileAsyncRequestBody(SimplePublisher<AsyncRequestBod
                                                                         .path(path)
                                                                         .position(position)
                                                                         .numBytesToRead(numBytesToReadForThisChunk)
+                                                                        .chunkSizeInBytes(bufferPerAsyncRequestBody)
                                                                         .build();
         return new FileAsyncRequestBodyWrapper(fileAsyncRequestBody, simplePublisher);
     }

services-custom/s3-transfer-manager/src/it/java/software/amazon/awssdk/transfer/s3/S3TransferManagerUploadIntegrationTest.java

@@ -24,15 +24,19 @@
 import java.util.HashMap;
 import java.util.Map;
 import java.util.concurrent.CancellationException;
+import java.util.concurrent.TimeUnit;
 import org.apache.commons.lang3.RandomStringUtils;
 import org.junit.jupiter.api.AfterAll;
 import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.Timeout;
 import org.junit.jupiter.params.ParameterizedTest;
 import org.junit.jupiter.params.provider.MethodSource;
 import software.amazon.awssdk.core.ResponseInputStream;
 import software.amazon.awssdk.core.async.AsyncRequestBody;
 import software.amazon.awssdk.core.internal.async.FileAsyncRequestBody;
 import software.amazon.awssdk.core.sync.ResponseTransformer;
+import software.amazon.awssdk.services.s3.S3AsyncClient;
 import software.amazon.awssdk.services.s3.model.ChecksumAlgorithm;
 import software.amazon.awssdk.services.s3.model.GetObjectResponse;
 import software.amazon.awssdk.testutils.RandomTempFile;
@@ -93,6 +97,34 @@ void upload_file_SentCorrectly(S3TransferManager tm) throws IOException {
         assertListenerForSuccessfulTransferComplete(transferListener);
     }
 
+    // This is a test for an issue where the file upload hangs (no chunk
+    // uploads are initiated) if apiCallBufferSizeInBytes is less than the
+    // publisher chunk size.
+    // Note: Only applicable to the Java based TM because file uploads are
+    // done completely by CRT for the CRT based transfer manager, and does
+    // not hit the same code path.
+    @Test
+    @Timeout(value = 10, unit = TimeUnit.MINUTES)
+    public void uploadFile_apiBufferSizeLessThanFileAsyncPublisherChunkSize_sentCorrectly() {
+        try (
+            S3AsyncClient s3Async = s3AsyncClientBuilder()
+                .multipartConfiguration(c -> c.apiCallBufferSizeInBytes(SizeConstant.KB))
+                .build();
+
+            S3TransferManager tm = S3TransferManager.builder()
+                                                    .s3Client(s3Async)
+                                                    .build();
+        ) {
+            FileUpload fileUpload =
+                tm.uploadFile(u -> u.putObjectRequest(p -> p.bucket(TEST_BUCKET)
+                                                            .key(TEST_KEY))
+                                    .source(testFile.toPath())
+                                    .build());
+
+            fileUpload.completionFuture().join();
+        }
+    }
+
     private static void assertListenerForSuccessfulTransferComplete(CaptureTransferListener transferListener) {
         assertThat(transferListener.isTransferInitiated()).isTrue();
         assertThat(transferListener.isTransferComplete()).isTrue();