Stream retry support part 2: Introduce a new split method in AsyncRequestBody that returns an SdkP… #6346

zoewangg · 2025-08-13T23:12:04Z

…ublisher of ClosableAsyncRequestBody and use it in s3 multipart client

Motivation and Context

This PR introduces a new split method, splitClosable in AsyncRequestBody that returns an SdkPublisher of ClosableAsyncRequestBody instead of AsyncRequestBody, enabling proper resource management for split request bodies. The change is primarily implemented to support streaming retry for S3 multipart client for retry Support retries for individual parts of a multipart upload in Java SDK V2 #6198
Introduces ClosableAsyncRequestBody interface extending AsyncRequestBody with close capability
Updates S3 multipart client implementation to use the new splitClosable method

Testing

Added wiremock tests

Screenshots (if appropriate)

Types of changes

Bug fix (non-breaking change which fixes an issue)
New feature (non-breaking change which adds functionality)

Checklist

I have read the CONTRIBUTING document
Local run of mvn install succeeds
My code follows the code style of this project
My change requires a change to the Javadoc documentation
I have updated the Javadoc documentation accordingly
I have added tests to cover my changes
All new and existing tests passed
I have added a changelog entry. Adding a new entry must be accomplished by running the scripts/new-change script and following the instructions. Commit the new file created by the script in .changes/next-release with your changes.
My change is to implement 1.11 parity feature and I have updated LaunchChangelog

License

I confirm that this pull request can be released under the Apache 2 license

Copilot

Pull Request Overview

This PR introduces a new splitV2 method in AsyncRequestBody that returns an SdkPublisher of ClosableAsyncRequestBody instead of AsyncRequestBody, enabling proper resource management for split request bodies. This change is designed to support streaming retry for S3 multipart uploads.

Adds ClosableAsyncRequestBody interface extending AsyncRequestBody with close capability
Updates S3 multipart client implementation to use the new splitV2 method
Maintains backward compatibility by deprecating the original split method

Reviewed Changes

Copilot reviewed 17 out of 18 changed files in this pull request and generated 4 comments.

Show a summary per file

File	Description
`ClosableAsyncRequestBody.java`	New interface extending AsyncRequestBody with close capability
`AsyncRequestBody.java`	Adds splitV2 method and deprecates original split method
`SplittingPublisher.java`	Updates to return ClosableAsyncRequestBody and improve resource management
`UploadWithUnknownContentLengthHelper.java`	Updated to use splitV2 and properly close request bodies
`UploadWithKnownContentLengthHelper.java`	Updated to use splitV2 method
`KnownContentLengthAsyncRequestBodySubscriber.java`	Updated to handle ClosableAsyncRequestBody and close resources
Test files	Updated tests to use new interfaces and verify retry behavior

_{Tip: Customize your code reviews with copilot-instructions.md. Create the file or learn how to get started.
You can also share your feedback on Copilot code review for a chance to win a $100 gift card. Take the survey.}

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

...tware/amazon/awssdk/services/s3/internal/multipart/UploadWithUnknownContentLengthHelper.java

...azon/awssdk/services/s3/internal/multipart/KnownContentLengthAsyncRequestBodySubscriber.java

…ublisher of ClosableAsyncRequestBody and use it in s3 multipart client

...-core/src/main/java/software/amazon/awssdk/core/async/listener/AsyncRequestBodyListener.java

joviegas · 2025-08-18T15:03:02Z

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

    }

+    private void stubUploadFailsInitialAttemptCalls(int partNumber, ResponseDefinitionBuilder responseDefinitionBuilder) {


Can we have end to end test cases with Wiremock/stubs to test end to end to retryable error cases ?
Or can have separate task to cover end to end test w.r.t Multipart client error cases with different Part sizes and part numbers.

I've added wiremock test for retryable error cases and this method is actually used in that test

https://github.com/aws/aws-sdk-java-v2/pull/6346/files#diff-1202818b28c2d2d4382e4113baa11a572895fa9bd35c8b0f336dc5f0e546d44bR149

...core/src/main/java/software/amazon/awssdk/core/async/BufferedSplittableAsyncRequestBody.java

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

joviegas · 2025-08-21T19:12:05Z

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

+
+        @Override
+        public void close() {
+            // no op


Why no op comment would be helpful ?

Basically making it more explicit that this is deliberately left empty and not by mistake.

...rc/main/java/software/amazon/awssdk/core/internal/async/NonRetryableSubAsyncRequestBody.java

joviegas · 2025-08-21T19:23:32Z

...rc/main/java/software/amazon/awssdk/core/internal/async/NonRetryableSubAsyncRequestBody.java

+        if (subscribeCalled.compareAndSet(false, true)) {
+            delegate.subscribe(s);
+        } else {
+            s.onSubscribe(new NoopSubscription(s));


Why doe we need to do this ?

Per reactive spec, onSubscribe must be signaled

Publisher.subscribe MUST call onSubscribe on the provided Subscriber prior to any other signals to that Subscriber

joviegas · 2025-08-21T19:24:40Z

...rc/main/java/software/amazon/awssdk/core/internal/async/NonRetryableSubAsyncRequestBody.java

+            delegate.subscribe(s);
+        } else {
+            s.onSubscribe(new NoopSubscription(s));
+            s.onError(NonRetryableException.create(


Is retry only the use case case where this re subscription can occur ?

Yes for the SDK use case, but I suppose it could be some other use case if customers use split directly (although we don't expect it to be used by users directly). I can update the message to make it more clear

joviegas · 2025-08-21T19:28:54Z

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

+                log.debug(() -> "requesting data after closing" + partNumber);
+            }
+        } catch (Throwable e) {
+            log.warn(() -> String.format("Unexpected error thrown from cleaning up AsyncRequestBody for part number %d, "


Why is this just WARN and not error log ?

I used warn because it's not a immediate request failure per se. WDYT?

joviegas · 2025-08-21T20:55:42Z

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

+        log.debug(() -> "Received complete() for part number: " + partNumber);
+        // ByteBuffersAsyncRequestBody MUST be created before we complete the current
+        // request because retry may happen right after
+        bufferedAsyncRequestBody = ByteBuffersAsyncRequestBody.of(buffers, bufferedLength);


We are passing buffers as it to ByteBuffersAsyncRequestBody , should we pass a copy new ArrayList<>(buffers) and make it threadsafe too using synchronized (buffersLock)

Extra copy/buffering is not necessary because it's entirely used internally and we know we will not modify it. Yup, will add synchronized

joviegas · 2025-08-21T21:01:10Z

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

+            // For unknown content length, we always create a new DownstreamBody because we don't know if there is data
+            // left or not, so we need to only send the body if there is actually data
+            long bufferedLength = currentBody.receivedBytesLength();
+            if (bufferedLength == 0) {


If possible just in response to this comment can you tell me how we will end up bufferedLength == 0 here ? As in for touch files () byte files)

It's explained above :) Will move it down and clarify

For unknown content length, we always create a new DownstreamBody because we don't know if there is data eft or not, so we need to only send the body if there is actually data

...core/src/main/java/software/amazon/awssdk/core/async/BufferedSplittableAsyncRequestBody.java

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

dagnir · 2025-08-21T22:03:56Z

...rc/main/java/software/amazon/awssdk/core/internal/async/NonRetryableSubAsyncRequestBody.java

+            onNumBytesConsumed.accept((long) length);
+            if (t != null) {
+                error(t);
+            }


Just for my understanding, why is onNumBytesConsumed always updated, instead of only when t == null?

The data is considered consumed once the future completes (exceptionally or not) since there's no retry anyway.

...rc/main/java/software/amazon/awssdk/core/internal/async/NonRetryableSubAsyncRequestBody.java

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

sonarqubecloud · 2025-08-22T02:58:11Z

Quality Gate failed

Failed conditions
76.6% Coverage on New Code (required ≥ 80%)
8.9% Duplication on New Code (required ≤ 3%)
C Reliability Rating on New Code (required ≥ A)

See analysis details on SonarQube Cloud

Catch issues before they fail your Quality Gate with our IDE extension SonarQube for IDE

zoewangg requested a review from Copilot August 13, 2025 23:12

This comment was marked as outdated.

Sign in to view

zoewangg force-pushed the zoewang/splitV2 branch from 93b3757 to 01a4cdb Compare August 14, 2025 00:45

zoewangg requested a review from Copilot August 14, 2025 18:38

This comment was marked as outdated.

Sign in to view

zoewangg changed the title ~~Introduce a new split method in AsyncRequestBody that returns an SdkP…~~ Stream retry support part 2: Introduce a new split method in AsyncRequestBody that returns an SdkP… Aug 14, 2025

zoewangg force-pushed the zoewang/splitV2 branch from 01a4cdb to 8f3a1ac Compare August 14, 2025 19:55

zoewangg requested a review from Copilot August 14, 2025 19:57

Copilot AI reviewed Aug 14, 2025

View reviewed changes

zoewangg force-pushed the zoewang/splitV2 branch from 8f3a1ac to e0a046e Compare August 14, 2025 22:04

Introduce a new split method in AsyncRequestBody that returns an SdkP…

d94cec4

…ublisher of ClosableAsyncRequestBody and use it in s3 multipart client

zoewangg force-pushed the zoewang/splitV2 branch from e0a046e to d94cec4 Compare August 14, 2025 22:53

zoewangg marked this pull request as ready for review August 14, 2025 22:54

zoewangg requested a review from a team as a code owner August 14, 2025 22:54

zoewangg force-pushed the zoewang/splitV2 branch from 3814adb to 3ffa942 Compare August 15, 2025 19:22

Fix build and add more tests

3c91cc9

zoewangg force-pushed the zoewang/splitV2 branch from 3ffa942 to 3c91cc9 Compare August 15, 2025 19:38

joviegas reviewed Aug 18, 2025

View reviewed changes

zoewangg force-pushed the zoewang/splitV2 branch 3 times, most recently from 3b244ec to 29f911a Compare August 21, 2025 03:07

Refactoring based on API surface are review

004733b

zoewangg force-pushed the zoewang/splitV2 branch from 29f911a to 004733b Compare August 21, 2025 16:33

joviegas reviewed Aug 21, 2025

View reviewed changes

dagnir requested changes Aug 21, 2025

View reviewed changes

zoewangg force-pushed the zoewang/splitV2 branch from 57cbc1f to 43ee3ba Compare August 22, 2025 00:49

Add more tests and address comments

0c2000e

zoewangg force-pushed the zoewang/splitV2 branch from 43ee3ba to 0c2000e Compare August 22, 2025 02:15

		}

		private void stubUploadFailsInitialAttemptCalls(int partNumber, ResponseDefinitionBuilder responseDefinitionBuilder) {

Stream retry support part 2: Introduce a new split method in AsyncRequestBody that returns an SdkP… #6346

Are you sure you want to change the base?

Stream retry support part 2: Introduce a new split method in AsyncRequestBody that returns an SdkP… #6346

Uh oh!

Conversation

zoewangg commented Aug 13, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Motivation and Context

Testing

Screenshots (if appropriate)

Types of changes

Checklist

License

Uh oh!

This comment was marked as outdated.

Uh oh!

This comment was marked as outdated.

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

Pull Request Overview

Reviewed Changes

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

zoewangg Aug 20, 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!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

zoewangg Aug 21, 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!

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!

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!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

sonarqubecloud bot commented Aug 22, 2025

Quality Gate failed

Uh oh!

Uh oh!

zoewangg commented Aug 13, 2025 •

edited

Loading

zoewangg Aug 20, 2025 •

edited

Loading

zoewangg Aug 21, 2025 •

edited

Loading