From d4efe6e488b10aa9eca4490f12e55599418445b7 Mon Sep 17 00:00:00 2001 From: Rishabh Jain Date: Tue, 21 Jul 2026 13:11:15 -0700 Subject: [PATCH] fix(s3): preserve original cause in ParallelMultipartDownloaderSubscriber.onError onError cancelled the in-flight part requests before completing resultFuture with the failure cause. Each part future is wired to resultFuture via CompletableFutureUtils.forwardExceptionTo, so cancelling first raced a CancellationException onto resultFuture and discarded the original cause; callers observing the download's completion future saw a bare CancellationException with no root cause, and the trigger was never logged. Complete resultFuture with the original throwable first, then cancel the in-flight parts, and log the cause at debug level, matching the sibling ParallelPresignedUrlMultipartDownloaderSubscriber. Add a unit test covering the onError cause-preservation ordering. --- .../next-release/bugfix-AmazonS3-4e9e1fe.json | 6 ++ ...ParallelMultipartDownloaderSubscriber.java | 5 +- ...llelMultipartDownloaderSubscriberTest.java | 71 +++++++++++++++++++ 3 files changed, 81 insertions(+), 1 deletion(-) create mode 100644 .changes/next-release/bugfix-AmazonS3-4e9e1fe.json create mode 100644 services/s3/src/test/java/software/amazon/awssdk/services/s3/internal/multipart/ParallelMultipartDownloaderSubscriberTest.java diff --git a/.changes/next-release/bugfix-AmazonS3-4e9e1fe.json b/.changes/next-release/bugfix-AmazonS3-4e9e1fe.json new file mode 100644 index 000000000000..cef38540d0d1 --- /dev/null +++ b/.changes/next-release/bugfix-AmazonS3-4e9e1fe.json @@ -0,0 +1,6 @@ +{ + "type": "bugfix", + "category": "Amazon S3", + "contributor": "rishabhjainps", + "description": "Fixed an issue in multipart download where a failed download could surface a bare CancellationException with no root cause. ParallelMultipartDownloaderSubscriber.onError now completes the result future with the original throwable before cancelling in-flight part requests, matching the sibling ParallelPresignedUrlMultipartDownloaderSubscriber, and logs the cause at debug level." +} diff --git a/services/s3/src/main/java/software/amazon/awssdk/services/s3/internal/multipart/ParallelMultipartDownloaderSubscriber.java b/services/s3/src/main/java/software/amazon/awssdk/services/s3/internal/multipart/ParallelMultipartDownloaderSubscriber.java index a0550d2a10f1..20072e1acb4a 100644 --- a/services/s3/src/main/java/software/amazon/awssdk/services/s3/internal/multipart/ParallelMultipartDownloaderSubscriber.java +++ b/services/s3/src/main/java/software/amazon/awssdk/services/s3/internal/multipart/ParallelMultipartDownloaderSubscriber.java @@ -415,9 +415,12 @@ public void onError(Throwable t) { // Signal received from the publisher this is subscribed to // (in the case of file download, that's FileAsyncResponseTransformerPublisher) // Failed state, something really wrong has happened, cancel everything + // Complete the result future with the original cause before cancelling in-flight parts. + // Cancelling first races a CancellationException onto resultFuture and masks t. + log.debug(() -> "Error in parallel multipart download", t); + resultFuture.completeExceptionally(t); inFlightRequests.values().forEach(future -> future.cancel(true)); inFlightRequests.clear(); - resultFuture.completeExceptionally(t); } @Override diff --git a/services/s3/src/test/java/software/amazon/awssdk/services/s3/internal/multipart/ParallelMultipartDownloaderSubscriberTest.java b/services/s3/src/test/java/software/amazon/awssdk/services/s3/internal/multipart/ParallelMultipartDownloaderSubscriberTest.java new file mode 100644 index 000000000000..362daecceb44 --- /dev/null +++ b/services/s3/src/test/java/software/amazon/awssdk/services/s3/internal/multipart/ParallelMultipartDownloaderSubscriberTest.java @@ -0,0 +1,71 @@ +/* + * Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. + * + * Licensed under the Apache License, Version 2.0 (the "License"). + * You may not use this file except in compliance with the License. + * A copy of the License is located at + * + * http://aws.amazon.com/apache2.0 + * + * or in the "license" file accompanying this file. This file is distributed + * on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either + * express or implied. See the License for the specific language governing + * permissions and limitations under the License. + */ + +package software.amazon.awssdk.services.s3.internal.multipart; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; + +import java.util.concurrent.CancellationException; +import java.util.concurrent.CompletableFuture; +import org.junit.jupiter.api.Test; +import org.reactivestreams.Subscription; +import software.amazon.awssdk.core.async.AsyncResponseTransformer; +import software.amazon.awssdk.services.s3.S3AsyncClient; +import software.amazon.awssdk.services.s3.model.GetObjectRequest; +import software.amazon.awssdk.services.s3.model.GetObjectResponse; + +/** + * Unit tests for {@link ParallelMultipartDownloaderSubscriber}. + */ +class ParallelMultipartDownloaderSubscriberTest { + + @Test + void onError_withInFlightRequests_shouldCompleteResultFutureWithOriginalCause() { + S3AsyncClient s3 = mock(S3AsyncClient.class); + CompletableFuture resultFuture = new CompletableFuture<>(); + GetObjectRequest request = GetObjectRequest.builder() + .bucket("test-bucket") + .key("test-key") + .build(); + ParallelMultipartDownloaderSubscriber subscriber = + new ParallelMultipartDownloaderSubscriber(s3, request, resultFuture, 10); + + CompletableFuture firstPartFuture = new CompletableFuture<>(); + CompletableFuture secondPartFuture = new CompletableFuture<>(); + when(s3.getObject(any(GetObjectRequest.class), any(AsyncResponseTransformer.class))) + .thenReturn(firstPartFuture, secondPartFuture); + + subscriber.onSubscribe(mock(Subscription.class)); + subscriber.onNext(mock(AsyncResponseTransformer.class)); + firstPartFuture.complete(GetObjectResponse.builder() + .partsCount(3) + .eTag("etag") + .build()); + // Second part is now in flight and never completes on its own. + subscriber.onNext(mock(AsyncResponseTransformer.class)); + + RuntimeException cause = new RuntimeException("original failure"); + subscriber.onError(cause); + + // The caller-facing future must carry the original cause, not a CancellationException + // raced onto it by the in-flight part cancellation. + Throwable thrown = resultFuture.handle((r, t) -> t).join(); + assertThat(thrown).isNotInstanceOf(CancellationException.class); + assertThat(thrown).isSameAs(cause); + } +}