Enable setting multipart to true/false by defining it as a custom client context parameter

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

@@ -157,6 +157,9 @@
  *         // Wait for the transfer to complete
  *         CompletedCopy completedCopy = copy.completionFuture().join();
  * }
+ * <b> Warns the user if a multi-part operation disabled client is provided for TransferManager to use
+ * If no client is provided, a multi-part enabled async client is instantiated and used.
+ * In the event of a multi-part disabled defaultS3AsyncClient being used, a warning is logged </b>
  */
 @SdkPublicApi
 @ThreadSafe

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

@@ -52,11 +52,8 @@ public static S3TransferManager createTransferManager(DefaultBuilder tmBuilder)
             return new CrtS3TransferManager(transferConfiguration, s3AsyncClient, isDefaultS3AsyncClient);
         }
 
-        if (s3AsyncClient.getClass().getName().equals("software.amazon.awssdk.services.s3.DefaultS3AsyncClient")) {
-            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.getClass().getName()
+                          .equals("software.amazon.awssdk.services.s3.internal.multipart.MultipartS3AsyncClient")) {
             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 +65,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/TransferManagerLoggingTest.java

@@ -47,21 +47,61 @@ void transferManager_withCrtClient_shouldNotLogWarnMessages() {
     }
 
     @Test
-    void transferManager_withJavaClient_shouldLogWarnMessage() {
+    void transferManager_withJavaClientMultiPartNotSet_shouldLogDebugMessage() {
 
 
         try (S3AsyncClient s3Crt = S3AsyncClient.builder()
                                                 .region(Region.US_WEST_2)
                                                 .credentialsProvider(() -> AwsBasicCredentials.create("foo", "bar"))
                                                 .build();
-             LogCaptor logCaptor = LogCaptor.create(Level.WARN);
+             LogCaptor logCaptor = LogCaptor.create(Level.DEBUG);
              S3TransferManager tm = S3TransferManager.builder().s3Client(s3Crt).build()) {
             List<LogEvent> events = logCaptor.loggedEvents();
-            assertLogged(events, Level.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.");
+            assertLogged(events, Level.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.");
+        }
+    }
+
+    @Test
+    void transferManager_withJavaClientMultiPartEqualsFalse_shouldLogDebugMessage() {
+
+
+        try (S3AsyncClient s3Crt = S3AsyncClient.builder()
+                                                .region(Region.US_WEST_2)
+                                                .credentialsProvider(() -> AwsBasicCredentials.create("foo", "bar"))
+                                                .multipartEnabled(false)
+                                                .build();
+             LogCaptor logCaptor = LogCaptor.create(Level.DEBUG);
+             S3TransferManager tm = S3TransferManager.builder().s3Client(s3Crt).build()) {
+            List<LogEvent> events = logCaptor.loggedEvents();
+            assertLogged(events, Level.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.");
+        }
+    }
+
+    @Test
+    void transferManager_withDefaultClient_shouldNotLogDebugMessage() {
+
+
+        try (LogCaptor logCaptor = LogCaptor.create(Level.DEBUG);
+             S3TransferManager tm = S3TransferManager.builder().build()) {
+            List<LogEvent> events = logCaptor.loggedEvents();
+            assertThat(events).isEmpty();
+        }
+    }
+
+    @Test
+    void transferManager_withMultiPartEnabledJavaClient_shouldNotLogDebugMessage() {
+
+        try (S3AsyncClient s3Crt = S3AsyncClient.builder()
+                                                .region(Region.US_WEST_2)
+                                                .credentialsProvider(() -> AwsBasicCredentials.create("foo", "bar"))
+                                                .multipartEnabled(true)
+                                                .build();
+             LogCaptor logCaptor = LogCaptor.create(Level.DEBUG);
+             S3TransferManager tm = S3TransferManager.builder().s3Client(s3Crt).build()) {
+            List<LogEvent> events = logCaptor.loggedEvents();
+            assertThat(events).isEmpty();
         }
     }