Q:

How to fix "MissingBackpressureException"

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.

A:

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();

