Consumer buffer of flowable leaks memory about rxjava HOT 3 CLOSED

1. Is your publishSubjectController serialized? Concurrent onNext access can lead to hangs or leaks.
2. groupBy may hold open groups longer than anticipated, especially if the source has some form of rolling-grouping.
3. Calling subscribe in forEach is a code smell as it breaks the chain. That can also cause leaks as those are no longer part of the same sequence to be automatically cleaned up.

from rxjava.

@akarnokd Thank you for quick reply.

1. Its not serialized. Mainly because of high throughput requirement (and like I mentioned, this code worked well until very recently - maybe broken now due to increased load). I was able to simulate hang due to non-serialized publishsubject (under a much higher load and thread count), but it does not create consumer buffer bloating.

2. There is no rolling-grouping. The incoming data can belong to 1 of 4 groups/topics (with a fairly similar probability). But, even then, are grouped elements cached in consumer buffer ?

3. Point taken. I ll move the buffering+filtration to a flatmap.

from rxjava.

If you call onNext concurrently, you must serialize the subject.

toFlowable(BackpressureStrategy.BUFFER) has an unbounded buffer so if you can't drain it fast enough in downstream operators, it can lead to buffer bloat on its own. You'll have to rethink the dataflow, to use Flowables and some form of bounded data generation.

from rxjava.