Fix CRT S3 getObject file handling

.changes/next-release/bugfix-AWSS3-9d4c2a1.json

@@ -0,0 +1,6 @@
+{
+    "type": "bugfix",
+    "category": "Amazon S3",
+    "contributor": "",
+    "description": "Prevent the CRT-based S3 async client from replacing a pre-existing destination when using `getObject(request, Path)`. Failed or cancelled downloads now delete files they create. Existing destinations surface `FileAlreadyExistsException`; callers that require replacement can use `AsyncResponseTransformer.toFile` with `FileTransformerConfiguration.defaultCreateOrReplaceExisting()`. This fixes an inconsistency introduced in 2.32.11."
+}

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

@@ -91,6 +91,8 @@ public final class DefaultS3CrtAsyncClient extends DelegatingS3AsyncClient imple
     public static final ExecutionAttribute<Path> RESPONSE_FILE_PATH = new ExecutionAttribute<>("responseFilePath");
     public static final ExecutionAttribute<S3MetaRequestOptions.ResponseFileOption> RESPONSE_FILE_OPTION =
         new ExecutionAttribute<>("responseFileOption");
+    public static final ExecutionAttribute<Boolean> RESPONSE_FILE_DELETE_ON_FAILURE =
+        new ExecutionAttribute<>("responseFileDeleteOnFailure");
     private static final String CRT_CLIENT_CLASSPATH = "software.amazon.awssdk.crt.s3.S3Client";
     private final CopyObjectHelper copyObjectHelper;
 
@@ -124,10 +126,12 @@ public CompletableFuture<GetObjectResponse> getObject(GetObjectRequest getObject
 
         AwsRequestOverrideConfiguration overrideConfig =
             getObjectRequest.overrideConfiguration()
-                            .map(config -> config.toBuilder().putExecutionAttribute(RESPONSE_FILE_PATH, destinationPath))
-                            .orElseGet(() -> AwsRequestOverrideConfiguration.builder()
-                                                                            .putExecutionAttribute(RESPONSE_FILE_PATH,
-                                                                                                   destinationPath))
+                            .map(AwsRequestOverrideConfiguration::toBuilder)
+                            .orElseGet(AwsRequestOverrideConfiguration::builder)
+                            .putExecutionAttribute(RESPONSE_FILE_PATH, destinationPath)
+                            .putExecutionAttribute(RESPONSE_FILE_OPTION,
+                                                   S3MetaRequestOptions.ResponseFileOption.CREATE_NEW)
+                            .putExecutionAttribute(RESPONSE_FILE_DELETE_ON_FAILURE, true)
                             .build();
 
         return getObject(getObjectRequest.toBuilder().overrideConfiguration(overrideConfig).build(), responseTransformer);
@@ -439,7 +443,9 @@ public void afterMarshalling(Context.AfterMarshalling context,
                    .put(S3InternalSdkHttpExecutionAttribute.RESPONSE_FILE_PATH,
                         executionAttributes.getAttribute(RESPONSE_FILE_PATH))
                    .put(S3InternalSdkHttpExecutionAttribute.RESPONSE_FILE_OPTION,
-                        executionAttributes.getAttribute(RESPONSE_FILE_OPTION));
+                        executionAttributes.getAttribute(RESPONSE_FILE_OPTION))
+                   .put(S3InternalSdkHttpExecutionAttribute.RESPONSE_FILE_DELETE_ON_FAILURE,
+                        executionAttributes.getAttribute(RESPONSE_FILE_DELETE_ON_FAILURE));
 
             SdkRequest request = context.request();
             if (request instanceof AwsRequest) {

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

@@ -24,6 +24,7 @@
 import static software.amazon.awssdk.services.s3.internal.crt.S3InternalSdkHttpExecutionAttribute.OPERATION_NAME;
 import static software.amazon.awssdk.services.s3.internal.crt.S3InternalSdkHttpExecutionAttribute.REQUEST_CHECKSUM_CALCULATION;
 import static software.amazon.awssdk.services.s3.internal.crt.S3InternalSdkHttpExecutionAttribute.RESPONSE_CHECKSUM_VALIDATION;
+import static software.amazon.awssdk.services.s3.internal.crt.S3InternalSdkHttpExecutionAttribute.RESPONSE_FILE_DELETE_ON_FAILURE;
 import static software.amazon.awssdk.services.s3.internal.crt.S3InternalSdkHttpExecutionAttribute.RESPONSE_FILE_OPTION;
 import static software.amazon.awssdk.services.s3.internal.crt.S3InternalSdkHttpExecutionAttribute.RESPONSE_FILE_PATH;
 import static software.amazon.awssdk.services.s3.internal.crt.S3InternalSdkHttpExecutionAttribute.SIGNING_NAME;
@@ -32,6 +33,7 @@
 import static software.amazon.awssdk.utils.FunctionalUtils.invokeSafely;
 
 import java.net.URI;
+import java.nio.file.FileAlreadyExistsException;
 import java.nio.file.Path;
 import java.time.Duration;
 import java.util.ArrayList;
@@ -43,6 +45,7 @@
 import software.amazon.awssdk.core.checksums.RequestChecksumCalculation;
 import software.amazon.awssdk.core.checksums.ResponseChecksumValidation;
 import software.amazon.awssdk.core.interceptor.trait.HttpChecksum;
+import software.amazon.awssdk.crt.CrtRuntimeException;
 import software.amazon.awssdk.crt.auth.credentials.CredentialsProvider;
 import software.amazon.awssdk.crt.auth.signing.AwsSigningConfig;
 import software.amazon.awssdk.crt.http.HttpHeader;
@@ -159,6 +162,7 @@ public CompletableFuture<Void> execute(AsyncExecuteRequest asyncRequest) {
 
         Path responseFilePath = httpExecutionAttributes.getAttribute(RESPONSE_FILE_PATH);
         S3MetaRequestOptions.ResponseFileOption responseFileOption = httpExecutionAttributes.getAttribute(RESPONSE_FILE_OPTION);
+        Boolean responseFileDeleteOnFailure = httpExecutionAttributes.getAttribute(RESPONSE_FILE_DELETE_ON_FAILURE);
 
         S3CrtResponseHandlerAdapter responseHandler =
             new S3CrtResponseHandlerAdapter(
@@ -186,6 +190,9 @@ public CompletableFuture<Void> execute(AsyncExecuteRequest asyncRequest) {
         if (responseFileOption != null) {
             requestOptions = requestOptions.withResponseFileOption(responseFileOption);
         }
+        if (responseFileDeleteOnFailure != null) {
+            requestOptions = requestOptions.withResponseFileDeleteOnFailure(responseFileDeleteOnFailure);
+        }
 
         CrtCredentialsProviderAdapter requestCredentialsAdapter =
             httpExecutionAttributes.getAttribute(S3InternalSdkHttpExecutionAttribute.CRT_CREDENTIALS_PROVIDER_ADAPTER);
@@ -204,6 +211,16 @@ public CompletableFuture<Void> execute(AsyncExecuteRequest asyncRequest) {
             if (requestCredentialsAdapter != null) {
                 requestCredentialsAdapter.close();
             }
+            if (isResponseFileAlreadyExistsError(t, responseFilePath)) {
+                FileAlreadyExistsException fileAlreadyExistsException =
+                    new FileAlreadyExistsException(responseFilePath.toString());
+                fileAlreadyExistsException.addSuppressed(t);
+                // Native initialization failed before creating a meta-request. Complete its placeholder first because
+                // completing executeFuture invokes the response adapter synchronously, which otherwise waits for a timeout.
+                s3MetaRequestFuture.complete(null);
+                executeFuture.completeExceptionally(fileAlreadyExistsException);
+                return executeFuture;
+            }
             throw t;
         } finally {
             signingConfig.close();
@@ -216,6 +233,12 @@ public CompletableFuture<Void> execute(AsyncExecuteRequest asyncRequest) {
         return executeFuture;
     }
 
+    private static boolean isResponseFileAlreadyExistsError(Throwable throwable, Path responseFilePath) {
+        return responseFilePath != null
+               && throwable instanceof CrtRuntimeException
+               && "AWS_ERROR_S3_RECV_FILE_ALREADY_EXISTS".equals(((CrtRuntimeException) throwable).errorName);
+    }
+
     private AwsSigningConfig awsSigningConfig(Region signingRegion, SdkHttpExecutionAttributes httpExecutionAttributes) {
         CrtCredentialsProviderAdapter requestAdapter =
             httpExecutionAttributes.getAttribute(S3InternalSdkHttpExecutionAttribute.CRT_CREDENTIALS_PROVIDER_ADAPTER);

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

@@ -64,6 +64,9 @@ public final class S3InternalSdkHttpExecutionAttribute<T> extends SdkHttpExecuti
     public static final S3InternalSdkHttpExecutionAttribute<S3MetaRequestOptions.ResponseFileOption> RESPONSE_FILE_OPTION =
         new S3InternalSdkHttpExecutionAttribute<>(S3MetaRequestOptions.ResponseFileOption.class);
 
+    public static final S3InternalSdkHttpExecutionAttribute<Boolean> RESPONSE_FILE_DELETE_ON_FAILURE =
+        new S3InternalSdkHttpExecutionAttribute<>(Boolean.class);
+
     public static final S3InternalSdkHttpExecutionAttribute<CrtCredentialsProviderAdapter> CRT_CREDENTIALS_PROVIDER_ADAPTER =
         new S3InternalSdkHttpExecutionAttribute<>(CrtCredentialsProviderAdapter.class);
 

services/s3/src/test/java/software/amazon/awssdk/services/s3/crt/CrtDownloadErrorTest.java

@@ -15,27 +15,39 @@
 
 package software.amazon.awssdk.services.s3.crt;
 
+import static com.github.tomakehurst.wiremock.client.WireMock.anyRequestedFor;
+import static com.github.tomakehurst.wiremock.client.WireMock.anyUrl;
+import static com.github.tomakehurst.wiremock.client.WireMock.exactly;
 import static com.github.tomakehurst.wiremock.client.WireMock.get;
 import static com.github.tomakehurst.wiremock.client.WireMock.head;
 import static com.github.tomakehurst.wiremock.client.WireMock.stubFor;
 import static com.github.tomakehurst.wiremock.client.WireMock.urlPathEqualTo;
+import static com.github.tomakehurst.wiremock.client.WireMock.verify;
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
 
 import com.github.tomakehurst.wiremock.client.WireMock;
 import com.github.tomakehurst.wiremock.junit5.WireMockRuntimeInfo;
 import com.github.tomakehurst.wiremock.junit5.WireMockTest;
 import java.net.URI;
 import java.nio.charset.StandardCharsets;
+import java.nio.file.FileAlreadyExistsException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.concurrent.CompletableFuture;
 import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.BeforeAll;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.Timeout;
+import org.junit.jupiter.api.io.TempDir;
+import software.amazon.awssdk.core.FileTransformerConfiguration;
 import software.amazon.awssdk.core.async.AsyncResponseTransformer;
 import software.amazon.awssdk.crt.Log;
 import software.amazon.awssdk.regions.Region;
 import software.amazon.awssdk.services.s3.S3AsyncClient;
+import software.amazon.awssdk.services.s3.model.GetObjectResponse;
 import software.amazon.awssdk.services.s3.model.S3Exception;
 
 @WireMockTest
@@ -44,6 +56,8 @@ public class CrtDownloadErrorTest {
     private static final String BUCKET = "my-bucket";
     private static final String KEY = "my-key";
     private S3AsyncClient s3;
+    @TempDir
+    private Path tempDir;
 
     @BeforeAll
     public static void setUpBeforeAll() {
@@ -120,6 +134,60 @@ public void getObject_headObjectOk_getObjectOk_operationSucceeds() {
         assertThat(objectContent.getBytes(StandardCharsets.UTF_8)).isEqualTo(content);
     }
 
+    @Test
+    public void getObjectToPath_success_writesFile() throws Exception {
+        byte[] content = "hello".getBytes(StandardCharsets.UTF_8);
+        stubSuccessfulDownload(content);
+        Path destination = tempDir.resolve("download");
+
+        s3.getObject(r -> r.bucket(BUCKET).key(KEY), destination).join();
+
+        assertThat(Files.readAllBytes(destination)).isEqualTo(content);
+    }
+
+    @Test
+    public void getObjectWithReplaceTransformer_existingFile_replacesFile() throws Exception {
+        byte[] content = "hello".getBytes(StandardCharsets.UTF_8);
+        stubSuccessfulDownload(content);
+        Path destination = tempDir.resolve("download");
+        Files.write(destination, "original".getBytes(StandardCharsets.UTF_8));
+
+        s3.getObject(r -> r.bucket(BUCKET).key(KEY),
+                     AsyncResponseTransformer.toFile(
+                         destination, FileTransformerConfiguration.defaultCreateOrReplaceExisting())).join();
+
+        assertThat(Files.readAllBytes(destination)).isEqualTo(content);
+    }
+
+    @Test
+    public void getObjectToPath_existingFile_failsAsynchronouslyAndPreservesFile() throws Exception {
+        byte[] content = "hello".getBytes(StandardCharsets.UTF_8);
+        stubSuccessfulDownload(content);
+        Path destination = tempDir.resolve("download");
+        byte[] originalContent = "original".getBytes(StandardCharsets.UTF_8);
+        Files.write(destination, originalContent);
+
+        CompletableFuture<GetObjectResponse> future =
+            assertDoesNotThrow(() -> s3.getObject(r -> r.bucket(BUCKET).key(KEY), destination));
+
+        assertThatThrownBy(future::join)
+            .hasRootCauseInstanceOf(FileAlreadyExistsException.class)
+            .hasRootCauseMessage(destination.toString());
+        assertThat(Files.readAllBytes(destination)).isEqualTo(originalContent);
+        verify(exactly(0), anyRequestedFor(anyUrl()));
+    }
+
+    @Test
+    public void getObjectToPath_failedDownload_deletesFile() {
+        String requestPath = String.format("/%s/%s", BUCKET, KEY);
+        stubFor(head(urlPathEqualTo(requestPath)).willReturn(WireMock.aResponse().withStatus(404)));
+        Path destination = tempDir.resolve("download");
+
+        assertThatThrownBy(s3.getObject(r -> r.bucket(BUCKET).key(KEY), destination)::join)
+            .hasCauseInstanceOf(S3Exception.class);
+        assertThat(destination).doesNotExist();
+    }
+
     @Test
     public void getObject_headObjectThrows_operationThrows() {
         String path = String.format("/%s/%s", BUCKET, KEY);
@@ -132,4 +200,15 @@ public void getObject_headObjectThrows_operationThrows() {
             .hasCauseInstanceOf(S3Exception.class)
             .hasMessageContaining("Status Code: 403");
     }
+
+    private void stubSuccessfulDownload(byte[] content) {
+        String path = String.format("/%s/%s", BUCKET, KEY);
+        stubFor(head(urlPathEqualTo(path))
+                    .willReturn(WireMock.aResponse()
+                                        .withStatus(200)
+                                        .withHeader("ETag", "etag")
+                                        .withHeader("Content-Length", Integer.toString(content.length))));
+        stubFor(get(urlPathEqualTo(path))
+                    .willReturn(WireMock.aResponse().withStatus(200).withBody(content)));
+    }
 }

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

@@ -18,6 +18,7 @@
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.assertj.core.api.Assertions.assertThatThrownBy;
 
+import java.nio.file.Path;
 import java.util.concurrent.atomic.AtomicReference;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.params.ParameterizedTest;
@@ -33,6 +34,7 @@
 import software.amazon.awssdk.core.interceptor.ExecutionAttributes;
 import software.amazon.awssdk.core.interceptor.ExecutionInterceptor;
 import software.amazon.awssdk.core.interceptor.SdkInternalExecutionAttribute;
+import software.amazon.awssdk.crt.s3.S3MetaRequestOptions.ResponseFileOption;
 import software.amazon.awssdk.http.SdkHttpExecutionAttributes;
 import software.amazon.awssdk.identity.spi.AwsCredentialsIdentity;
 import software.amazon.awssdk.identity.spi.IdentityProvider;
@@ -41,6 +43,7 @@
 import software.amazon.awssdk.services.s3.S3AsyncClient;
 import software.amazon.awssdk.services.s3.endpoints.S3ClientContextParams;
 import software.amazon.awssdk.services.s3.internal.crossregion.S3CrossRegionAsyncClient;
+import software.amazon.awssdk.testutils.RandomTempFile;
 import software.amazon.awssdk.utils.AttributeMap;
 import software.amazon.awssdk.utils.MapUtils;
 
@@ -90,6 +93,40 @@ public void beforeExecution(Context.BeforeExecution context, ExecutionAttributes
         }
     }
 
+    @Test
+    void getObjectWithPath_shouldConfigureResponseFile() {
+        AtomicReference<SdkHttpExecutionAttributes> capturedAttributes = new AtomicReference<>();
+        ExecutionInterceptor captor = new ExecutionInterceptor() {
+            @Override
+            public void beforeTransmission(Context.BeforeTransmission context, ExecutionAttributes executionAttributes) {
+                capturedAttributes.set(
+                    executionAttributes.getAttribute(SdkInternalExecutionAttribute.SDK_HTTP_EXECUTION_ATTRIBUTES));
+                throw new RuntimeException("STOP");
+            }
+        };
+
+        DefaultS3CrtAsyncClient.DefaultS3CrtClientBuilder builder =
+            (DefaultS3CrtAsyncClient.DefaultS3CrtClientBuilder) S3CrtAsyncClient.builder();
+        builder.addExecutionInterceptor(captor);
+        Path destination = RandomTempFile.randomUncreatedFile().toPath();
+
+        try (S3AsyncClient client = builder.region(Region.US_EAST_1)
+                                           .credentialsProvider(StaticCredentialsProvider.create(
+                                               AwsBasicCredentials.create("key", "secret")))
+                                           .build()) {
+            assertThatThrownBy(() -> client.getObject(r -> r.bucket("bucket").key("key"), destination).join())
+                .hasMessageContaining("STOP");
+        }
+
+        SdkHttpExecutionAttributes attributes = capturedAttributes.get();
+        assertThat(attributes.getAttribute(S3InternalSdkHttpExecutionAttribute.RESPONSE_FILE_PATH))
+            .isEqualTo(destination);
+        assertThat(attributes.getAttribute(S3InternalSdkHttpExecutionAttribute.RESPONSE_FILE_OPTION))
+            .isEqualTo(ResponseFileOption.CREATE_NEW);
+        assertThat(attributes.getAttribute(S3InternalSdkHttpExecutionAttribute.RESPONSE_FILE_DELETE_ON_FAILURE))
+            .isTrue();
+    }
+
     @ParameterizedTest
     @ValueSource(longs = {0, -1L})
     void invalidConfig_shouldThrowException(long value) {