Fixed the thread safety issue with CRT based S3 client (#3806)

Author: zoewangg

services/s3/src/main/java/software/amazon/awssdk/services/s3/internal/crt/CopyObjectHelper.java

@@ -56,11 +56,11 @@ public final class CopyObjectHelper {
     private static final long MAX_UPLOAD_PARTS = 10_000;
 
     private final S3AsyncClient s3AsyncClient;
-    private final S3NativeClientConfiguration s3NativeClientConfiguration;
+    private final long partSizeInBytes;
 
-    public CopyObjectHelper(S3AsyncClient s3AsyncClient, S3NativeClientConfiguration s3NativeClientConfiguration) {
+    public CopyObjectHelper(S3AsyncClient s3AsyncClient, long partSizeInBytes) {
         this.s3AsyncClient = s3AsyncClient;
-        this.s3NativeClientConfiguration = s3NativeClientConfiguration;
+        this.partSizeInBytes = partSizeInBytes;
     }
 
     public CompletableFuture<CopyObjectResponse> copyObject(CopyObjectRequest copyObjectRequest) {
@@ -92,9 +92,7 @@ private void doCopyObject(CopyObjectRequest copyObjectRequest, CompletableFuture
                               HeadObjectResponse headObjectResponse) {
         Long contentLength = headObjectResponse.contentLength();
 
-        long partSize = s3NativeClientConfiguration.partSizeBytes();
-
-        if (contentLength <= partSize) {
+        if (contentLength <= partSizeInBytes) {
             log.debug(() -> "Starting the copy as a single copy part request");
             copyInOneChunk(copyObjectRequest, returnFuture);
         } else {
@@ -132,7 +130,7 @@ private void doCopyInParts(CopyObjectRequest copyObjectRequest,
                                Long contentLength,
                                CompletableFuture<CopyObjectResponse> returnFuture,
                                String uploadId) {
-        long optimalPartSize = calculateOptimalPartSizeForCopy(s3NativeClientConfiguration.partSizeBytes());
+        long optimalPartSize = calculateOptimalPartSizeForCopy(partSizeInBytes);
 
         int partCount = determinePartCount(contentLength, optimalPartSize);
 
@@ -281,8 +279,7 @@ private long calculateOptimalPartSizeForCopy(long contentLengthOfSource) {
         double optimalPartSize = contentLengthOfSource / (double) MAX_UPLOAD_PARTS;
 
         optimalPartSize = Math.ceil(optimalPartSize);
-        return (long) Math.max(optimalPartSize,
-                               s3NativeClientConfiguration.partSizeBytes());
+        return (long) Math.max(optimalPartSize, partSizeInBytes);
     }
 
     private void copyInOneChunk(CopyObjectRequest copyObjectRequest,

services/s3/src/main/java/software/amazon/awssdk/services/s3/internal/crt/DefaultS3CrtAsyncClient.java

@@ -18,6 +18,7 @@
 import static software.amazon.awssdk.core.interceptor.SdkInternalExecutionAttribute.SDK_HTTP_EXECUTION_ATTRIBUTES;
 import static software.amazon.awssdk.services.s3.internal.crt.S3InternalSdkHttpExecutionAttribute.HTTP_CHECKSUM;
 import static software.amazon.awssdk.services.s3.internal.crt.S3InternalSdkHttpExecutionAttribute.OPERATION_NAME;
+import static software.amazon.awssdk.services.s3.internal.crt.S3NativeClientConfiguration.DEFAULT_PART_SIZE_IN_BYTES;
 
 import java.net.URI;
 import java.util.concurrent.CompletableFuture;
@@ -52,12 +53,13 @@
 @SdkInternalApi
 public final class DefaultS3CrtAsyncClient extends DelegatingS3AsyncClient implements S3CrtAsyncClient {
     private static final String CRT_CLIENT_CLASSPATH = "software.amazon.awssdk.crt.s3.S3Client";
-    private static S3NativeClientConfiguration s3NativeClientConfiguration;
     private final CopyObjectHelper copyObjectHelper;
 
     private DefaultS3CrtAsyncClient(DefaultS3CrtClientBuilder builder) {
         super(initializeS3AsyncClient(builder));
-        this.copyObjectHelper = new CopyObjectHelper((S3AsyncClient) delegate(), s3NativeClientConfiguration);
+        long partSizeInBytes = builder.minimalPartSizeInBytes == null ? DEFAULT_PART_SIZE_IN_BYTES :
+                               builder.minimalPartSizeInBytes;
+        this.copyObjectHelper = new CopyObjectHelper((S3AsyncClient) delegate(),  partSizeInBytes);
     }
 
     @Override
@@ -93,7 +95,7 @@ private static S3CrtAsyncHttpClient.Builder initializeS3CrtAsyncHttpClient(Defau
         Validate.isPositiveOrNull(builder.targetThroughputInGbps, "targetThroughputInGbps");
         Validate.isPositiveOrNull(builder.minimalPartSizeInBytes, "minimalPartSizeInBytes");
 
-        s3NativeClientConfiguration =
+        S3NativeClientConfiguration s3NativeClientConfiguration =
             S3NativeClientConfiguration.builder()
                                        .checksumValidationEnabled(builder.checksumValidationEnabled)
                                        .targetThroughputInGbps(builder.targetThroughputInGbps)
@@ -104,6 +106,7 @@ private static S3CrtAsyncHttpClient.Builder initializeS3CrtAsyncHttpClient(Defau
                                        .credentialsProvider(builder.credentialsProvider)
                                        .readBufferSizeInBytes(builder.readBufferSizeInBytes)
                                        .build();
+
         return S3CrtAsyncHttpClient.builder()
                                    .s3ClientConfiguration(s3NativeClientConfiguration);
     }

services/s3/src/main/java/software/amazon/awssdk/services/s3/internal/crt/S3NativeClientConfiguration.java

@@ -29,7 +29,7 @@
  */
 @SdkInternalApi
 public class S3NativeClientConfiguration implements SdkAutoCloseable {
-    private static final long DEFAULT_PART_SIZE_IN_BYTES = 8L * 1024 * 1024;
+    static final long DEFAULT_PART_SIZE_IN_BYTES = 8L * 1024 * 1024;
     private static final long DEFAULT_TARGET_THROUGHPUT_IN_GBPS = 10;
 
     private final String signingRegion;

services/s3/src/test/java/software/amazon/awssdk/services/s3/internal/crt/CopyObjectHelperTest.java

@@ -58,15 +58,11 @@ class CopyObjectHelperTest {
     private static final String MULTIPART_ID = "multipartId";
     private S3AsyncClient s3AsyncClient;
     private CopyObjectHelper copyHelper;
-    private S3NativeClientConfiguration s3NativeClientConfiguration;
 
     @BeforeEach
     public void setUp() {
-        s3NativeClientConfiguration = S3NativeClientConfiguration.builder()
-                                                                 .partSizeInBytes(1024L)
-                                                                 .build();
         s3AsyncClient = Mockito.mock(S3AsyncClient.class);
-        copyHelper = new CopyObjectHelper(s3AsyncClient, s3NativeClientConfiguration);
+        copyHelper = new CopyObjectHelper(s3AsyncClient, 1024L);
     }
 
     @Test