add ParallelConfiguration for maxInFlightParts

Author: L-Applin

services/s3/src/it/java/software/amazon/awssdk/services/s3/multipart/S3MultipartClientFileDownloadIntegrationTest.java

@@ -87,7 +87,6 @@ public void init() {
         this.interceptor = new TestInterceptor();
         this.s3Client = S3AsyncClient.builder()
                                      .multipartEnabled(true)
-                                     .multipartConfiguration(c -> c.maxInflightDownloads(50))
                                      .overrideConfiguration(o -> o.addExecutionInterceptor(this.interceptor))
                                      .credentialsProvider(CREDENTIALS_PROVIDER_CHAIN)
                                      .build();

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

@@ -17,6 +17,7 @@
 
 import software.amazon.awssdk.annotations.SdkInternalApi;
 import software.amazon.awssdk.services.s3.multipart.MultipartConfiguration;
+import software.amazon.awssdk.services.s3.multipart.ParallelConfiguration;
 import software.amazon.awssdk.utils.Validate;
 
 /**
@@ -26,11 +27,12 @@
 public final class MultipartConfigurationResolver {
 
     private static final long DEFAULT_MIN_PART_SIZE = 8L * 1024 * 1024;
-    private static final int DEFAULT_MAX_IN_FLIGHT = 50;
+    private static final int DEFAULT_MAX_IN_FLIGHT_PARTS = 50;
+
     private final long minimalPartSizeInBytes;
     private final long apiCallBufferSize;
     private final long thresholdInBytes;
-    private final int maxInFlight;
+    private final int maxInFlightParts;
 
     public MultipartConfigurationResolver(MultipartConfiguration multipartConfiguration) {
         Validate.notNull(multipartConfiguration, "multipartConfiguration");
@@ -39,7 +41,13 @@ public MultipartConfigurationResolver(MultipartConfiguration multipartConfigurat
         this.apiCallBufferSize = Validate.getOrDefault(multipartConfiguration.apiCallBufferSizeInBytes(),
                                                        () -> minimalPartSizeInBytes * 4);
         this.thresholdInBytes = Validate.getOrDefault(multipartConfiguration.thresholdInBytes(), () -> minimalPartSizeInBytes);
-        this.maxInFlight = Validate.getOrDefault(multipartConfiguration.maxInflightDownloads(), () -> DEFAULT_MAX_IN_FLIGHT);
+        ParallelConfiguration parallelConfiguration = multipartConfiguration.parallelConfiguration();
+        if (parallelConfiguration == null) {
+            this.maxInFlightParts = DEFAULT_MAX_IN_FLIGHT_PARTS;
+        } else {
+            this.maxInFlightParts = Validate.getOrDefault(multipartConfiguration.parallelConfiguration().maxInFlightParts(),
+                                                          () -> DEFAULT_MAX_IN_FLIGHT_PARTS);
+        }
     }
 
     public long minimalPartSizeInBytes() {
@@ -54,7 +62,7 @@ public long apiCallBufferSize() {
         return apiCallBufferSize;
     }
 
-    public int maxInFlight() {
-        return maxInFlight;
+    public int maxInFlightParts() {
+        return maxInFlightParts;
     }
 }

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

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

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

@@ -15,6 +15,7 @@
 
 package software.amazon.awssdk.services.s3.multipart;
 
+import java.util.function.Consumer;
 import software.amazon.awssdk.annotations.SdkPublicApi;
 import software.amazon.awssdk.core.async.AsyncRequestBody;
 import software.amazon.awssdk.core.async.AsyncResponseTransformer;
@@ -44,13 +45,13 @@ public final class MultipartConfiguration implements ToCopyableBuilder<Multipart
     private final Long thresholdInBytes;
     private final Long minimumPartSizeInBytes;
     private final Long apiCallBufferSizeInBytes;
-    private final Integer maxInflightDownloads;
+    private final ParallelConfiguration parallelConfiguration;;
 
     private MultipartConfiguration(DefaultMultipartConfigBuilder builder) {
         this.thresholdInBytes = builder.thresholdInBytes;
         this.minimumPartSizeInBytes = builder.minimumPartSizeInBytes;
         this.apiCallBufferSizeInBytes = builder.apiCallBufferSizeInBytes;
-        this.maxInflightDownloads = builder.maxInflightDownloads;
+        this.parallelConfiguration = builder.parallelConfiguration;
     }
 
     public static Builder builder() {
@@ -91,8 +92,12 @@ public Long apiCallBufferSizeInBytes() {
         return this.apiCallBufferSizeInBytes;
     }
 
-    public Integer maxInflightDownloads() {
-        return this.maxInflightDownloads;
+    /**
+     *
+     * @return
+     */
+    public ParallelConfiguration parallelConfiguration() {
+        return this.parallelConfiguration;
     }
 
     /**
@@ -172,52 +177,65 @@ public interface Builder extends CopyableBuilder<Builder, MultipartConfiguration
 
         /**
          * todo
-         * @param maxInflightDownloads
+         * @param
          * @return
          */
-        Builder maxInflightDownloads(Integer maxInflightDownloads);
+        Builder parallelConfiguration(ParallelConfiguration parallelConfiguration);
+
+        Builder parallelConfiguration(Consumer<ParallelConfiguration.Builder> consumer);
 
         /**
          * todo
          * @return
          */
-        Integer maxInflightDownloads();
+        ParallelConfiguration parallelConfiguration();
 
     }
 
     private static class DefaultMultipartConfigBuilder implements Builder {
         private Long thresholdInBytes;
         private Long minimumPartSizeInBytes;
         private Long apiCallBufferSizeInBytes;
-        private Integer maxInflightDownloads;
+        private ParallelConfiguration parallelConfiguration;
 
+        @Override
         public Builder thresholdInBytes(Long thresholdInBytes) {
             this.thresholdInBytes = thresholdInBytes;
             return this;
         }
 
+        @Override
         public Long thresholdInBytes() {
             return this.thresholdInBytes;
         }
 
+        @Override
         public Builder minimumPartSizeInBytes(Long minimumPartSizeInBytes) {
             this.minimumPartSizeInBytes = minimumPartSizeInBytes;
             return this;
         }
 
+        @Override
         public Long minimumPartSizeInBytes() {
             return this.minimumPartSizeInBytes;
         }
 
         @Override
-        public Builder maxInflightDownloads(Integer maxInflightDownloads) {
-            this.maxInflightDownloads = maxInflightDownloads;
+        public Builder parallelConfiguration(ParallelConfiguration parallelConfiguration) {
+            this.parallelConfiguration = parallelConfiguration;
             return this;
         }
 
         @Override
-        public Integer maxInflightDownloads() {
-            return this.maxInflightDownloads;
+        public Builder parallelConfiguration(Consumer<ParallelConfiguration.Builder> configuration) {
+            ParallelConfiguration.Builder builder = ParallelConfiguration.builder();
+            configuration.accept(builder);
+            return parallelConfiguration(builder.build());
+        }
+
+        @Override
+        public ParallelConfiguration parallelConfiguration() {
+            return this.parallelConfiguration;
         }
 
         @Override

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

@@ -0,0 +1,61 @@
+/*
+ * 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.services.s3.multipart;
+
+import software.amazon.awssdk.annotations.SdkPublicApi;
+import software.amazon.awssdk.utils.builder.CopyableBuilder;
+import software.amazon.awssdk.utils.builder.ToCopyableBuilder;
+
+@SdkPublicApi
+public class ParallelConfiguration implements ToCopyableBuilder<ParallelConfiguration.Builder, ParallelConfiguration> {
+
+    private final Integer maxInFlightParts;
+
+    public ParallelConfiguration(Builder builder) {
+        this.maxInFlightParts = builder.maxInFlightParts;
+    }
+
+    public static Builder builder() {
+        return new Builder();
+    }
+
+    public Integer maxInFlightParts() {
+        return maxInFlightParts;
+    }
+
+    @Override
+    public Builder toBuilder() {
+        return builder().maxInFlightParts(maxInFlightParts);
+    }
+
+    public static class Builder implements CopyableBuilder<Builder, ParallelConfiguration> {
+        private int maxInFlightParts;
+
+        public Builder maxInFlightParts(int maxInFlightParts) {
+            this.maxInFlightParts = maxInFlightParts;
+            return this;
+        }
+
+        public int maxInFlightParts() {
+            return maxInFlightParts;
+        }
+
+        @Override
+        public ParallelConfiguration build() {
+            return new ParallelConfiguration(this);
+        }
+    }
+}

services/s3/src/test/java/software/amazon/awssdk/services/s3/internal/multipart/MultipartConfigurationResolverTest.java

@@ -19,6 +19,7 @@
 
 import org.junit.jupiter.api.Test;
 import software.amazon.awssdk.services.s3.multipart.MultipartConfiguration;
+import software.amazon.awssdk.services.s3.multipart.ParallelConfiguration;
 
 public class MultipartConfigurationResolverTest {
 
@@ -59,17 +60,32 @@ void resolveApiCallBufferSize_valueNotProvided_shouldComputeBasedOnPartSize() {
         assertThat(resolver.apiCallBufferSize()).isEqualTo(40L);
     }
 
+    @Test
+    void resolveMaxInFlightParts_valueProvidedWithBuilder_shouldHonor() {
+        MultipartConfiguration configuration =
+            MultipartConfiguration.builder()
+                                  .parallelConfiguration(p -> p.maxInFlightParts(1))
+                                  .build();
+        MultipartConfigurationResolver resolver = new MultipartConfigurationResolver(configuration);
+        assertThat(resolver.maxInFlightParts()).isEqualTo(1);
+    }
+
     @Test
     void valueProvidedForAllFields_shouldHonor() {
-        MultipartConfiguration configuration = MultipartConfiguration.builder()
-                                                                     .minimumPartSizeInBytes(10L)
-                                                                     .thresholdInBytes(8L)
-                                                                     .apiCallBufferSizeInBytes(3L)
-                                                                     .build();
+        MultipartConfiguration configuration =
+            MultipartConfiguration.builder()
+                                  .minimumPartSizeInBytes(10L)
+                                  .thresholdInBytes(8L)
+                                  .apiCallBufferSizeInBytes(3L)
+                                  .parallelConfiguration(ParallelConfiguration.builder()
+                                                                              .maxInFlightParts(1)
+                                                                              .build())
+                                  .build();
         MultipartConfigurationResolver resolver = new MultipartConfigurationResolver(configuration);
         assertThat(resolver.minimalPartSizeInBytes()).isEqualTo(10L);
         assertThat(resolver.thresholdInBytes()).isEqualTo(8L);
         assertThat(resolver.apiCallBufferSize()).isEqualTo(3L);
+        assertThat(resolver.maxInFlightParts()).isEqualTo(1);
     }
 
     @Test
@@ -79,5 +95,7 @@ void noValueProvided_shouldUseDefault() {
         assertThat(resolver.minimalPartSizeInBytes()).isEqualTo(8L * 1024 * 1024);
         assertThat(resolver.thresholdInBytes()).isEqualTo(8L * 1024 * 1024);
         assertThat(resolver.apiCallBufferSize()).isEqualTo(8L * 1024 * 1024 * 4);
+        assertThat(resolver.maxInFlightParts()).isEqualTo(50);
     }
+
 }