Async commit is not working

[architmmmec]

   /**
     * Consumes data from Kafka and returns a flux of receiver record
     * @return receiver record
     */
    public Flux<ReceiverRecord<String, byte[]>> consumeFromSource() {
        ReceiverOptions<String, byte[]> receiverOptions = getReceiverOptions();
        return KafkaReceiver.create(receiverOptions).receive(Integer.parseInt(prefetch))
                .map(receiverRecord -> {
                    try {
                        if (receiverRecord.receiverOffset().offset() % commitBatchSize == 0) {
                            receiverRecord.receiverOffset().commit();
                            log.debug("Committed offset {},{}", receiverRecord.receiverOffset().topicPartition(),
                                    receiverRecord.receiverOffset().offset());
                        }
                    } catch (Exception exception) {
                        log.error("Exception while committing", exception);
                    }
                  
                    return receiverRecord;
                });
    }

This code is not working, can some please this

[garyrussell]

"not working" is not enough information. Please elaborate. What behavior are you seeing?

[architmmmec]

Offsets are not getting committed @garyrussell

[garyrussell]

You need to subscribe (e.g. .block()) on the Mono returned by commit - see its javadoc.

[aridechettali]

But block() is when you wanted a sync commit, As per the doc we can have async commit by just calling receiverOffset().commit();
https://projectreactor.io/docs/kafka/release/reference/#concurrent-ordered

[garyrussell]

That's a bug in the documentation; you always have to subscribe to a Mono to make it run.

You can run the commit on a different thread; add .publishOn(...).

See https://projectreactor.io/docs/kafka/snapshot/reference/#kafka-source

[aridechettali]

Got it thanks

[garyrussell]

So here's the final answer - you need block() if you set the commit interval to zero - because then, the commit event is never scheduled, subscribing to the Mono is required to schedule the commit task to run.

When there is a commit interval, the commit task is scheduled periodically. Calling commit() simply updates the offsets to be committed and they will be picked up by the scheduled task; there is no need to subscribe to the Mono.

[koldat]

You can use Mono as it was designed (receiverRecord.receiverOffset().commit().then(Mono.just(receiverRecord))):

        return KafkaReceiver.create(receiverOptions).receive()
                .flatMap(receiverRecord -> {
                    try {
                        if (receiverRecord.receiverOffset().offset() % commitBatchSize == 0) {
                            log.debug("Committed offset {},{}", receiverRecord.receiverOffset().topicPartition(),
                                    receiverRecord.receiverOffset().offset());
                            return receiverRecord.receiverOffset().commit()
                                .then(Mono.just(receiverRecord));
                        }
                    } catch (Exception exception) {
                        log.error("Exception while committing", exception);
                    }
                  
                    return Mono.just(receiverRecord);
                });