[Kafka] reset offset

reset offset 

테스트를 하면서 consumer group의 offset을 reset 해야 하는 경우도 있고, 데이터 마이그레이션으로 데이터 누락 가능성을 방지하기 위해 offset을 reset 해야 하는 경우 등이 있다.

 

consumer group의 offset을 reset 하기 위해서는 해당 consumer group이 inactive 상태여야 가능하다. 즉, consumer group의 모든 consumer application을 종료해야 한다.

(신규로 생성될 consumer group인 경우는 상관 없다.)

 

command

kafka-consumer-groups.sh --bootstrap-server \\
{kafka broker endpoint} --group {consumer group} \\
--reset-offsets {offset option} --topic {topic} -execute

• -topic 대신 --all-topics를 지정하면 모든 토픽에 대해서 실행한다.
• -execute 옵션을 제거하고 실행하면 실제 반영되지 않고 어떻게 변할지 결과만 출력할 수 있다. ← dry run

 

오프셋 위치 설정 option

option	description
--to-offset <Long: offset>	지정한 offset 위치로 변경한다.
--to-earliest	처리할 수 있는 가장 처음 시간으로 되돌린다.
--to-latest	해당 topic의 가장 최근 offset으로 변경한다.
--to-datetime <String: datetime>	특정 시간대의 offset으로 되돌린다. Format YYYY-MM-DDTHH:mm:SS.sss
--shift-by <Long: number-of offsets>	현재 위치부터 설정한 n값에 따라 offset 위치를 변경한다. n: +, - 모두 가능
--to-current	컨슈머 그룹이 마지막으로 읽은 offset 위치로 변경한다.
--by-duration	현재 시간 기준으로 옵션에 작성한 시간 기준별로 offset을 변경한다. Format PnDTnHnMnS

 

 

🔗 참고 링크

Kadeck Blog | How to reset consumer offsets in Apache Kafka

Kafka Streams Application Reset Tool | Confluent Documentation