У меня есть функция @kafkalistener, которая потребляет сообщения. Слушатель обладает свойствами, установленными в Application.yml: < /p>
Одна тема < /li>
ack-mode: manual_immediate < /li>
enable-auto-commit: false < /li>
max-poll-rcords: 1 < /li>
< /br /> max-poll-records: 1 < /li> < /br /> max-poll-records: Самый ранний < /li>
< /ul>
Поэтому я хочу обработать одно сообщение за раз, признавая его, если все работает нормально.
Если я теряю соединение DB, то я должен прекратить обработку сообщений, и последнее, что можно увеличить смещение. {testkafka-0 = offsetAndmetAdata {offset = 10, inderepoch = null, metadata = '' '}} < /li>
< /ul>
После подключения DB, и я создаю новое сообщение, последнее составное снятие внезапно прыгает до 17, но функция слушателя не вызывается для обработки предыдущего 10-16 сообщений. /> Текущий смещение: 17, последнее смещение: {testkafka-0 = offsetAndmetadata {offset = 17, inderepoch = null, metadata = ''}} < /li>
< /ul>
Можете ли вы мне помочь, почему это может произойти? Я думал, что он автоматически запускает функцию n-times. < /P>
@KafkaListener(topics = {TOPIC}, groupId = "mxsmart-center")
public void handle(ConsumerRecord record, Consumer consumer, Acknowledgment ack) {
log.info("listener started" + record.value().getCreatedAt());
TopicPartition partition = new TopicPartition(record.topic(), record.partition());
// Fetch end offset (the next offset to be written)
Map endOffsets = consumer.committed(Collections.singleton(partition));
log.info("Current offset: {}, Latest offset: {}", record.offset(), endOffsets);
if (isDBUp() && isMessageValid(record.value())) {
try {
ack.acknowledge();
log.info("Acknowledged");
} catch (Exception e) {
lastKnownState = ConnectionState.DOWN;
log.error("Error while processing message: ", e);
throw e;
}
} else {
throw new RuntimeException("DB is not in UP state, skipping message");
}
}
Подробнее здесь: https://stackoverflow.com/questions/796 ... ot-process
