How to fix "MissingBackpressureException"

Question

I am using RxJava2 Flowables by subscribing to a stream of events from a PublishSubject.It's being used in enterprise level application and we don't have the choice of dropping any events. I am using version RxJava 2.2.8

I am using BackpressureStrategy.BUFFER as I don't want to lose any of my events.

Also, I buffer again for 50000 or 3 minutes whichever is earlier. This I do as I want to consolidate events and then process them.

But I get the following errors in a few minutes of my run

io.reactivex.exceptions.MissingBackpressureException: Could not emit buffer due to lack of requests
at io.reactivex.internal.subscribers.QueueDrainSubscriber.fastPathOrderedEmitMax(QueueDrainSubscriber.java:121)
at io.reactivex.internal.operators.flowable.FlowableBufferTimed$BufferExactBoundedSubscriber.run(FlowableBufferTimed.java:569)
at io.reactivex.Scheduler$Worker$PeriodicTask.run(Scheduler.java:479)
at io.reactivex.internal.schedulers.ScheduledRunnable.run(ScheduledRunnable.java:66)

I tried increasing the buffer size by setting up, but there is no change in the behavior.

System.setProperty("rx2.buffer-size", "524288");

Also If I buffer for a longer time instead of 3 minutes, I get the exception after much longer time probably because my downstream performs better when the events are consolidated more. However, I don't have that choice because these are live events and needs processing immediately(in 3-5 minutes).

I also tried thread.sleep() before invoking the "subscription.next" in case of error but still getting the same results.

keySubject.hide()
.toFlowable(BackpressureStrategy.BUFFER)
.parallel()
.runOn(Schedulers.computation())
.map(e -> e.getContents())
.flatMap(s -> Flowable.fromIterable(s))
.sequential()
.buffer(3,TimeUnit.MINUTES,50000)
.subscribe(new Subscriber<List<String>>() {

@Override
  public void onSubscribe(Subscription var1) {
   innerSubscription = var1;
innerSubscription.request(1L);
 }

@Override
public void onNext(List<String> logs) {
    Subscription.request(1L);

///   Do some logic here

}

I want to know How do I handle the backpressure to avoid this exception? Is this exception because of ".buffer" method Is there a way for me to check the status of these buffers. Also why even if I increase the rx2.buffer-size, I still get the exception in the same amount of time. Ideally, the system should run longer with a higher buffer size if the exception is because if buffer getting full.

Any help on the reason for this message "Could not emit buffer due to lack of requests at " will be great.

Answer 1

The thing is, why do you use a subject that isn't backpressure-aware? Are you using that as a poor man's event bus? Also, assuming e.getContents() is a simple getter I believe you can replace this whole block

.toFlowable(BackpressureStrategy.BUFFER)
.parallel()
.runOn(Schedulers.computation())
.map(e -> e.getContents())
.flatMap(s -> Flowable.fromIterable(s))
.sequential()
.buffer(3,TimeUnit.MINUTES,50000)
.subscribe(new Subscriber<List<String>>() { ... });

with

.flatMapIterable(e -> e.getContents())
.buffer(3,TimeUnit.MINUTES,50000)
.rebatchRequests(1)
.observeOn(Schedulers.computation())
.doOnNext(s -> /* Do some logic here */)
.subscribe();