Skip to content

Commit 3ceee7f

Browse files
davidh44davidh44
committed
Fix bug in MultipartS3AsyncClient GetObject retryable errors
1 parent bfb1030 commit 3ceee7f

File tree

8 files changed

+414
-33
lines changed

8 files changed

+414
-33
lines changed

‎.changes/next-release/bugfix-AmazonS3-263fed5.json

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,6 @@
1+
{
2+
"type": "bugfix",
3+
"category": "Amazon S3",
4+
"contributor": "",
5+
"description": "Fix MultipartS3AsyncClient GetObject retryable errors"
6+
}

‎core/sdk-core/src/main/java/software/amazon/awssdk/core/internal/async/SplittingTransformer.java

Lines changed: 79 additions & 28 deletions
Original file line numberDiff line numberDiff line change
@@ -25,6 +25,10 @@
2525
import software.amazon.awssdk.core.SplittingTransformerConfiguration;
2626
import software.amazon.awssdk.core.async.AsyncResponseTransformer;
2727
import software.amazon.awssdk.core.async.SdkPublisher;
28+
import software.amazon.awssdk.core.exception.SdkException;
29+
import software.amazon.awssdk.core.exception.SdkServiceException;
30+
import software.amazon.awssdk.core.internal.retry.SdkDefaultRetrySetting;
31+
import software.amazon.awssdk.core.retry.RetryUtils;
2832
import software.amazon.awssdk.utils.CompletableFutureUtils;
2933
import software.amazon.awssdk.utils.Logger;
3034
import software.amazon.awssdk.utils.Validate;
@@ -54,16 +58,6 @@ public class SplittingTransformer<ResponseT, ResultT> implements SdkPublisher<As
5458
*/
5559
private final AsyncResponseTransformer<ResponseT, ResultT> upstreamResponseTransformer;
5660

57-
/**
58-
* Set to true once {@code .prepare()} is called on the upstreamResponseTransformer
59-
*/
60-
private final AtomicBoolean preparedCalled = new AtomicBoolean(false);
61-
62-
/**
63-
* Set to true once {@code .onResponse()} is called on the upstreamResponseTransformer
64-
*/
65-
private final AtomicBoolean onResponseCalled = new AtomicBoolean(false);
66-
6761
/**
6862
* Set to true once {@code .onStream()} is called on the upstreamResponseTransformer
6963
*/
@@ -111,6 +105,24 @@ public class SplittingTransformer<ResponseT, ResultT> implements SdkPublisher<As
111105

112106
private final Object cancelLock = new Object();
113107

108+
/**
109+
* Keeps track of the upstream future returned by {@code upstreamResponseTransformer.prepare()}. If an error occurs, we
110+
* forward this to the {@code resultFuture}.
111+
*/
112+
private volatile CompletableFuture<ResultT> upstreamFuture;
113+
114+
/**
115+
* Tracks whether an {@code IndividualTransformer} has been created for the first part yet. Errors will only be retried for
116+
* the first part.
117+
*/
118+
private final AtomicBoolean isFirstIndividualTransformer = new AtomicBoolean(true);
119+
120+
/**
121+
* Tracks whether an {@code IndividualPartSubscriber} has been created for the first part yet. Errors will only be retried for
122+
* the first part.
123+
*/
124+
private final AtomicBoolean isFirstIndividualSubscriber = new AtomicBoolean(true);
125+
114126
private SplittingTransformer(AsyncResponseTransformer<ResponseT, ResultT> upstreamResponseTransformer,
115127
Long maximumBufferSizeInBytes,
116128
CompletableFuture<ResultT> resultFuture) {
@@ -198,7 +210,7 @@ private boolean doEmit() {
198210
}
199211
if (outstandingDemand.get() > 0) {
200212
demand = outstandingDemand.decrementAndGet();
201-
downstreamSubscriber.onNext(new IndividualTransformer());
213+
downstreamSubscriber.onNext(new IndividualTransformer(isFirstIndividualTransformer.compareAndSet(true, false)));
202214
}
203215
}
204216
return false;
@@ -230,6 +242,7 @@ private void handleSubscriptionCancel() {
230242
} else {
231243
log.trace(() -> "calling downstreamSubscriber.onComplete()");
232244
downstreamSubscriber.onComplete();
245+
CompletableFutureUtils.forwardResultTo(upstreamFuture, resultFuture);
233246
}
234247
downstreamSubscriber = null;
235248
});
@@ -259,28 +272,27 @@ private void handleFutureCancel(Throwable e) {
259272
* body publisher.
260273
*/
261274
private class IndividualTransformer implements AsyncResponseTransformer<ResponseT, ResponseT> {
275+
private final boolean isFirstPart;
262276
private ResponseT response;
263277
private CompletableFuture<ResponseT> individualFuture;
264278

279+
IndividualTransformer(boolean isFirstPart) {
280+
this.isFirstPart = isFirstPart;
281+
}
282+
265283
@Override
266284
public CompletableFuture<ResponseT> prepare() {
267285
this.individualFuture = new CompletableFuture<>();
268-
if (preparedCalled.compareAndSet(false, true)) {
286+
287+
if (isFirstPart) {
269288
if (isCancelled.get()) {
270289
return individualFuture;
271290
}
272291
log.trace(() -> "calling prepare on the upstream transformer");
273-
CompletableFuture<ResultT> upstreamFuture = upstreamResponseTransformer.prepare();
274-
if (!resultFuture.isDone()) {
275-
CompletableFutureUtils.forwardResultTo(upstreamFuture, resultFuture);
276-
}
292+
upstreamFuture = upstreamResponseTransformer.prepare();
293+
277294
}
278-
resultFuture.whenComplete((r, e) -> {
279-
if (e == null) {
280-
return;
281-
}
282-
individualFuture.completeExceptionally(e);
283-
});
295+
284296
individualFuture.whenComplete((r, e) -> {
285297
if (isCancelled.get()) {
286298
handleSubscriptionCancel();
@@ -291,7 +303,7 @@ public CompletableFuture<ResponseT> prepare() {
291303

292304
@Override
293305
public void onResponse(ResponseT response) {
294-
if (onResponseCalled.compareAndSet(false, true)) {
306+
if (isFirstPart) {
295307
log.trace(() -> "calling onResponse on the upstream transformer");
296308
upstreamResponseTransformer.onResponse(response);
297309
}
@@ -304,7 +316,8 @@ public void onStream(SdkPublisher<ByteBuffer> publisher) {
304316
return;
305317
}
306318
synchronized (cancelLock) {
307-
if (onStreamCalled.compareAndSet(false, true)) {
319+
if (isFirstPart) {
320+
onStreamCalled.set(true);
308321
log.trace(() -> "calling onStream on the upstream transformer");
309322
upstreamResponseTransformer.onStream(upstreamSubscriber -> publisherToUpstream.subscribe(
310323
DelegatingBufferingSubscriber.builder()
@@ -314,29 +327,65 @@ public void onStream(SdkPublisher<ByteBuffer> publisher) {
314327
);
315328
}
316329
}
317-
publisher.subscribe(new IndividualPartSubscriber<>(this.individualFuture, response));
330+
331+
if (!resultFuture.isDone()) {
332+
CompletableFutureUtils.forwardResultTo(upstreamFuture, resultFuture);
333+
}
334+
335+
publisher.subscribe(new IndividualPartSubscriber<>(this.individualFuture, response,
336+
isFirstIndividualSubscriber.compareAndSet(true, false)));
318337
}
319338

320339
@Override
321340
public void exceptionOccurred(Throwable error) {
322-
publisherToUpstream.error(error);
323341
log.trace(() -> "calling exceptionOccurred on the upstream transformer");
324342
upstreamResponseTransformer.exceptionOccurred(error);
343+
344+
if (!isFirstPart || onStreamCalled.get()) {
345+
publisherToUpstream.error(error);
346+
}
347+
348+
if (!isRetryableError(error)) {
349+
log.trace(() -> "error is non-retryable, forwarding upstream future to result future");
350+
CompletableFutureUtils.forwardResultTo(upstreamFuture, resultFuture);
351+
}
325352
}
326353
}
327354

355+
private boolean isRetryableError(Throwable error) {
356+
if (error instanceof SdkException) {
357+
SdkException ex = (SdkException) error;
358+
return retryOnStatusCodes(ex)
359+
|| RetryUtils.isRetryableException(ex)
360+
|| RetryUtils.isClockSkewException(ex)
361+
|| RetryUtils.isClockSkewException(ex);
362+
363+
}
364+
return false;
365+
}
366+
367+
private static boolean retryOnStatusCodes(Throwable ex) {
368+
if (ex instanceof SdkServiceException) {
369+
SdkServiceException failure = (SdkServiceException) ex;
370+
return SdkDefaultRetrySetting.RETRYABLE_STATUS_CODES.contains(failure.statusCode());
371+
}
372+
return false;
373+
}
374+
328375
/**
329376
* the Subscriber for each of the individual request's ByteBuffer publisher
330377
*/
331378
class IndividualPartSubscriber<T> implements Subscriber<ByteBuffer> {
332379

333380
private final CompletableFuture<T> future;
334381
private final T response;
382+
private final boolean isFirstPart;
335383
private Subscription subscription;
336384

337-
IndividualPartSubscriber(CompletableFuture<T> future, T response) {
385+
IndividualPartSubscriber(CompletableFuture<T> future, T response, boolean isFirstPart) {
338386
this.future = future;
339387
this.response = response;
388+
this.isFirstPart = isFirstPart;
340389
}
341390

342391
@Override
@@ -376,7 +425,9 @@ public void onComplete() {
376425
}
377426

378427
private void handleError(Throwable t) {
379-
publisherToUpstream.error(t);
428+
if (!isFirstPart) {
429+
publisherToUpstream.error(t);
430+
}
380431
future.completeExceptionally(t);
381432
}
382433
}

‎core/sdk-core/src/test/java/software/amazon/awssdk/core/internal/async/IndividualPartSubscriberTckTest.java

Lines changed: 1 addition & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -18,7 +18,6 @@
1818

1919
import java.nio.ByteBuffer;
2020
import java.util.concurrent.CompletableFuture;
21-
import java.util.concurrent.atomic.AtomicInteger;
2221
import org.reactivestreams.Subscriber;
2322
import org.reactivestreams.Subscription;
2423
import org.reactivestreams.tck.SubscriberWhiteboxVerification;
@@ -45,7 +44,7 @@ public Subscriber<ByteBuffer> createSubscriber(WhiteboxSubscriberProbe<ByteBuffe
4544
.maximumBufferSizeInBytes(32L)
4645
.resultFuture(new CompletableFuture<>())
4746
.build();
48-
return transformer.new IndividualPartSubscriber<ByteBuffer>(future, ByteBuffer.wrap(new byte[0])) {
47+
return transformer.new IndividualPartSubscriber<ByteBuffer>(future, ByteBuffer.wrap(new byte[0]), true) {
4948
@Override
5049
public void onSubscribe(Subscription s) {
5150
super.onSubscribe(s);

‎services/s3/pom.xml

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -128,6 +128,11 @@
128128
<artifactId>retries-spi</artifactId>
129129
<version>${awsjavasdk.version}</version>
130130
</dependency>
131+
<dependency>
132+
<groupId>software.amazon.awssdk</groupId>
133+
<artifactId>retries</artifactId>
134+
<version>${awsjavasdk.version}</version>
135+
</dependency>
131136
<!-- Test Dependencies -->
132137
<dependency>
133138
<artifactId>commons-io</artifactId>

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

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -23,6 +23,7 @@
2323
import software.amazon.awssdk.services.s3.S3AsyncClient;
2424
import software.amazon.awssdk.services.s3.model.GetObjectRequest;
2525
import software.amazon.awssdk.services.s3.model.GetObjectResponse;
26+
import software.amazon.awssdk.utils.CompletableFutureUtils;
2627
import software.amazon.awssdk.utils.Logger;
2728

2829
@SdkInternalApi
@@ -49,6 +50,7 @@ public <T> CompletableFuture<T> downloadObject(
4950
.build());
5051
MultipartDownloaderSubscriber subscriber = subscriber(getObjectRequest);
5152
split.publisher().subscribe(subscriber);
53+
CompletableFutureUtils.forwardExceptionTo(subscriber.future(), split.resultFuture());
5254
return split.resultFuture();
5355
}
5456

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

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -15,6 +15,8 @@
1515

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

18+
import java.util.ArrayList;
19+
import java.util.List;
1820
import java.util.concurrent.CompletableFuture;
1921
import java.util.concurrent.atomic.AtomicInteger;
2022
import org.reactivestreams.Subscriber;
@@ -79,6 +81,8 @@ public class MultipartDownloaderSubscriber implements Subscriber<AsyncResponseTr
7981
*/
8082
private final Object lock = new Object();
8183

84+
private final List<CompletableFuture<GetObjectResponse>> getObjectFutures = new ArrayList<>();
85+
8286
public MultipartDownloaderSubscriber(S3AsyncClient s3, GetObjectRequest getObjectRequest) {
8387
this(s3, getObjectRequest, 0);
8488
}
@@ -119,6 +123,7 @@ public void onNext(AsyncResponseTransformer<GetObjectResponse, GetObjectResponse
119123
GetObjectRequest actualRequest = nextRequest(nextPartToGet);
120124
log.debug(() -> "Sending GetObjectRequest for next part with partNumber=" + nextPartToGet);
121125
CompletableFuture<GetObjectResponse> getObjectFuture = s3.getObject(actualRequest, asyncResponseTransformer);
126+
getObjectFutures.add(getObjectFuture);
122127
getObjectFuture.whenComplete((response, error) -> {
123128
if (error != null) {
124129
log.debug(() -> "Error encountered during GetObjectRequest with partNumber=" + nextPartToGet);
@@ -166,6 +171,7 @@ private void requestMoreIfNeeded(GetObjectResponse response) {
166171

167172
@Override
168173
public void onError(Throwable t) {
174+
getObjectFutures.forEach(future -> future.cancel(true));
169175
future.completeExceptionally(t);
170176
}
171177

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

Lines changed: 2 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -38,9 +38,8 @@
3838
import software.amazon.awssdk.utils.Validate;
3939

4040
/**
41-
* An {@link S3AsyncClient} that automatically converts PUT, COPY requests to their respective multipart call. CRC32 will be
42-
* enabled for the PUT and COPY requests, unless the the checksum is specified or checksum validation is disabled.
43-
* Note: GET is not yet supported.
41+
* An {@link S3AsyncClient} that automatically converts PUT, COPY, and GET requests to their respective multipart call. CRC32
42+
* will be enabled for the requests, unless the checksum is specified or checksum validation is disabled.
4443
*
4544
* @see MultipartConfiguration
4645
*/

0 commit comments

Comments
 (0)