EmittingSubscription.doEmit() NPEs when cancel() nulls downstreamSubscriber concurrently, failing single-part multipart downloads #7351

Describe the bug

EmittingSubscription is annotated @ThreadSafe, but cancel() and doEmit() race on the plain (non-volatile)
downstreamSubscriber field. cancel() nulls the field; doEmit() checks isCancelled at the top of its loop and
dereferences downstreamSubscriber several statements later. A cancel() that lands in that window makes the emitting
thread throw a bare NullPointerException.

core/sdk-core/src/main/java/software/amazon/awssdk/core/internal/async/EmittingSubscription.java (current master):

private Subscriber<? super T> downstreamSubscriber;   // not volatile

@Override
public void cancel() {
    isCancelled.set(true);
    downstreamSubscriber = null;      // (B)
    onCancel.run();
}

private boolean doEmit() {
    long demand = outstandingDemand.get();
    while (demand > 0) {
        if (isCancelled.get()) {      // (A) checked here...
            return true;
        }
        if (outstandingDemand.get() > 0) {
            demand = outstandingDemand.decrementAndGet();
            T value;
            try {
                value = supplier.get();
            } catch (Exception e) {
                downstreamSubscriber.onError(e);   // ...and dereferenced here
                return true;
            }
            downstreamSubscriber.onNext(value);    // (C) NPE if (B) ran between (A) and (C)
        }
    }
    return false;
}

request(long n) has the same hazard on its n <= 0 branch.

The S3 parallel multipart download path drives exactly this interleaving, on every single-part object:

1. MultipartS3AsyncClient.getObject → DownloadObjectHelper.downloadObject → FileAsyncResponseTransformerPublisher
   (the only user of EmittingSubscription) → ParallelMultipartDownloaderSubscriber.
2. ParallelMultipartDownloaderSubscriber.onSubscribe calls subscription.request(maxInFlightParts) synchronously on
   the calling thread, so that thread is inside doEmit()'s loop emitting transformers.
3. Part 1's response arrives on an SDK response thread. isMultipartObject(res) sees partsCount == null — a
   single-part object — and calls subscription.cancel():

private boolean isMultipartObject(GetObjectResponse response) {
    if (response.partsCount() == null || response.partsCount() == 1) {
        log.debug(() -> "Single Part object detected, skipping multipart download");
        subscription.cancel();
        resultFuture.complete(response);
        return false;
    }
    return true;
}

Whether the download survives is decided by whether the emitting thread has finished its maxInFlightParts iterations
before that cancel arrives, i.e. purely by scheduling and response latency.

Regression Issue

• Select this option if this issue appears to be a regression.

Before parallel download, DownloadObjectHelper used downloadPartsSerially / MultipartDownloaderSubscriber, which
does not involve EmittingSubscription. Applications that had multipartEnabled(true) and downloaded single-part
objects worked before #6589 (2025-12-02, "Parallel multipart download") and can fail after it. The file has exactly one
commit and is unchanged since.

Expected Behavior

A cancel() concurrent with an in-progress emit terminates the emit loop quietly. Per Reactive Streams §2.8 a
Subscriber must be prepared to receive onNext signals after calling cancel(), so even emitting one extra element
to the cancelled subscriber would be spec-legal. Throwing a NullPointerException out of Subscription.request(long)
is not, and it is not something a caller can classify or recover from.

Concretely: S3TransferManager.downloadFile(...) / S3AsyncClient.getObject(...) on a multipartEnabled(true) client
should download a single-part object successfully every time.

Current Behavior

The emitting thread throws out of Subscription.request(long). From the deterministic reproduction below, run against
the released sdk-core 2.49.1 artifact (line numbers are the shipped ones, and identical on master):

java.lang.NullPointerException: Cannot invoke "org.reactivestreams.Subscriber.onNext(Object)" because "this.downstreamSubscriber" is null
    at software.amazon.awssdk.core.internal.async.EmittingSubscription.doEmit(EmittingSubscription.java:112)
    at software.amazon.awssdk.core.internal.async.EmittingSubscription.emit(EmittingSubscription.java:87)
    at software.amazon.awssdk.core.internal.async.EmittingSubscription.request(EmittingSubscription.java:71)

In the S3 download path the remaining frames are
ParallelMultipartDownloaderSubscriber.onSubscribe → FileAsyncResponseTransformerPublisher.subscribe →
DownloadObjectHelper.downloadPartsNonSerially.

We see this as an intermittent download failure carrying a bare NullPointerException with no SDK exception wrapping
it, so it cannot be classified as retryable by any normal means — we ended up matching on the top stack frame being
EmittingSubscription to decide it is safe to retry, which is obviously not something we want to keep.

Reproduction Steps

The MWE below is deterministic — it uses a latch to hold the emitting thread inside supplier.get() while another
thread cancels, which is the interleaving described above. I ran this logic against the released 2.49.1 sdk-core jar: it
throws the NPE above every time, and passes (delivering exactly one onNext) once the patch in Possible Solution is
applied.

package software.amazon.awssdk.core.internal.async;

import static org.assertj.core.api.Assertions.assertThatCode;

import java.util.concurrent.CountDownLatch;
import java.util.concurrent.atomic.AtomicBoolean;
import org.junit.jupiter.api.Test;
import org.reactivestreams.Subscriber;
import org.reactivestreams.Subscription;

class EmittingSubscriptionReproTest {

    @Test
    void concurrentCancelDuringEmit_doesNotThrowNpe() throws Exception {
        CountDownLatch supplierEntered = new CountDownLatch(1);
        CountDownLatch cancelCompleted = new CountDownLatch(1);
        AtomicBoolean firstCall = new AtomicBoolean(true);

        Subscriber<Object> downstream = new Subscriber<Object>() {
            @Override public void onSubscribe(Subscription s) { }
            @Override public void onNext(Object o) { }
            @Override public void onError(Throwable t) { }
            @Override public void onComplete() { }
        };

        EmittingSubscription<Object> subscription =
            EmittingSubscription.builder()
                                .downstreamSubscriber(downstream)
                                .onCancel(() -> { })
                                .supplier(() -> {
                                    if (firstCall.compareAndSet(true, false)) {
                                        supplierEntered.countDown();
                                        await(cancelCompleted);
                                    }
                                    return new Object();
                                })
                                .build();

        Thread canceller = new Thread(() -> {
            await(supplierEntered);
            subscription.cancel();
            cancelCompleted.countDown();
        });
        canceller.start();

        assertThatCode(() -> subscription.request(2)).doesNotThrowAnyException();
        canceller.join();
    }

    private static void await(CountDownLatch latch) {
        try {
            latch.await();
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            throw new RuntimeException(e);
        }
    }
}

End to end, any single-part object reproduces it, with the probability set by how fast the first part response comes
back relative to the emit loop. It shows up readily against a low-latency endpoint (for us, a LocalStack container in
CI) and intermittently against real S3:

S3AsyncClient s3 = S3AsyncClient.builder()
        .multipartEnabled(true)
        .multipartConfiguration(c -> c.minimumPartSizeInBytes(16 * 1024 * 1024L)
                                      .thresholdInBytes(16 * 1024 * 1024L))
        .build();
S3TransferManager tm = S3TransferManager.builder().s3Client(s3).build();

// any object NOT uploaded via MPU, so GetObject returns partsCount == null
tm.downloadFile(DownloadFileRequest.builder()
                                   .getObjectRequest(b -> b.bucket("<bucket>").key("<small-object>"))
                                   .destination(Paths.get("/tmp/out.bin"))
                                   .build())
  .completionFuture()
  .join();

Possible Solution

Make the field volatile and read it into a local inside doEmit(), so a concurrent cancel() cannot null it between
the check and the dereference:

private volatile Subscriber<? super T> downstreamSubscriber;

private boolean doEmit() {
    long demand = outstandingDemand.get();

    while (demand > 0) {
        Subscriber<? super T> subscriber = downstreamSubscriber;
        if (isCancelled.get() || subscriber == null) {
            return true;
        }
        if (outstandingDemand.get() > 0) {
            demand = outstandingDemand.decrementAndGet();
            T value;
            try {
                value = supplier.get();
            } catch (Exception e) {
                subscriber.onError(e);
                return true;
            }
            subscriber.onNext(value);
        }
    }
    return false;
}

plus the same read-into-local guard on the request(long n) n <= 0 branch. This can still deliver one onNext to a
subscriber that cancelled during supplier.get(), which §2.8 explicitly permits.

The simpler alternative is to stop nulling downstreamSubscriber in cancel() and let isCancelled alone gate the
loop — I assumed the nulling is deliberate (releasing the reference for GC, mirroring
FileAsyncResponseTransformerPublisher.onCancel), so the patch above preserves it.

I would be happy to open a PR with this fix and the test.

Additional Information/Context

Why every single-part download is exposed, not just large ones. MultipartS3AsyncClient.getObject delegates
unconditionally:

public <ReturnT> CompletableFuture<ReturnT> getObject(
    GetObjectRequest getObjectRequest, AsyncResponseTransformer<GetObjectResponse, ReturnT> asyncResponseTransformer) {
    return downloadObjectHelper.downloadObject(getObjectRequest, asyncResponseTransformer);
}

multipartConfiguration.thresholdInBytes is consumed only by the copy and upload helpers, never on the download path.
So on a multipart-enabled client every getObject — including a 1 KB object — takes the parallel path, gets a part-1
GET, and then hits isMultipartObject → cancel(). The race window is entered on every single-part download rather
than rarely. I am not filing that as a separate defect since it may well be intentional, but it is what turns this race
from a corner case into something that can bite any user of multipartEnabled(true).

Sibling hazard in the same path. FileAsyncResponseTransformerPublisher.subscriber is also a non-volatile field
nulled in onCancel(), and two of its dereferences are unguarded — the "Content length header is missing" branch in
IndividualFileTransformer.onResponse, and handleError. The other two uses do a check-then-use on the same
non-volatile field. Same defect class, same feature; happy to include it in the PR or leave it out.

Related but distinct: #7174 reports that ParallelMultipartDownloaderSubscriber.onError cancels the in-flight part
futures before completing resultFuture, so the original cause is lost. Different class, different defect; mentioning
it only because both live in the parallel download path added by #6589.

AWS Java SDK version used

2.49.1 (via quarkus-amazon-services BOM); EmittingSubscription verified byte-identical on master after 2.54.12

JDK version used

JDK 21 and JDK 25 (Temurin)

Operating System and version

Linux (Ubuntu 24.04 in CI, Manjaro locally)