Propagate content-type from AsyncRequestBody for multipart uploads (Issue #6607)

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

@@ -0,0 +1,6 @@
+{
+    "type": "bugfix",
+    "category": "Amazon S3",
+    "contributor": "",
+    "description": "Fixed multipart uploads not propagating content-type from AsyncRequestBody when using S3AsyncClient with multipartEnabled(true). See Issue [#6607](https://github.com/aws/aws-sdk-java-v2/issues/6607)"
+}

pom.xml

@@ -187,8 +187,8 @@
         <jre.version>1.8</jre.version>
         <httpcomponents.httpclient.version>4.5.13</httpcomponents.httpclient.version>
         <httpcomponents.httpcore.version>4.4.16</httpcomponents.httpcore.version>
-        <httpcomponents.client5.version>5.6</httpcomponents.client5.version>
-        <httpcomponents.core5.version>5.4</httpcomponents.core5.version>
+        <httpcomponents.client5.version>5.5</httpcomponents.client5.version>
+        <httpcomponents.core5.version>5.3.4</httpcomponents.core5.version>
         <!-- Reactive Streams version -->
         <reactive-streams.version>1.0.4</reactive-streams.version>
 

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

@@ -60,6 +60,13 @@ public UploadObjectHelper(S3AsyncClient s3AsyncClient,
 
     public CompletableFuture<PutObjectResponse> uploadObject(PutObjectRequest putObjectRequest,
                                                              AsyncRequestBody asyncRequestBody) {
+
+        // Propagate content-type from AsyncRequestBody if not explicitly set on the request
+        if (putObjectRequest.contentType() == null && asyncRequestBody.contentType() != null) {
+            putObjectRequest = putObjectRequest.toBuilder()
+                                               .contentType(asyncRequestBody.contentType())
+                                               .build();
+        }
         Long contentLength = asyncRequestBody.contentLength().orElseGet(putObjectRequest::contentLength);
 
         if (contentLength == null) {

services/s3/src/test/java/software/amazon/awssdk/services/s3/internal/multipart/S3MultipartClientContentTypeTest.java

@@ -0,0 +1,212 @@
+/*
+ * 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.services.s3.internal.multipart;
+
+import static com.github.tomakehurst.wiremock.client.WireMock.aResponse;
+import static com.github.tomakehurst.wiremock.client.WireMock.anyUrl;
+import static com.github.tomakehurst.wiremock.client.WireMock.containing;
+import static com.github.tomakehurst.wiremock.client.WireMock.findAll;
+import static com.github.tomakehurst.wiremock.client.WireMock.post;
+import static com.github.tomakehurst.wiremock.client.WireMock.postRequestedFor;
+import static com.github.tomakehurst.wiremock.client.WireMock.put;
+import static com.github.tomakehurst.wiremock.client.WireMock.putRequestedFor;
+import static com.github.tomakehurst.wiremock.client.WireMock.stubFor;
+import static com.github.tomakehurst.wiremock.client.WireMock.urlPathEqualTo;
+import static org.assertj.core.api.Assertions.assertThat;
+
+import com.github.tomakehurst.wiremock.junit5.WireMockRuntimeInfo;
+import com.github.tomakehurst.wiremock.junit5.WireMockTest;
+import com.github.tomakehurst.wiremock.verification.LoggedRequest;
+import java.io.File;
+import java.io.IOException;
+import java.net.URI;
+import java.nio.ByteBuffer;
+import java.nio.file.Files;
+import java.util.List;
+import java.util.Optional;
+import java.util.stream.Stream;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.TestInstance;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.Arguments;
+import org.junit.jupiter.params.provider.MethodSource;
+import org.reactivestreams.Subscriber;
+import software.amazon.awssdk.auth.credentials.AwsBasicCredentials;
+import software.amazon.awssdk.auth.credentials.StaticCredentialsProvider;
+import software.amazon.awssdk.core.async.AsyncRequestBody;
+import software.amazon.awssdk.regions.Region;
+import software.amazon.awssdk.services.s3.S3AsyncClient;
+
+/**
+ * Functional tests to verify content-type propagation from AsyncRequestBody for multipart uploads.
+ * <p>
+ * These tests verify that when using S3AsyncClient with multipartEnabled(true), the content-type
+ * from AsyncRequestBody is correctly propagated to the CreateMultipartUpload request, ensuring
+ * consistent behavior between single-part and multipart uploads.
+ */
+@WireMockTest
+@TestInstance(TestInstance.Lifecycle.PER_CLASS)
+class S3MultipartClientContentTypeTest {
+
+    private static final String BUCKET = "test-bucket";
+    private static final String KEY = "test-key";
+    private static final long MULTIPART_THRESHOLD = 10 * 1024; // 10KB
+    private static final long PART_SIZE = 5 * 1024; // 5KB
+
+    private static final String CREATE_MULTIPART_RESPONSE =
+        "<InitiateMultipartUploadResult><Bucket>" + BUCKET + "</Bucket>"
+        + "<Key>" + KEY + "</Key><UploadId>upload-id</UploadId></InitiateMultipartUploadResult>";
+
+    private static final String COMPLETE_MULTIPART_RESPONSE =
+        "<CompleteMultipartUploadResult><Bucket>" + BUCKET + "</Bucket>"
+        + "<Key>" + KEY + "</Key><ETag>\"etag\"</ETag></CompleteMultipartUploadResult>";
+
+    private S3AsyncClient s3Client;
+
+    private static File createTempFile(String prefix, String suffix, int size) throws IOException {
+        File file = File.createTempFile(prefix, suffix);
+        file.deleteOnExit();
+        Files.write(file.toPath(), new byte[size]);
+        return file;
+    }
+
+    static Stream<Arguments> uploadScenarios() throws IOException {
+        File singlePartHtmlFile = createTempFile("single", ".html", 5 * 1024);
+        File multipartHtmlFile = createTempFile("multi", ".html", 15 * 1024);
+
+        return Stream.of(
+            Arguments.of("singlePart_htmlFile", singlePartHtmlFile, false, "text/html"),
+            Arguments.of("multipart_htmlFile", multipartHtmlFile, true, "text/html")
+        );
+    }
+
+    @BeforeEach
+    void setup(WireMockRuntimeInfo wiremock) {
+        s3Client = S3AsyncClient.builder()
+                                .region(Region.US_EAST_1)
+                                .endpointOverride(URI.create(wiremock.getHttpBaseUrl()))
+                                .credentialsProvider(StaticCredentialsProvider.create(
+                                    AwsBasicCredentials.create("key", "secret")))
+                                .forcePathStyle(true)
+                                .multipartEnabled(true)
+                                .multipartConfiguration(c -> c.thresholdInBytes(MULTIPART_THRESHOLD)
+                                                              .minimumPartSizeInBytes(PART_SIZE))
+                                .build();
+    }
+
+    @AfterEach
+    void teardown() {
+        if (s3Client != null) {
+            s3Client.close();
+        }
+    }
+
+    @ParameterizedTest(name = "{0}")
+    @MethodSource("uploadScenarios")
+    void putObject_withContentTypeFromRequestBody_shouldPropagateContentType(String scenario, File file,
+                                                                              boolean isMultipart,
+                                                                              String expectedContentType) {
+        stubSuccessfulResponses(isMultipart);
+
+        s3Client.putObject(r -> r.bucket(BUCKET).key(KEY), AsyncRequestBody.fromFile(file)).join();
+
+        if (isMultipart) {
+            assertCreateMultipartUploadContentType(expectedContentType);
+        } else {
+            assertPutObjectContentType(expectedContentType);
+        }
+    }
+
+    @ParameterizedTest(name = "{0}")
+    @MethodSource("uploadScenarios")
+    void putObject_withExplicitContentType_shouldNotOverride(String scenario, File file,
+                                                              boolean isMultipart, String ignoredContentType) {
+        stubSuccessfulResponses(isMultipart);
+        String explicitContentType = "custom/type";
+
+        s3Client.putObject(r -> r.bucket(BUCKET).key(KEY).contentType(explicitContentType),
+                          AsyncRequestBody.fromFile(file)).join();
+
+        if (isMultipart) {
+            assertCreateMultipartUploadContentType(explicitContentType);
+        } else {
+            assertPutObjectContentType(explicitContentType);
+        }
+    }
+
+    @Test
+    void putObject_withNullContentTypeFromRequestBody_shouldUseDefaultContentType() {
+        stubSuccessfulResponses(true);
+        byte[] content = new byte[15 * 1024]; // 15KB - above threshold
+
+        AsyncRequestBody bodyWithNullContentType = new AsyncRequestBody() {
+            @Override
+            public Optional<Long> contentLength() {
+                return Optional.of((long) content.length);
+            }
+
+            @Override
+            public String contentType() {
+                return null;
+            }
+
+            @Override
+            public void subscribe(Subscriber<? super ByteBuffer> subscriber) {
+                AsyncRequestBody.fromBytes(content).subscribe(subscriber);
+            }
+        };
+
+        s3Client.putObject(r -> r.bucket(BUCKET).key(KEY), bodyWithNullContentType).join();
+
+        // When AsyncRequestBody.contentType() is null, should use SDK default
+        assertCreateMultipartUploadContentType("binary/octet-stream");
+    }
+
+    private void stubSuccessfulResponses(boolean isMultipart) {
+        if (isMultipart) {
+            stubFor(post(anyUrl()).willReturn(aResponse().withStatus(200).withBody(CREATE_MULTIPART_RESPONSE)));
+            stubFor(put(anyUrl()).willReturn(aResponse().withStatus(200).withHeader("ETag", "\"etag\"")));
+            stubFor(post(urlPathEqualTo("/" + BUCKET + "/" + KEY))
+                        .withQueryParam("uploadId", containing("upload-id"))
+                        .willReturn(aResponse().withStatus(200).withBody(COMPLETE_MULTIPART_RESPONSE)));
+        } else {
+            stubFor(put(anyUrl()).willReturn(aResponse().withStatus(200).withHeader("ETag", "\"etag\"")));
+        }
+    }
+
+    private void assertCreateMultipartUploadContentType(String expectedContentType) {
+        List<LoggedRequest> requests = findAll(postRequestedFor(urlPathEqualTo("/" + BUCKET + "/" + KEY))
+                                                   .withQueryParam("uploads", containing("")));
+        assertThat(requests)
+            .as("Expected CreateMultipartUpload request")
+            .hasSize(1);
+        assertThat(requests.get(0).getHeader("Content-Type"))
+            .as("Content-Type header in CreateMultipartUpload request")
+            .isEqualTo(expectedContentType);
+    }
+
+    private void assertPutObjectContentType(String expectedContentType) {
+        List<LoggedRequest> requests = findAll(putRequestedFor(urlPathEqualTo("/" + BUCKET + "/" + KEY)));
+        assertThat(requests)
+            .as("Expected PutObject request")
+            .hasSize(1);
+        assertThat(requests.get(0).getHeader("Content-Type"))
+            .as("Content-Type header in PutObject request")
+            .isEqualTo(expectedContentType);
+    }
+}

services/s3/src/test/java/software/amazon/awssdk/services/s3/internal/multipart/UploadObjectHelperTest.java

@@ -144,9 +144,18 @@ void uploadObject_unKnownContentLengthDoesNotExceedPartSize_shouldUploadInOneChu
 
         CompletableFuture<PutObjectResponse> completedFuture =
             CompletableFuture.completedFuture(PutObjectResponse.builder().build());
-        when(s3AsyncClient.putObject(putObjectRequest, asyncRequestBody)).thenReturn(completedFuture);
+        when(s3AsyncClient.putObject(any(PutObjectRequest.class), any(AsyncRequestBody.class))).thenReturn(completedFuture);
         uploadHelper.uploadObject(putObjectRequest, asyncRequestBody).join();
-        Mockito.verify(s3AsyncClient).putObject(putObjectRequest, asyncRequestBody);
+
+        ArgumentCaptor<PutObjectRequest> requestCaptor = ArgumentCaptor.forClass(PutObjectRequest.class);
+        Mockito.verify(s3AsyncClient).putObject(requestCaptor.capture(), any(AsyncRequestBody.class));
+
+        PutObjectRequest capturedRequest = requestCaptor.getValue();
+        assertThat(capturedRequest.bucket()).isEqualTo(BUCKET);
+        assertThat(capturedRequest.key()).isEqualTo(KEY);
+        assertThat(capturedRequest.contentLength()).isEqualTo(contentLength);
+        // Content-type should be enriched from AsyncRequestBody (default is application/octet-stream)
+        assertThat(capturedRequest.contentType()).isEqualTo("application/octet-stream");
     }
 
     @ParameterizedTest
@@ -414,6 +423,48 @@ void uploadObject_partsFinishedOutOfOrder_shouldSortThemInCompleteMultipart() {
         assertThat(actualRequest.multipartUpload().parts()).isEqualTo(completedParts(numTotalParts));
     }
 
+    @Test
+    void uploadObject_contentTypeNotSetOnRequest_shouldUseContentTypeFromRequestBody() {
+        PutObjectRequest putObjectRequest = PutObjectRequest.builder()
+                                                            .bucket(BUCKET)
+                                                            .key(KEY)
+                                                            .build();
+
+        stubSuccessfulCreateMultipartCall(UPLOAD_ID, s3AsyncClient);
+        stubSuccessfulUploadPartCalls(s3AsyncClient);
+        stubSuccessfulCompleteMultipartCall(BUCKET, KEY, s3AsyncClient);
+
+        uploadHelper.uploadObject(putObjectRequest, AsyncRequestBody.fromFile(testFile)).join();
+
+        ArgumentCaptor<CreateMultipartUploadRequest> requestCaptor =
+            ArgumentCaptor.forClass(CreateMultipartUploadRequest.class);
+        verify(s3AsyncClient).createMultipartUpload(requestCaptor.capture());
+
+        assertThat(requestCaptor.getValue().contentType()).isEqualTo("application/octet-stream");
+    }
+
+    @Test
+    void uploadObject_contentTypeSetOnRequest_shouldNotOverride() {
+        String explicitContentType = "my/content-type";
+        PutObjectRequest putObjectRequest = PutObjectRequest.builder()
+                                                            .bucket(BUCKET)
+                                                            .key(KEY)
+                                                            .contentType(explicitContentType)
+                                                            .build();
+
+        stubSuccessfulCreateMultipartCall(UPLOAD_ID, s3AsyncClient);
+        stubSuccessfulUploadPartCalls(s3AsyncClient);
+        stubSuccessfulCompleteMultipartCall(BUCKET, KEY, s3AsyncClient);
+
+        uploadHelper.uploadObject(putObjectRequest, AsyncRequestBody.fromFile(testFile)).join();
+
+        ArgumentCaptor<CreateMultipartUploadRequest> requestCaptor =
+            ArgumentCaptor.forClass(CreateMultipartUploadRequest.class);
+        verify(s3AsyncClient).createMultipartUpload(requestCaptor.capture());
+
+        assertThat(requestCaptor.getValue().contentType()).isEqualTo(explicitContentType);
+    }
+
     private List<CompletedPart> completedParts(int totalNumParts) {
         return IntStream.range(1, totalNumParts + 1).mapToObj(i -> CompletedPart.builder().partNumber(i).build()).collect(Collectors.toList());
     }

services/s3/src/test/java/software/amazon/awssdk/services/s3/internal/multipart/UploadWithUnknownContentLengthHelperTest.java

@@ -18,7 +18,6 @@
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.assertj.core.api.Assertions.assertThatThrownBy;
 import static org.mockito.ArgumentMatchers.any;
-import static org.mockito.ArgumentMatchers.eq;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.times;
 import static org.mockito.Mockito.verify;
@@ -28,6 +27,7 @@
 import static software.amazon.awssdk.services.s3.internal.multipart.utils.MultipartUploadTestUtils.stubSuccessfulPutObjectCall;
 import static software.amazon.awssdk.services.s3.internal.multipart.utils.MultipartUploadTestUtils.stubSuccessfulUploadPartCalls;
 
+import java.io.File;
 import java.io.FileInputStream;
 import java.io.FileNotFoundException;
 import java.io.IOException;
@@ -53,6 +53,7 @@
 import software.amazon.awssdk.services.s3.S3AsyncClient;
 import software.amazon.awssdk.services.s3.model.CompleteMultipartUploadRequest;
 import software.amazon.awssdk.services.s3.model.CompletedPart;
+import software.amazon.awssdk.services.s3.model.CreateMultipartUploadRequest;
 import software.amazon.awssdk.services.s3.model.PutObjectRequest;
 import software.amazon.awssdk.services.s3.model.PutObjectResponse;
 import software.amazon.awssdk.services.s3.model.UploadPartRequest;
@@ -255,4 +256,50 @@ private void verifyCompleteMultipartUploadRequest() {
         CompleteMultipartUploadRequest actualRequest = completeMpuArgumentCaptor.getValue();
         assertThat(actualRequest.multipartUpload().parts()).isEqualTo(createCompletedParts(NUM_TOTAL_PARTS));
     }
+
+    @Test
+    void uploadObject_unknownContentLength_contentTypeNotSet_shouldUseContentTypeFromRequestBody() throws IOException {
+        File tempFile = File.createTempFile("test-file", ".txt");
+        tempFile.deleteOnExit();
+        byte[] data = new byte[(int) (PART_SIZE * 2 + 1024)];
+        java.nio.file.Files.write(tempFile.toPath(), data);
+
+        stubSuccessfulCreateMultipartCall(UPLOAD_ID, s3AsyncClient);
+        stubSuccessfulUploadPartCalls(s3AsyncClient);
+        stubSuccessfulCompleteMultipartCall(BUCKET, KEY, s3AsyncClient);
+
+        AsyncRequestBody fileBody = AsyncRequestBody.fromFile(tempFile);
+        AsyncRequestBody unknownLengthBody = new AsyncRequestBody() {
+            @Override
+            public java.util.Optional<Long> contentLength() {
+                return java.util.Optional.empty();
+            }
+
+            @Override
+            public String contentType() {
+                return fileBody.contentType();
+            }
+
+            @Override
+            public void subscribe(org.reactivestreams.Subscriber<? super java.nio.ByteBuffer> s) {
+                fileBody.subscribe(s);
+            }
+        };
+
+        PutObjectRequest putObjectRequest = createPutObjectRequest();
+
+        UploadObjectHelper uploadObjectHelper = new UploadObjectHelper(s3AsyncClient,
+            new MultipartConfigurationResolver(software.amazon.awssdk.services.s3.multipart.MultipartConfiguration.builder()
+                .minimumPartSizeInBytes(PART_SIZE)
+                .thresholdInBytes(PART_SIZE)
+                .build()));
+
+        uploadObjectHelper.uploadObject(putObjectRequest, unknownLengthBody).join();
+
+        ArgumentCaptor<CreateMultipartUploadRequest> requestCaptor =
+            ArgumentCaptor.forClass(CreateMultipartUploadRequest.class);
+        verify(s3AsyncClient).createMultipartUpload(requestCaptor.capture());
+
+        assertThat(requestCaptor.getValue().contentType()).isEqualTo("text/plain");
+    }
 }