Reset subscriberRef in prepare

Fix a bug in EventStreamAsyncResponseTransformer where the subscriber ref is not
reset between calls to prepare(), causing retries to fail when the
EventStreamResponseHandler attempts to subscribe to a new stream on a request
retry.

Author: dagnir

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

@@ -0,0 +1,5 @@
+{
+    "category": "AWS SDK for Java v2", 
+    "type": "bugfix", 
+    "description": "Fix a bug in `EventStreamAsyncResponseTransformer` where the reference to the current stream `Subscriber` is not reset in `prepare`, causing an `IllegalStateException` to be thrown when attemping to subscribe to the event stream upon a retry."
+}

core/aws-core/src/main/java/software/amazon/awssdk/awscore/eventstream/EventStreamAsyncResponseTransformer.java

@@ -185,6 +185,7 @@ private EventStreamAsyncResponseTransformer(
     @Override
     public CompletableFuture<Void> prepare() {
         transformFuture = new CompletableFuture<>();
+        subscriberRef.set(null);
         isDone = false;
         return transformFuture;
     }

core/aws-core/src/test/java/software/amazon/awssdk/awscore/eventstream/EventStreamAsyncResponseTransformerTest.java

@@ -26,6 +26,7 @@
 import java.util.concurrent.CompletionException;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.Executors;
+import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicLong;
 import java.util.function.Consumer;
@@ -182,6 +183,42 @@ public void prepareReturnsNewFuture() {
         assertThat(transformer.prepare()).isNotEqualTo(cf1);
     }
 
+    @Test(timeout = 2000)
+    public void prepareResetsSubscriberRef() throws InterruptedException {
+        CountDownLatch latch = new CountDownLatch(2);
+        AtomicBoolean exceptionThrown = new AtomicBoolean(false);
+
+        AsyncResponseTransformer<SdkResponse, Void> transformer =
+                EventStreamAsyncResponseTransformer.builder()
+                        .eventStreamResponseHandler(
+                                onEventStream(p -> {
+                                    try {
+                                        p.subscribe(e -> {});
+                                    } catch (Throwable t) {
+                                        exceptionThrown.set(true);
+                                    } finally {
+                                        latch.countDown();
+                                    }
+                                }))
+                        .eventResponseHandler((r, e) -> null)
+                        .executor(Executors.newFixedThreadPool(2))
+                        .future(new CompletableFuture<>())
+                        .build();
+
+        Flowable<ByteBuffer> bytePublisher = Flowable.empty();
+
+        CompletableFuture<Void> transformFuture = transformer.prepare();
+        transformer.onStream(SdkPublisher.adapt(bytePublisher));
+        transformFuture.join();
+
+        transformFuture = transformer.prepare();
+        transformer.onStream(SdkPublisher.adapt(bytePublisher));
+        transformFuture.join();
+
+        latch.await();
+        assertThat(exceptionThrown).isFalse();
+    }
+
     @Test
     public void erroneousExtraExceptionOccurredDoesNotSurfaceException() {
         AtomicLong numExceptions = new AtomicLong(0);