Fixed an issue in SequentialSubscriber where future never got completed when the consumer threw error (#5945)

Author: zoewangg

.changes/next-release/bugfix-AWSSDKforJavav2-86d3eef.json

@@ -0,0 +1,6 @@
+{
+    "type": "bugfix",
+    "category": "AWS SDK for Java v2",
+    "contributor": "",
+    "description": "Updated the SDK to handle error thrown from consumer subscribed to paginator publisher, which caused the request to hang for pagination operations"
+}

utils/pom.xml

@@ -125,6 +125,11 @@
             <artifactId>commons-lang</artifactId>
             <scope>test</scope>
         </dependency>
+        <dependency>
+            <groupId>io.reactivex.rxjava2</groupId>
+            <artifactId>rxjava</artifactId>
+            <scope>test</scope>
+        </dependency>
     </dependencies>
 
     <build>

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

@@ -40,16 +40,25 @@ public SequentialSubscriber(Consumer<T> consumer,
 
     @Override
     public void onSubscribe(Subscription subscription) {
+        if (this.subscription != null) {
+            subscription.cancel();
+        }
+
         this.subscription = subscription;
         subscription.request(1);
     }
 
     @Override
     public void onNext(T t) {
+        if (t == null) {
+            NullPointerException exception = new NullPointerException("onNext(null) is not allowed.");
+            future.completeExceptionally(exception);
+            throw exception;
+        }
         try {
             consumer.accept(t);
             subscription.request(1);
-        } catch (RuntimeException e) {
+        } catch (Throwable e) {
             // Handle the consumer throwing an exception
             subscription.cancel();
             future.completeExceptionally(e);

utils/src/test/java/software/amazon/awssdk/utils/async/SequentialSubscriberTckTest.java

@@ -0,0 +1,73 @@
+/*
+ * 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.utils.async;
+
+import java.util.concurrent.CompletableFuture;
+import org.reactivestreams.Subscriber;
+import org.reactivestreams.Subscription;
+import org.reactivestreams.tck.SubscriberWhiteboxVerification;
+import org.reactivestreams.tck.TestEnvironment;
+
+public class SequentialSubscriberTckTest extends SubscriberWhiteboxVerification<Integer>  {
+    protected SequentialSubscriberTckTest() {
+        super(new TestEnvironment());
+    }
+
+    @Override
+    public Subscriber<Integer> createSubscriber(WhiteboxSubscriberProbe<Integer> probe) {
+        CompletableFuture<Void> future = new CompletableFuture<>();
+        return new SequentialSubscriber<Integer>(i -> {}, future) {
+            @Override
+            public void onError(Throwable throwable) {
+                super.onError(throwable);
+                probe.registerOnError(throwable);
+            }
+
+            @Override
+            public void onSubscribe(Subscription subscription) {
+                super.onSubscribe(subscription);
+                probe.registerOnSubscribe(new SubscriberPuppet() {
+                    @Override
+                    public void triggerRequest(long elements) {
+                        subscription.request(elements);
+                    }
+
+                    @Override
+                    public void signalCancel() {
+                        subscription.cancel();
+                    }
+                });
+            }
+
+            @Override
+            public void onNext(Integer nextItems) {
+                super.onNext(nextItems);
+                probe.registerOnNext(nextItems);
+            }
+
+            @Override
+            public void onComplete() {
+                super.onComplete();
+                probe.registerOnComplete();
+            }
+        };
+    }
+
+    @Override
+    public Integer createElement(int i) {
+        return i;
+    }
+}

utils/src/test/java/software/amazon/awssdk/utils/async/SequentialSubscriberTest.java

@@ -0,0 +1,40 @@
+/*
+ * 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.utils.async;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+import io.reactivex.Flowable;
+import java.util.concurrent.CompletableFuture;
+import org.junit.jupiter.api.Test;
+
+public class SequentialSubscriberTest {
+
+    @Test
+    void consumerThrowsException_shouldCompleteFutureExceptionally() {
+        CompletableFuture<Void> future = new CompletableFuture<>();
+        AssertionError error = new AssertionError("boom");
+        SequentialSubscriber<Integer> subscriber =
+            new SequentialSubscriber<>(i -> {
+                throw error;
+            }, future);
+
+        Flowable.fromArray(1, 2).subscribe(subscriber);
+        assertThat(future).isCompletedExceptionally();
+        assertThatThrownBy(() -> future.join()).hasRootCause(error);
+    }
+}