Splitting transformer #4826

L-Applin · 2024-01-11T17:32:05Z

Implementation of the Splitting transformer.

Split an AsyncResponseTranformer into multiple ones, retreived via a Publisher of AsyncResponseTranformer. When requested, the publisher will publish a new AsyncResponseTranformer to its subscriber, but the data received in this individual AsyncResponseTranformer will be buffered and then send to the upstream AsyncResponseTranformer (the one that was split).

Also includes a draft of the subscriber that will use the individually published AsyncResponseTransformer, but is still in a draft state.

The intent is to use the SplittingTransformer to perform multipart get request for s3 objects. The SplittingTranformer would publish an AsyncResponseTranformer instance when requested by the subscriber, which would perform the call to s3 using that AsyncResponseTranformer. By using the split() method on an instance of an existing AsyncResponseTranformer to create a SplittingTranformer, the behaviour of the splitting tranformer can be fine tuned to the different implementations of the AsyncResponseTranformer interface if required.

Diagram

L-Applin · 2024-01-12T16:07:43Z

...ava/software/amazon/awssdk/services/s3/internal/multipart/MultipartDownloaderSubscriber.java

+
+// [WIP]
+// Still work in progress, currently only used to help manual testing, please ignore
+public class MultipartDownloaderSubscriber implements Subscriber<AsyncResponseTransformer<GetObjectResponse, GetObjectResponse>> {


Ignore this class for now, this is only a stub that was used for manually testing with s3.

L-Applin · 2024-01-12T16:08:24Z

.../software/amazon/awssdk/services/s3/internal/multipart/MultipartDownloadIntegrationTest.java

+import software.amazon.awssdk.services.s3.model.GetObjectResponse;
+
+// WIP - please ignore for now, only used in manually testing
+class MultipartDownloadIntegrationTest {


Ignore for now, only used for manually testing with s3 objects.

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

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

cenedhryn · 2024-01-17T17:50:06Z

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

+    default SplitAsyncResponseTransformer<ResponseT, ResultT> split(long bufferSize) {
+        CompletableFuture<ResultT> future = new CompletableFuture<>();


Why can't the split() method return a SplittingTransformer instead of a SplitAsyncResponseTransformer? Can't the splitting transformer create the future and make it available to the user in the same way?

I was trying to avoid returning the concrete type SplittingTransformer and instead return the interface type SdkPublisher<AsyncResponseTransformer<T>>. If we return a SplittingTransformer here, we would need to make it @SdkPublicAPI

Is there any reasonSplitAsyncResponseTransformer doesn't implement AsyncResponseTransformer interface?

SplitAsyncResponseTransformer is just a holder for the publisher and a completable future, the class implementing AsyncResponseTransformer is SplittingTransformer.IndividualTransformer

Should we make it implement AsyncResponseTransformer?

no, I don't think so. The usage would be:

SplitAsyncResponseTransformer<GetObjectResponse, GetObjectResponse> split = transformer.split(...); split.publisher().subscribe(downloaderSubscriber); CompletableFuture<GetObjectResponse> future = split.preparedFuture(); // do whatever we want with the future GetObjectResponse response = future.join();

It just seems a bit odd that the static factory method returns a class that doesn't implement the same interface. Can we make it implement it? We did the same for toPublisher method

https://github.com/aws/aws-sdk-java-v2/blob/master/core/sdk-core/src/main/java/software/amazon/awssdk/core/async/AsyncResponseTransformer.java#L235

.split() is not a static method though, it is an instance method called on a specific instance of an AsyncResponseTransformer just like the split() method on AsyncRequestBody. We could instead return just the SdkPublisher<AsyncResponseTransformer<ResponseT, ResponseT>> like AsyncRequestBody.split. I would need to find another way to deal with the future that needs to be returned from the service call.

I see, I'll need to think a bit more on this. Let's keep this conversation open for now.

Can we add a TODO to revisit the naming? It's a bit odd that this class is suffixed with AsyncResponseTransformer but doesn't implement it

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

...-core/src/test/java/software/amazon/awssdk/core/internal/async/SplittingTransformerTest.java

utils/src/main/java/software/amazon/awssdk/utils/async/DelegatingBufferingSubscriber.java

zoewangg

Left some initial comments. Still going through the PR

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

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

utils/src/main/java/software/amazon/awssdk/utils/async/DelegatingBufferingSubscriber.java

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

...-core/src/test/java/software/amazon/awssdk/core/internal/async/SplittingTransformerTest.java

- optional maxElement for SplittingTransformer - white box verification tests for DelegatingBufferingSubscriber and IndividualPartSubscriber - make DelegatingBufferingSubscriber.onNext iterative

zoewangg

Looking at DelegatingBufferingSubscriber now

zoewangg · 2024-02-06T21:11:29Z

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

+    default SplitAsyncResponseTransformer<ResponseT, ResultT> split(long bufferSize) {
+        CompletableFuture<ResultT> future = new CompletableFuture<>();


Is there any reasonSplitAsyncResponseTransformer doesn't implement AsyncResponseTransformer interface?

zoewangg · 2024-02-06T21:16:23Z

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

+                                long bufferSize, CompletableFuture<ResultT> returnFuture) {
+        this.upstreamResponseTransformer = Validate.paramNotNull(upstreamResponseTransformer, "asyncRequestBody");
+        this.returnFuture = Validate.paramNotNull(returnFuture, "returnFuture");
+        this.maximumBufferSize = NumericUtils.saturatedCast(bufferSize);


Can we support long?

we might need to allocate a ByteBuffer of that size in the case we receive a bigger ByteBuffer from the publisher of the individual request body. See DelegatingBufferingSubscriber#L89

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

utils/src/main/java/software/amazon/awssdk/utils/async/DelegatingBufferingSubscriber.java

zoewangg · 2024-02-06T22:24:23Z

utils/src/main/java/software/amazon/awssdk/utils/async/DelegatingBufferingSubscriber.java

+    @Override
+    public void onNext(ByteBuffer byteBuffer) {
+        while (bufferOverflows(byteBuffer)) {
+            flushStorageToDelegate();


We need to track downstream subscription; if there's no demand from downstream, we shouldn't call subscriber.onNext

Example: https://github.com/aws/aws-sdk-java-v2/blob/master/utils/src/main/java/software/amazon/awssdk/utils/async/FlatteningSubscriber.java

It's going to be super complex and hard to maintain if we add that logic... Let me see if we can reuse some of the logic in some way.

Right, let me know if you think of something better

Well, can't think of another way, we just need to implement it :(

zoewangg · 2024-02-06T22:27:01Z

utils/src/main/java/software/amazon/awssdk/utils/async/DelegatingBufferingSubscriber.java

+        super.onSubscribe(subscription);
+        this.subscription = subscription;
+        publisherToStorage.subscribe(storage);
+        subscription.request(maximumBuffer);


The number is actually number of ByteBuffer instead of number of bytes.

- SplittingTransformer builder - DelegatingBufferingSubscriber protected API - remove spec comments - refactor emit method

…maximumBufferSize

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

zoewangg · 2024-02-09T00:01:39Z

utils/src/main/java/software/amazon/awssdk/utils/async/DelegatingBufferingSubscriber.java

+    @Override
+    public void onNext(ByteBuffer byteBuffer) {
+        while (bufferOverflows(byteBuffer)) {
+            flushStorageToDelegate();


Well, can't think of another way, we just need to implement it :(

zoewangg · 2024-02-09T00:08:03Z

utils/src/main/java/software/amazon/awssdk/utils/async/DelegatingBufferingSubscriber.java

+
+    @Override
+    public void onNext(ByteBuffer byteBuffer) {
+        while (bufferOverflows(byteBuffer)) {


I think we should trigger a flush as long as we have data instead of buffer is full. See how FlatteningSubscriber is implemented https://github.com/aws/aws-sdk-java-v2/blob/master/utils/src/main/java/software/amazon/awssdk/utils/async/FlatteningSubscriber.java#L81

utils/src/main/java/software/amazon/awssdk/utils/async/DelegatingBufferingSubscriber.java

Move common logic for FlatteningSubscriber and DelegatingBufferingSubscriber to new class AbstractFlatteningSubscriber

L-Applin · 2024-02-12T17:17:48Z

To answer most concerns about DelegatingBufferingSubscriber, it was modified to reuse the logic in FlatteningSubscriber for its implementation

sonarcloud · 2024-02-12T17:18:24Z

Quality Gate failed

Failed conditions
79.0% Coverage on New Code (required ≥ 80%)
C Reliability Rating on New Code (required ≥ A)

See analysis details on SonarCloud

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

zoewangg · 2024-02-12T23:04:25Z

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

+         * @param maximumBufferSize the amount of data buffered and the size of the chunk of data
+         * @return an instance of this builder
+         */
+        public Builder<ResponseT, ResultT> maximumBufferSize(long maximumBufferSize) {


Should it be Long?

zoewangg · 2024-02-12T23:09:30Z

utils/src/main/java/software/amazon/awssdk/utils/async/DelegatingBufferingSubscriber.java

+     */
+    @Override
+    protected boolean onNextNeeded() {
+        return super.onNextNeeded() && storage.peek().isPresent() && storage.peek().get().type() == ON_NEXT;


Instead of doing super.onNextNeeded(), suggesting using template methods:

In the parent class:

private boolean onNextNeeded() { return downstreamDemand.get() > 0 && additionalOnNextNeededCheck(); } /** Can be overridden by subclasses to provide additional checks */ protected boolean additionalOnNextNeededCheck() { return true; }

zoewangg · 2024-02-12T23:36:38Z

utils/src/main/java/software/amazon/awssdk/utils/async/DelegatingBufferingSubscriber.java

+     */
+    private final StoringSubscriber<ByteBuffer> storage = new StoringSubscriber<>(Integer.MAX_VALUE);
+
+    public DelegatingBufferingSubscriber(long maximumBuffer, Subscriber<? super ByteBuffer> subscriber) {


Since this is a protected API, we should use builder

zoewangg · 2024-02-12T23:37:05Z

utils/src/main/java/software/amazon/awssdk/utils/async/DelegatingBufferingSubscriber.java

+
+    public DelegatingBufferingSubscriber(long maximumBuffer, Subscriber<? super ByteBuffer> subscriber) {
+        super(subscriber);
+        this.maximumBuffer = maximumBuffer;


nit: maximumBufferInBytes

zoewangg · 2024-02-12T23:38:49Z

utils/src/main/java/software/amazon/awssdk/utils/async/DelegatingBufferingSubscriber.java

+
+    @Override
+    public void onNext(ByteBuffer item) {
+        if (currentlyBuffered.get() + item.remaining() > maximumBuffer) {


I think we should trigger a flush as long as we have data instead of buffer is full.

zoewangg · 2024-02-12T23:41:12Z

utils/src/main/java/software/amazon/awssdk/utils/async/AbstractFlatteningSubscriber.java

+     * Returns true if we need to call onNext downstream. If this is executed outside the handling-state-update condition, the
+     * result is subject to change.
+     */
+    protected boolean onNextNeeded() {


See my previous comment, suggesting template methods to allow subclasses to add additional checks. Can we make those methods that are intended to be overridden by subclasses package private?

zoewangg · 2024-02-12T23:42:29Z

utils/src/main/java/software/amazon/awssdk/utils/async/AbstractFlatteningSubscriber.java

+     * result is subject to change.
+     */
+    protected boolean onCompleteNeeded() {
+        return onCompleteCalledByUpstream && !terminalCallMadeDownstream;


Same here, suggesting template methods.

private boolean onCompleteNeeded() { return onCompleteCalledByUpstream && !terminalCallMadeDownstream && additionalOnCompleteNeededCheck() ; } boolean additionalOnCompleteNeededCheck() { return true; }

zoewangg · 2024-02-12T23:45:49Z

utils/src/main/java/software/amazon/awssdk/utils/async/AbstractFlatteningSubscriber.java

+     *
+     * @param item the value with which onNext was called.
+     */
+    protected abstract void doWithItem(T item);


I can't think of a better naming for this one, can we make this package private so that we can change it in the future?

zoewangg · 2024-02-12T23:49:17Z

utils/src/main/java/software/amazon/awssdk/utils/async/AbstractFlatteningSubscriber.java

+import software.amazon.awssdk.utils.Logger;
+
+@SdkInternalApi
+public abstract class AbstractFlatteningSubscriber<T, U> extends DelegatingSubscriber<T, U> {


This is a protected API. nit: this no longer does flattening, maybe BaseSubscriberConverter/Adaptor?

L-Applin added 8 commits January 9, 2024 09:21

working - no buffering

fe243ae

create DelegatingBufferingSubscriber and use it in SplittingTransformer

e6d2ab9

split now returns SplitAsyncResponseTransformer

a73ec72

clean up, doc, more tests

f76307a

clean up

54223e9

added todo for buffer size

076f5c1

DelegatingBufferingSubscriber Internal API and some clean up

57c0140

fix checkstyle

4b10284

L-Applin marked this pull request as ready for review January 12, 2024 16:14

L-Applin requested a review from a team as a code owner January 12, 2024 16:14

L-Applin commented Jan 12, 2024

View reviewed changes

L-Applin added 4 commits January 12, 2024 11:42

fix checkstyle

c53a980

added demand management in SplittingTransformer

deeb551

clean-up

1508140

fix checkstyle

99098de

cenedhryn reviewed Jan 16, 2024

View reviewed changes

cenedhryn reviewed Jan 17, 2024

View reviewed changes

.../sdk-core/src/main/java/software/amazon/awssdk/core/internal/async/SplittingTransformer.java Outdated Show resolved Hide resolved

cenedhryn reviewed Jan 17, 2024

View reviewed changes

zoewangg reviewed Jan 23, 2024

View reviewed changes

L-Applin added 3 commits January 26, 2024 10:24

TCK tests

1e18772

TCK tests

173b2d1

- optional maxElement for SplittingTransformer - white box verification tests for DelegatingBufferingSubscriber and IndividualPartSubscriber - make DelegatingBufferingSubscriber.onNext iterative

checkstyle

5d4e7e8

zoewangg reviewed Feb 6, 2024

View reviewed changes

cenedhryn reviewed Feb 6, 2024

View reviewed changes

.../sdk-core/src/main/java/software/amazon/awssdk/core/internal/async/SplittingTransformer.java Outdated Show resolved Hide resolved

.../sdk-core/src/main/java/software/amazon/awssdk/core/internal/async/SplittingTransformer.java Outdated Show resolved Hide resolved

zoewangg reviewed Feb 6, 2024

View reviewed changes

L-Applin added 3 commits February 7, 2024 16:21

Some PR comment

60fa61e

- SplittingTransformer builder - DelegatingBufferingSubscriber protected API - remove spec comments - refactor emit method

checkstyle

ac0f2ce

removed maxElements, fix checkstyle, rename bufferSize in builder to …

7b0dfac

…maximumBufferSize

zoewangg reviewed Feb 8, 2024

View reviewed changes

.../sdk-core/src/main/java/software/amazon/awssdk/core/internal/async/SplittingTransformer.java Outdated Show resolved Hide resolved

zoewangg reviewed Feb 9, 2024

View reviewed changes

utils/src/main/java/software/amazon/awssdk/utils/async/DelegatingBufferingSubscriber.java Outdated Show resolved Hide resolved

L-Applin added 2 commits February 9, 2024 16:58

Use FlatteningSubscriber as base for DelegatingBufferingSubscriber.

3d2baaf

Move common logic for FlatteningSubscriber and DelegatingBufferingSubscriber to new class AbstractFlatteningSubscriber

japicmp fix, small javadoc

8f1c7e9

L-Applin merged commit 8c18e39 into feature/master/s3mpu Feb 12, 2024
15 of 16 checks passed

zoewangg reviewed Feb 12, 2024

View reviewed changes

L-Applin mentioned this pull request Feb 14, 2024

Implementation of MultipartDownloaderSubscriber #4931

Merged

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Splitting transformer #4826

Splitting transformer #4826

L-Applin commented Jan 11, 2024 •

edited

Loading

L-Applin Jan 12, 2024

L-Applin Jan 12, 2024

cenedhryn Jan 17, 2024

L-Applin Jan 17, 2024

zoewangg Feb 6, 2024

L-Applin Feb 7, 2024

zoewangg Feb 8, 2024

L-Applin Feb 9, 2024

zoewangg Feb 9, 2024

L-Applin Feb 9, 2024 •

edited

Loading

zoewangg Feb 12, 2024

zoewangg Feb 12, 2024

zoewangg left a comment

zoewangg left a comment

zoewangg Feb 6, 2024

zoewangg Feb 6, 2024

L-Applin Feb 7, 2024

zoewangg Feb 6, 2024 •

edited

Loading

L-Applin Feb 7, 2024

zoewangg Feb 9, 2024

zoewangg Feb 6, 2024

zoewangg Feb 9, 2024

zoewangg Feb 9, 2024

L-Applin commented Feb 12, 2024

sonarcloud bot commented Feb 12, 2024

zoewangg Feb 12, 2024

zoewangg Feb 12, 2024

zoewangg Feb 12, 2024

zoewangg Feb 12, 2024

zoewangg Feb 12, 2024

zoewangg Feb 12, 2024

zoewangg Feb 12, 2024

zoewangg Feb 12, 2024

zoewangg Feb 12, 2024

		default SplitAsyncResponseTransformer<ResponseT, ResultT> split(long bufferSize) {
		CompletableFuture<ResultT> future = new CompletableFuture<>();

Splitting transformer #4826

Splitting transformer #4826

Conversation

L-Applin commented Jan 11, 2024 • edited Loading

Diagram

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

L-Applin Feb 9, 2024 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

zoewangg left a comment

Choose a reason for hiding this comment

zoewangg left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

zoewangg Feb 6, 2024 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

L-Applin commented Feb 12, 2024

sonarcloud bot commented Feb 12, 2024

Quality Gate failed

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

L-Applin commented Jan 11, 2024 •

edited

Loading

L-Applin Feb 9, 2024 •

edited

Loading

zoewangg Feb 6, 2024 •

edited

Loading