Flux#bufferWhen memory leak suspicion using Flux#buffer(timespan, timeshift) when timespan != timeshift

[rreynaud]

Expected behavior

expect Flux.buffer(Duration.ofMillis(3000), Duration.ofMillis(2000)) behaves as Flux.buffer(Duration.ofMillis(3000), Duration.ofMillis(3000))

Actual behavior

When timespan is different than timeshift, heap dumps assert that buffered elements are never garbagged before flux completion, while when timespan equals timeshift, they are...

Instances are retained by reactor.core.publisher.FluxBufferWhen$BufferStartEndMainSubscriber

Steps to reproduce

static class Wrapper {
    final int i;

    Wrapper(int i) {this.i = i;}

    @Override
    public String toString() {
        return "Wrapper{" +
                "i=" + i +
                '}';
    }
}

public void test() throws InterruptedException {
    final CountDownLatch latch = new CountDownLatch(1);
    final UnicastProcessor<Wrapper> processor = UnicastProcessor.create();
    processor.buffer(Duration.ofMillis(3000), Duration.ofMillis(2000))
             .doOnNext(t -> System.out.println(String.format("tuple %s", t)))
             .subscribe();

    Flux.range(1, Integer.MAX_VALUE)
        .delayElements(Duration.ofMillis(10))
        .doOnNext(i -> processor.onNext(new Wrapper(i)))
        .doOnComplete(processor::onComplete)
        .subscribe();

    latch.await(10, TimeUnit.MINUTES);
}

Reactor Core version

3.1.2.RELEASE

JVM version (e.g. java -version)

1.8.151

OS version (e.g. uname -a)