Mulitpart download workaround

.changes/next-release/feature-AWSSDKforJavav2-787c575.json

@@ -0,0 +1,6 @@
+{
+    "type": "feature",
+    "category": "S3 Transfer Manager",
+    "contributor": "",
+    "description": "Make Transfer Manager work by default with S3AsyncClient when multipart configuration is enabled."
+}

.changes/next-release/feature-S3TransferManager-2e987ba.json

@@ -0,0 +1,6 @@
+{
+    "type": "feature",
+    "category": "S3 Transfer Manager",
+    "contributor": "",
+    "description": "Enable multipart configuration by default when creating a new S3TranferManager instance using the .create() method"
+}

services-custom/s3-transfer-manager/src/main/java/software/amazon/awssdk/transfer/s3/internal/GenericS3TransferManager.java

@@ -33,6 +33,7 @@
 import software.amazon.awssdk.core.exception.SdkClientException;
 import software.amazon.awssdk.core.exception.SdkException;
 import software.amazon.awssdk.core.internal.async.FileAsyncRequestBody;
+import software.amazon.awssdk.services.s3.DelegatingS3AsyncClient;
 import software.amazon.awssdk.services.s3.S3AsyncClient;
 import software.amazon.awssdk.services.s3.internal.multipart.MultipartS3AsyncClient;
 import software.amazon.awssdk.services.s3.internal.resource.S3AccessPointResource;
@@ -110,11 +111,11 @@ class GenericS3TransferManager implements S3TransferManager {
     }
 
     @SdkTestInternalApi
-    GenericS3TransferManager(S3AsyncClient s3CrtAsyncClient,
+    GenericS3TransferManager(S3AsyncClient s3AsyncClient,
                              UploadDirectoryHelper uploadDirectoryHelper,
                              TransferManagerConfiguration configuration,
                              DownloadDirectoryHelper downloadDirectoryHelper) {
-        this.s3AsyncClient = s3CrtAsyncClient;
+        this.s3AsyncClient = s3AsyncClient;
         this.isDefaultS3AsyncClient = false;
         this.transferConfiguration = configuration;
         this.uploadDirectoryHelper = uploadDirectoryHelper;
@@ -138,13 +139,13 @@ public Upload upload(UploadRequest uploadRequest) {
         try {
             assertNotUnsupportedArn(uploadRequest.putObjectRequest().bucket(), "upload");
 
-            CompletableFuture<PutObjectResponse> crtFuture =
+            CompletableFuture<PutObjectResponse> future =
                 s3AsyncClient.putObject(uploadRequest.putObjectRequest(), requestBody);
 
-            // Forward upload cancellation to CRT future
-            CompletableFutureUtils.forwardExceptionTo(returnFuture, crtFuture);
+            // Forward upload cancellation to future
+            CompletableFutureUtils.forwardExceptionTo(returnFuture, future);
 
-            CompletableFutureUtils.forwardTransformedResultTo(crtFuture, returnFuture,
+            CompletableFutureUtils.forwardTransformedResultTo(future, returnFuture,
                                                               r -> CompletedUpload.builder()
                                                                                   .response(r)
                                                                                   .build());
@@ -298,13 +299,12 @@ public <ResultT> Download<ResultT> download(DownloadRequest<ResultT> downloadReq
         try {
             assertNotUnsupportedArn(downloadRequest.getObjectRequest().bucket(), "download");
 
-            CompletableFuture<ResultT> crtFuture =
-                s3AsyncClient.getObject(downloadRequest.getObjectRequest(), responseTransformer);
+            CompletableFuture<ResultT> future = doGetObject(downloadRequest.getObjectRequest(), responseTransformer);
 
-            // Forward download cancellation to CRT future
-            CompletableFutureUtils.forwardExceptionTo(returnFuture, crtFuture);
+            // Forward download cancellation to future
+            CompletableFutureUtils.forwardExceptionTo(returnFuture, future);
 
-            CompletableFutureUtils.forwardTransformedResultTo(crtFuture, returnFuture,
+            CompletableFutureUtils.forwardTransformedResultTo(future, returnFuture,
                                                               r -> CompletedDownload.builder()
                                                                                     .result(r)
                                                                                     .build());
@@ -341,14 +341,12 @@ private TransferProgressUpdater doDownloadFile(
 
             assertNotUnsupportedArn(downloadRequest.getObjectRequest().bucket(), "download");
 
-            CompletableFuture<GetObjectResponse> crtFuture =
-                s3AsyncClient.getObject(downloadRequest.getObjectRequest(),
-                                        responseTransformer);
+            CompletableFuture<GetObjectResponse> future = doGetObject(downloadRequest.getObjectRequest(), responseTransformer);
 
-            // Forward download cancellation to CRT future
-            CompletableFutureUtils.forwardExceptionTo(returnFuture, crtFuture);
+            // Forward download cancellation to future
+            CompletableFutureUtils.forwardExceptionTo(returnFuture, future);
 
-            CompletableFutureUtils.forwardTransformedResultTo(crtFuture, returnFuture,
+            CompletableFutureUtils.forwardTransformedResultTo(future, returnFuture,
                                                               res -> CompletedFileDownload.builder()
                                                                                           .response(res)
                                                                                           .build());
@@ -450,13 +448,13 @@ public Copy copy(CopyRequest copyRequest) {
             assertNotUnsupportedArn(copyRequest.copyObjectRequest().sourceBucket(), "copy sourceBucket");
             assertNotUnsupportedArn(copyRequest.copyObjectRequest().destinationBucket(), "copy destinationBucket");
 
-            CompletableFuture<CopyObjectResponse> crtFuture =
+            CompletableFuture<CopyObjectResponse> future =
                 s3AsyncClient.copyObject(copyRequest.copyObjectRequest());
 
-            // Forward transfer cancellation to CRT future
-            CompletableFutureUtils.forwardExceptionTo(returnFuture, crtFuture);
+            // Forward transfer cancellation to future
+            CompletableFutureUtils.forwardExceptionTo(returnFuture, future);
 
-            CompletableFutureUtils.forwardTransformedResultTo(crtFuture, returnFuture,
+            CompletableFutureUtils.forwardTransformedResultTo(future, returnFuture,
                                                               r -> CompletedCopy.builder()
                                                                                 .response(r)
                                                                                 .build());
@@ -511,4 +509,14 @@ private static boolean isMrapArn(Arn arn) {
 
         return !s3EndpointResource.region().isPresent();
     }
+
+    // TODO remove once MultipartS3AsyncClient is complete
+    private <ResultT> CompletableFuture<ResultT> doGetObject(
+        GetObjectRequest getObjectRequest, AsyncResponseTransformer<GetObjectResponse, ResultT> asyncResponseTransformer) {
+        S3AsyncClient clientToUse = s3AsyncClient;
+        if (s3AsyncClient instanceof MultipartS3AsyncClient) {
+            clientToUse = (S3AsyncClient) ((DelegatingS3AsyncClient) s3AsyncClient).delegate();
+        }
+        return clientToUse.getObject(getObjectRequest, asyncResponseTransformer);
+    }
 }

services-custom/s3-transfer-manager/src/main/java/software/amazon/awssdk/transfer/s3/internal/TransferManagerFactory.java

@@ -21,6 +21,7 @@
 import software.amazon.awssdk.core.internal.util.ClassLoaderHelper;
 import software.amazon.awssdk.services.s3.S3AsyncClient;
 import software.amazon.awssdk.services.s3.internal.crt.S3CrtAsyncClient;
+import software.amazon.awssdk.services.s3.internal.multipart.MultipartS3AsyncClient;
 import software.amazon.awssdk.transfer.s3.S3TransferManager;
 import software.amazon.awssdk.utils.Logger;
 
@@ -56,6 +57,10 @@ public static S3TransferManager createTransferManager(DefaultBuilder tmBuilder)
             log.warn(() -> "The provided DefaultS3AsyncClient is not an instance of S3CrtAsyncClient, and thus multipart"
                            + " upload/download feature is not enabled and resumable file upload is not supported. To benefit "
                            + "from maximum throughput, consider using S3AsyncClient.crtBuilder().build() instead.");
+        } else if (s3AsyncClient instanceof MultipartS3AsyncClient) {
+            log.warn(() -> "The provided S3AsyncClient is an instance of MultipartS3AsyncClient, and thus multipart"
+                           + " download feature is not enabled. To benefit from all features, "
+                           + "consider using S3AsyncClient.crtBuilder().build() instead.");
         } else {
             log.debug(() -> "The provided S3AsyncClient is not an instance of S3CrtAsyncClient, and thus multipart"
                             + " upload/download feature may not be enabled and resumable file upload may not be supported.");
@@ -68,7 +73,7 @@ private static Supplier<S3AsyncClient> defaultS3AsyncClient() {
         if (crtInClasspath()) {
             return S3AsyncClient::crtCreate;
         }
-        return S3AsyncClient::create;
+        return S3AsyncClient.builder().multipartEnabled(true)::build;
     }
 
     private static boolean crtInClasspath() {

services-custom/s3-transfer-manager/src/test/java/software/amazon/awssdk/transfer/s3/internal/MultipartDownloadJavaBasedTest.java

@@ -0,0 +1,67 @@
+/*
+ * Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License").
+ * You may not use this file except in compliance with the License.
+ * A copy of the License is located at
+ *
+ *  http://aws.amazon.com/apache2.0
+ *
+ * or in the "license" file accompanying this file. This file is distributed
+ * on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either
+ * express or implied. See the License for the specific language governing
+ * permissions and limitations under the License.
+ */
+
+package software.amazon.awssdk.transfer.s3.internal;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+import java.nio.file.Paths;
+import java.util.concurrent.CompletableFuture;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import software.amazon.awssdk.core.async.AsyncResponseTransformer;
+import software.amazon.awssdk.services.s3.S3AsyncClient;
+import software.amazon.awssdk.services.s3.internal.multipart.MultipartS3AsyncClient;
+import software.amazon.awssdk.services.s3.model.GetObjectRequest;
+import software.amazon.awssdk.services.s3.model.GetObjectResponse;
+import software.amazon.awssdk.services.s3.multipart.MultipartConfiguration;
+import software.amazon.awssdk.transfer.s3.S3TransferManager;
+import software.amazon.awssdk.transfer.s3.model.CompletedFileDownload;
+
+class MultipartDownloadJavaBasedTest {
+    private S3AsyncClient mockDelegate;
+    private MultipartS3AsyncClient s3Multi;
+    private S3TransferManager tm;
+    private UploadDirectoryHelper uploadDirectoryHelper;
+    private DownloadDirectoryHelper downloadDirectoryHelper;
+    private TransferManagerConfiguration configuration;
+
+    @BeforeEach
+    public void methodSetup() {
+        mockDelegate = mock(S3AsyncClient.class);
+        s3Multi = MultipartS3AsyncClient.create(mockDelegate, MultipartConfiguration.builder().build());
+        uploadDirectoryHelper = mock(UploadDirectoryHelper.class);
+        configuration = mock(TransferManagerConfiguration.class);
+        downloadDirectoryHelper = mock(DownloadDirectoryHelper.class);
+        tm = new GenericS3TransferManager(s3Multi, uploadDirectoryHelper, configuration, downloadDirectoryHelper);
+    }
+
+    @Test
+    void usingMultipartDownload_shouldNotThrowException() {
+        GetObjectResponse response = GetObjectResponse.builder().build();
+        when(mockDelegate.getObject(any(GetObjectRequest.class), any(AsyncResponseTransformer.class)))
+            .thenReturn(CompletableFuture.completedFuture(response));
+
+        CompletedFileDownload completedFileDownload = tm.downloadFile(d -> d.getObjectRequest(g -> g.bucket("bucket")
+                                                                                                    .key("key"))
+                                                                            .destination(Paths.get(".")))
+                                                        .completionFuture()
+                                                        .join();
+        assertThat(completedFileDownload.response()).isEqualTo(response);
+    }
+}