Support Async pagination #419

varunnvs92 · 2018-02-02T19:41:33Z

Add support for Async Pagination
#185

Description

Generates extra methods in client and interface for paginated methods. These methods return a custom publisher that can be used to request a stream of data without the customer explicitly making service calls.
Added RxJava2 as test dependency to show a test case on using it. I am not fully satisfied with the async user experience using the pub-sub interfaces. Using third party reactive streams implementations will help in making experience better as there are lot of helper methods.

Edit1

Added resume method in the response that takes in last successful page and returns a new instance of the response class. The new instance can be used to iterate/request data after the last successful page

Motivation and Context

#185

Testing

Added unit tests
See ListObjectsV2PaginatorsIntegrationTest.java for integration test and how customer uses

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 read the README document
I have added tests to cover my changes
All new and existing tests passed
A short description of the change has been added to the CHANGELOG

License

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

zoewangg · 2018-02-03T01:12:15Z

codegen/src/main/java/software/amazon/awssdk/codegen/docs/AsyncOperationDocProvider.java

+
+        @Override
+        protected String appendToDescription() {
+            return opModel.isPaginated() ? paginationDocs.getDocsForAsyncOperation() : null;


return null will cause a Null Dereference risk on fortify, we can probably return Optional here

spfink · 2018-02-23T18:48:26Z

codegen/src/main/java/software/amazon/awssdk/codegen/docs/PaginationDocs.java

+                             + "see {@link $T#$L($T)}. If there are errors in your "
+                             + "request, you will see the failures only after you start streaming the data.</p>",
+                             getPublisherType(), SUBSCRIBE_METHOD_NAME, getSubscriberType())
+                         .add(getAsyncCodeSnippets())


Extra space here

spfink · 2018-02-23T18:49:29Z

codegen/src/main/java/software/amazon/awssdk/codegen/docs/PaginationDocs.java

+    /**
+     * @return A Poet {@link ClassName} for the reactive streams {@link Publisher}.
+     */
+    private ClassName getPublisherType() {


Do we really need this and the one below in separate methods?

These are used 2-3 places in the class. I thought it would be better to have them in separate method rather than calling the ClassName.get() method.

spfink · 2018-02-23T18:49:58Z

...en/src/main/java/software/amazon/awssdk/codegen/emitters/tasks/PaginatorsGeneratorTasks.java

+        return new PoetGeneratorTask(paginatorsClassDir, model.getFileHeader(), classSpec);
+    }
+
+    private Stream<GeneratorTask> createASyncTasks() {


ASync -> Async

spfink · 2018-02-23T18:50:25Z

codegen/src/main/java/software/amazon/awssdk/codegen/poet/PoetExtensions.java

     */
    @ReviewBeforeRelease("Naming of response shape for paginated APIs")
    public ClassName getResponseClassForPaginatedSyncOperation(String operationName) {
-        return ClassName.get(model.getMetadata().getFullPaginatorsPackageName(), operationName + "Paginator");
+        return ClassName.get(model.getMetadata().getFullPaginatorsPackageName(), operationName + "Iterable");


Why the change in name here?

The method ends with "Paginator", so didn't want same name for response type. For example,
ListTablesIterable listTablesPaginator(ListTablesRequest);
over
ListTablesPaginator listTablesPaginator(ListTablesRequest);

This also conveys the return type is a Iterable. It also matches the convention for async method.
// Async method declaration
ListTablesPublisher listTablesPaginator(ListTablesRequest);

spfink · 2018-02-23T18:51:27Z

codegen/src/main/java/software/amazon/awssdk/codegen/poet/client/AsyncClientInterface.java

+        return paginatedTraditionalMethodBody(builder, opModel).build();
+    }
+
+    protected MethodSpec.Builder paginatedTraditionalMethodBody(MethodSpec.Builder builder, OperationModel operationModel) {


Why is this a separate method?

This is overriden in AsyncClientClass

spfink · 2018-02-23T18:54:23Z

...n/src/test/resources/software/amazon/awssdk/codegen/poet/client/test-async-client-class.java

+     * 1) Using the forEach helper method. This uses @
+     * {@link software.amazon.awssdk.core.pagination.async.SequentialSubscriber} internally
+     *
+     * <pre>


Does this all look good in javadoc?

Formatting is good. But classes are displayed with FQCN. So it looks little verbose.

spfink · 2018-02-23T18:56:33Z

core/src/main/java/software/amazon/awssdk/core/pagination/async/PaginatedItemsPublisher.java

+        subscriber.onSubscribe(new ItemsSubscription(subscriber));
+    }
+
+    private class ItemsSubscription implements org.reactivestreams.Subscription {


Why are we using the full fqcn import?

spfink · 2018-02-23T18:57:01Z

pom.xml

@@ -247,6 +247,11 @@
                <artifactId>netty-reactive-streams-http</artifactId>
                <version>2.0.0</version>
            </dependency>
+            <dependency>
+                <groupId>io.reactivex.rxjava2</groupId>


What's the size of this dependency?

Size is 2.1MB. I added it to show that customer can use 3rd party implementations for additional helper methods. I can remove it if we decide so.

Is this a test dependency?

Yes, I have marked the scope as test in services pom. Do we need it here too?

No just where it's used is fine. Just making sure it's a test dependency.

shorea

Still looking at it, will review more later.

shorea · 2018-02-23T21:44:56Z

codegen/src/main/java/software/amazon/awssdk/codegen/docs/AsyncOperationDocProvider.java

+
+        @Override
+        protected String appendToDescription() {
+            return opModel.isPaginated() ? paginationDocs.getDocsForAsyncOperation() : "";


Do you need to check isPaginated? Would this be invoked if the operation isn't paginated?

Good catch, removed the check.

shorea · 2018-02-23T21:49:28Z

codegen/src/main/java/software/amazon/awssdk/codegen/docs/PaginationDocs.java

+                        .add("<p>When the operation is called, an instance of this class is returned.  At this point, "
+                             + "no service calls are made yet and so there is no guarantee that the request is valid. "
+                             + "The subscribe method should be called as a request to stream data. For more info, "
+                             + "see {@link $T#$L($T)}. If there are errors in your "


This is a question for both sync and async pagination. How do we handle retryable/transient errors while paginating? Is the iterable/publisher considered terminated at that point and the customer has to make a new request? Is there a way they can get the request object to reconnect (i.e. without getting the last response tokens manually)

As discussed offline, added the resume() method in the Response.

shorea · 2018-02-23T21:50:18Z

codegen/src/main/java/software/amazon/awssdk/codegen/docs/PaginationDocs.java

+                             syncResponsePageType())
+                        .add("<p>When the operation is called, an instance of this class is returned.  At this point, "
+                             + "no service calls are made yet and so there is no guarantee that the request is valid. "
+                             + "The subscribe method should be called as a request to stream data. For more info, "


Can we add some documentation on what multiple subscribes means? I.E. each one is a separate iteration from the starting request.

shorea · 2018-02-23T21:50:53Z

codegen/src/main/java/software/amazon/awssdk/codegen/docs/PaginationDocs.java

+
+        return CodeBlock.builder()
+                        .add("\n\n<p>The following are few ways to use the response class:</p>")
+                        .add("1) Using the forEach helper method. This uses @{@link $T} internally",


Do they need to care what it uses internally?

Nope, removed.

shorea · 2018-02-23T21:51:42Z

codegen/src/main/java/software/amazon/awssdk/codegen/docs/PaginationDocs.java

+                        .add(buildCode(CodeBlock.builder()
+                                                .add(callOperationOnClient)
+                                                .add(CodeBlock.builder()
+                                                              .addStatement("CompletableFuture<Void> future = publisher"


Should the future here be fulfilled with the last response? Would that be useful or confusing?

Where would you do it? Can you show me sample code

shorea · 2018-02-23T21:54:00Z

...en/src/main/java/software/amazon/awssdk/codegen/emitters/tasks/PaginatorsGeneratorTasks.java

+                     .collect(Collectors.toList());
+    }
+
+    private Stream<GeneratorTask> createSyncTasks() {


You should be able to reduce a little duplication here and create a method that takes in this::createSyncTask this::createAsyncTask respectively.

shorea · 2018-02-23T21:55:44Z

codegen/src/main/java/software/amazon/awssdk/codegen/poet/client/AsyncClientClass.java

@@ -128,6 +129,14 @@ private MethodSpec closeMethod() {

    }

+    @Override
+    protected MethodSpec.Builder paginatedTraditionalMethodBody(MethodSpec.Builder builder, OperationModel opModel) {


What does traditional mean in this context? Non-simple?

Yep, a method that takes in a request and returns a response.

Do you need to use "traditional" here (vs nothing) and in other cases? Are there any cases beyond paginated methods and paginated simple methods?

Got it. There is no need for the "traditional" part as its the only method body for paginated operations. Will remove.

shorea · 2018-02-24T00:28:09Z

...n/src/test/resources/software/amazon/awssdk/codegen/poet/client/test-async-client-class.java

+     * {@link #paginatedOperationWithResultKey(software.amazon.awssdk.services.json.model.PaginatedOperationWithResultKeyRequest)}
+     * operation.</b>
+     * </p>
+     *


Why no onError and onComplete in the custom Subscriber example?

As the body of the methods are empty, didn't think it would be useful to add all of them and clutter the docs. Anyways when customers tries to add code like this, compiler will give errors about the missing methods

shorea · 2018-02-24T00:30:37Z

...t/resources/software/amazon/awssdk/codegen/poet/client/test-json-async-client-interface.java

+     * </p>
+     *
+     * @param paginatedOperationWithResultKeyRequest
+     * @return A Java Future containing the result of the PaginatedOperationWithResultKey operation returned by the


This is incorrect

Fixed the return statement.
Needs some more changes to fix the next line (prefix for async throws statement)

shorea · 2018-02-24T00:33:00Z

...t/resources/software/amazon/awssdk/codegen/poet/client/test-json-async-client-interface.java

+     * @see <a href="http://docs.aws.amazon.com/goto/WebAPI/json-service-2010-05-08/PaginatedOperationWithResultKey"
+     *      target="_top">AWS API Documentation</a>
+     */
+    default PaginatedOperationWithResultKeyPublisher paginatedOperationWithResultKeyPaginator(


Is there a test that exercises the no arg simple method?

shorea · 2018-02-24T00:52:05Z

core/src/main/java/software/amazon/awssdk/core/pagination/async/PaginatedItemsPublisher.java

+
+            outstandingRequests.addAndGet(n);
+
+            synchronized (this) {


Can we use compareAndSet here and lose the lock?

shorea · 2018-02-24T00:53:56Z

core/src/main/java/software/amazon/awssdk/core/pagination/async/PaginatedItemsPublisher.java

+        }
+
+        private void handleRequestsRecursively() {
+            if (currentPage != null && !nextPageFetcher.hasNextPage(currentPage) &&


This expression is a bit hard to read. Some methods may help?

if(!(hasNextPage() || hasMoreItems())

shorea · 2018-02-24T00:57:42Z

core/src/main/java/software/amazon/awssdk/core/pagination/async/PaginatedItemsPublisher.java

+        }
+
+        private void fetchNextPage() {
+            CompletableFuture<ResponseT> future = nextPageFetcher.nextPage(currentPage);


Inline future

shorea · 2018-02-24T00:59:11Z

core/src/main/java/software/amazon/awssdk/core/pagination/async/PaginatedItemsPublisher.java

+                    sendNextElement();
+                }
+                if (error != null) {
+                    subscriber.onError(error);


Is there something that tells the publisher to stop publishing items?

Added a isTerminated flag

shorea · 2018-02-24T01:01:28Z

core/src/main/java/software/amazon/awssdk/core/pagination/async/PaginatedItemsPublisher.java

+
+        @Override
+        public void cancel() {
+            outstandingRequests.set(0);


Would future calls to request(n) re-activate the subscription? Should we guard against that?

Added a isTerminated flag

shorea · 2018-02-24T01:04:47Z

core/src/main/java/software/amazon/awssdk/core/pagination/async/PaginatedItemsPublisher.java

+ * @param <ResponseT> The type of a single response page
+ * @param <ItemT> The type of paginated member in a response page
+ */
+public class PaginatedItemsPublisher<ResponseT, ItemT> implements SdkPublisher<ItemT> {


There's a lot of duplicated code from the Response publisher. Is there a way to consolidate?

This code from Rx might be relevant for making this more abstract

https://github.com/ReactiveX/RxJava/blob/2.x/src/main/java/io/reactivex/internal/operators/flowable/FlowableGenerate.java

Didn't use the above link. But abstracted out the common code.

shorea · 2018-02-24T01:05:28Z

core/src/main/java/software/amazon/awssdk/core/pagination/async/ResponsesSubscription.java

+
+        outstandingRequests.getAndDecrement();
+
+        CompletableFuture<ResponseT> future = nextPageFetcher.nextPage(currentPage);


shorea · 2018-02-24T01:37:36Z

core/src/main/java/software/amazon/awssdk/core/pagination/async/ResponsesSubscription.java

+            return;
+        }
+
+        if (outstandingRequests.get() <= 0) {


I'm thinking there's a race condition between this and the code that initiates a new task.

Evaluation order
If(outstandingRequests.get() <= 0) evalues to true
customer requests more demand
outstandingRequests incremented
we check if task is still running, see that it is, and bail
we set isTaskRunning to false here and now we've dropped a request

I think we would need to acquire the same lock to prevent this.

Modified the code a bit. Still looking for race conditions

spfink · 2018-03-01T18:02:23Z

codegen/src/main/java/software/amazon/awssdk/codegen/poet/client/AsyncClientClass.java

@@ -128,6 +129,14 @@ private MethodSpec closeMethod() {

    }

+    @Override
+    protected MethodSpec.Builder paginatedTraditionalMethodBody(MethodSpec.Builder builder, OperationModel opModel) {


Do you need to use "traditional" here (vs nothing) and in other cases? Are there any cases beyond paginated methods and paginated simple methods?

spfink · 2018-03-01T21:16:05Z

.../software/amazon/awssdk/codegen/poet/paginators/PaginatedOperationWithResultKeyIterable.java

+            return new PaginatedOperationWithResultKeyIterable(client, firstRequest.toBuilder()
+                                                                                   .nextToken(lastSuccessfulPage.nextToken()).build());
+        }
+        return new PaginatedOperationWithResultKeyIterable(client, firstRequest) {


What's the behavior here? To return an empty list if no additional pages are available?

If customer tries to resume from last page (for some reason), the iterator method would return empty iterator as there would be no more pages.

spfink · 2018-03-01T21:16:16Z

...software/amazon/awssdk/codegen/poet/paginators/PaginatedOperationWithResultKeyPublisher.java

+
+    private final AsyncPageFetcher nextPageFetcher;
+
+    private boolean isLastPage;


What's this used for?

In resume method, we return EmptySubscription (EmptyIterator for sync) on subscribe if there are no more pages. This works when customer subscribes to get pages. If you try to subscribe to get Items (PaginatedItemsPublisher class), there is currently no way to recognize if input is last page and return EmptySubscription. The two options I see are:

Pass a value to PaginatedItemsPublisher to indicate its last page (current approach)

Override all the items methods in resume method to use EmptySubscription (if there are no more pages). I felt this will bloat all the publisher classes and chose option 1.
The resume method in option 2 will look like:

public PaginatedOperationWithResultKeyPublisher resume(final PaginatedOperationWithResultKeyResponse lastSuccessfulPage) { if (nextPageFetcher.hasNextPage(lastSuccessfulPage)) { return new PaginatedOperationWithResultKeyPublisher(client, firstRequest.toBuilder() .nextToken(lastSuccessfulPage.nextToken()).build()); } return new PaginatedOperationWithResultKeyPublisher(client, firstRequest) { @Override public void subscribe(Subscriber<? super PaginatedOperationWithResultKeyResponse> subscriber) { subscriber.onSubscribe(new EmptySubscription(subscriber)); } @Override public SdkPublisher<SimpleStruct> items() { return new EmptyPublisher<>(); // A new publisher that creates EmptySubscription when subscribed to } //More overriden methods if there are multiple result keys }; }

shorea

I'll finish looking at this tonight.

shorea · 2018-03-02T01:20:55Z

core/src/main/java/software/amazon/awssdk/core/pagination/async/EmptySubscription.java

+            throw new IllegalArgumentException("Non-positive request signals are illegal");
+        }
+
+        subscriber.onComplete();


Race condition here, if I queue up two requests then I can get two completes.

This could should fix it.

if(isTerminated.compareAndSet(false, true)) { subscriber.onComplete(); }

shorea · 2018-03-02T01:21:46Z

...tware/amazon/awssdk/codegen/poet/paginators/PaginatedOperationWithoutResultKeyPublisher.java

+        }.withLastPage(true);
+    }
+
+    PaginatedOperationWithoutResultKeyPublisher withLastPage(boolean isLastPage) {


Why do you need a setter for this? Can't you just set it in the constructor?

It can be set in constructor too. But having it in setter might hide it better constructor.

Actually thinking again, having a private constructor is better than the setter. Will fix it.

zoewangg reviewed Feb 3, 2018

View reviewed changes

varunnvs92 force-pushed the varunkn/AsyncPagination branch 3 times, most recently from f7c140d to 6beb521 Compare February 16, 2018 01:02

varunnvs92 force-pushed the varunkn/AsyncPagination branch from 6beb521 to 4b7f0cd Compare February 22, 2018 20:11

spfink reviewed Feb 23, 2018

View reviewed changes

shorea reviewed Feb 23, 2018

View reviewed changes

shorea reviewed Feb 24, 2018

View reviewed changes

varunnvs92 force-pushed the varunkn/AsyncPagination branch 2 times, most recently from de3b0e6 to cdcba96 Compare March 1, 2018 20:38

spfink reviewed Mar 1, 2018

View reviewed changes

varunnvs92 force-pushed the varunkn/AsyncPagination branch from cdcba96 to bc78fb0 Compare March 1, 2018 22:55

shorea reviewed Mar 2, 2018

View reviewed changes

varunnvs92 force-pushed the varunkn/AsyncPagination branch from bc78fb0 to 8558232 Compare March 2, 2018 03:53

shorea approved these changes Mar 3, 2018

View reviewed changes

varunnvs92 force-pushed the varunkn/AsyncPagination branch from 60e8177 to 813a7c2 Compare March 5, 2018 19:51

Support Async pagination

3b12e24

varunnvs92 force-pushed the varunkn/AsyncPagination branch from 813a7c2 to 3b12e24 Compare March 5, 2018 19:55

varunnvs92 mentioned this pull request Mar 5, 2018

Use Builders in async pagination classes #435

Closed

varunnvs92 merged commit 7544074 into master Mar 5, 2018

varunnvs92 deleted the varunkn/AsyncPagination branch May 15, 2018 17:00


		outstandingRequests.getAndDecrement();

		CompletableFuture<ResponseT> future = nextPageFetcher.nextPage(currentPage);


		private final AsyncPageFetcher nextPageFetcher;

		private boolean isLastPage;

Support Async pagination #419

Support Async pagination #419

Conversation

varunnvs92 commented Feb 2, 2018 • edited Loading

Description

Motivation and Context

Testing

Types of changes

Checklist

License

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

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

varunnvs92 Feb 23, 2018 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

shorea 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

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

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

varunnvs92 Feb 27, 2018 • 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

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

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

shorea 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

varunnvs92 commented Feb 2, 2018 •

edited

Loading

varunnvs92 Feb 23, 2018 •

edited

Loading

varunnvs92 Feb 27, 2018 •

edited

Loading