FlowableOnBackpressureLatest can signal backpressure issue with concurrent Subscription.request and Subscriber.onNext

[wburns]

I have been having very rare issues where an entry is signaled as overflowing back pressure, but when I checked all of my downstream Publishers they all have proper back pressure requests.

It seems that if you have a Subscription.request call inside BackpressureLatestSubscriber it will "obtain" the drain loop exclusive AtomicInteger. Then if a concurrent Subscriber.onNext call is done in BackpressureLatestSubscriberit will set the current AtomicReference and then try to enter the drain loop but immediately exits and then when it comes a second time it will see current variable is set and say it is a back pressure issue. The BackpressureLatestSubscriber should verify the requested count to determine if there is a back pressure issue instead of just the single current variable being set or not.

Note this requires a downstream processor to complete "async" so that the subscribe is called on the thread that completed the publisher while another thread is adding onNext call.

processorDisposer = requestProcessor.onBackpressureLatest(supplier -> {
                // Sometimes this is invoked despite rebatchRequests having requested more values
            })
            .rebatchRequests(indexerConfig.rebatchRequestsSize())
            // This is completed asynchronously from the invoking thread
            .flatMap(Supplier::get, indexerConfig.maxConcurrency())
            .doFinally(submittedTasks::clear)
            .subscribe(Functions.emptyConsumer(),
                  t -> log.fatal("Indexing encountered a non recoverable error", t));

[wburns]

To show provided links it seems that if you have a request on thread 1 come into https://github.com/ReactiveX/RxJava/blob/3.x/src/main/java/io/reactivex/rxjava3/internal/operators/flowable/AbstractBackpressureThrottlingSubscriber.java#L83 and then if on thread 2 comes in at https://github.com/ReactiveX/RxJava/blob/3.x/src/main/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableOnBackpressureLatest.java#L48 comes in it is possible the current variable is set by 2 but then it can't invoke the drain loop. Then if thread 2 does another onNext call the current variable may be set (depends if thread 1 sees it in the loop) and signals that there is missing back pressure.

[wburns]

This also affects onBackpressureReduce as well. onBackpressureDrop checks the request properly and is unaffected. onBackpressureBuffer I can't tell unfortunately, but might be fine.

[akarnokd]

This is an expected behavior:

Drops all but the latest item emitted by the current Flowable if the downstream is not ready to receive new items (indicated by a lack of Subscription.request(long) calls from it) and emits this latest item when the downstream becomes ready.

Note that due to the nature of how backpressure requests are propagated through subscribeOn/observeOn, requesting more than 1 from downstream doesn't guarantee a continuous delivery of onNext events.

Why are you using onBackpressureLatest?

[wburns]

This is an expected behavior:

I would say it depends what downstream becomes ready actually means. Downstream in my case had 250 request outstanding and because it requested another 750 it got in this state, which seems odd as it had already processed 750 entries.

For example on onBackpressureDrop it states

Drops items from the current Flowable if the downstream is not ready to receive new items (indicated by a lack of Subscription.request(long) calls from it)

But onBackpressureLatest and onBackpressureDrop don't behave the same here as onBackpressureLatest has to be "ready" and not concurrently handling a request.

Why are you using onBackpressureLatest?

Good question and while investigating this issue I realized onBackpressureDrop does what I need and doesn't have the issue here. Thus I changed my code and I haven't seen the error in the past few hours :)