KafkaFluxTest.java

{
        scheduler = Schedulers.newElastic("FLUX_DEFER", 10, true);

        ReceiverOptions<Integer, String> options = receiverOptions.subscription(Collections.singleton(topic))
                .addAssignListener(partitions -> log.info("Partitions Assigned {}", printPartitions(partitions)))
                .addRevokeListener(partitions -> log.info("Partitions Revoked {}", printPartitions(partitions)))
                //.assignment(Collections.singleton(new TopicPartition(topic, 9)))  // <-- ** EASY TEST ONLY **
                .commitInterval(Duration.ZERO)
                .commitBatchSize(0);

        KafkaReceiver<Integer, String> receiver = KafkaReceiver.create(options);

        disposable = Flux.defer(receiver::receive)
                .groupBy(m -> m.receiverOffset().topicPartition())
                .flatMap(partitionFlux ->
                        partitionFlux.publishOn(scheduler)
                                .concatMap(m -> consumeMessage(m)
                                        .thenEmpty(m.receiverOffset().commit())))
                .retryBackoff(numRetries, Duration.ofSeconds(2), Duration.ofHours(2))
                .doOnError(err -> {
                    handleError(err);
                }).retry()
                .doOnCancel(() -> close()).subscribe();
    }