cleanup and changelog

Author: L-Applin

.changes/next-release/feature-S3-12a0b55.json

@@ -0,0 +1,6 @@
+{
+    "type": "feature",
+    "category": "S3",
+    "contributor": "",
+    "description": "Add support for parallel download for individual part-get for multipart GetObject in s3 async client and Transfer Manager"
+}

.idea/codeStyles/Project.xml

@@ -179,7 +179,6 @@ public void exceptionOccurred(Throwable throwable) {
             }
         }
         if (cf != null) {
-            log.info(() -> "completing file future exceptionally");
             cf.completeExceptionally(throwable);
         } else {
             log.warn(() -> "An exception occurred before the call to prepare() was able to instantiate the CompletableFuture."
@@ -192,6 +191,9 @@ public String name() {
         return TransformerType.FILE.getName();
     }
 
+    /**
+     * {@link Subscriber} implementation that writes chunks to a file.
+     */
     static class FileSubscriber implements Subscriber<ByteBuffer> {
         private final AtomicLong position;
         private final AsynchronousFileChannel fileChannel;
@@ -205,7 +207,7 @@ static class FileSubscriber implements Subscriber<ByteBuffer> {
         private Subscription subscription;
 
         FileSubscriber(AsynchronousFileChannel fileChannel, Path path, CompletableFuture<Void> future,
-                              Consumer<Throwable> onErrorMethod, long startingPosition) {
+                       Consumer<Throwable> onErrorMethod, long startingPosition) {
             this.fileChannel = fileChannel;
             this.path = path;
             this.future = future;

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

@@ -57,4 +57,4 @@ public long maxElementsFromPublisher() {
         return Long.MAX_VALUE;
     }
 
-}
+}

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

@@ -31,9 +31,11 @@
 import org.reactivestreams.Subscription;
 import org.reactivestreams.tck.SubscriberWhiteboxVerification;
 import org.reactivestreams.tck.TestEnvironment;
+import software.amazon.awssdk.core.internal.async.FileAsyncResponseTransformer.FileSubscriber;
+
 
 /**
- * TCK verification test for {@link FileAsyncResponseTransformer.FileSubscriber}.
+ * TCK verification test for {@link FileSubscriber}.
  */
 public class FileSubscriberTckTest extends SubscriberWhiteboxVerification<ByteBuffer> {
     private static final byte[] CONTENT = new byte[16];
@@ -46,7 +48,7 @@ public FileSubscriberTckTest() {
     @Override
     public Subscriber<ByteBuffer> createSubscriber(WhiteboxSubscriberProbe<ByteBuffer> whiteboxSubscriberProbe) {
         Path tempFile = getNewTempFile();
-        return new FileAsyncResponseTransformer.FileSubscriber(openChannel(tempFile), tempFile, new CompletableFuture<>(), (t) -> {}, 0) {
+        return new FileSubscriber(openChannel(tempFile), tempFile, new CompletableFuture<>(), (t) -> {}, 0) {
             @Override
             public void onSubscribe(Subscription s) {
                 super.onSubscribe(s);

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

@@ -62,7 +62,6 @@ private <T> CompletableFuture<T> downloadPartsNonSerially(
         GetObjectRequest getObjectRequest,
         AsyncResponseTransformer.SplitResult<GetObjectResponse, T> split,
         int maxInFlight) {
-        // TODO pause & resume
         NonLinearMultipartDownloaderSubscriber subscriber = new NonLinearMultipartDownloaderSubscriber(
             s3AsyncClient, getObjectRequest, (CompletableFuture<GetObjectResponse>) split.resultFuture(), maxInFlight);
         split.publisher().subscribe(subscriber);

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

@@ -62,10 +62,10 @@ private MultipartS3AsyncClient(S3AsyncClient delegate, MultipartConfiguration mu
         long minPartSizeInBytes = resolver.minimalPartSizeInBytes();
         long threshold = resolver.thresholdInBytes();
         long apiCallBufferSize = resolver.apiCallBufferSize();
-        int maxInFLight = resolver.maxInFlightParts();
+        int maxInFlightParts = resolver.maxInFlightParts();
         mpuHelper = new UploadObjectHelper(delegate, resolver);
         copyObjectHelper = new CopyObjectHelper(delegate, minPartSizeInBytes, threshold);
-        downloadObjectHelper = new DownloadObjectHelper(delegate, apiCallBufferSize, maxInFLight);
+        downloadObjectHelper = new DownloadObjectHelper(delegate, apiCallBufferSize, maxInFlightParts);
         this.checksumEnabled = checksumEnabled;
     }
 

services/s3/src/main/java/software/amazon/awssdk/services/s3/internal/multipart/DownloadObjectHelper.java

@@ -163,12 +163,12 @@ private void request(int amount) {
     @Override
     public void onNext(AsyncResponseTransformer<GetObjectResponse, GetObjectResponse> asyncResponseTransformer) {
         outstandingDemand.decrementAndGet();
-        log.trace(() -> "=== On Next ===\nTotal in flight parts: " + inFlightRequests.size()
-                        + "\nOutstanding Demand : " + outstandingDemand.get()
-                        + "\nTotal completed parts: " + completedParts
-                        + "\nTotal transformers requested: " + transformersRequested.get()
-                        + "\nTotal pending transformers: " + pendingTransformers.size()
-                        + "\nCurrent in flight requests: " + inFlightRequests.keySet());
+        log.trace(() -> "On Next - Total in flight parts: " + inFlightRequests.size()
+                        + " - Demand : " + outstandingDemand.get()
+                        + " - Total completed parts: " + completedParts
+                        + " - Total transformers requested: " + transformersRequested.get()
+                        + " - Total pending transformers: " + pendingTransformers.size()
+                        + " - Current in flight requests: " + inFlightRequests.keySet());
         if (asyncResponseTransformer == null) {
             subscription.cancel();
             throw new NullPointerException("onNext must not be called with null asyncResponseTransformer");

services/s3/src/main/java/software/amazon/awssdk/services/s3/internal/multipart/MultipartS3AsyncClient.java

@@ -16,9 +16,15 @@
 package software.amazon.awssdk.services.s3.multipart;
 
 import software.amazon.awssdk.annotations.SdkPublicApi;
+import software.amazon.awssdk.core.async.AsyncResponseTransformer;
+import software.amazon.awssdk.services.s3.S3AsyncClient;
 import software.amazon.awssdk.utils.builder.CopyableBuilder;
 import software.amazon.awssdk.utils.builder.ToCopyableBuilder;
 
+/**
+ * Class that holds configuration properties related to multipart operations for a {@link S3AsyncClient}, related specifically
+ * to non-linear, parallel operations, that is, when the {@link AsyncResponseTransformer} supports non-serial split.
+ */
 @SdkPublicApi
 public class ParallelConfiguration implements ToCopyableBuilder<ParallelConfiguration.Builder, ParallelConfiguration> {
 
@@ -32,6 +38,10 @@ public static Builder builder() {
         return new Builder();
     }
 
+    /**
+     * The maximum number of concurrent GetObject the that are allowed for multipart download.
+     * @return The value for the maximum number of concurrent GetObject the that are allowed for multipart download.
+     */
     public Integer maxInFlightParts() {
         return maxInFlightParts;
     }