diff --git a/.changes/next-release/bugfix-AmazonS3-b7503bb.json b/.changes/next-release/bugfix-AmazonS3-b7503bb.json new file mode 100644 index 000000000000..453439d6bd81 --- /dev/null +++ b/.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)" +} diff --git a/core/http-auth-aws/src/main/java/software/amazon/awssdk/http/auth/aws/internal/signer/io/InMemoryPublisher.java b/core/http-auth-aws/src/main/java/software/amazon/awssdk/http/auth/aws/internal/signer/io/InMemoryPublisher.java index f094ce61ad11..5a7fecd9c96b 100644 --- a/core/http-auth-aws/src/main/java/software/amazon/awssdk/http/auth/aws/internal/signer/io/InMemoryPublisher.java +++ b/core/http-auth-aws/src/main/java/software/amazon/awssdk/http/auth/aws/internal/signer/io/InMemoryPublisher.java @@ -56,6 +56,12 @@ public void subscribe(Subscriber 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 +76,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); } diff --git a/core/http-auth-aws/src/test/java/software/amazon/awssdk/http/auth/aws/internal/signer/io/InMemoryPublisherTckTest.java b/core/http-auth-aws/src/test/java/software/amazon/awssdk/http/auth/aws/internal/signer/io/InMemoryPublisherTckTest.java new file mode 100644 index 000000000000..922a1d0e551a --- /dev/null +++ b/core/http-auth-aws/src/test/java/software/amazon/awssdk/http/auth/aws/internal/signer/io/InMemoryPublisherTckTest.java @@ -0,0 +1,52 @@ +/* + * 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 { + + public InMemoryPublisherTckTest() { + super(new TestEnvironment(1000)); + } + + @Override + public Publisher createPublisher(long elements) { + List data = new ArrayList<>(); + for (long i = 0; i < elements; i++) { + data.add(ByteBuffer.wrap(new byte[]{(byte) (i % 127)})); + } + return new InMemoryPublisher(data); + } + + @Override + public Publisher createFailedPublisher() { + return null; + } + + @Override + public long maxElementsFromPublisher() { + return 1024L; + } +} diff --git a/core/http-auth-aws/src/test/java/software/amazon/awssdk/http/auth/aws/internal/signer/io/InMemoryPublisherTest.java b/core/http-auth-aws/src/test/java/software/amazon/awssdk/http/auth/aws/internal/signer/io/InMemoryPublisherTest.java new file mode 100644 index 000000000000..9078bf29c6fa --- /dev/null +++ b/core/http-auth-aws/src/test/java/software/amazon/awssdk/http/auth/aws/internal/signer/io/InMemoryPublisherTest.java @@ -0,0 +1,167 @@ +/* + * 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 subscribe_deliversAllData() throws Exception { + byte[] bytes = "test data".getBytes(StandardCharsets.UTF_8); + List data = Arrays.asList(ByteBuffer.wrap(bytes)); + InMemoryPublisher publisher = new InMemoryPublisher(data); + + List received = new ArrayList<>(); + CountDownLatch completed = new CountDownLatch(1); + + publisher.subscribe(new Subscriber() { + @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 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); + + List received = new ArrayList<>(); + CountDownLatch completed = new CountDownLatch(1); + AtomicBoolean error = new AtomicBoolean(false); + + publisher.subscribe(new Subscriber() { + 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 data = Arrays.asList(ByteBuffer.wrap("x".getBytes(StandardCharsets.UTF_8))); + InMemoryPublisher publisher = new InMemoryPublisher(data); + + publisher.subscribe(new Subscriber() { + @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() { + @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(); + } + + @Test + public void subscribe_requestNonPositive_signalsError() throws Exception { + List data = Arrays.asList(ByteBuffer.wrap("x".getBytes(StandardCharsets.UTF_8))); + InMemoryPublisher publisher = new InMemoryPublisher(data); + + AtomicBoolean gotError = new AtomicBoolean(false); + CountDownLatch completed = new CountDownLatch(1); + + publisher.subscribe(new Subscriber() { + @Override + public void onSubscribe(Subscription s) { + s.request(0); + } + + @Override public void onNext(ByteBuffer b) { } + + @Override + public void onError(Throwable t) { + gotError.set(t instanceof IllegalArgumentException); + completed.countDown(); + } + + @Override public void onComplete() { completed.countDown(); } + }); + + assertThat(completed.await(5, TimeUnit.SECONDS)).isTrue(); + assertThat(gotError.get()).isTrue(); + } +} diff --git a/services/s3/src/it/java/software/amazon/awssdk/services/s3/PutObjectIntegrationTest.java b/services/s3/src/it/java/software/amazon/awssdk/services/s3/PutObjectIntegrationTest.java index dbe6681d027b..875f03353116 100644 --- a/services/s3/src/it/java/software/amazon/awssdk/services/s3/PutObjectIntegrationTest.java +++ b/services/s3/src/it/java/software/amazon/awssdk/services/s3/PutObjectIntegrationTest.java @@ -36,6 +36,7 @@ import java.util.List; import java.util.Random; import java.util.concurrent.CompletableFuture; +import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicReference; import java.util.stream.Stream; import org.apache.commons.lang3.RandomStringUtils; @@ -57,6 +58,7 @@ import software.amazon.awssdk.core.metrics.CoreMetric; import software.amazon.awssdk.core.sync.RequestBody; import software.amazon.awssdk.http.ContentStreamProvider; +import software.amazon.awssdk.http.crt.AwsCrtAsyncHttpClient; import software.amazon.awssdk.metrics.MetricCollection; import software.amazon.awssdk.metrics.MetricPublisher; import software.amazon.awssdk.services.s3.model.GetObjectResponse; @@ -94,6 +96,23 @@ public static Stream s3Clients() { Arguments.of(crtClientBuilder().build())); } + @Test + public void crtAsyncClient_chunkedEncodingDisabled_shouldNotHang() throws Exception { + try (S3AsyncClient s3Async = S3AsyncClient.builder() + .region(DEFAULT_REGION) + .credentialsProvider(CREDENTIALS_PROVIDER_CHAIN) + .httpClientBuilder(AwsCrtAsyncHttpClient.builder()) + .serviceConfiguration(S3Configuration.builder() + .chunkedEncodingEnabled(false) + .build()) + .build()) { + PutObjectResponse response = s3Async.putObject(r -> r.bucket(BUCKET).key(SYNC_KEY), + AsyncRequestBody.fromString("TESTING")) + .get(30, TimeUnit.SECONDS); + assertThat(response.eTag()).isNotNull(); + } + } + @Test public void objectInputStreamsAreClosed() { TestContentProvider provider = new TestContentProvider(CONTENT);