Stream retry support part 1: Add BufferingAsyncRequestBody that buffers the entire content and sup…

[zoewangg]

…ports multiple concurrent subscribers

Motivation and Context

#6198

This is the first PR to support retry in stream based transfer in s3 multipart client.

This PR adds BufferingAsyncRequestBody that is essentially a multicast publisher and can be subscribed multiple times until it's been closed

This PR made the following changes to existing ByteBuffersAsyncRequestBody:

• implement SdkClosable interface to support clearing buffer
• it's now compliant to the reactive streams spec

[Copilot]

Pull Request Overview

This PR introduces BufferingAsyncRequestBody, a new implementation of AsyncRequestBody that buffers content in memory and supports multiple concurrent subscribers for stream retry functionality in the AWS SDK.

• Adds a thread-safe, multicast publisher that can buffer data incrementally and replay it to multiple subscribers
• Implements comprehensive test coverage including concurrency tests and reactive streams TCK compliance
• Provides resource management through proper cleanup of buffered data and subscriber notifications

Reviewed Changes

Copilot reviewed 3 out of 3 changed files in this pull request and generated 3 comments.

File	Description
BufferingAsyncRequestBody.java	Core implementation of the buffering async request body with multicast publisher capabilities
BufferingAsyncRequestBodyTest.java	Comprehensive unit tests covering functionality, edge cases, and thread safety
BufferingAsyncRequestBodyTckTest.java	Reactive Streams TCK compliance test suite

Comments suppressed due to low confidence (1)

core/sdk-core/src/main/java/software/amazon/awssdk/core/internal/async/BufferingAsyncRequestBody.java:208

• [nitpick] The error message should be consistent with the one used in the subscribe method. Consider using 'AsyncRequestBody has been closed' to match the error message on line 172.

            subscriptions.forEach(s -> s.notifyError(new IllegalStateException("The publisher has been closed")));

[sonarqubecloud]

Quality Gate passed

Issues
18 New issues
0 Accepted issues

Measures
0 Security Hotspots
92.4% Coverage on New Code
0.0% Duplication on New Code

See analysis details on SonarQube Cloud

[dagnir]

It's kind of weird to me that contentLength could return an empty optional, but we require all the data to be buffered first before we start publishing. Should we just require all of the data to be ready at construction time?

[zoewangg]

Hmm, good point! It'll also reduce the complexity

[zoewangg]

Removed this new class and updated the existing ByteBuffersAsyncRequestBody to make it compliant to the reactive stream spec and implement closable interface; it already supports multicast today