Fixed an issue in that could cause NPE to be thrown when close and onSubscribed get invoked concurrently. (#4460)

Author: zoewangg

.changes/next-release/bugfix-AWSSDKforJavav2-4c44582.json

@@ -0,0 +1,6 @@
+{
+    "type": "bugfix",
+    "category": "AWS SDK for Java v2",
+    "contributor": "",
+    "description": "Fixed an issue in `InputStreamSubscriber` that could cause NPE to be thrown when close and onSubscribed get invoked concurrently. See [#4081](https://github.com/aws/aws-sdk-java-v2/issues/4081)"
+}

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

@@ -45,6 +45,8 @@ public final class InputStreamSubscriber extends InputStream implements Subscrib
     private final AtomicBoolean drainingCallQueue = new AtomicBoolean(false);
     private final Queue<QueueEntry> callQueue = new ConcurrentLinkedQueue<>();
 
+    private final Object subscribeLock = new Object();
+
     private Subscription subscription;
 
     private boolean done = false;
@@ -55,13 +57,15 @@ public InputStreamSubscriber() {
 
     @Override
     public void onSubscribe(Subscription s) {
-        if (!inputStreamState.compareAndSet(State.UNINITIALIZED, State.READABLE)) {
-            close();
-            return;
-        }
+        synchronized (subscribeLock) {
+            if (!inputStreamState.compareAndSet(State.UNINITIALIZED, State.READABLE)) {
+                close();
+                return;
+            }
 
-        this.subscription = new CancelWatcher(s);
-        delegate.onSubscribe(subscription);
+            this.subscription = new CancelWatcher(s);
+            delegate.onSubscribe(subscription);
+        }
     }
 
     @Override
@@ -120,12 +124,14 @@ public int read(byte[] bytes, int off, int len) {
 
     @Override
     public void close() {
-        if (inputStreamState.compareAndSet(State.UNINITIALIZED, State.CLOSED)) {
-            delegate.onSubscribe(new NoOpSubscription());
-            delegate.onError(new CancellationException());
-        } else if (inputStreamState.compareAndSet(State.READABLE, State.CLOSED)) {
-            subscription.cancel();
-            onError(new CancellationException());
+        synchronized (subscribeLock) {
+            if (inputStreamState.compareAndSet(State.UNINITIALIZED, State.CLOSED)) {
+                delegate.onSubscribe(new NoOpSubscription());
+                delegate.onError(new CancellationException());
+            } else if (inputStreamState.compareAndSet(State.READABLE, State.CLOSED)) {
+                subscription.cancel();
+                onError(new CancellationException());
+            }
         }
     }