izian
commented
3 years ago A Java class to show, in complete isolation, the failure scenario for the SDK class in question which discards data from Dynamo results
package com.mycompany.hello;
import org.reactivestreams.Publisher;
import org.reactivestreams.Subscriber;
import org.reactivestreams.Subscription;
import software.amazon.awssdk.utils.async.FlatteningSubscriber;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Iterator;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
public class FlatteningSubscriberTest {
static List<String> ITEMS_ON_PAGE = Arrays.asList("1","2","3","4","5","6","7","8","9","10"); //10 per page
static List<List<String>> PAGES = Arrays.asList(ITEMS_ON_PAGE, ITEMS_ON_PAGE, ITEMS_ON_PAGE, ITEMS_ON_PAGE, ITEMS_ON_PAGE); //5 pages
public static void main(String... args) throws Exception {
MySubscriber mySubscriber = new MySubscriber();
FlatteningSubscriber<String> flatteningSubscriber = new FlatteningSubscriber<>(mySubscriber);
Iterator<List<String>> pagesIterator = PAGES.iterator();
SubPub subPub = new SubPub(pagesIterator);
subPub.subscribe(flatteningSubscriber);
flatteningSubscriber.onSubscribe(subPub);
mySubscriber.doRequest(4L); //Initial request from a Flux flatMap
mySubscriber.doRequest(10L); //double tap from a Flux publishOn or flatMap etc
while (mySubscriber.getCollected().size() < 14L){ //wait for the demand to be met
Thread.sleep(100L);
}
//HERE we have 14 collected and 6 left on the currentBatch
System.out.println("Seen demand met");
Thread.sleep(500L); //wait before sending more demand (something writing results to S3)
System.out.println("Making massive demand to finish");
// BUT by now, another page has pushed those 6 remaining to the GC pile, a new page of 10 loaded
mySubscriber.doRequest(1000L); //just to say we want all that's left please
while (pagesIterator.hasNext()){
Thread.sleep(50L);
}
Thread.sleep(500L);
System.out.println("Got " + mySubscriber.getCollected().size() + " of " + PAGES.stream().mapToLong(List::size).sum());
System.out.println("Got " + mySubscriber.getCollected());
//We end up with 44 collected results, 6 are DISCARDED!
subPub.getExecutor().shutdown();
}
private static class MySubscriber implements Subscriber<String> {
public Subscription subscription;
List<String> collected = new ArrayList<>();
public List<String> getCollected() {
return collected;
}
@Override
public void onSubscribe(Subscription subscription) {
this.subscription = subscription;
}
public void doRequest(long amount) {
subscription.request(amount);
}
@Override
public void onNext(String string) {
collected.add(string);
}
@Override
public void onError(Throwable throwable) {
}
@Override
public void onComplete() {
}
}
private static class SubPub implements Subscription, Publisher<Iterable<String>> {
ExecutorService executor = Executors.newSingleThreadExecutor();
Subscriber<? super Iterable<String>> subscriber;
Iterator<List<String>> pagesIterator;
public ExecutorService getExecutor() {
return executor;
}
public SubPub(Iterator<List<String>> pagesIterator) {
this.pagesIterator = pagesIterator;
}
@Override
public void subscribe(Subscriber<? super Iterable<String>> subscriber) {
this.subscriber = subscriber;
}
@Override
public void request(long l) {
System.out.println("Page Request");
executor.execute(() -> {
try {
Thread.sleep(50L);
}
catch (InterruptedException e) {
e.printStackTrace();
}
if(pagesIterator.hasNext()){
System.out.println("Supplying Request");
subscriber.onNext(pagesIterator.next());
}
});
}
@Override
public void cancel() {
}
}
}
millems
github-actions[bot]
Version 2.17.28 When using the
flatMapIterablemethod onQueryPublisher(the return fromqueryPaginatormethod on the Async DDB client) there is a bug in the underlyingFlatteningSubscriberin AWS SDK.When we take the publisher produced from this, and feed it into reactor, we see, by how we happened to have set up our Flux, that because of some un-related flatMap / publish; reactor makes an initial
requestcall forXand then immediately anotherrequestcall forY.What this seems to do is, enter your
requestmethod on the in-line implementation ofSubscriberinFlatteningSubscribertwice, back to back. This means the first one makes a request "up" to the mapper, and the second one immediately then hit's theelseand goes tofulfillDemand(). This then sees that thecurrentBatchis empty, of course because we've not even had a page back yet in this double tap. So... it also does a request "up" for another page!now our demand is at
X+Yand we sit and wait, whenonNextis called by the mapper because a QueryResponse came back with a page of results, we are now destined to failure if the back pressure builds. The upstream demand is now 2 pages, which I think the code makes no account for because it wouldn't normally happen unless you hit therequestmethod multiple times before the first page, or if an empty page happened to end at the exact amount of demand being met.In my example X is 8 and Y is 256 making 264 demand. onNext is called for the first page, as I said, and also called again with the subsequent page, but is locked out. page 1 in my case had 79 results, and when the batch was empty and demand was high, it then requested another (3rd) page at the end of
fulfillDemandthen it unlocks and the queued uponNextfor page 2 comes in, and entersfulfillDemandin my case page 2 had 21 results. so at the end, 100 have been sent onward and the batch is empty so it, too, requests another page, the 4th now. it then exits and unlocks for the queued up 3rd page to come in toonNextwhere it enters againfulfillDemandthis page has 225 results. so the 164 remaining demand is satisfied, demand is left at 0 because no more requests have come in as the downstream is saturated with IO backpressure.so we have 61 items left on this 3rd page in the currentBatch list. as soon as demand hit 0, the
fulfillDemandcompleted and so did theonNextwhich called it for page 3 but a 4th page was requested by the 2nd page completing, and page 4 comes inonNextand .... destroys the currentBatch by replacing it with the 4th page, not checking if it was empty and adding to it because you presume that it would be empty to get a request, wrong in the case of more than 1requestcall happening before the 1st page of results...so, those 61 items? gone. repeatable, every time, but I can't give you a query that you can run on any arbritrary database to see this. But I just read the code and figured this would happen when I went looking for my missing 61 orders! My solution, was not to use the AWS SDK V2
flatMapIterableon theQueryPublisherand instead hand over that publisher to project reactor right away and hadFlux.flatMapIterabletake over the work, because it manages and not replaces it's own queue, it would seem, instead of a batch.I could try make some code but, if you read and understand this, this relies on
So; The cause is that a double (or more) tap to
requestbefore the first page hitsonNextsets you up for failure, by requesting more pages. And when a page comes in no check is made for emptycurrentBatchit is lost if not empty.Resolution: simplest would be to concatenate currentBatch with the incoming iter, if it happens not to be empty as it should be. Or, disallow more requests upstream if we didn't yet get a page, record that a request is made but not satisfied yet, and add the delta demand but don't push more demand up. That seems less easy but more logical, if a request has been made up for
1(page/iterable) then until one comes down don't request more because the code cannot handle more than one hittingonNextif the demand ever gets met.