New receiver source: Flux<Flux<ReceiverRecord<K, V>>>

[justinplus]

It would be nice if reactor-kafka receiver adds a new API:

Flux<Flux<ReceiverRecord<K, V>>> receiveMannualAck() 

Normally, to achieve high throughput, there would be multiple parallel processors in a Kafka consumer.

In this case, existing Flux<ReceiverRecord<K, V>> receive() is hard for multiple parallel processors to manage the offsets. And another API, Flux<Flux<ConsumerRecord<K, V>>> receiveAutoAck() which auto ack when the inner Flux is terminated, is hard to used to design and implement some smart commit strategies especially in multiple topic-partition subscription scenario.

[turbospaces]

I agree with @justinplus and also looking for something like this:

@Override
    public Flux<Flux<ReceiverRecord<K, V>>> receiveManualAck() {
        this.ackMode = AckMode.MANUAL_ACK;
        Flux<ConsumerRecords<K, V>> flux = withDoOnRequest(createConsumerFlux());
        return flux.map(consumerRecords -> Flux.fromIterable(consumerRecords).map( r -> {
            TopicPartition topicPartition = new TopicPartition(r.topic(), r.partition());
            CommittableOffset committableOffset = new CommittableOffset(topicPartition, r.offset());
            return new ReceiverRecord<>(r, committableOffset);
        } ));
    }