Как повторно использовать сообщения с самого начала?

Question

Я использую auto.offset.reset=earliest в своем коде и использовал коммит смещения в kafka с помощью приведенного ниже кода.

val offsetRanges=rdd.asInstanceOf[HasOffsetRanges].offsetRanges
inputStream.asInstanceOf[CanCommitOffsets].commitAsync(offsetRanges)

Теперь, когда я запускаю свою программу, она не получает новые сообщения, так как все сообщенияCommit.

Я тестирую этот код в QA, поэтому хочу сбросить смещение на начало, но похоже, что ранние не работают, не читают новые сообщения и в теме нет новых сообщений.Я хочу прочитать сообщения с начала для целей тестирования.

Может ли кто-нибудь помочь, если ранние не получат сообщения с начала, если они зафиксированы?

Answer 1

Свойство auto.offset.reset используется только в том случае, если для раздела нет зафиксированного смещения.Вы можете сбросить смещение для всей группы, используя kafka-consumer-groups (входит в состав Kafka):

kafka-consumer-groups --bootstrap-server <kafkahost:port> --group <group_id> --topic <topic_name> --reset-offsets --to-earliest --execute