Skip to content

Commit 5d57416

Browse files
joviegasjoviegas
authored
S3CrossRegion Sync and Async Clients Redirect implementation (#4089)
* S3CrossRegionSyncClient Redirect implementation * Added implementation for Async client Decorator * Updated older Cross region test cases * Added paramterized test * Async Exception checged to completableException * Updated test cases and changes the Exception handling when HeadBucket Call fails * Handled Anna-karin's comments * Removed async execution of HeadBucket and attached it to the completableFuture of main request * Handled Zoe's comments * Added test case when Redirected after the Region is cached * Changed region constant to Region Type in Tests
1 parent f7e4188 commit 5d57416

File tree

11 files changed

+1290
-143
lines changed

11 files changed

+1290
-143
lines changed

‎services/s3/src/it/java/software/amazon/awssdk/services/s3/S3IntegrationTestBase.java

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -93,6 +93,7 @@ private static void createBucket(String bucketName, int retryCount) {
9393
.build())
9494
.build());
9595
} catch (S3Exception e) {
96+
e.printStackTrace();
9697
System.err.println("Error attempting to create bucket: " + bucketName);
9798
if (e.awsErrorDetails().errorCode().equals("BucketAlreadyOwnedByYou")) {
9899
System.err.printf("%s bucket already exists, likely leaked by a previous run\n", bucketName);

‎services/s3/src/main/java/software/amazon/awssdk/services/s3/internal/crossregion/S3CrossRegionAsyncClient.java

Lines changed: 96 additions & 46 deletions
Original file line numberDiff line numberDiff line change
@@ -15,29 +15,38 @@
1515

1616
package software.amazon.awssdk.services.s3.internal.crossregion;
1717

18+
import static software.amazon.awssdk.services.s3.internal.crossregion.utils.CrossRegionUtils.getBucketRegionFromException;
19+
import static software.amazon.awssdk.services.s3.internal.crossregion.utils.CrossRegionUtils.isS3RedirectException;
20+
import static software.amazon.awssdk.services.s3.internal.crossregion.utils.CrossRegionUtils.requestWithDecoratedEndpointProvider;
1821
import static software.amazon.awssdk.services.s3.internal.crossregion.utils.CrossRegionUtils.updateUserAgentInConfig;
1922

23+
import java.util.Map;
2024
import java.util.Optional;
2125
import java.util.concurrent.CompletableFuture;
26+
import java.util.concurrent.ConcurrentHashMap;
27+
import java.util.function.BiConsumer;
2228
import java.util.function.Function;
2329
import software.amazon.awssdk.annotations.SdkInternalApi;
2430
import software.amazon.awssdk.awscore.AwsRequestOverrideConfiguration;
25-
import software.amazon.awssdk.endpoints.Endpoint;
31+
import software.amazon.awssdk.regions.Region;
2632
import software.amazon.awssdk.services.s3.DelegatingS3AsyncClient;
2733
import software.amazon.awssdk.services.s3.S3AsyncClient;
28-
import software.amazon.awssdk.services.s3.endpoints.S3EndpointParams;
29-
import software.amazon.awssdk.services.s3.endpoints.S3EndpointProvider;
34+
import software.amazon.awssdk.services.s3.model.S3Exception;
3035
import software.amazon.awssdk.services.s3.model.S3Request;
36+
import software.amazon.awssdk.utils.CompletableFutureUtils;
3137

3238
@SdkInternalApi
3339
public final class S3CrossRegionAsyncClient extends DelegatingS3AsyncClient {
40+
41+
private final Map<String, Region> bucketToRegionCache = new ConcurrentHashMap<>();
42+
3443
public S3CrossRegionAsyncClient(S3AsyncClient s3Client) {
3544
super(s3Client);
3645
}
3746

3847
@Override
39-
protected <T extends S3Request, ReturnT> CompletableFuture<ReturnT>
40-
invokeOperation(T request, Function<T, CompletableFuture<ReturnT>> operation) {
48+
protected <T extends S3Request, ReturnT> CompletableFuture<ReturnT> invokeOperation(
49+
T request, Function<T, CompletableFuture<ReturnT>> operation) {
4150

4251
Optional<String> bucket = request.getValueForField("Bucket", String.class);
4352

@@ -47,53 +56,94 @@ public S3CrossRegionAsyncClient(S3AsyncClient s3Client) {
4756
if (!bucket.isPresent()) {
4857
return operation.apply(userAgentUpdatedRequest);
4958
}
50-
51-
return operation.apply(requestWithDecoratedEndpointProvider(userAgentUpdatedRequest, bucket.get()))
52-
.whenComplete((r, t) -> handleOperationFailure(t, bucket.get()));
59+
String bucketName = bucket.get();
60+
61+
CompletableFuture<ReturnT> returnFuture = new CompletableFuture<>();
62+
CompletableFuture<ReturnT> apiOperationFuture = bucketToRegionCache.containsKey(bucketName) ?
63+
operation.apply(
64+
requestWithDecoratedEndpointProvider(
65+
userAgentUpdatedRequest,
66+
() -> bucketToRegionCache.get(bucketName),
67+
serviceClientConfiguration().endpointProvider().get()
68+
)
69+
) :
70+
operation.apply(userAgentUpdatedRequest);
71+
72+
apiOperationFuture.whenComplete(redirectToCrossRegionIfRedirectException(operation,
73+
userAgentUpdatedRequest,
74+
bucketName,
75+
returnFuture));
76+
return returnFuture;
5377
}
5478

55-
private void handleOperationFailure(Throwable t, String bucket) {
56-
//TODO: handle failure case
79+
private <T extends S3Request, ReturnT> BiConsumer<ReturnT, Throwable> redirectToCrossRegionIfRedirectException(
80+
Function<T, CompletableFuture<ReturnT>> operation,
81+
T userAgentUpdatedRequest, String bucketName,
82+
CompletableFuture<ReturnT> returnFuture) {
83+
84+
return (response, throwable) -> {
85+
if (throwable != null) {
86+
if (isS3RedirectException(throwable)) {
87+
bucketToRegionCache.remove(bucketName);
88+
requestWithCrossRegion(userAgentUpdatedRequest, operation, bucketName, returnFuture, throwable);
89+
} else {
90+
returnFuture.completeExceptionally(throwable);
91+
}
92+
} else {
93+
returnFuture.complete(response);
94+
}
95+
};
5796
}
5897

59-
//Cannot avoid unchecked cast without upstream changes to supply builder function
60-
@SuppressWarnings("unchecked")
61-
private <T extends S3Request> T requestWithDecoratedEndpointProvider(T request, String bucket) {
62-
return (T) request.toBuilder()
63-
.overrideConfiguration(getOrCreateConfigWithEndpointProvider(request, bucket))
64-
.build();
98+
private <T extends S3Request, ReturnT> void requestWithCrossRegion(T request,
99+
Function<T, CompletableFuture<ReturnT>> operation,
100+
String bucketName,
101+
CompletableFuture<ReturnT> returnFuture,
102+
Throwable throwable) {
103+
104+
Optional<String> bucketRegionFromException = getBucketRegionFromException((S3Exception) throwable.getCause());
105+
if (bucketRegionFromException.isPresent()) {
106+
sendRequestWithRightRegion(request, operation, bucketName, returnFuture, bucketRegionFromException);
107+
} else {
108+
fetchRegionAndSendRequest(request, operation, bucketName, returnFuture);
109+
}
65110
}
66111

67-
//TODO: optimize shared sync/async code
68-
private AwsRequestOverrideConfiguration getOrCreateConfigWithEndpointProvider(S3Request request, String bucket) {
69-
AwsRequestOverrideConfiguration requestOverrideConfig =
70-
request.overrideConfiguration().orElseGet(() -> AwsRequestOverrideConfiguration.builder().build());
71-
72-
S3EndpointProvider delegateEndpointProvider = (S3EndpointProvider)
73-
requestOverrideConfig.endpointProvider().orElseGet(() -> serviceClientConfiguration().endpointProvider().get());
74-
75-
return requestOverrideConfig.toBuilder()
76-
.endpointProvider(BucketEndpointProvider.create(delegateEndpointProvider, bucket))
77-
.build();
112+
private <T extends S3Request, ReturnT> void fetchRegionAndSendRequest(T request,
113+
Function<T, CompletableFuture<ReturnT>> operation,
114+
String bucketName,
115+
CompletableFuture<ReturnT> returnFuture) {
116+
// // TODO: will fix the casts with separate PR
117+
((S3AsyncClient) delegate()).headBucket(b -> b.bucket(bucketName)).whenComplete((response,
118+
throwable) -> {
119+
if (throwable != null) {
120+
if (isS3RedirectException(throwable)) {
121+
bucketToRegionCache.remove(bucketName);
122+
Optional<String> bucketRegion = getBucketRegionFromException((S3Exception) throwable.getCause());
123+
if (bucketRegion.isPresent()) {
124+
sendRequestWithRightRegion(request, operation, bucketName, returnFuture, bucketRegion);
125+
} else {
126+
returnFuture.completeExceptionally(throwable);
127+
}
128+
} else {
129+
returnFuture.completeExceptionally(throwable);
130+
}
131+
}
132+
});
78133
}
79134

80-
//TODO: add cross region logic
81-
static final class BucketEndpointProvider implements S3EndpointProvider {
82-
private final S3EndpointProvider delegate;
83-
private final String bucket;
84-
85-
private BucketEndpointProvider(S3EndpointProvider delegate, String bucket) {
86-
this.delegate = delegate;
87-
this.bucket = bucket;
88-
}
89-
90-
public static BucketEndpointProvider create(S3EndpointProvider delegate, String bucket) {
91-
return new BucketEndpointProvider(delegate, bucket);
92-
}
93-
94-
@Override
95-
public CompletableFuture<Endpoint> resolveEndpoint(S3EndpointParams endpointParams) {
96-
return delegate.resolveEndpoint(endpointParams);
97-
}
135+
private <T extends S3Request, ReturnT> void sendRequestWithRightRegion(T request,
136+
Function<T, CompletableFuture<ReturnT>> operation,
137+
String bucketName,
138+
CompletableFuture<ReturnT> returnFuture,
139+
Optional<String> bucketRegionFromException) {
140+
String region = bucketRegionFromException.get();
141+
bucketToRegionCache.put(bucketName, Region.of(region));
142+
CompletableFuture<ReturnT> newFuture = operation.apply(
143+
requestWithDecoratedEndpointProvider(request,
144+
() -> Region.of(region),
145+
serviceClientConfiguration().endpointProvider().get()));
146+
CompletableFutureUtils.forwardResultTo(newFuture, returnFuture);
147+
CompletableFutureUtils.forwardExceptionTo(returnFuture, newFuture);
98148
}
99-
}
149+
}

‎services/s3/src/main/java/software/amazon/awssdk/services/s3/internal/crossregion/S3CrossRegionSyncClient.java

Lines changed: 57 additions & 49 deletions
Original file line numberDiff line numberDiff line change
@@ -15,85 +15,93 @@
1515

1616
package software.amazon.awssdk.services.s3.internal.crossregion;
1717

18+
import static software.amazon.awssdk.services.s3.internal.crossregion.utils.CrossRegionUtils.getBucketRegionFromException;
19+
import static software.amazon.awssdk.services.s3.internal.crossregion.utils.CrossRegionUtils.isS3RedirectException;
20+
import static software.amazon.awssdk.services.s3.internal.crossregion.utils.CrossRegionUtils.requestWithDecoratedEndpointProvider;
1821
import static software.amazon.awssdk.services.s3.internal.crossregion.utils.CrossRegionUtils.updateUserAgentInConfig;
1922

23+
import java.util.Map;
2024
import java.util.Optional;
21-
import java.util.concurrent.CompletableFuture;
25+
import java.util.concurrent.ConcurrentHashMap;
2226
import java.util.function.Function;
2327
import software.amazon.awssdk.annotations.SdkInternalApi;
2428
import software.amazon.awssdk.awscore.AwsRequestOverrideConfiguration;
25-
import software.amazon.awssdk.endpoints.Endpoint;
29+
import software.amazon.awssdk.regions.Region;
2630
import software.amazon.awssdk.services.s3.DelegatingS3Client;
2731
import software.amazon.awssdk.services.s3.S3Client;
28-
import software.amazon.awssdk.services.s3.endpoints.S3EndpointParams;
29-
import software.amazon.awssdk.services.s3.endpoints.S3EndpointProvider;
32+
import software.amazon.awssdk.services.s3.model.HeadBucketRequest;
33+
import software.amazon.awssdk.services.s3.model.S3Exception;
3034
import software.amazon.awssdk.services.s3.model.S3Request;
3135

36+
/**
37+
* Decorator S3 Sync client that will fetch the region name whenever there is Redirect 301 error due to cross region bucket
38+
* access.
39+
*/
3240
@SdkInternalApi
3341
public final class S3CrossRegionSyncClient extends DelegatingS3Client {
42+
43+
private final Map<String, Region> bucketToRegionCache = new ConcurrentHashMap<>();
44+
3445
public S3CrossRegionSyncClient(S3Client s3Client) {
3546
super(s3Client);
3647
}
3748

49+
private static <T extends S3Request> Optional<String> bucketNameFromRequest(T request) {
50+
return request.getValueForField("Bucket", String.class);
51+
}
52+
3853
@Override
3954
protected <T extends S3Request, ReturnT> ReturnT invokeOperation(T request, Function<T, ReturnT> operation) {
4055

41-
Optional<String> bucket = request.getValueForField("Bucket", String.class);
56+
Optional<String> bucketRequest = bucketNameFromRequest(request);
4257

4358
AwsRequestOverrideConfiguration overrideConfiguration = updateUserAgentInConfig(request);
4459
T userAgentUpdatedRequest = (T) request.toBuilder().overrideConfiguration(overrideConfiguration).build();
4560

46-
if (bucket.isPresent()) {
47-
try {
48-
return operation.apply(requestWithDecoratedEndpointProvider(userAgentUpdatedRequest, bucket.get()));
49-
} catch (Exception e) {
50-
handleOperationFailure(e, bucket.get());
61+
62+
if (!bucketRequest.isPresent()) {
63+
return operation.apply(userAgentUpdatedRequest);
64+
}
65+
String bucketName = bucketRequest.get();
66+
try {
67+
if (bucketToRegionCache.containsKey(bucketName)) {
68+
return operation.apply(
69+
requestWithDecoratedEndpointProvider(userAgentUpdatedRequest,
70+
() -> bucketToRegionCache.get(bucketName),
71+
serviceClientConfiguration().endpointProvider().get()));
72+
}
73+
return operation.apply(userAgentUpdatedRequest);
74+
} catch (S3Exception exception) {
75+
if (isS3RedirectException(exception)) {
76+
updateCacheFromRedirectException(exception, bucketName);
77+
return operation.apply(
78+
requestWithDecoratedEndpointProvider(
79+
userAgentUpdatedRequest,
80+
() -> bucketToRegionCache.computeIfAbsent(bucketName, this::fetchBucketRegion),
81+
serviceClientConfiguration().endpointProvider().get()));
5182
}
83+
throw exception;
5284
}
53-
54-
return operation.apply(userAgentUpdatedRequest);
55-
}
56-
57-
private void handleOperationFailure(Throwable t, String bucket) {
58-
//TODO: handle failure case
5985
}
6086

61-
@SuppressWarnings("unchecked")
62-
private <T extends S3Request> T requestWithDecoratedEndpointProvider(T request, String bucket) {
63-
return (T) request.toBuilder()
64-
.overrideConfiguration(getOrCreateConfigWithEndpointProvider(request, bucket))
65-
.build();
87+
private void updateCacheFromRedirectException(S3Exception exception, String bucketName) {
88+
Optional<String> regionStr = getBucketRegionFromException(exception);
89+
// If redirected, clear previous values due to region change.
90+
bucketToRegionCache.remove(bucketName);
91+
regionStr.ifPresent(region -> bucketToRegionCache.put(bucketName, Region.of(region)));
6692
}
6793

68-
//TODO: optimize shared sync/async code
69-
private AwsRequestOverrideConfiguration getOrCreateConfigWithEndpointProvider(S3Request request, String bucket) {
70-
AwsRequestOverrideConfiguration requestOverrideConfig =
71-
request.overrideConfiguration().orElseGet(() -> AwsRequestOverrideConfiguration.builder().build());
72-
73-
S3EndpointProvider delegateEndpointProvider = (S3EndpointProvider)
74-
requestOverrideConfig.endpointProvider().orElseGet(() -> serviceClientConfiguration().endpointProvider().get());
75-
76-
return requestOverrideConfig.toBuilder()
77-
.endpointProvider(BucketEndpointProvider.create(delegateEndpointProvider, bucket))
78-
.build();
79-
}
80-
81-
static final class BucketEndpointProvider implements S3EndpointProvider {
82-
private final S3EndpointProvider delegate;
83-
private final String bucket;
84-
85-
private BucketEndpointProvider(S3EndpointProvider delegate, String bucket) {
86-
this.delegate = delegate;
87-
this.bucket = bucket;
94+
private Region fetchBucketRegion(String bucketName) {
95+
try {
96+
((S3Client) delegate()).headBucket(HeadBucketRequest.builder().bucket(bucketName).build());
97+
} catch (S3Exception exception) {
98+
if (isS3RedirectException(exception)) {
99+
return Region.of(getBucketRegionFromException(exception).orElseThrow(() -> exception));
100+
}
101+
throw exception;
88102
}
103+
return null;
104+
}
89105

90-
public static BucketEndpointProvider create(S3EndpointProvider delegate, String bucket) {
91-
return new BucketEndpointProvider(delegate, bucket);
92-
}
93106

94-
@Override
95-
public CompletableFuture<Endpoint> resolveEndpoint(S3EndpointParams endpointParams) {
96-
return delegate.resolveEndpoint(endpointParams);
97-
}
98-
}
99107
}

‎services/s3/src/main/java/software/amazon/awssdk/services/s3/internal/crossregion/endpointprovider/BucketEndpointProvider.java

Lines changed: 50 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,50 @@
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.crossregion.endpointprovider;
17+
18+
import java.util.concurrent.CompletableFuture;
19+
import java.util.function.Supplier;
20+
import software.amazon.awssdk.annotations.SdkInternalApi;
21+
import software.amazon.awssdk.endpoints.Endpoint;
22+
import software.amazon.awssdk.regions.Region;
23+
import software.amazon.awssdk.services.s3.endpoints.S3EndpointParams;
24+
import software.amazon.awssdk.services.s3.endpoints.S3EndpointProvider;
25+
26+
/**
27+
* Decorator S3EndpointProvider which updates the region with the one that is supplied during its instantiation.
28+
*/
29+
@SdkInternalApi
30+
public class BucketEndpointProvider implements S3EndpointProvider {
31+
private final S3EndpointProvider delegateEndPointProvider;
32+
private final Supplier<Region> regionSupplier;
33+
34+
private BucketEndpointProvider(S3EndpointProvider delegateEndPointProvider, Supplier<Region> regionSupplier) {
35+
this.delegateEndPointProvider = delegateEndPointProvider;
36+
this.regionSupplier = regionSupplier;
37+
}
38+
39+
public static BucketEndpointProvider create(S3EndpointProvider delegateEndPointProvider, Supplier<Region> regionSupplier) {
40+
return new BucketEndpointProvider(delegateEndPointProvider, regionSupplier);
41+
}
42+
43+
@Override
44+
public CompletableFuture<Endpoint> resolveEndpoint(S3EndpointParams endpointParams) {
45+
Region crossRegion = regionSupplier.get();
46+
return delegateEndPointProvider.resolveEndpoint(
47+
crossRegion != null ? endpointParams.copy(c -> c.region(crossRegion)) : endpointParams);
48+
}
49+
}
50+

0 commit comments

Comments
 (0)