import reactor.core.publisher.Flux;
import reactor.core.scheduler.Scheduler;
import reactor.core.scheduler.Schedulers;

public class GroupByIssue {

    public static void main(String[] args) throws InterruptedException {
        int parallelism = 2; // issue can be reproduced with any parallelism

        Scheduler process = Schedulers.newParallel("process", parallelism, true);

        final long start = System.nanoTime();

        Flux.range(0, 4_000_000)
            .subscribeOn(Schedulers.newSingle("range", true))
            .groupBy(i -> i % 2)
            .flatMap(g ->
                g.key() == 0
                    ? g //.hide()  /* adding hide here fixes the hang issue */
                        .parallel(parallelism)
                        .runOn(process)
                        .map(i -> i)
                        .sequential()
                    : g.map(i -> i) // no need to use hide
            )
            .doOnNext(i -> print(i))
            .then()
            .block();

        System.out.printf("elapsed: %d\n", (System.nanoTime() - start) / 1_000_000);
    }

    private static void print(int current) {
        if (current % 100000 == 0) {
            System.out.println("processed: " + current);
        }
    }
}