FluxGroupBy silently drops onNext signals #2352

I have a use case where I'm using Flux::groupBy with a relatively high cardinality of grouping (1024-4096 concurrent groups at any given time). FWIW, the use case itself is similar to a sort of deduplication functionality on an infinite stream.

Each "group" is limited in time (.take(Duration)) and size (.take(Long)) and collected as a List. Under high load, it appears some items are being dropped.

Thus far, I think I can point to where the relevant pieces of code are:

• FluxGroupBy::onNext may be retrieving a UnicastGroupedFlux racing to termination:
  https://github.com/reactor/reactor-core/blob/master/reactor-core/src/main/java/reactor/core/publisher/FluxGroupBy.java#L194-L212
• UnicastGroupedFlux::drainRegular may be short circuit returning after emitting non-zero number of elements
  https://github.com/reactor/reactor-core/blob/master/reactor-core/src/main/java/reactor/core/publisher/FluxGroupBy.java#L545-L554
• UnicastGroupedFlux::checkTerminated silently clears its queue when checking termination:
  https://github.com/reactor/reactor-core/blob/master/reactor-core/src/main/java/reactor/core/publisher/FluxGroupBy.java#L642-L647

Expected Behavior

Items are not dropped by groupBy operator

Actual Behavior

Items emitted from upstream of groupBy operator are not emitted downstream, and are silently being dropped.

Steps to Reproduce

I wrote the following test to show a simplified version of what's going on with my use case:

    @Test
    public void test() throws Exception {
        // Hooks.onNextDroppedFail(); // This actually has no effect; Was thinking it would cause failure with CancelException

        AtomicLong upstream = new AtomicLong(0L);
        AtomicLong downstream = new AtomicLong(0L);
        CountDownLatch latch = new CountDownLatch(1);

        Flux.fromStream(Stream.iterate(0L, (last) -> (long) (Math.random() * 4096)))
            .flatMap(number -> Flux.concat(
                Mono.just(number),
                Mono.just(number).delayElement(Duration.ofMillis((int) (Math.random() * 2000)))),
                4096)
            .take(Duration.ofSeconds(30L))
            .doOnNext(next -> upstream.incrementAndGet())
            .publishOn(Schedulers.elastic())
            .groupBy(Function.identity())
            .flatMap(groupFlux -> groupFlux.take(Duration.ofSeconds(1L))
                .take(2)
                .collectList(), 16384)
            .doOnNext(batch -> {
                try {
                    Thread.sleep(1L);
                } catch (Exception e) {

                }
            })
            .map(Collection::size)
            .subscribe(downstream::addAndGet, System.err::println, latch::countDown);

        latch.await();
        assertEquals(upstream.get(), downstream.get());
    }

Interestingly, the test sometimes passes without the doOnNext with Thread.sleep(1L). I had to add that to consistently get it to fail (and mimic small real-world processing overhead of each downstream item)

Possible Solution

I'm not sure. Still trying to figure out exactly what the problem is in groupBy

Your Environment

• Reactor version(s) used: 3.3.9.RELEASE
• JVM version: 1.8.0_172
• OS and version (eg uname -a): Mac OS Catalina 10.15.6 (Darwin Kernel Version 19.6.0 x86_64)