Address comments and refactor tests

Author: davidh44

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

@@ -16,8 +16,6 @@
 package software.amazon.awssdk.core.internal.async;
 
 import java.nio.ByteBuffer;
-import java.util.ArrayList;
-import java.util.List;
 import java.util.Map;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ConcurrentHashMap;
@@ -42,7 +40,7 @@ public class ByteArraySplittingTransformer<ResponseT> implements SdkPublisher<As
     private final AsyncResponseTransformer<ResponseT, ResponseBytes<ResponseT>> upstreamResponseTransformer;
     private final CompletableFuture<ResponseBytes<ResponseT>> resultFuture;
     private Subscriber<? super AsyncResponseTransformer<ResponseT, ResponseT>> downstreamSubscriber;
-    private final AtomicInteger onNextSignalsSent = new AtomicInteger(0);
+    private final AtomicInteger nextPartNumber = new AtomicInteger(1);
     private final AtomicReference<ResponseT> responseT = new AtomicReference<>();
 
     private final SimplePublisher<ByteBuffer> publisherToUpstream = new SimplePublisher<>();
@@ -138,7 +136,7 @@ private boolean doEmit() {
             }
             if (outstandingDemand.get() > 0) {
                 demand = outstandingDemand.decrementAndGet();
-                downstreamSubscriber.onNext(new IndividualTransformer(onNextSignalsSent.incrementAndGet()));
+                downstreamSubscriber.onNext(new IndividualTransformer(nextPartNumber.getAndIncrement()));
             }
         }
         return false;
@@ -193,25 +191,19 @@ private void handleSubscriptionCancel() {
     }
 
     private final class IndividualTransformer implements AsyncResponseTransformer<ResponseT, ResponseT> {
-        private final int onNextCount;
+        private final int partNumber;
         private final ByteArrayAsyncResponseTransformer<ResponseT> delegate = new ByteArrayAsyncResponseTransformer<>();
 
-        private CompletableFuture<ResponseT> future;
-        private final List<CompletableFuture<ResponseBytes<ResponseT>>> delegatePrepareFutures = new ArrayList<>();
-
-        private IndividualTransformer(int onNextCount) {
-            this.onNextCount = onNextCount;
+        private IndividualTransformer(int partNumber) {
+            this.partNumber = partNumber;
         }
 
         @Override
         public CompletableFuture<ResponseT> prepare() {
-            future = new CompletableFuture<>();
             CompletableFuture<ResponseBytes<ResponseT>> prepare = delegate.prepare();
-            CompletableFutureUtils.forwardExceptionTo(prepare, future);
-            delegatePrepareFutures.add(prepare);
-            return prepare.thenApply(responseTResponseBytes -> {
-                buffers.put(onNextCount, responseTResponseBytes.asByteBuffer());
-                return responseTResponseBytes.response();
+            return prepare.thenApply(responseBytes -> {
+                buffers.put(partNumber, responseBytes.asByteBuffer());
+                return responseBytes.response();
             });
         }
 

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

@@ -0,0 +1,49 @@
+/*
+ * 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.core.internal.async;
+
+import java.util.concurrent.CompletableFuture;
+import org.reactivestreams.Publisher;
+import org.reactivestreams.tck.PublisherVerification;
+import org.reactivestreams.tck.TestEnvironment;
+import software.amazon.awssdk.core.ResponseBytes;
+import software.amazon.awssdk.core.async.AsyncResponseTransformer;
+import software.amazon.awssdk.core.async.SdkPublisher;
+
+public class ByteArraySplittingTransformerTckTest extends PublisherVerification<AsyncResponseTransformer<Object, Object>> {
+
+    public ByteArraySplittingTransformerTckTest() {
+        super(new TestEnvironment());
+    }
+
+    @Override
+    public Publisher<AsyncResponseTransformer<Object, Object>> createPublisher(long l) {
+        CompletableFuture<ResponseBytes<Object>> future = new CompletableFuture<>();
+        AsyncResponseTransformer<Object, ResponseBytes<Object>> upstreamTransformer = AsyncResponseTransformer.toBytes();
+        ByteArraySplittingTransformer<Object> transformer = new ByteArraySplittingTransformer<>(upstreamTransformer, future);
+        return SdkPublisher.adapt(transformer).limit(Math.toIntExact(l));
+    }
+
+    @Override
+    public Publisher<AsyncResponseTransformer<Object, Object>> createFailedPublisher() {
+        return null;
+    }
+
+    @Override
+    public long maxElementsFromPublisher() {
+        return Long.MAX_VALUE;
+    }
+}

services/s3/src/main/java/software/amazon/awssdk/services/s3/multipart/MultipartConfiguration.java

@@ -42,8 +42,8 @@
  * This means the S3 client will only download multiple parts if the object itself was uploaded as a
  * {@link S3AsyncClient#createMultipartUpload(CreateMultipartUploadRequest) multipart object}
  * <p>
- * When performing multipart download, retry is only supported for downloading to byte array, i.e., when providing a
- * {@link ByteArrayAsyncResponseTransformer}
+ * When performing multipart download, retry is only supported when using an {@link AsyncResponseTransformer} implementation
+ * that downloads the object into memory such, as {@link AsyncResponseTransformer#toBytes()}
  */
 @SdkPublicApi
 public final class MultipartConfiguration implements ToCopyableBuilder<MultipartConfiguration.Builder, MultipartConfiguration> {
@@ -91,8 +91,8 @@ public Long minimumPartSizeInBytes() {
     /**
      * The maximum memory, in bytes, that the SDK will use to buffer requests content into memory.
      * <p>
-     * This setting is not supported and will be ignored when downloading to a byte array, i.e., when providing a
-     * {@link ByteArrayAsyncResponseTransformer}.
+     * This setting does not apply if you are using an {@link AsyncResponseTransformer} implementation that downloads the
+     * object into memory such as {@link AsyncResponseTransformer#toBytes}
      *
      * @return the value of the configured maximum memory usage.
      */

services/s3/src/main/resources/codegen-resources/customization.config

@@ -289,7 +289,7 @@
   "useS3ExpressSessionAuth": true,
   "multipartCustomization": {
     "multipartConfigurationClass": "software.amazon.awssdk.services.s3.multipart.MultipartConfiguration",
-    "multipartConfigMethodDoc": "Configuration for multipart operation of this client.<p>When performing multipart download, retry is only supported for downloading to byte array, i.e., when providing a {@code ByteArrayAsyncResponseTransformer}",
+    "multipartConfigMethodDoc": "Configuration for multipart operation of this client.<p>When performing multipart download, retry is only supported when using an {@code AsyncResponseTransformer} implementation that downloads the object into memory, such as {@code AsyncResponseTransformer#toBytes()}",
     "multipartEnableMethodDoc": "Enables automatic conversion of GET, PUT and COPY methods to their equivalent multipart operation. CRC32 checksum will be enabled for PUT, unless the checksum is specified or checksum validation is disabled.<p>When performing multipart download, retry is only supported for downloading to byte array, i.e., when providing a {@code ByteArrayAsyncResponseTransformer}",
     "contextParamEnabledKey": "S3AsyncClientDecorator.MULTIPART_ENABLED_KEY",
     "contextParamConfigKey": "S3AsyncClientDecorator.MULTIPART_CONFIGURATION_KEY"