Preserve source object metadata during multipart copy

.changes/next-release/bugfix-AmazonS3-aee65f7.json

@@ -0,0 +1,6 @@
+{
+    "type": "bugfix",
+    "category": "Amazon S3",
+    "contributor": "",
+    "description": "The multipart S3 client now honors MetadataDirective COPY by preserving source object metadata on the destination during multipart copy."
+}

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

@@ -637,9 +637,10 @@ default DirectoryDownload downloadDirectory(Consumer<DownloadDirectoryRequest.Bu
      * Depending on the underlying S3Client, {@link S3TransferManager} may intelligently use plain {@link CopyObjectRequest}s
      * for smaller objects, and multiple parallel {@link UploadPartCopyRequest}s for larger objects. If multipart copy is
      * supported by the underlying S3Client, this behavior can be configured via
-     * {@link S3CrtAsyncClientBuilder#minimumPartSizeInBytes(Long)}. Note that for multipart copy request, existing metadata
-     * stored in the source object is NOT copied to the destination object; if required, you can retrieve the metadata from the
-     * source object and set it explicitly in the @link CopyObjectRequest.Builder#metadata(Map)}.
+     * {@link S3CrtAsyncClientBuilder#minimumPartSizeInBytes(Long)}. Note that for multipart copy requests, source object
+     * metadata is not copied by default. To preserve source metadata, set
+     * {@link CopyObjectRequest.Builder#metadataDirective(MetadataDirective)} to {@code COPY}. To provide your own
+     * metadata instead, set it to {@code REPLACE}.
      *
      * <p>
      * While this API supports {@link TransferListener}s, they will not receive {@code bytesTransferred} callback-updates due to

services/s3/src/main/java/software/amazon/awssdk/services/s3/internal/multipart/CopyObjectHelper.java

@@ -34,6 +34,7 @@
 import software.amazon.awssdk.services.s3.model.CreateMultipartUploadRequest;
 import software.amazon.awssdk.services.s3.model.CreateMultipartUploadResponse;
 import software.amazon.awssdk.services.s3.model.HeadObjectResponse;
+import software.amazon.awssdk.services.s3.model.MetadataDirective;
 import software.amazon.awssdk.services.s3.model.UploadPartCopyRequest;
 import software.amazon.awssdk.services.s3.model.UploadPartCopyResponse;
 import software.amazon.awssdk.utils.CompletableFutureUtils;
@@ -95,15 +96,34 @@ private void doCopyObject(CopyObjectRequest copyObjectRequest, CompletableFuture
             copyInOneChunk(copyObjectRequest, returnFuture);
         } else {
             log.debug(() -> "Starting the copy as multipart copy request");
-            copyInParts(copyObjectRequest, contentLength, returnFuture);
+            copyInParts(copyObjectRequest, contentLength, returnFuture, headObjectResponse);
         }
     }
 
+    /**
+     * Performs a multipart copy, forwarding source object metadata by default to match
+     * {@code CopyObject} with {@code MetadataDirective: COPY} semantics.
+     */
     private void copyInParts(CopyObjectRequest copyObjectRequest,
                              Long contentLength,
-                             CompletableFuture<CopyObjectResponse> returnFuture) {
+                             CompletableFuture<CopyObjectResponse> returnFuture,
+                             HeadObjectResponse headObjectResponse) {
 
         CreateMultipartUploadRequest request = SdkPojoConversionUtils.toCreateMultipartUploadRequest(copyObjectRequest);
+
+        // Preserve source metadata for multipart copy since CreateMultipartUpload has no MetadataDirective support.
+        if (copyObjectRequest.metadataDirective() == MetadataDirective.COPY) {
+            request = request.toBuilder()
+                             .metadata(headObjectResponse.metadata())
+                             .contentType(headObjectResponse.contentType())
+                             .cacheControl(headObjectResponse.cacheControl())
+                             .contentDisposition(headObjectResponse.contentDisposition())
+                             .contentEncoding(headObjectResponse.contentEncoding())
+                             .contentLanguage(headObjectResponse.contentLanguage())
+                             .expires(headObjectResponse.expires())
+                             .build();
+        }
+
         CompletableFuture<CreateMultipartUploadResponse> createMultipartUploadFuture =
             s3AsyncClient.createMultipartUpload(request);
 

services/s3/src/test/java/software/amazon/awssdk/services/s3/internal/crt/CopyObjectHelperTest.java

@@ -23,10 +23,18 @@
 import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.when;
 
+import java.time.Instant;
+import java.util.Collections;
+import java.util.HashMap;
 import java.util.List;
+import java.util.Map;
 import java.util.concurrent.CompletableFuture;
+import java.util.stream.Stream;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.Arguments;
+import org.junit.jupiter.params.provider.MethodSource;
 import org.mockito.ArgumentCaptor;
 import org.mockito.Mockito;
 import org.mockito.invocation.InvocationOnMock;
@@ -46,6 +54,7 @@
 import software.amazon.awssdk.services.s3.model.CreateMultipartUploadResponse;
 import software.amazon.awssdk.services.s3.model.HeadObjectRequest;
 import software.amazon.awssdk.services.s3.model.HeadObjectResponse;
+import software.amazon.awssdk.services.s3.model.MetadataDirective;
 import software.amazon.awssdk.services.s3.model.NoSuchBucketException;
 import software.amazon.awssdk.services.s3.model.UploadPartCopyRequest;
 import software.amazon.awssdk.services.s3.model.UploadPartCopyResponse;
@@ -380,6 +389,122 @@ private void stubSuccessfulCompleteMultipartCall() {
             .thenReturn(completeMultipartUploadFuture);
     }
 
+    @ParameterizedTest
+    @MethodSource("metadataDirectiveCopyProvider")
+    void multiPartCopy_metadataDirectiveCopyOrUnset_shouldForwardSourceMetadata(MetadataDirective directive) {
+        CopyObjectRequest.Builder requestBuilder = CopyObjectRequest.builder()
+                                                                    .sourceBucket(SOURCE_BUCKET)
+                                                                    .sourceKey(SOURCE_KEY)
+                                                                    .destinationBucket(DESTINATION_BUCKET)
+                                                                    .destinationKey(DESTINATION_KEY);
+        if (directive != null) {
+            requestBuilder.metadataDirective(directive);
+        }
+        CopyObjectRequest copyObjectRequest = requestBuilder.build();
+
+        stubSuccessfulHeadObjectCallWithMetadata(4000L);
+        stubSuccessfulCreateMulipartCall();
+        stubSuccessfulUploadPartCopyCalls();
+        stubSuccessfulCompleteMultipartCall();
+
+        copyHelper.copyObject(copyObjectRequest).join();
+
+        ArgumentCaptor<CreateMultipartUploadRequest> captor = ArgumentCaptor.forClass(CreateMultipartUploadRequest.class);
+        verify(s3AsyncClient).createMultipartUpload(captor.capture());
+
+        CreateMultipartUploadRequest actualRequest = captor.getValue();
+        assertThat(actualRequest.metadata()).containsEntry("customKey", "customValue");
+        assertThat(actualRequest.contentType()).isEqualTo("application/zip");
+        assertThat(actualRequest.cacheControl()).isEqualTo("max-age=86400");
+        assertThat(actualRequest.contentDisposition()).isEqualTo("attachment");
+        assertThat(actualRequest.contentEncoding()).isEqualTo("gzip");
+        assertThat(actualRequest.contentLanguage()).isEqualTo("en-US");
+        assertThat(actualRequest.expires()).isEqualTo(Instant.ofEpochSecond(1700000000L));
+    }
+
+    private static Stream<Arguments> metadataDirectiveCopyProvider() {
+        return Stream.of(
+            Arguments.of(MetadataDirective.COPY)
+        );
+    }
+
+    @Test
+    void multiPartCopy_metadataDirectiveReplace_shouldNotForwardSourceMetadata() {
+        CopyObjectRequest copyObjectRequest = CopyObjectRequest.builder()
+                                                               .sourceBucket(SOURCE_BUCKET)
+                                                               .sourceKey(SOURCE_KEY)
+                                                               .destinationBucket(DESTINATION_BUCKET)
+                                                               .destinationKey(DESTINATION_KEY)
+                                                               .metadataDirective(MetadataDirective.REPLACE)
+                                                               .metadata(Collections.singletonMap("newKey", "newValue"))
+                                                               .contentType("text/plain")
+                                                               .build();
+
+        stubSuccessfulHeadObjectCallWithMetadata(4000L);
+        stubSuccessfulCreateMulipartCall();
+        stubSuccessfulUploadPartCopyCalls();
+        stubSuccessfulCompleteMultipartCall();
+
+        copyHelper.copyObject(copyObjectRequest).join();
+
+        ArgumentCaptor<CreateMultipartUploadRequest> captor = ArgumentCaptor.forClass(CreateMultipartUploadRequest.class);
+        verify(s3AsyncClient).createMultipartUpload(captor.capture());
+
+        CreateMultipartUploadRequest actualRequest = captor.getValue();
+        assertThat(actualRequest.metadata()).containsEntry("newKey", "newValue");
+        assertThat(actualRequest.metadata()).doesNotContainKey("customKey");
+        assertThat(actualRequest.contentType()).isEqualTo("text/plain");
+    }
+
+    @Test
+    void multiPartCopy_metadataDirectiveCopyWithCustomerMetadata_sourceMetadataShouldWin() {
+        CopyObjectRequest copyObjectRequest = CopyObjectRequest.builder()
+                                                               .sourceBucket(SOURCE_BUCKET)
+                                                               .sourceKey(SOURCE_KEY)
+                                                               .destinationBucket(DESTINATION_BUCKET)
+                                                               .destinationKey(DESTINATION_KEY)
+                                                               .metadataDirective(MetadataDirective.COPY)
+                                                               .metadata(Collections.singletonMap("ignored", "value"))
+                                                               .contentType("text/html")
+                                                               .build();
+
+        stubSuccessfulHeadObjectCallWithMetadata(4000L);
+        stubSuccessfulCreateMulipartCall();
+        stubSuccessfulUploadPartCopyCalls();
+        stubSuccessfulCompleteMultipartCall();
+
+        copyHelper.copyObject(copyObjectRequest).join();
+
+        ArgumentCaptor<CreateMultipartUploadRequest> captor = ArgumentCaptor.forClass(CreateMultipartUploadRequest.class);
+        verify(s3AsyncClient).createMultipartUpload(captor.capture());
+
+        CreateMultipartUploadRequest actualRequest = captor.getValue();
+        // Source metadata wins when directive is COPY, matching S3 CopyObject API behavior
+        assertThat(actualRequest.metadata()).containsEntry("customKey", "customValue");
+        assertThat(actualRequest.metadata()).doesNotContainKey("ignored");
+        assertThat(actualRequest.contentType()).isEqualTo("application/zip");
+    }
+
+    private void stubSuccessfulHeadObjectCallWithMetadata(long contentLength) {
+        Map<String, String> metadata = new HashMap<>();
+        metadata.put("customKey", "customValue");
+
+        CompletableFuture<HeadObjectResponse> headFuture =
+            CompletableFuture.completedFuture(HeadObjectResponse.builder()
+                                                                .contentLength(contentLength)
+                                                                .metadata(metadata)
+                                                                .contentType("application/zip")
+                                                                .cacheControl("max-age=86400")
+                                                                .contentDisposition("attachment")
+                                                                .contentEncoding("gzip")
+                                                                .contentLanguage("en-US")
+                                                                .expires(Instant.ofEpochSecond(1700000000L))
+                                                                .build());
+
+        when(s3AsyncClient.headObject(any(HeadObjectRequest.class)))
+            .thenReturn(headFuture);
+    }
+
     private void stubSuccessfulHeadObjectCall(long contentLength) {
         CompletableFuture<HeadObjectResponse> headFuture =
             CompletableFuture.completedFuture(HeadObjectResponse.builder()