Address comments

Author: davidh44

.changes/next-release/bugfix-AmazonS3-263fed5.json

@@ -2,5 +2,5 @@
     "type": "bugfix",
     "category": "Amazon S3",
     "contributor": "",
-    "description": "Fix bug in MultipartS3AsyncClient GET where retryable errors may not be retried, and if retried, successful responses are incorrectly processed with the initial error."
+    "description": "Fix a bug in the Java based multipart client where retryable errors from getObject may not be retried correctly."
 }

core/sdk-core/src/main/java/software/amazon/awssdk/core/internal/async/SplittingTransformer.java

@@ -307,6 +307,7 @@ public void onStream(SdkPublisher<ByteBuffer> publisher) {
             }
             synchronized (cancelLock) {
                 if (partNumber == 1) {
+                    CompletableFutureUtils.forwardResultTo(upstreamFuture, resultFuture);
                     onStreamCalled = true;
                     log.trace(() -> "calling onStream on the upstream transformer");
                     upstreamResponseTransformer.onStream(upstreamSubscriber -> publisherToUpstream.subscribe(
@@ -317,9 +318,7 @@ public void onStream(SdkPublisher<ByteBuffer> publisher) {
                     );
                 }
             }
-
-            CompletableFutureUtils.forwardResultTo(upstreamFuture, resultFuture);
-            publisher.subscribe(new IndividualPartSubscriber<>(this.individualFuture, response, partNumber));
+            publisher.subscribe(new IndividualPartSubscriber<>(this.individualFuture, response));
         }
 
         @Override
@@ -347,13 +346,11 @@ class IndividualPartSubscriber<T> implements Subscriber<ByteBuffer> {
 
         private final CompletableFuture<T> future;
         private final T response;
-        private final int partNumber;
         private Subscription subscription;
 
-        IndividualPartSubscriber(CompletableFuture<T> future, T response, int partNumber) {
+        IndividualPartSubscriber(CompletableFuture<T> future, T response) {
             this.future = future;
             this.response = response;
-            this.partNumber = partNumber;
         }
 
         @Override
@@ -393,9 +390,7 @@ public void onComplete() {
         }
 
         private void handleError(Throwable t) {
-            if (partNumber > 1) {
-                publisherToUpstream.error(t);
-            }
+            publisherToUpstream.error(t);
             future.completeExceptionally(t);
         }
     }

core/sdk-core/src/test/java/software/amazon/awssdk/core/internal/async/IndividualPartSubscriberTckTest.java

@@ -44,7 +44,7 @@ public Subscriber<ByteBuffer> createSubscriber(WhiteboxSubscriberProbe<ByteBuffe
                                 .maximumBufferSizeInBytes(32L)
                                 .resultFuture(new CompletableFuture<>())
                                 .build();
-        return transformer.new IndividualPartSubscriber<ByteBuffer>(future, ByteBuffer.wrap(new byte[0]), 1) {
+        return transformer.new IndividualPartSubscriber<ByteBuffer>(future, ByteBuffer.wrap(new byte[0])) {
             @Override
             public void onSubscribe(Subscription s) {
                 super.onSubscribe(s);