From e8d78727e50ea855b059ebea965257c46429a2b1 Mon Sep 17 00:00:00 2001 From: John Viegas Date: Fri, 4 Sep 2026 15:49:00 -0700 Subject: [PATCH 1/3] Fix NPE race in EmittingSubscription when cancel() nulls subscriber during emit --- .../bugfix-AWSSDKforJavav2-c89f928.json | 6 + .../internal/async/EmittingSubscription.java | 7 +- ...FileAsyncResponseTransformerPublisher.java | 28 ++-- .../async/EmittingSubscriptionTest.java | 148 ++++++++++++++++++ 4 files changed, 178 insertions(+), 11 deletions(-) create mode 100644 .changes/next-release/bugfix-AWSSDKforJavav2-c89f928.json create mode 100644 core/sdk-core/src/test/java/software/amazon/awssdk/core/internal/async/EmittingSubscriptionTest.java diff --git a/.changes/next-release/bugfix-AWSSDKforJavav2-c89f928.json b/.changes/next-release/bugfix-AWSSDKforJavav2-c89f928.json new file mode 100644 index 000000000000..b6ec7f3a9cd1 --- /dev/null +++ b/.changes/next-release/bugfix-AWSSDKforJavav2-c89f928.json @@ -0,0 +1,6 @@ +{ + "type": "bugfix", + "category": "AWS SDK for Java v2", + "contributor": "", + "description": "Fixed an intermittent NullPointerException when downloading a single-part object with a multipart-enabled async S3 client. A race between the emit loop and subscription cancellation could dereference a cleared subscriber reference; the reference is now stable and reads are null-safe." +} diff --git a/core/sdk-core/src/main/java/software/amazon/awssdk/core/internal/async/EmittingSubscription.java b/core/sdk-core/src/main/java/software/amazon/awssdk/core/internal/async/EmittingSubscription.java index 25f1a78705ca..c9b4fde319c8 100644 --- a/core/sdk-core/src/main/java/software/amazon/awssdk/core/internal/async/EmittingSubscription.java +++ b/core/sdk-core/src/main/java/software/amazon/awssdk/core/internal/async/EmittingSubscription.java @@ -28,6 +28,10 @@ * Subscription which can emit {@link Subscriber#onNext(T)} signals to a subscriber, based on the demand received with the * {@link Subscription#request(long)}. It tracks the outstandingDemand that has not yet been fulfilled and used a Supplier * passed to it to create the object it needs to emit. + *

+ * Thread safe: {@link #request(long)} and {@link #cancel()} may run concurrently. The subscriber reference is + * {@code final} and never cleared, and {@link #cancel()} only sets the {@code isCancelled} flag that the emit loop + * checks before each signal. Per Reactive Streams rule 2.8, one {@code onNext} may still arrive after {@link #cancel()}. * @param the type of object to emit to the subscriber. */ @SdkInternalApi @@ -35,7 +39,7 @@ public final class EmittingSubscription implements Subscription { private static final Logger log = Logger.loggerFor(EmittingSubscription.class); - private Subscriber downstreamSubscriber; + private final Subscriber downstreamSubscriber; private final AtomicBoolean emitting; private final AtomicLong outstandingDemand; private final Runnable onCancel; @@ -74,7 +78,6 @@ public void request(long n) { @Override public void cancel() { isCancelled.set(true); - downstreamSubscriber = null; onCancel.run(); } diff --git a/core/sdk-core/src/main/java/software/amazon/awssdk/core/internal/async/FileAsyncResponseTransformerPublisher.java b/core/sdk-core/src/main/java/software/amazon/awssdk/core/internal/async/FileAsyncResponseTransformerPublisher.java index f6c215a45165..ba811809c536 100644 --- a/core/sdk-core/src/main/java/software/amazon/awssdk/core/internal/async/FileAsyncResponseTransformerPublisher.java +++ b/core/sdk-core/src/main/java/software/amazon/awssdk/core/internal/async/FileAsyncResponseTransformerPublisher.java @@ -41,7 +41,7 @@ public class FileAsyncResponseTransformerPublisher private final Path path; private final FileTransformerConfiguration initialConfig; - private Subscriber subscriber; + private volatile Subscriber subscriber; private final AtomicLong transformerCount; @@ -70,9 +70,22 @@ private AsyncResponseTransformer createTransformer() { } private void onCancel() { + // Drop the subscriber reference on cancel, per Reactive Streams rule 3.13. Reads snapshot the volatile field and + // null-check it, so this clear cannot race into a NullPointerException. subscriber = null; } + /** + * Signals an error to the downstream subscriber if still present. Reads the volatile field once so a concurrent + * {@link #onCancel()} cannot cause a {@link NullPointerException}. + */ + private void signalError(Throwable e) { + Subscriber currentSubscriber = subscriber; + if (currentSubscriber != null) { + currentSubscriber.onError(e); + } + } + /** * This is the AsyncResponseTransformer that will be used for each individual requests. *

@@ -107,13 +120,13 @@ public void onResponse(T response) { Optional contentLength = response.sdkHttpResponse().firstMatchingHeader("content-length"); long transformerCount = FileAsyncResponseTransformerPublisher.this.transformerCount.get(); // Error out if content range header is missing and this is not the initial request - if (subscriber != null && transformerCount > 0) { - subscriber.onError(new IllegalStateException("Content range header is missing")); + if (transformerCount > 0) { + signalError(new IllegalStateException("Content range header is missing")); return; } if (!contentLength.isPresent()) { - subscriber.onError(new IllegalStateException("Content length header is missing")); + signalError(new IllegalStateException("Content length header is missing")); return; } String totalLength = contentLength.get(); @@ -125,10 +138,7 @@ public void onResponse(T response) { String contentRange = contentRangeOpt.get(); Optional> contentRangePair = ContentRangeParser.range(contentRange); if (!contentRangePair.isPresent()) { - if (subscriber != null) { - IllegalStateException e = new IllegalStateException("Could not parse content range header " + contentRange); - handleError(e); - } + handleError(new IllegalStateException("Could not parse content range header " + contentRange)); return; } @@ -141,7 +151,7 @@ public void onResponse(T response) { } private void handleError(Throwable e) { - subscriber.onError(e); + signalError(e); future.completeExceptionally(e); } diff --git a/core/sdk-core/src/test/java/software/amazon/awssdk/core/internal/async/EmittingSubscriptionTest.java b/core/sdk-core/src/test/java/software/amazon/awssdk/core/internal/async/EmittingSubscriptionTest.java new file mode 100644 index 000000000000..1d040476cf95 --- /dev/null +++ b/core/sdk-core/src/test/java/software/amazon/awssdk/core/internal/async/EmittingSubscriptionTest.java @@ -0,0 +1,148 @@ +/* + * 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.core.internal.async; + +import static org.assertj.core.api.Assertions.assertThat; + +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicReference; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.Timeout; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.ValueSource; +import org.reactivestreams.Subscriber; +import org.reactivestreams.Subscription; + +class EmittingSubscriptionTest { + + @Test + void request_withPositiveDemand_emitsOneItemPerRequested() { + RecordingSubscriber subscriber = new RecordingSubscriber(); + EmittingSubscription subscription = subscriptionFor(subscriber); + + subscription.request(3); + + assertThat(subscriber.onNextCount.get()).isEqualTo(3); + assertThat(subscriber.onError.get()).isNull(); + } + + @ParameterizedTest + @ValueSource(longs = {0L, -1L, Long.MIN_VALUE}) + void request_withNonPositiveDemand_signalsIllegalArgumentException(long demand) { + RecordingSubscriber subscriber = new RecordingSubscriber(); + EmittingSubscription subscription = subscriptionFor(subscriber); + + subscription.request(demand); + + assertThat(subscriber.onError.get()).isInstanceOf(IllegalArgumentException.class); + assertThat(subscriber.onNextCount.get()).isZero(); + } + + @Test + void cancel_thenRequest_emitsNothing() { + RecordingSubscriber subscriber = new RecordingSubscriber(); + EmittingSubscription subscription = subscriptionFor(subscriber); + + subscription.cancel(); + subscription.request(5); + + assertThat(subscriber.onNextCount.get()).isZero(); + assertThat(subscriber.onError.get()).isNull(); + } + + /** + * A {@code cancel()} concurrent with an in-flight emit loop must not throw. + */ + @Test + @Timeout(value = 30, unit = TimeUnit.SECONDS) + void cancel_concurrentWithEmit_neverThrowsNullPointerException() throws InterruptedException { + int rounds = 1_000; + // A large demand keeps the emit loop running long enough to overlap the cancel. + long demandPerRound = 10_000; + + for (int round = 0; round < rounds; round++) { + RecordingSubscriber subscriber = new RecordingSubscriber(); + EmittingSubscription subscription = subscriptionFor(subscriber); + + AtomicReference emitFailure = new AtomicReference<>(); + CountDownLatch startSignal = new CountDownLatch(1); + + Thread emitter = new Thread(() -> { + awaitQuietly(startSignal); + try { + subscription.request(demandPerRound); + } catch (Throwable t) { + emitFailure.set(t); + } + }); + Thread canceller = new Thread(() -> { + awaitQuietly(startSignal); + subscription.cancel(); + }); + + emitter.start(); + canceller.start(); + startSignal.countDown(); + emitter.join(); + canceller.join(); + + assertThat(emitFailure.get()) + .withFailMessage("cancel() racing the emit loop threw: %s", emitFailure.get()) + .isNull(); + } + } + + private static EmittingSubscription subscriptionFor(Subscriber subscriber) { + return EmittingSubscription.builder() + .downstreamSubscriber(subscriber) + .onCancel(() -> { }) + .supplier(Object::new) + .build(); + } + + private static void awaitQuietly(CountDownLatch latch) { + try { + latch.await(); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + } + + private static final class RecordingSubscriber implements Subscriber { + private final AtomicInteger onNextCount = new AtomicInteger(); + private final AtomicReference onError = new AtomicReference<>(); + + @Override + public void onSubscribe(Subscription s) { + } + + @Override + public void onNext(Object item) { + onNextCount.incrementAndGet(); + } + + @Override + public void onError(Throwable t) { + onError.compareAndSet(null, t); + } + + @Override + public void onComplete() { + } + } +} From 014529b4090958d1fae81162913ec3fc6f3932fd Mon Sep 17 00:00:00 2001 From: John Viegas Date: Tue, 8 Sep 2026 13:39:55 -0700 Subject: [PATCH 2/3] Handled review comments --- .changes/next-release/bugfix-AWSSDKforJavav2-c89f928.json | 2 +- .../awssdk/core/internal/async/EmittingSubscription.java | 4 ---- 2 files changed, 1 insertion(+), 5 deletions(-) diff --git a/.changes/next-release/bugfix-AWSSDKforJavav2-c89f928.json b/.changes/next-release/bugfix-AWSSDKforJavav2-c89f928.json index b6ec7f3a9cd1..181e4932f120 100644 --- a/.changes/next-release/bugfix-AWSSDKforJavav2-c89f928.json +++ b/.changes/next-release/bugfix-AWSSDKforJavav2-c89f928.json @@ -1,6 +1,6 @@ { "type": "bugfix", "category": "AWS SDK for Java v2", - "contributor": "", + "contributor": "cthiebault", "description": "Fixed an intermittent NullPointerException when downloading a single-part object with a multipart-enabled async S3 client. A race between the emit loop and subscription cancellation could dereference a cleared subscriber reference; the reference is now stable and reads are null-safe." } diff --git a/core/sdk-core/src/main/java/software/amazon/awssdk/core/internal/async/EmittingSubscription.java b/core/sdk-core/src/main/java/software/amazon/awssdk/core/internal/async/EmittingSubscription.java index c9b4fde319c8..efd16ed28c04 100644 --- a/core/sdk-core/src/main/java/software/amazon/awssdk/core/internal/async/EmittingSubscription.java +++ b/core/sdk-core/src/main/java/software/amazon/awssdk/core/internal/async/EmittingSubscription.java @@ -28,10 +28,6 @@ * Subscription which can emit {@link Subscriber#onNext(T)} signals to a subscriber, based on the demand received with the * {@link Subscription#request(long)}. It tracks the outstandingDemand that has not yet been fulfilled and used a Supplier * passed to it to create the object it needs to emit. - *

- * Thread safe: {@link #request(long)} and {@link #cancel()} may run concurrently. The subscriber reference is - * {@code final} and never cleared, and {@link #cancel()} only sets the {@code isCancelled} flag that the emit loop - * checks before each signal. Per Reactive Streams rule 2.8, one {@code onNext} may still arrive after {@link #cancel()}. * @param the type of object to emit to the subscriber. */ @SdkInternalApi From dcf676562354f71ef275b11095f90c4bcc2874d7 Mon Sep 17 00:00:00 2001 From: John Viegas Date: Tue, 8 Sep 2026 14:35:49 -0700 Subject: [PATCH 3/3] Handled review comments to add lock to cancel() --- .../multipart/ParallelMultipartDownloaderSubscriber.java | 8 ++++++-- 1 file changed, 6 insertions(+), 2 deletions(-) diff --git a/services/s3/src/main/java/software/amazon/awssdk/services/s3/internal/multipart/ParallelMultipartDownloaderSubscriber.java b/services/s3/src/main/java/software/amazon/awssdk/services/s3/internal/multipart/ParallelMultipartDownloaderSubscriber.java index a0550d2a10f1..7c895388cd82 100644 --- a/services/s3/src/main/java/software/amazon/awssdk/services/s3/internal/multipart/ParallelMultipartDownloaderSubscriber.java +++ b/services/s3/src/main/java/software/amazon/awssdk/services/s3/internal/multipart/ParallelMultipartDownloaderSubscriber.java @@ -198,7 +198,9 @@ public void onSubscribe(Subscription s) { return; } this.subscription = s; - subscription.request(maxInFlightParts); + synchronized (subscriptionLock) { + subscription.request(maxInFlightParts); + } } @Override @@ -352,7 +354,9 @@ private boolean isMultipartObject(GetObjectResponse response) { if (response.partsCount() == null || response.partsCount() == 1) { // Single part object detected, skip multipart and complete everything now log.debug(() -> "Single Part object detected, skipping multipart download"); - subscription.cancel(); + synchronized (subscriptionLock) { + subscription.cancel(); + } resultFuture.complete(response); return false; }