Sinks.many().replay().limit(Duration.ZERO) leaks memory

[progys]

Memory is leaked using replay sink with zero limit duration: Sinks.many().replay().limit(Duration.ZERO)

Expected Behavior

I have a use case of one publisher and many subscribers (want to honor downstream individually for each subscriber). Previously I used ReplayProcessor with 0 history size. However, this API was deprecated and also once I upgraded reactor from 3.3.x to 3.4.14 it started to behave differently. Thus, I moved to use Sinks.many().replay().limit(Duration.ZERO) as this was recommended in Java Doc as a replacement for ReplayProcessor with zero history.

In my understanding it should work like this:
a) Multicast
b) Without Subscriber: 0 elements pushed to this sink are remembered, even when there is no subscriber. Older elements are discarded
c) Backpressure : this sink honors downstream demand of individual subscribers.
d) Replaying: 0 elements pushed to this sink are replayed to new subscribers. Older elements are discarded.

Is there any other way to achieve this? For me the most important is b) and c). So that publishing does not fail without any subscribers and that each subscribers downstream demand is honored individually.

Expected, but not happening:
Older items (history) should be discarded so that GC could free up memory.

Actual Behavior

Older items (history) are discarded (not published to subscribers), but memory retains nodes with older items. This prevents memory being reclaimed by garbage collector. See the heap dump analysis of the live system. It contains 2 GB of history items from this sink (there are more than 300K items there, did not expand it for the sake of example):

Steps to Reproduce

Run this test and connect with any memory profiler to see how memory keeps growing and instances are not getting released.

@Test
void reproCase() {
  Sinks.Many<Integer> sink = Sinks.many().replay().limit(Duration.ZERO);

  for (int i=0; i<1000_000_000; i++){
    sink.tryEmitNext(i);
  }
}

Possible Solution

Run this test and connect with any memory profiler to see how memory is reclaimed properly. Also, it is much faster than the one above.

@Test
void solutionCase() {
  Sinks.Many<Integer> sink = Sinks.many().replay().limit(1, Duration.ZERO); //this eliminates the leak, memory is reclaimed

  for (int i=0; i<1000_000_000; i++){
    sink.tryEmitNext(i);
  }
}

Environment

• Reactor version(s) used: 3.4.14
• JVM version (java -version):
  java version "1.8.0_291"
  Java(TM) SE Runtime Environment (build 1.8.0_291-b26)
  Java HotSpot(TM) 64-Bit Server VM (build 25.291-b26, mixed mode)
• OS and version (eg uname -a): Windows 10 Enterprise

[progys]

thanks for picking it up so quickly! 👍