Compliance with Reactive Streams specs

core/http-auth-aws/pom.xml

@@ -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>

core/http-auth-aws/src/test/java/software/amazon/awssdk/http/auth/aws/internal/signer/chunkedencoding/ChunkedEncodedPublisherTckTest.java

@@ -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();
     }

core/sdk-core/pom.xml

@@ -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>
@@ -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>

http-client-spi/pom.xml

@@ -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>

http-clients/aws-crt-client/pom.xml

@@ -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>

http-clients/netty-nio-client/pom.xml

@@ -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>
@@ -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>

pom.xml

@@ -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>
@@ -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>
@@ -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>

services-custom/s3-transfer-manager/pom.xml

@@ -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>

services-custom/s3-transfer-manager/src/main/java/software/amazon/awssdk/transfer/s3/internal/AsyncBufferingSubscriber.java

@@ -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;
@@ -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) {
+                        if (subscription != null) {
+                            subscription.cancel();
+                        }
                     }
                 }
             }
@@ -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;
 
@@ -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);

services-custom/s3-transfer-manager/src/test/java/software/amazon/awssdk/transfer/s3/internal/AsyncBufferingSubscriberTest.java

@@ -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);
     }
 

services/s3/pom.xml

@@ -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>

utils/pom.xml

@@ -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>

utils/src/main/java/software/amazon/awssdk/utils/async/SimplePublisher.java

@@ -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;
 
@@ -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();
     }
 
@@ -275,6 +295,7 @@ private void doProcessQueue() {
 
                         log.trace(() -> "Calling onComplete()");
                         subscriber.onComplete();
+                        subscriber = null;
                         break;
                     case ON_ERROR:
 
@@ -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?
@@ -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).
+            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;
@@ -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();