Kafka-реактор для повторного чтения того же сообщения Kafka, если обработка сообщения не удалась ⇐ JAVA

[Anonymous]

При чтении из Kafka мне нужна гарантия того, что мое сообщение Kafka будет успешно сохранено в базе данных, прежде чем я обработаю следующее сообщение из Kafka. Я использую java-spring kafka-reactor. Ниже приведен простой пример.

Код: Выделить всё

kafkaReceiver
.receive()
.delayUntil(record -> saveInDb(record.value())
.doOnNext(v -> record.receiverOffset().commit()
.subscribeOn(Schedulers.boundedElastic())
.subscribe()))
.retry()
.subscribe();

private Mono saveInDb(Dto message) {
// save in database here
}

Если операция saveInDb завершается неудачей (например, из-за оптимистических блокировок и т. д.), то kafkaReceiver закрывает KafkaConsumer. Но .retry() создаст совершенно новый KafkaConsumer. Новый KafkaConsumer снова прочитает то же сообщение (поскольку оно не было зафиксировано) и попытается сохранить его в базе данных.
Меня беспокоит то, что мы каждый раз закрываем и заново создаем KafkaConsumer. time saveInDb не удается. Для меня это выглядит как накладные расходы на операцию закрытия-восстановления. Скажем, если база данных будет недоступна в течение некоторого времени, мы закроем и воссоздадим KafkaConsumer много раз. Согласится ли брокер Kafka на это и не заблокирует меня (моего потребителя) за такие частые операции?
Можно ли как-нибудь просто повторить попытку только операции saveInDb, не закрывая KafkaConsumer?

Подробнее здесь: https://stackoverflow.com/questions/783 ... ing-failed