反应式 Kafka 接收器轮询新记录，即使旧记录也未确认

如何解决反应式 Kafka 接收器轮询新记录，即使旧记录也未确认

我正在尝试使用 Reactive Kafka Receiver 来处理 Kafka 消息。我希望消费者只有在我调查的最后一个被确认的情况下才会给我新消息。但是，即使我不运行 ack 我也会得到新的记录。这是一种理想的行为还是我缺少一些配置？

void consumeMessages() {
    kafkaConsumer(() -> true).subscribe();
}

Flux<Void> kafkaConsumer(Booleansupplier repeatCondition) {
    return reactiveConsumer
        .receive()
        .doOnError(error -> log.warn("Received error when get kafka message",error))
        .doOnNext(receivedRecord -> log.debug("Received event {}",receivedRecord.value()))
        .flatMap(this::handleReceivedRecord)
        .onErrorResume(error -> Flux.empty())
        .repeat(repeatCondition);
}


Flux<Void> handleReceivedRecord(ReceiverRecord<String,CustomEvent> receivedRecord) {
    return Mono.just(receivedRecord)
        .flatMapMany(
            record ->
                handleCustomEvent(record.value())
                    .doOnComplete(() -> {
                        log.debug("Committing offset for record: {}",record);
                        //record.receiverOffset().ackNowledge();
                    })
                    .onErrorMap(error -> new ReceiverRecordException(record,error)))
        .doOnError(error -> log.error("Error when processing CustomEvent",error))
        .retrywhen(
            Retry.backoff(consumerProperties.retries(),consumerProperties.minBackOff())
                .transientErrors(true))
        .onErrorResume(
            error -> {
                ReceiverRecordException ex = (ReceiverRecordException) error.getCause();
                log.warn("Retries exhausted for " + ex.getRecord().value());
                ex.getRecord().receiverOffset().ackNowledge();
                return Mono.empty();
            });
}