Fix PutObject with CRT client chunkedEncodingEnabled(false)

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

@@ -0,0 +1,6 @@
+{
+    "type": "bugfix",
+    "category": "Amazon S3",
+    "contributor": "",
+    "description": "Fix bug where S3 PutObject would hang indefinitely when using AwsCrtAsyncHttpClient with chunkedEncodingEnabled(false)"
+}

core/http-auth-aws/src/main/java/software/amazon/awssdk/http/auth/aws/internal/signer/FlexibleChecksummer.java

@@ -81,9 +81,10 @@ public CompletableFuture<Publisher<ByteBuffer>> checksum(Publisher<ByteBuffer> p
         }
 
         payload.subscribe(checksumSubscriber);
-        CompletableFuture<Publisher<ByteBuffer>> result = checksumSubscriber.completeFuture();
-        result.thenRun(() -> addChecksums(request));
-        return result;
+        return checksumSubscriber.completeFuture().thenApply(checksummedPayload -> {
+            addChecksums(request);
+            return checksummedPayload;
+        });
     }
 
     private void addChecksums(SdkHttpRequest.Builder request) {

core/http-auth-aws/src/main/java/software/amazon/awssdk/http/auth/aws/internal/signer/io/ChecksumSubscriber.java

@@ -95,7 +95,8 @@ public void onError(Throwable throwable) {
 
     @Override
     public void onComplete() {
-        checksumming.complete(new InMemoryPublisher(bufferedPayload));
+        long totalBytes = bufferedPayload.stream().mapToLong(ByteBuffer::remaining).sum();
+        checksumming.complete(new InMemoryPublisher(bufferedPayload, totalBytes));
     }
 
     public CompletableFuture<Publisher<ByteBuffer>> completeFuture() {

core/http-auth-aws/src/main/java/software/amazon/awssdk/http/auth/aws/internal/signer/io/InMemoryPublisher.java

@@ -18,24 +18,33 @@
 import java.nio.ByteBuffer;
 import java.util.ArrayList;
 import java.util.List;
+import java.util.Optional;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicLong;
-import org.reactivestreams.Publisher;
 import org.reactivestreams.Subscriber;
 import org.reactivestreams.Subscription;
 import software.amazon.awssdk.annotations.SdkInternalApi;
+import software.amazon.awssdk.http.async.SdkHttpContentPublisher;
 import software.amazon.awssdk.utils.Validate;
 
 /**
- * Temporarily used for buffering all data into memory.
+ * A content-length-aware publisher that replays buffered data. Used by {@link ChecksumSubscriber} to replay the payload after
+ * checksumming.
  */
 @SdkInternalApi
-public class InMemoryPublisher implements Publisher<ByteBuffer> {
+public class InMemoryPublisher implements SdkHttpContentPublisher {
     private final AtomicBoolean subscribed = new AtomicBoolean(false);
     private final List<ByteBuffer> data;
+    private final long length;
 
-    public InMemoryPublisher(List<ByteBuffer> data) {
+    public InMemoryPublisher(List<ByteBuffer> data, long length) {
         this.data = new ArrayList<>(Validate.noNullElements(data, "Data must not contain null elements."));
+        this.length = length;
+    }
+
+    @Override
+    public Optional<Long> contentLength() {
+        return Optional.of(length);
     }
 
     @Override
@@ -56,6 +65,12 @@ public void subscribe(Subscriber<? super ByteBuffer> s) {
 
             @Override
             public void request(long n) {
+                if (n <= 0) {
+                    finish(() -> s.onError(
+                        new IllegalArgumentException("n > 0 required but it was " + n)));
+                    return;
+                }
+
                 if (done.get()) {
                     return;
                 }
@@ -70,12 +85,15 @@ public void request(long n) {
 
             private void fulfillDemand() {
                 do {
-                    if (sending.compareAndSet(false, true)) {
-                        try {
+                    if (!sending.compareAndSet(false, true)) {
+                        return;
+                    }
+                    try {
+                        while (!done.get() && demand.get() > 0) {
                             send();
-                        } finally {
-                            sending.set(false);
                         }
+                    } finally {
+                        sending.set(false);
                     }
                 } while (!done.get() && demand.get() > 0);
             }

core/http-auth-aws/src/test/java/software/amazon/awssdk/http/auth/aws/internal/signer/io/InMemoryPublisherTckTest.java

@@ -0,0 +1,55 @@
+/*
+ * 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.http.auth.aws.internal.signer.io;
+
+import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.List;
+import org.reactivestreams.Publisher;
+import org.reactivestreams.tck.PublisherVerification;
+import org.reactivestreams.tck.TestEnvironment;
+
+/**
+ * TCK verification test for {@link InMemoryPublisher}.
+ */
+public class InMemoryPublisherTckTest extends PublisherVerification<ByteBuffer> {
+
+    public InMemoryPublisherTckTest() {
+        super(new TestEnvironment(1000));
+    }
+
+    @Override
+    public Publisher<ByteBuffer> createPublisher(long elements) {
+        List<ByteBuffer> data = new ArrayList<>();
+        long totalLength = 0;
+        for (long i = 0; i < elements; i++) {
+            byte[] content = {(byte) (i % 127)};
+            data.add(ByteBuffer.wrap(content));
+            totalLength += content.length;
+        }
+        return new InMemoryPublisher(data, totalLength);
+    }
+
+    @Override
+    public Publisher<ByteBuffer> createFailedPublisher() {
+        return null;
+    }
+
+    @Override
+    public long maxElementsFromPublisher() {
+        return 1024L;
+    }
+}

core/http-auth-aws/src/test/java/software/amazon/awssdk/http/auth/aws/internal/signer/io/InMemoryPublisherTest.java

@@ -0,0 +1,149 @@
+/*
+ * 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.http.auth.aws.internal.signer.io;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import java.nio.ByteBuffer;
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import org.junit.jupiter.api.Test;
+import org.reactivestreams.Subscriber;
+import org.reactivestreams.Subscription;
+
+public class InMemoryPublisherTest {
+
+    @Test
+    public void contentLength_returnsCorrectValue() {
+        List<ByteBuffer> data = Arrays.asList(
+            ByteBuffer.wrap("hello".getBytes(StandardCharsets.UTF_8)),
+            ByteBuffer.wrap(" world".getBytes(StandardCharsets.UTF_8))
+        );
+        InMemoryPublisher publisher = new InMemoryPublisher(data, 11);
+
+        assertThat(publisher.contentLength()).hasValue(11L);
+    }
+
+    @Test
+    public void subscribe_deliversAllData() throws Exception {
+        byte[] bytes = "test data".getBytes(StandardCharsets.UTF_8);
+        List<ByteBuffer> data = Arrays.asList(ByteBuffer.wrap(bytes));
+        InMemoryPublisher publisher = new InMemoryPublisher(data, bytes.length);
+
+        List<ByteBuffer> received = new ArrayList<>();
+        CountDownLatch completed = new CountDownLatch(1);
+
+        publisher.subscribe(new Subscriber<ByteBuffer>() {
+            @Override
+            public void onSubscribe(Subscription s) {
+                s.request(Long.MAX_VALUE);
+            }
+
+            @Override
+            public void onNext(ByteBuffer byteBuffer) {
+                received.add(byteBuffer);
+            }
+
+            @Override
+            public void onError(Throwable t) {
+            }
+
+            @Override
+            public void onComplete() {
+                completed.countDown();
+            }
+        });
+
+        assertThat(completed.await(5, TimeUnit.SECONDS)).isTrue();
+        assertThat(received).hasSize(1);
+    }
+
+    @Test
+    public void subscribe_reEntrantRequestFromOnNext_doesNotDeadlock() throws Exception {
+        List<ByteBuffer> data = Arrays.asList(
+            ByteBuffer.wrap("a".getBytes(StandardCharsets.UTF_8)),
+            ByteBuffer.wrap("b".getBytes(StandardCharsets.UTF_8)),
+            ByteBuffer.wrap("c".getBytes(StandardCharsets.UTF_8))
+        );
+        InMemoryPublisher publisher = new InMemoryPublisher(data, 3);
+
+        List<ByteBuffer> received = new ArrayList<>();
+        CountDownLatch completed = new CountDownLatch(1);
+        AtomicBoolean error = new AtomicBoolean(false);
+
+        publisher.subscribe(new Subscriber<ByteBuffer>() {
+            private Subscription subscription;
+
+            @Override
+            public void onSubscribe(Subscription s) {
+                this.subscription = s;
+                s.request(1);
+            }
+
+            @Override
+            public void onNext(ByteBuffer byteBuffer) {
+                received.add(byteBuffer);
+                // Re-entrant request — this is what ByteBufferStoringSubscriber does
+                subscription.request(1);
+            }
+
+            @Override
+            public void onError(Throwable t) {
+                error.set(true);
+                completed.countDown();
+            }
+
+            @Override
+            public void onComplete() {
+                completed.countDown();
+            }
+        });
+
+        assertThat(completed.await(5, TimeUnit.SECONDS))
+            .as("Should complete without deadlocking")
+            .isTrue();
+        assertThat(error.get()).isFalse();
+        assertThat(received).hasSize(3);
+    }
+
+    @Test
+    public void subscribe_secondSubscription_getsError() {
+        List<ByteBuffer> data = Arrays.asList(ByteBuffer.wrap("x".getBytes(StandardCharsets.UTF_8)));
+        InMemoryPublisher publisher = new InMemoryPublisher(data, 1);
+
+        publisher.subscribe(new Subscriber<ByteBuffer>() {
+            @Override public void onSubscribe(Subscription s) { s.request(1); }
+            @Override public void onNext(ByteBuffer b) { }
+            @Override public void onError(Throwable t) { }
+            @Override public void onComplete() { }
+        });
+
+        AtomicBoolean gotError = new AtomicBoolean(false);
+        publisher.subscribe(new Subscriber<ByteBuffer>() {
+            @Override public void onSubscribe(Subscription s) { }
+            @Override public void onNext(ByteBuffer b) { }
+            @Override public void onError(Throwable t) { gotError.set(true); }
+            @Override public void onComplete() { }
+        });
+
+        assertThat(gotError.get()).isTrue();
+    }
+}

core/sdk-core/src/main/java/software/amazon/awssdk/core/internal/http/pipeline/stages/AsyncSigningStage.java

@@ -21,6 +21,7 @@
 import java.util.Optional;
 import java.util.concurrent.CompletableFuture;
 import org.reactivestreams.Publisher;
+import org.reactivestreams.Subscriber;
 import software.amazon.awssdk.annotations.SdkInternalApi;
 import software.amazon.awssdk.core.SelectedAuthScheme;
 import software.amazon.awssdk.core.async.AsyncRequestBody;
@@ -38,6 +39,7 @@
 import software.amazon.awssdk.core.signer.Signer;
 import software.amazon.awssdk.http.SdkHttpFullRequest;
 import software.amazon.awssdk.http.SdkHttpRequest;
+import software.amazon.awssdk.http.async.SdkHttpContentPublisher;
 import software.amazon.awssdk.http.auth.spi.scheme.AuthSchemeOption;
 import software.amazon.awssdk.http.auth.spi.signer.AsyncSignRequest;
 import software.amazon.awssdk.http.auth.spi.signer.AsyncSignedRequest;
@@ -154,6 +156,19 @@ private static void updateAsyncRequestBodyInContexts(RequestExecutionContext con
             Publisher<ByteBuffer> signedPayload = optionalPayload.get();
             if (signedPayload instanceof AsyncRequestBody) {
                 newAsyncRequestBody = (AsyncRequestBody) signedPayload;
+            } else if (signedPayload instanceof SdkHttpContentPublisher) {
+                SdkHttpContentPublisher contentPublisher = (SdkHttpContentPublisher) signedPayload;
+                newAsyncRequestBody = new AsyncRequestBody() {
+                    @Override
+                    public Optional<Long> contentLength() {
+                        return contentPublisher.contentLength();
+                    }
+
+                    @Override
+                    public void subscribe(Subscriber<? super ByteBuffer> s) {
+                        contentPublisher.subscribe(s);
+                    }
+                };
             } else {
                 newAsyncRequestBody = AsyncRequestBody.fromPublisher(signedPayload);
             }

core/sdk-core/src/main/java/software/amazon/awssdk/core/internal/http/pipeline/stages/MakeAsyncHttpRequestStage.java

@@ -18,6 +18,7 @@
 import static software.amazon.awssdk.core.interceptor.SdkInternalExecutionAttribute.SDK_HTTP_EXECUTION_ATTRIBUTES;
 import static software.amazon.awssdk.core.internal.http.timers.TimerUtils.resolveTimeoutInMillis;
 import static software.amazon.awssdk.http.Header.CONTENT_LENGTH;
+import static software.amazon.awssdk.http.Header.TRANSFER_ENCODING;
 
 import java.nio.ByteBuffer;
 import java.time.Duration;
@@ -246,6 +247,10 @@ private boolean shouldSetContentLength(SdkHttpFullRequest request, SdkHttpConten
             return false;
         }
 
+        if (request.firstMatchingHeader(TRANSFER_ENCODING).isPresent()) {
+            return false;
+        }
+
         return Optional.ofNullable(requestProvider).flatMap(SdkHttpContentPublisher::contentLength).isPresent();
     }