Skip to content

Compliance with Reactive Streams specs #7191

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
davidh44 wants to merge 3 commits into
base: master
Choose a base branch
from
Open

Compliance with Reactive Streams specs #7191

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
4db8286
Compliance with Reactive Streams specs
davidh44 Jul 23, 2026
db82b95
Move dependency to parent pom
davidh44 Jul 24, 2026
4d5a585
Compliance with Reactive Streams specs
davidh44 Jul 24, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 0 additions & 5 deletions core/http-auth-aws/pom.xml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -126,11 +126,6 @@
<artifactId>equalsverifier</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.reactivestreams</groupId>
<artifactId>reactive-streams-tck</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>software.amazon.awssdk</groupId>
<artifactId>test-utils</artifactId>
Expand Down
5 changes: 5 additions & 0 deletions .../awssdk/http/auth/aws/internal/signer/chunkedencoding/ChunkedEncodedPublisherTckTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -61,9 +61,14 @@ private Publisher<ByteBuffer> createChunkedPublisher(long chunksToProduce) {

Publisher<ByteBuffer> inputPublisher = Flowable.fromIterable(elements);

// Required by the builder; must match the data the input publisher produces. Production sets this from the
// x-amz-decoded-content-length header (see AwsChunkedV4PayloadSigner).
long contentLength = (long) totalElements * INPUT_STREAM_ELEMENT_SIZE;

return ChunkedEncodedPublisher.builder()
.chunkSize(CHUNK_SIZE)
.publisher(inputPublisher)
.contentLength(contentLength)
.addEmptyTrailingChunk(false)
.build();
}
Expand Down
32 changes: 0 additions & 32 deletions core/sdk-core/pom.xml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -193,11 +193,6 @@
<version>${awsjavasdk.version}</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.reactivestreams</groupId>
<artifactId>reactive-streams-tck</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>com.google.jimfs</groupId>
<artifactId>jimfs</artifactId>
Expand Down Expand Up @@ -232,33 +227,6 @@
</dependencies>
<build>
<plugins>
<!-- The Reactive Streams TCK tests are based on TestNG. See http://maven.apache.org/surefire/maven-surefire-plugin/examples/testng.html#Running_TestNG_and_JUnit_Tests -->
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-surefire-plugin</artifactId>
<version>${maven.surefire.version}</version>
<configuration>
<properties>
<property>
<name>junit</name>
<value>false</value>
</property>
</properties>
<threadCount>1</threadCount>
</configuration>
<dependencies>
<dependency>
<groupId>org.apache.maven.surefire</groupId>
<artifactId>surefire-junit-platform</artifactId>
<version>${maven.surefire.version}</version>
</dependency>
<dependency>
<groupId>org.apache.maven.surefire</groupId>
<artifactId>surefire-testng</artifactId>
<version>${maven.surefire.version}</version>
</dependency>
</dependencies>
</plugin>
<plugin>
<groupId>org.codehaus.mojo</groupId>
<artifactId>build-helper-maven-plugin</artifactId>
Expand Down
5 changes: 0 additions & 5 deletions http-client-spi/pom.xml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -75,11 +75,6 @@
<artifactId>assertj-core</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.reactivestreams</groupId>
<artifactId>reactive-streams-tck</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>nl.jqno.equalsverifier</groupId>
<artifactId>equalsverifier</artifactId>
Expand Down
5 changes: 0 additions & 5 deletions http-clients/aws-crt-client/pom.xml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -112,11 +112,6 @@
<artifactId>assertj-core</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.reactivestreams</groupId>
<artifactId>reactive-streams-tck</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.apache.logging.log4j</groupId>
<artifactId>log4j-api</artifactId>
Expand Down
32 changes: 0 additions & 32 deletions http-clients/netty-nio-client/pom.xml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -147,11 +147,6 @@
<artifactId>assertj-core</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.reactivestreams</groupId>
<artifactId>reactive-streams-tck</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.apache.logging.log4j</groupId>
<artifactId>log4j-api</artifactId>
Expand Down Expand Up @@ -254,33 +249,6 @@

<build>
<plugins>
<!-- The Reactive Streams TCK tests are based on TestNG. See http://maven.apache.org/surefire/maven-surefire-plugin/examples/testng.html#Running_TestNG_and_JUnit_Tests -->
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-surefire-plugin</artifactId>
<version>${maven.surefire.version}</version>
<configuration>
<properties>
<property>
<name>junit</name>
<value>false</value>
</property>
</properties>
<threadCount>1</threadCount>
</configuration>
<dependencies>
<dependency>
<groupId>org.apache.maven.surefire</groupId>
<artifactId>surefire-junit-platform</artifactId>
<version>${maven.surefire.version}</version>
</dependency>
<dependency>
<groupId>org.apache.maven.surefire</groupId>
<artifactId>surefire-testng</artifactId>
<version>${maven.surefire.version}</version>
</dependency>
</dependencies>
</plugin>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-jar-plugin</artifactId>
Expand Down
21 changes: 21 additions & 0 deletions pom.xml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -213,6 +213,15 @@

<dependencies>
<!-- Internal dependencies are managed in the bom-internal module. -->

<!-- reactive-streams-tck brings TestNG, which is needed globally so the surefire-testng provider
can discover TCK tests in any module without per-module surefire configuration. -->
<dependency>
<groupId>org.reactivestreams</groupId>
<artifactId>reactive-streams-tck</artifactId>
<version>${reactive-streams.version}</version>
<scope>test</scope>
</dependency>
</dependencies>

<build>
Expand Down Expand Up @@ -302,6 +311,13 @@
<include>**/*TestCase.java</include>
</includes>
<skipTests>${skip.unit.tests}</skipTests>
<!-- Prevent TestNG from re-running JUnit tests -->
<properties>
<property>
<name>junit</name>
<value>false</value>
</property>
</properties>
</configuration>
<!-- Have to explicitly set surefire provider because reactivestreamsTCK is using TestNG-->
<dependencies>
Expand All @@ -310,6 +326,11 @@
<artifactId>surefire-junit-platform</artifactId>
<version>${maven.surefire.version}</version>
</dependency>
<dependency>
<groupId>org.apache.maven.surefire</groupId>
<artifactId>surefire-testng</artifactId>
<version>${maven.surefire.version}</version>
</dependency>
</dependencies>
</plugin>
<plugin>
Expand Down
5 changes: 0 additions & 5 deletions services-custom/s3-transfer-manager/pom.xml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -185,11 +185,6 @@
<artifactId>commons-lang3</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.reactivestreams</groupId>
<artifactId>reactive-streams-tck</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>com.google.jimfs</groupId>
<artifactId>jimfs</artifactId>
Expand Down
17 changes: 14 additions & 3 deletions ...r/src/main/java/software/amazon/awssdk/transfer/s3/internal/AsyncBufferingSubscriber.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -40,6 +40,7 @@ public class AsyncBufferingSubscriber<T> implements Subscriber<T> {
private final int maxConcurrentExecutions;
private final AtomicInteger numRequestsInFlight;
private volatile boolean upstreamDone;
private volatile boolean onErrorInvoked;
private volatile Subscription subscription;

private final Set<CompletableFuture<?>> requestsInFlight;
Expand All @@ -56,9 +57,14 @@ public AsyncBufferingSubscriber(Function<T, CompletableFuture<?>> consumer,
returnFuture.whenComplete((r, t) -> {
if (t != null) {
requestsInFlight.forEach(f -> f.cancel(true));
synchronized (this) {
if (subscription != null) {
subscription.cancel();
// Skip cancelling when the failure came from onError: upstream has already terminated, and cancelling here
// would call Subscription::cancel from within onError (Reactive Streams rule 2.3). Still cancel on an
// external abort.
if (!onErrorInvoked) {
synchronized (this) {

@zoewangg zoewangg Jul 24, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Could there be a race condition where onErrorInvoked turns to true (onError invoked) right after the check?

@davidh44 davidh44 Jul 24, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

No, Rule 2.3 only prohibits calling cancel() from within the onError call stack (same thread). That's the case this flag guards: onError → sets flag → calls completeExceptionally() → handler runs inline on the same thread → sees flag → skips cancel. In the scenario you're describing (different thread sets the flag after the check), the handler is running because of an external abort, not because of onError — so calling cancel() there is correct and not a rule 2.3 violation.

if (subscription != null) {
subscription.cancel();
}
}
}
}
Expand All @@ -79,6 +85,8 @@ public void onSubscribe(Subscription subscription) {

@Override
public void onNext(T item) {
// Reactive Streams rule 2.13: onNext must throw NullPointerException on a null element.
Validate.paramNotNull(item, "item");
numRequestsInFlight.incrementAndGet();
CompletableFuture<?> currentRequest;

Expand Down Expand Up @@ -111,6 +119,9 @@ public void onNext(T item) {

@Override
public void onError(Throwable t) {
// Set before completing the future: completeExceptionally may run the whenComplete handler synchronously, and it
// must see this flag to avoid cancelling the subscription from within onError (see constructor).
onErrorInvoked = true;
// Need to complete future exceptionally first to prevent
// accidental successful completion by a concurrent checkForCompletion.
returnFuture.completeExceptionally(t);
Expand Down
9 changes: 3 additions & 6 deletions ...c/test/java/software/amazon/awssdk/transfer/s3/internal/AsyncBufferingSubscriberTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -128,12 +128,9 @@ public void consumerFunctionThrows_shouldCancelSubscriptionAndCompleteFutureExce
subscriber.onSubscribe(mockSubscription);
subscriber.onNext("item");

/*
subscription.cancel() now exists in two codepaths:
- in onNext() catch block.
- in future.whenComplete()
*/
verify(mockSubscription, times(2)).cancel();
// Cancelled once, from the onNext() catch block. The whenComplete() handler does not cancel again, which would
// violate Reactive Streams rule 2.3 (cancel from within onError).
verify(mockSubscription, times(1)).cancel();
assertThatThrownBy(future::join).hasCause(exception);
}

Expand Down
5 changes: 0 additions & 5 deletions services/s3/pom.xml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -180,11 +180,6 @@
<artifactId>wiremock-jre8-standalone</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.reactivestreams</groupId>
<artifactId>reactive-streams-tck</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>software.amazon.awssdk</groupId>
<artifactId>auth-crt</artifactId>
Expand Down
5 changes: 0 additions & 5 deletions utils/pom.xml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -94,11 +94,6 @@
<artifactId>commons-io</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.reactivestreams</groupId>
<artifactId>reactive-streams-tck</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.apache.logging.log4j</groupId>
<artifactId>log4j-api</artifactId>
Expand Down
37 changes: 34 additions & 3 deletions utils/src/main/java/software/amazon/awssdk/utils/async/SimplePublisher.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -95,8 +95,23 @@ public final class SimplePublisher<T> implements Publisher<T> {
*/
private final FailureMessage failureMessage = new FailureMessage();

/**
* Whether {@link #subscribe(Subscriber)} has already been called, used to reject a second subscription. Tracked
* separately from {@link #subscriber} because that field is cleared on termination and must not re-open subscription.
*/
private final AtomicBoolean subscribed = new AtomicBoolean(false);

/**
* True while the subscriber's {@code onSubscribe} is executing. Used to prevent {@link #processEventQueue()} from
* delivering {@code onNext} signals before {@code onSubscribe} has returned, as required by Reactive Streams rule 1.03.
*/
private volatile boolean onSubscribeInProgress = false;

/**
* The subscriber provided via {@link #subscribe(Subscriber)}. This publisher only supports a single subscriber.
*
* <p>Cleared when the stream terminates (complete, error, or cancel) so the publisher does not retain the subscriber
* reference, per Reactive Streams rule 3.13.
*/
private Subscriber<? super T> subscriber;

Expand Down Expand Up @@ -196,14 +211,19 @@ public CompletableFuture<Void> error(Throwable error) {
*/
@Override
public void subscribe(Subscriber<? super T> s) {
if (subscriber != null) {
if (!subscribed.compareAndSet(false, true)) {
s.onSubscribe(new NoOpSubscription());
s.onError(new IllegalStateException("Only one subscription may be active at a time."));
return;
}

this.subscriber = s;
s.onSubscribe(new SubscriptionImpl());
onSubscribeInProgress = true;
try {
s.onSubscribe(new SubscriptionImpl());
} finally {
onSubscribeInProgress = false;
}
processEventQueue();
}

Expand Down Expand Up @@ -275,6 +295,7 @@ private void doProcessQueue() {

log.trace(() -> "Calling onComplete()");
subscriber.onComplete();
subscriber = null;
break;
case ON_ERROR:

Expand All @@ -283,9 +304,11 @@ private void doProcessQueue() {
onErrorEntry.failure));
log.trace(() -> "Calling onError() with " + onErrorEntry.failure, onErrorEntry.failure);
subscriber.onError(onErrorEntry.failure);
subscriber = null;
break;
case CANCEL:
failureMessage.trySet(() -> new CancellationException("subscription has been cancelled."));
subscriber = null;
break;
default:
// Should never happen. Famous last words?
Expand Down Expand Up @@ -317,6 +340,11 @@ private boolean shouldProcessQueueEntry(QueueEntry<T> entry) {
return false;
}

if (onSubscribeInProgress) {
// Do not deliver signals until onSubscribe has returned (Reactive Streams rule 1.03).

@zoewangg zoewangg Jul 24, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Trying to understand this: how could processEventQueue be invoked if onSubscribeInProgress?

@davidh44 davidh44 Jul 24, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The subscriber calls request(n) from within onSubscribe (rule 2.7 allows this, and most subscribers do it to signal initial demand). request(n) → processEventQueue(). Without this flag, if data was already queued, it would deliver onNext before onSubscribe returns — violating rule 1.03. A concurrent send() from another thread can also trigger processEventQueue() during onSubscribe.

return false;
}

if (entry.type() != ON_NEXT) {
// This event isn't an on-next event, so we don't need subscriber demand in order to process it.
return true;
Expand All @@ -338,7 +366,10 @@ private void panicAndDie(Throwable cause) {
// Create exception here instead of in supplier to preserve a more-useful stack trace.
RuntimeException failure = new IllegalStateException("Encountered fatal error in publisher", cause);
failureMessage.trySet(() -> failure);
subscriber.onError(cause instanceof Error ? cause : failure);
if (subscriber != null) {
subscriber.onError(cause instanceof Error ? cause : failure);
subscriber = null;
}

while (true) {
QueueEntry<T> entry = standardPriorityQueue.poll();
Expand Down
Loading