Olapplin/large object merge #6425

L-Applin · 2025-09-16T15:57:22Z

Implement parallel download for multipart GetObject in s3 Async Client and Transfer Manager.

Modifications

Add two new classes (Publisher/Subscriber) to orchestrate the non-linear multipart download: NonLinearMultipartDownloaderSubscriber and FileAsyncResponseTransformerPublisher. Note for reviewer: This is the core of the PR new functionality and review should probably start with those two classes.
Add support in Transfer-Manager module for Transfer Progress Updater.
- Note for reviewer: The AsyncResponseTransformer published by FileAsyncResponseTransformerPublisher needs to wrapped to publish progress to the progress updater. This is done in GenericS3TransferManager and TransferProgressUpdater
New public API, as discussed during design review
- supportNonSerial on SplitResult
- ParallelConfiguration new config class in MultipartConfiguration for the maxInFlightParts config
New internal API
- FileAsyncTransformer exposes getters for position, path and FileTransformerConfiguration

Testing

Added unit test
Added integration test
Manual tests using large objects

…in the onResponse callback. Keep track of all inflight requests.

…riber

…id-multi

L-Applin · 2025-09-16T16:02:48Z

...src/main/java/software/amazon/awssdk/core/internal/async/ThreadSafeEmittingSubscription.java

This is a copy of the DownstreamSubscription inner class in SplittingTransformer that has been moved to its own class to be reused.

Can we reuse the same class?

The DownstreamSubscription refers to a few of variable of SplittingTransformer, it is not a static inner class. I don't think we can.

L-Applin · 2025-09-16T16:16:52Z

utils/src/main/java/software/amazon/awssdk/utils/ContentRangeParser.java

Move to utils module because used in sdk-core

…ansformerPublisher to prevent copying all headers

…hod mocking

zoewangg

Still going through the PR.

core/sdk-core/src/main/java/software/amazon/awssdk/core/async/AsyncResponseTransformer.java

...e/src/main/java/software/amazon/awssdk/core/internal/async/FileAsyncResponseTransformer.java

...src/main/java/software/amazon/awssdk/core/internal/async/ThreadSafeEmittingSubscription.java

...e/src/main/java/software/amazon/awssdk/core/internal/async/FileAsyncResponseTransformer.java

...are/amazon/awssdk/services/s3/internal/multipart/NonLinearMultipartDownloaderSubscriber.java

zoewangg · 2025-09-17T22:08:35Z

...are/amazon/awssdk/services/s3/internal/multipart/NonLinearMultipartDownloaderSubscriber.java

+    /**
+     * Amount of demand requested but not yet fulfilled by the subscription
+     */
+    private final AtomicInteger outstandingDemand = new AtomicInteger(0);


Is there any reason we need to track outstanding demand in a subscriber?

We track the demand that we requested but that has not yet been fulfilled by on next. This is to prevent requesting more than the maxInFlight.

zoewangg · 2025-09-17T22:09:06Z

...are/amazon/awssdk/services/s3/internal/multipart/NonLinearMultipartDownloaderSubscriber.java

+ * is a 'one-shot' class, it should <em>NOT</em> be reused for more than one multipart download.
+ */
+@SdkInternalApi
+public class NonLinearMultipartDownloaderSubscriber


Nit: ParallelMultipartDownloadSubscriber?

since we use parallelSplitSupported we can rename that to ParallelMultipartDownloadSubscriber, yeah

...are/amazon/awssdk/services/s3/internal/multipart/NonLinearMultipartDownloaderSubscriber.java

zoewangg · 2025-09-17T22:14:26Z

...are/amazon/awssdk/services/s3/internal/multipart/NonLinearMultipartDownloaderSubscriber.java

+            return false;
+        }
+
+        firstPartLock.lock();


Is there any reason we need this? We only request more in whenComplete future of the first part, right?

MPU has similar logic where we need to wait for creatMPU to finish, can we use similar logic here? https://github.com/aws/aws-sdk-java-v2/blob/master/services/s3/src/main/java/software/amazon/awssdk/services/s3/internal/multipart/UploadWithUnknownContentLengthHelper.java#L191

We can't really reuse that logic because we need to wait for the first part to complete before we know if its a multipart object or not. Or maybe I don't exactly understand what you want to reuse, could you elaborate?

...n/java/software/amazon/awssdk/core/internal/async/FileAsyncResponseTransformerPublisher.java

core/sdk-core/src/main/java/software/amazon/awssdk/core/async/AsyncResponseTransformer.java

...n/java/software/amazon/awssdk/core/internal/async/FileAsyncResponseTransformerPublisher.java

...are/amazon/awssdk/services/s3/internal/multipart/NonLinearMultipartDownloaderSubscriber.java

…blisher

…ents - try/catch in ThreadSafeEmittingSubscription - negative totalParts check - NonLinearMultipartDownloaderSubscriberTckTest tck test

core/sdk-core/src/main/java/software/amazon/awssdk/core/async/AsyncResponseTransformer.java

joviegas · 2025-09-18T16:16:58Z

...a/software/amazon/awssdk/core/internal/async/DefaultAsyncResponseTransformerSplitResult.java


    private DefaultAsyncResponseTransformerSplitResult(Builder<ResponseT, ResultT> builder) {
        this.publisher = Validate.paramNotNull(
            builder.publisher(), "asyncResponseTransformerPublisher");
        this.future = Validate.paramNotNull(
            builder.resultFuture(), "future");
+        this.supportsNonSerial = Validate.getOrDefault(builder.supportsNonSerial(), () -> false);


Since these classes are all related to Async , and concurrency is good to have why do we disable concurrent splits by default ?

Because most implementation is AsyncResponseTransformer cannot support it. We send back data to customers serially and changing that would be a braking change. We would also probably need to update all our implemetation of AsyncResponseTransformer we have, which is out of scope

joviegas · 2025-09-18T17:55:28Z

...n/java/software/amazon/awssdk/core/internal/async/FileAsyncResponseTransformerPublisher.java

+
+        @Override
+        public void onResponse(T response) {
+            Optional<String> contentRangeList = response.sdkHttpResponse().firstMatchingHeader("x-amz-content-range");


Can we do something like

String contentRange = response.sdkHttpResponse() .firstMatchingHeader("x-amz-content-range") .orElseThrow(() -> new IllegalStateException("Content range header is missing"));

and do try catch for entire function

public void onResponse(T response) { try { // existing } catch (IllegalStateException e) { handleError(e.getMessage(), future); } } // Some common handleError function private void handleError(String errorMessage, CompletableFuture<T> future) { IllegalStateException exception = new IllegalStateException(errorMessage); if (subscriber != null) { subscriber.onError(exception); } if (future != null) { future.completeExceptionally(exception); } }

I usually don't like using exception as control flow (remember the createURI thing we had a while ago), but I can use a handle error method 👍

joviegas · 2025-09-18T18:02:13Z

...n/java/software/amazon/awssdk/core/internal/async/FileAsyncResponseTransformerPublisher.java

+
+    @Override
+    public void subscribe(Subscriber<? super AsyncResponseTransformer<T, T>> s) {
+        if (s == null) {


Can we use
this.subscriber =Validate.notNull(responseTransformer.path(), "subscriber");

joviegas · 2025-09-18T18:08:53Z

...n/java/software/amazon/awssdk/core/internal/async/FileAsyncResponseTransformerPublisher.java

+     * transformer will retry independently based on the retry configuration of the client it is used with. We only need to verify
+     * the completion state of the future of each individually
+     */
+    private class IndividualFileTransformer implements AsyncResponseTransformer<T, T> {


Can we move it to a separate class file ?

mmmh it is convenient to have it as an inner class as it references variable of the outer class. We use similar pattern in other places for multipart operations

...src/main/java/software/amazon/awssdk/core/internal/async/ThreadSafeEmittingSubscription.java

...are/amazon/awssdk/services/s3/internal/multipart/NonLinearMultipartDownloaderSubscriber.java

joviegas · 2025-09-18T20:07:48Z

...n/java/software/amazon/awssdk/core/internal/async/FileAsyncResponseTransformerPublisher.java

+
+        @Override
+        public void onResponse(T response) {
+            Optional<String> contentRangeList = response.sdkHttpResponse().firstMatchingHeader("x-amz-content-range");


Since we are depending on header sent by service will this cause issue for Third party tools like minio or gcp ? Since we are erring out if the header is not present it would be good to know its impact when used with Third party s3 like minio or gcp

...va/software/amazon/awssdk/services/s3/internal/multipart/MultipartConfigurationResolver.java

...are/amazon/awssdk/services/s3/internal/multipart/NonLinearMultipartDownloaderSubscriber.java

- renamed EmittingSubscription, mark it ThreadSafe - Added comments - some other renaming

L-Applin added 28 commits July 22, 2025 18:15

exploration

c5ed15e

hide cloudwatch logs for s3 regression tests

067db30

Merge remote-tracking branch 'origin/master'

41be0a6

Multipart dowload implementation

51357ee

Multipart dowload implementation

a9d25ec

Merge remote-tracking branch 'origin/master'

b70ab8b

Merge remote-tracking branch 'origin/master'

0a8f6c5

Update non-linear download logic

7c82e2b

request max-in-flight at first request

978f553

remove stateful calls to FileAsyncResponseTransformer by creating it …

3e29b8c

…in the onResponse callback. Keep track of all inflight requests.

Wiremock test for s3AsyncClient with AsyncResponseTransformer.toFile

c44a520

added 10_000 parts tests

418dcbd

add missing assertions to errorOnMiddlePart_retryable

db7d301

handle fileWriteOption

2f65a8c

prevent over-requesting and fix potential race condition in FileSubsc…

0bfbdb2

…riber

TM integration

0a20a9c

fix tests

e1aebb5

debugging for long stop before completing future

3a0a0b1

added outstanding demand tracking, assert with checksum in integ test

2f3a14e

Merge remote-tracking branch 'origin/master'

e49aadc

Fix pendingTransformer polling

aa0caf1

Make download method of TM work properly with progressUpdater

29f899d

remove logs

0aa1529

add ParallelConfiguration for maxInFlightParts

426d82d

javadoc and logs

5bdc142

Merge remote-tracking branch 'origin/master'

9551671

Merge remote-tracking branch 'origin/master' into feature/master/hagr…

7e19d14

…id-multi

Merge branch 'master' into feature/master/hagrid-multi

d37056d

L-Applin commented Sep 16, 2025

View reviewed changes

L-Applin added 5 commits September 16, 2025 12:21

cleanup and changelog

474111b

cleanup

9ad9c56

fix test utils after merge

861d64d

remove comment code in test

c684d4c

checkstyle

e34dcc3

L-Applin marked this pull request as ready for review September 16, 2025 17:04

L-Applin requested a review from a team as a code owner September 16, 2025 17:04

L-Applin added 7 commits September 16, 2025 13:11

rename initialPosition to position

8506c43

use firstMatchingHeader() instead of headers() in FileAsyncResponseTr…

f3d3773

…ansformerPublisher to prevent copying all headers

add handle error method

e1360f7

fix FileAsyncResponseTransformerPublisherTest wrong content-range met…

d1db3c8

…hod mocking

use succeedsWithin assertion for future

50631f5

fix checkstyle violations

8f36d18

fix checkstyle violations

a1f8d23

zoewangg reviewed Sep 17, 2025

View reviewed changes

S-Saranya1 reviewed Sep 17, 2025

View reviewed changes

...are/amazon/awssdk/services/s3/internal/multipart/NonLinearMultipartDownloaderSubscriber.java Outdated Show resolved Hide resolved

L-Applin added 5 commits September 17, 2025 23:32

PR comments - first pass

1716e46

PR comments - keep transformerCount in FileAsyncResponseTransformerPu…

289554e

…blisher

PR comments - keep transformerCount in FileAsyncResponseTransformerPu…

e025fb3

…blisher

rename supportsNonSerial to parallelSplitSupported, and other PR comm…

fe7985f

…ents - try/catch in ThreadSafeEmittingSubscription - negative totalParts check - NonLinearMultipartDownloaderSubscriberTckTest tck test

Added file config options to FileAsyncResponseTransformerPublisherTest

af94ef1

joviegas reviewed Sep 18, 2025

View reviewed changes

PR comments

9902ec8

- renamed EmittingSubscription, mark it ThreadSafe - Added comments - some other renaming

Olapplin/large object merge #6425

Are you sure you want to change the base?

Olapplin/large object merge #6425

Uh oh!

Conversation

L-Applin commented Sep 16, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Modifications

Testing

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

L-Applin Sep 18, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

L-Applin Sep 16, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

zoewangg left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

L-Applin commented Sep 16, 2025 •

edited

Loading

L-Applin Sep 18, 2025 •

edited

Loading

L-Applin Sep 16, 2025 •

edited

Loading