Skip to content

Commit d20116b

Browse files
joviegasjoviegas
committed
Merge branch 'master' into joviegas/ddb_eventual_consistency
2 parents 4f8bf60 + bfb1030 commit d20116b

10 files changed

Lines changed: 305 additions & 17 deletions

File tree

‎.changes/next-release/feature-AWSS3-233f74c.json‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,6 @@
1+
{
2+
"type": "feature",
3+
"category": "AWS S3",
4+
"contributor": "",
5+
"description": "Add support for using CRT's response file in the CRT based S3AsyncClient - CRT will directly write to the file when calling getObject with a Path."
6+
}

‎http-clients/apache-client/src/main/java/software/amazon/awssdk/http/apache/ApacheHttpClient.java‎

Lines changed: 5 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -92,9 +92,11 @@
9292
import software.amazon.awssdk.utils.Validate;
9393

9494
/**
95-
* An implementation of {@link SdkHttpClient} that uses Apache HTTP client to communicate with the service. This is the most
96-
* powerful synchronous client that adds an extra dependency and additional startup latency in exchange for more functionality,
97-
* like support for HTTP proxies.
95+
*
96+
* An implementation of {@link SdkHttpClient} that uses Apache HttpClient 4.x to communicate with the service. This is a
97+
* full-featured synchronous client that adds an extra dependency and higher startup latency compared to
98+
* <a href="https://docs.aws.amazon.com/sdk-for-java/latest/developer-guide/http-configuration-url.html">UrlConnectionHttpClient</a>
99+
* in exchange for more functionality, like support for HTTP proxies.
98100
*
99101
* <p>See software.amazon.awssdk.http.urlconnection.UrlConnectionHttpClient for an alternative implementation.</p>
100102
*

‎http-clients/apache5-client/src/main/java/software/amazon/awssdk/http/apache5/Apache5HttpClient.java‎

Lines changed: 16 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -101,14 +101,26 @@
101101
import software.amazon.awssdk.utils.Validate;
102102

103103
/**
104-
* An implementation of {@link SdkHttpClient} that uses Apache5 HTTP client to communicate with the service. This is the most
105-
* powerful synchronous client that adds an extra dependency and additional startup latency in exchange for more functionality,
106-
* like support for HTTP proxies.
104+
* An implementation of {@link SdkHttpClient} that uses Apache HttpClient 5.x to communicate with the service. This is a
105+
* full-featured synchronous client that adds an extra dependency and higher startup latency compared to
106+
* <a href="https://docs.aws.amazon.com/sdk-for-java/latest/developer-guide/http-configuration-url.html">UrlConnectionHttpClient</a>
107+
* in exchange for more functionality, like support for HTTP proxies.
107108
*
108-
* <p>See software.amazon.awssdk.http.urlconnection.UrlConnectionHttpClient for an alternative implementation.</p>
109+
* <p>This client uses Apache HttpClient 5.x, which provides the following
110+
* improvements over the Apache HttpClient 4.5.x based client:</p>
111+
* <ul>
112+
* <li>Modern Java ecosystem compatibility including virtual thread support for Java 21</li>
113+
* <li>Active maintenance with regular security updates</li>
114+
* <li>Enhanced logging flexibility through SLF4J (replacing problematic JCL dependencies)</li>
115+
* </ul>
116+
* <p><b>Note:</b> Performance characteristics between Apache 4.5.x and 5.x clients are similar.</p>
117+
* <p>See
118+
* <a href="https://docs.aws.amazon.com/sdk-for-java/latest/developer-guide/http-configuration-url.html">UrlConnectionHttpClient</a>
119+
* for a lighter-weight alternative implementation.</p>
109120
*
110121
* <p>This can be created via {@link #builder()}</p>
111122
*/
123+
112124
@SdkPreviewApi
113125
@SdkPublicApi
114126
public final class Apache5HttpClient implements SdkHttpClient {

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

Lines changed: 110 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,110 @@
1+
/*
2+
* Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
3+
*
4+
* Licensed under the Apache License, Version 2.0 (the "License").
5+
* You may not use this file except in compliance with the License.
6+
* A copy of the License is located at
7+
*
8+
* http://aws.amazon.com/apache2.0
9+
*
10+
* or in the "license" file accompanying this file. This file is distributed
11+
* on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either
12+
* express or implied. See the License for the specific language governing
13+
* permissions and limitations under the License.
14+
*/
15+
16+
package software.amazon.awssdk.services.s3.internal.crt;
17+
18+
import java.nio.ByteBuffer;
19+
import java.util.concurrent.CompletableFuture;
20+
import java.util.function.Consumer;
21+
import org.reactivestreams.Subscriber;
22+
import org.reactivestreams.Subscription;
23+
import software.amazon.awssdk.annotations.SdkInternalApi;
24+
import software.amazon.awssdk.core.async.AsyncResponseTransformer;
25+
import software.amazon.awssdk.core.async.SdkPublisher;
26+
import software.amazon.awssdk.utils.Logger;
27+
28+
/**
29+
* When the CRT Response File option is used in a request, the body is streamed directly to the file.
30+
* The S3CrtResponseHandlerAdapter in this case will never receive a response body but will call onStream
31+
* when the request is complete with a publisher that will complete immediately.
32+
* This transformer is effectively a no-op transformer that waits for the stream to complete and then
33+
* completes the future with the response.
34+
*
35+
* @param <ResponseT> Pojo response type.
36+
*/
37+
@SdkInternalApi
38+
public final class CrtResponseFileResponseTransformer<ResponseT> implements AsyncResponseTransformer<ResponseT,
39+
ResponseT> {
40+
41+
private static final Logger log = Logger.loggerFor(CrtResponseFileResponseTransformer.class);
42+
43+
private volatile CompletableFuture<Void> cf;
44+
private volatile ResponseT response;
45+
46+
@Override
47+
public CompletableFuture<ResponseT> prepare() {
48+
cf = new CompletableFuture<>();
49+
return cf.thenApply(ignored -> response);
50+
}
51+
52+
@Override
53+
public void onResponse(ResponseT response) {
54+
this.response = response;
55+
}
56+
57+
@Override
58+
public void onStream(SdkPublisher<ByteBuffer> publisher) {
59+
publisher.subscribe(new OnCompleteSubscriber(cf, this::exceptionOccurred));
60+
}
61+
62+
@Override
63+
public void exceptionOccurred(Throwable throwable) {
64+
if (cf != null) {
65+
cf.completeExceptionally(throwable);
66+
} else {
67+
log.warn(() -> "An exception occurred before the call to prepare() was able to instantiate the CompletableFuture. "
68+
+ "The future cannot be completed exceptionally because it is null");
69+
}
70+
}
71+
72+
private static final class OnCompleteSubscriber implements Subscriber<ByteBuffer> {
73+
74+
private final CompletableFuture<Void> future;
75+
private final Consumer<Throwable> onErrorMethod;
76+
private Subscription subscription;
77+
78+
private OnCompleteSubscriber(CompletableFuture<Void> future, Consumer<Throwable> onErrorMethod) {
79+
this.future = future;
80+
this.onErrorMethod = onErrorMethod;
81+
}
82+
83+
@Override
84+
public void onSubscribe(Subscription s) {
85+
if (this.subscription != null) {
86+
s.cancel();
87+
return;
88+
}
89+
this.subscription = s;
90+
// do not request data from the subscription since body is written directly to file
91+
}
92+
93+
@Override
94+
public void onNext(ByteBuffer byteBuffer) {
95+
// The response body is streamed directly to the file - this method should never be called.
96+
// ensure the future is completed exceptionally if this occurs
97+
onErrorMethod.accept(new IllegalStateException("OnCompleteSubscriber received unexpected call to onNext."));
98+
}
99+
100+
@Override
101+
public void onError(Throwable throwable) {
102+
onErrorMethod.accept(throwable);
103+
}
104+
105+
@Override
106+
public void onComplete() {
107+
future.complete(null);
108+
}
109+
}
110+
}

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

Lines changed: 28 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -42,6 +42,7 @@
4242
import software.amazon.awssdk.awscore.AwsRequestOverrideConfiguration;
4343
import software.amazon.awssdk.awscore.retry.AwsRetryStrategy;
4444
import software.amazon.awssdk.core.SdkRequest;
45+
import software.amazon.awssdk.core.async.AsyncResponseTransformer;
4546
import software.amazon.awssdk.core.checksums.ChecksumValidation;
4647
import software.amazon.awssdk.core.checksums.RequestChecksumCalculation;
4748
import software.amazon.awssdk.core.checksums.ResponseChecksumValidation;
@@ -57,6 +58,7 @@
5758
import software.amazon.awssdk.core.signer.NoOpSigner;
5859
import software.amazon.awssdk.crt.io.ExponentialBackoffRetryOptions;
5960
import software.amazon.awssdk.crt.io.StandardRetryOptions;
61+
import software.amazon.awssdk.crt.s3.S3MetaRequestOptions;
6062
import software.amazon.awssdk.http.SdkHttpExecutionAttributes;
6163
import software.amazon.awssdk.identity.spi.AwsCredentialsIdentity;
6264
import software.amazon.awssdk.identity.spi.IdentityProvider;
@@ -72,6 +74,8 @@
7274
import software.amazon.awssdk.services.s3.internal.s3express.S3ExpressUtils;
7375
import software.amazon.awssdk.services.s3.model.CopyObjectRequest;
7476
import software.amazon.awssdk.services.s3.model.CopyObjectResponse;
77+
import software.amazon.awssdk.services.s3.model.GetObjectRequest;
78+
import software.amazon.awssdk.services.s3.model.GetObjectResponse;
7579
import software.amazon.awssdk.services.s3.model.PutObjectRequest;
7680
import software.amazon.awssdk.services.s3.model.PutObjectResponse;
7781
import software.amazon.awssdk.utils.CollectionUtils;
@@ -80,6 +84,9 @@
8084
@SdkInternalApi
8185
public final class DefaultS3CrtAsyncClient extends DelegatingS3AsyncClient implements S3CrtAsyncClient {
8286
public static final ExecutionAttribute<Path> OBJECT_FILE_PATH = new ExecutionAttribute<>("objectFilePath");
87+
public static final ExecutionAttribute<Path> RESPONSE_FILE_PATH = new ExecutionAttribute<>("responseFilePath");
88+
public static final ExecutionAttribute<S3MetaRequestOptions.ResponseFileOption> RESPONSE_FILE_OPTION =
89+
new ExecutionAttribute<>("responseFileOption");
8390
private static final String CRT_CLIENT_CLASSPATH = "software.amazon.awssdk.crt.s3.S3Client";
8491
private final CopyObjectHelper copyObjectHelper;
8592

@@ -106,6 +113,22 @@ public CompletableFuture<PutObjectResponse> putObject(PutObjectRequest putObject
106113
new CrtContentLengthOnlyAsyncFileRequestBody(sourcePath));
107114
}
108115

116+
@Override
117+
public CompletableFuture<GetObjectResponse> getObject(GetObjectRequest getObjectRequest, Path destinationPath) {
118+
AsyncResponseTransformer<GetObjectResponse, GetObjectResponse> responseTransformer =
119+
new CrtResponseFileResponseTransformer<>();
120+
121+
AwsRequestOverrideConfiguration overrideConfig =
122+
getObjectRequest.overrideConfiguration()
123+
.map(config -> config.toBuilder().putExecutionAttribute(RESPONSE_FILE_PATH, destinationPath))
124+
.orElseGet(() -> AwsRequestOverrideConfiguration.builder()
125+
.putExecutionAttribute(RESPONSE_FILE_PATH,
126+
destinationPath))
127+
.build();
128+
129+
return getObject(getObjectRequest.toBuilder().overrideConfiguration(overrideConfig).build(), responseTransformer);
130+
}
131+
109132
@Override
110133
public CompletableFuture<CopyObjectResponse> copyObject(CopyObjectRequest copyObjectRequest) {
111134
return copyObjectHelper.copyObject(copyObjectRequest);
@@ -243,7 +266,7 @@ public DefaultS3CrtClientBuilder credentialsProvider(AwsCredentialsProvider cred
243266

244267
@Override
245268
public DefaultS3CrtClientBuilder credentialsProvider(
246-
IdentityProvider<? extends AwsCredentialsIdentity> credentialsProvider) {
269+
IdentityProvider<? extends AwsCredentialsIdentity> credentialsProvider) {
247270
this.credentialsProvider = credentialsProvider;
248271
return this;
249272
}
@@ -396,6 +419,10 @@ public void afterMarshalling(Context.AfterMarshalling context,
396419
executionAttributes.getAttribute(SdkInternalExecutionAttribute.REQUEST_CHECKSUM_CALCULATION))
397420
.put(RESPONSE_CHECKSUM_VALIDATION,
398421
executionAttributes.getAttribute(SdkInternalExecutionAttribute.RESPONSE_CHECKSUM_VALIDATION))
422+
.put(S3InternalSdkHttpExecutionAttribute.RESPONSE_FILE_PATH,
423+
executionAttributes.getAttribute(RESPONSE_FILE_PATH))
424+
.put(S3InternalSdkHttpExecutionAttribute.RESPONSE_FILE_OPTION,
425+
executionAttributes.getAttribute(RESPONSE_FILE_OPTION))
399426
.build();
400427

401428
// We rely on CRT to perform checksum validation, disable SDK flexible checksum implementation

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

Lines changed: 18 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -24,6 +24,8 @@
2424
import static software.amazon.awssdk.services.s3.internal.crt.S3InternalSdkHttpExecutionAttribute.OPERATION_NAME;
2525
import static software.amazon.awssdk.services.s3.internal.crt.S3InternalSdkHttpExecutionAttribute.REQUEST_CHECKSUM_CALCULATION;
2626
import static software.amazon.awssdk.services.s3.internal.crt.S3InternalSdkHttpExecutionAttribute.RESPONSE_CHECKSUM_VALIDATION;
27+
import static software.amazon.awssdk.services.s3.internal.crt.S3InternalSdkHttpExecutionAttribute.RESPONSE_FILE_OPTION;
28+
import static software.amazon.awssdk.services.s3.internal.crt.S3InternalSdkHttpExecutionAttribute.RESPONSE_FILE_PATH;
2729
import static software.amazon.awssdk.services.s3.internal.crt.S3InternalSdkHttpExecutionAttribute.SIGNING_NAME;
2830
import static software.amazon.awssdk.services.s3.internal.crt.S3InternalSdkHttpExecutionAttribute.SIGNING_REGION;
2931
import static software.amazon.awssdk.services.s3.internal.crt.S3InternalSdkHttpExecutionAttribute.USE_S3_EXPRESS_AUTH;
@@ -135,11 +137,6 @@ public CompletableFuture<Void> execute(AsyncExecuteRequest asyncRequest) {
135137
HttpRequest httpRequest = toCrtRequest(asyncRequest);
136138
SdkHttpExecutionAttributes httpExecutionAttributes = asyncRequest.httpExecutionAttributes();
137139
CompletableFuture<S3MetaRequestWrapper> s3MetaRequestFuture = new CompletableFuture<>();
138-
S3CrtResponseHandlerAdapter responseHandler =
139-
new S3CrtResponseHandlerAdapter(executeFuture,
140-
asyncRequest.responseHandler(),
141-
httpExecutionAttributes.getAttribute(CRT_PROGRESS_LISTENER),
142-
s3MetaRequestFuture);
143140

144141
String operationName = asyncRequest.httpExecutionAttributes().getAttribute(OPERATION_NAME);
145142
S3MetaRequestOptions.MetaRequestType requestType = requestType(operationName);
@@ -156,6 +153,16 @@ public CompletableFuture<Void> execute(AsyncExecuteRequest asyncRequest) {
156153
ChecksumConfig checksumConfig = checksumConfig(httpChecksum, requestType, requestChecksumCalculation,
157154
responseChecksumValidation);
158155

156+
Path responseFilePath = httpExecutionAttributes.getAttribute(RESPONSE_FILE_PATH);
157+
S3MetaRequestOptions.ResponseFileOption responseFileOption = httpExecutionAttributes.getAttribute(RESPONSE_FILE_OPTION);
158+
159+
S3CrtResponseHandlerAdapter responseHandler =
160+
new S3CrtResponseHandlerAdapter(
161+
executeFuture,
162+
asyncRequest.responseHandler(),
163+
httpExecutionAttributes.getAttribute(CRT_PROGRESS_LISTENER),
164+
s3MetaRequestFuture);
165+
159166
URI endpoint = getEndpoint(uri);
160167

161168
AwsSigningConfig signingConfig = awsSigningConfig(signingRegion, httpExecutionAttributes);
@@ -169,7 +176,12 @@ public CompletableFuture<Void> execute(AsyncExecuteRequest asyncRequest) {
169176
.withResumeToken(resumeToken)
170177
.withOperationName(operationName)
171178
.withRequestFilePath(requestFilePath)
172-
.withSigningConfig(signingConfig);
179+
.withSigningConfig(signingConfig)
180+
.withResponseFilePath(responseFilePath);
181+
182+
if (responseFileOption != null) {
183+
requestOptions = requestOptions.withResponseFileOption(responseFileOption);
184+
}
173185

174186
try {
175187
S3MetaRequestWrapper requestWrapper = new S3MetaRequestWrapper(crtS3Client.makeMetaRequest(requestOptions));

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

Lines changed: 7 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -21,6 +21,7 @@
2121
import software.amazon.awssdk.core.checksums.ResponseChecksumValidation;
2222
import software.amazon.awssdk.core.interceptor.trait.HttpChecksum;
2323
import software.amazon.awssdk.crt.s3.ResumeToken;
24+
import software.amazon.awssdk.crt.s3.S3MetaRequestOptions;
2425
import software.amazon.awssdk.http.SdkHttpExecutionAttribute;
2526
import software.amazon.awssdk.regions.Region;
2627

@@ -57,6 +58,12 @@ public final class S3InternalSdkHttpExecutionAttribute<T> extends SdkHttpExecuti
5758
public static final S3InternalSdkHttpExecutionAttribute<ResponseChecksumValidation> RESPONSE_CHECKSUM_VALIDATION =
5859
new S3InternalSdkHttpExecutionAttribute<>(ResponseChecksumValidation.class);
5960

61+
public static final S3InternalSdkHttpExecutionAttribute<Path> RESPONSE_FILE_PATH =
62+
new S3InternalSdkHttpExecutionAttribute<>(Path.class);
63+
64+
public static final S3InternalSdkHttpExecutionAttribute<S3MetaRequestOptions.ResponseFileOption> RESPONSE_FILE_OPTION =
65+
new S3InternalSdkHttpExecutionAttribute<>(S3MetaRequestOptions.ResponseFileOption.class);
66+
6067
private S3InternalSdkHttpExecutionAttribute(Class<T> valueClass) {
6168
super(valueClass);
6269
}

0 commit comments

Comments
 (0)