Reactor-core: Scaling problem with collectList

[msoub]

Hello,

When using Mono.collectList, I encounter a scaling problem with a certain amount of elements, whereas everything works fine with Flux.collectMultimap.
The code below reproduces the mentioned cases:

public class ReactiveTest {

    private final Flux<User> flux = Flux.merge(
        buildFlux("FR"),
        buildFlux("ES"),
        buildFlux("EN")
    );
    private final int count = 1000;

    @Test
    public void testCompute1() {
        StepVerifier.create(compute1(flux))
            .expectNextCount(count)
            .verifyComplete();
    }

    @Test
    public void testCompute2() {
        StepVerifier.create(compute2(flux))
            .expectNextCount(count)
            .verifyComplete();
    }

    private Flux<User> buildFlux(String language) {
        return buildFlux(new Region(language));
    }

    private Flux<User> buildFlux(Region region) {
        return Flux.fromStream(IntStream.rangeClosed(1, count).boxed())
            .map(id -> {
                Country country = new Country(id);
                return new User(country, region);
            });
    }

    private Flux<Group> compute1(Flux<User> userFlux) {
        return userFlux
            .groupBy(User::getCountry)
            .flatMap(f -> f
                .map(User::getRegion)
                .collectList()
                .map(regions -> new Group(f.key(), regions)));
    }

    private Flux<Group> compute2(Flux<User> userFlux) {
        return userFlux
            .collectMultimap(User::getCountry, User::getRegion)
            .flatMapIterable(Map::entrySet)
            .map(e -> new Group(e.getKey(), e.getValue()));
    }

    @Data
    private static class User {
        private final Country country;
        private final Region region;
    }

    @Data
    private static class Country {
        private final Integer id;
    }

    @Data
    private static class Region {
        private final String name;
    }

    @Data
    private static class Group {
        private final Country country;
        private final Collection<Region> regions;
    }

}

[simonbasle]

The problem is not with collectList but with your use of groupBy. You are essentially attempting to create 3000 groups (as you generate count countries for each region. Grouping by region would have been fine but this inverted use case sounds a bit strange.

Anyway, groupBy is not really made to split into a large amount of low-cardinality groups like that. This is amplified by the limited concurrency and prefetch of flatMap and the fact that collectList needs a onComplete signal to emit.

[msoub]

Thank you for the explanation.

To be honest this use case is not exactly the one I have in my production code but it reproduces the same problem.
Perhaps you should make it clear in the documentation regarding the usage of groupBy?

[smaldini]

I changed that for doc label to be documented in RC1 guide/javadoc @simonbasle

[simonbasle]

@msoub there was already a note with limitations on groupBy javadoc, as a last paragraph. I put a second warning towards the beginning about large amount of groups, plus a whole new chapter in the reference guide that mentions these limitations (see here)

don't hesitate to reopen if you feel this can be further documented (or open a PR if you have suggestions) 👍