컨슈머 그룹(consumer group) offset 관리 및 reset 방법

컨슈머 그룹(consumer group) offset 관리 및 reset 방법

1. Consumer offset - reset(리셋) 방법

• 운영중에 특정 offset으로 이동하여 다시 실행하고자 할경우에 사용할 수 있다.
• 혹은 테스트중에 컨슈머를 모두 리셋시키고 싶을경우에 사용한다.

1.1 컨슈머 삭제

컨슈머그룹을 바꿔주거나 컨슈머를 삭제고 다시 생성하는게 제이 빠른것 같다.

kafka-consumer-groups.sh --bootstrap-server <ip>:<port> --group <groupName> --topic <topic> --delete --execute

1.2 컨슈머 offset 변경방법 (리셋)

오프셋 변경방법에도 여러가지가 있다. 타임 단위로 이동할 수도 있고 , 초기상태로도 할 수있다.

즉 kafka 컨슈머 그룹의 offset을 리셋 하는방법은 크게 세가지로 분류된다.

1. Earlist
2. Latest
3. Specific

# 실행하기전 예상 결과값 확인
kafka-consumer-groups.sh --bootstrap-server <host:port> --group <group> --topic <topic> --reset-offsets --to-earliest --dry-run

# 실행
kafka-consumer-groups.sh --bootstrap-server <host:port> --group <group> --topic <topic> --reset-offsets --to-earliest --execute

• --dry-run 을 사용하여 offset을 실행했을때 예상되는 결과값을 확인하고
• —execute 옵션은 실제 offest을 진행함.

2. 특정 위치로 컨슈머그룹의 오프셋 이동하기

offset 의 위치를 특정위치로 되돌리는옵션종류

• --to-offset : 지정한 오프셋으로 이동 한다.
• --by-duration : 지정한 시간 간격만큼 오프셋 변경 PT 접두사와 시간간격을 표시하는 ISO 8061 기반 형식으로 지정됨, 예를들어 **‘PT1H’**는 한시간을 의미함
• --to-datetime : 지정한 날짜와 시간으로 오프셋이 이동된다. ‘yyyy-MM-dd’T’HH:mm:ss.SSS’ 형식으로 지정됨
• --to-latest: 가장 최신의 오프셋으로 설정됨
• --to-earliest: 가장 초기의 오프셋으로 설정됨

특정 컨슈머 그룹의 오프셋을 변경하기 위해서는 ‘kafka-consumer-groups.sh’ 을 사용한다.

2.1. 지정한 위치로 오프셋 이동하기 (절대값)

kafka-consumer-groups.sh --bootstrap-server <Kafka 서버 주소> --group <컨슈머 그룹 ID> --topic <토픽 이름> --reset-offsets --to-offset <새로운 오프셋> --execute

2.2 오프셋을 상대적인 위치로 이동

-shift-by <오프셋 증감 값> 을 사용하여 상대적으로 오프셋을 이동시킬수 있다.

kafka-consumer-groups.sh --bootstrap-server <Kafka 서버 주소> --group <컨슈머 그룹 ID> --topic <토픽 이름> --reset-offsets --shift-by <오프셋 증감 값> --execute

예를들어 100번의 offset을 150 번으로 보내고 싶을때 다음과 같이 작성하면, offset 100 → 150 으로 변경된다.

kafka-consumer-groups.sh --bootstrap-server localhost:9092 --group test-group-1 --topic test-topic --reset-offsets --shift-by 50 --execute

2.3. 컨슈머그룹에 속한 특정 파티션 offset 위치 변경

컨슈머그룹 전체에 대한 오프셋이동 뿐만 아니라 각 파티션에 대한 컨트롤도 가능하다.

그룹에서 사용했던 같은 shell 스크립트를 사용한다. kafka-consumer-groups.sh

그룹에서 사용하던 스크립트에 --partition 을 추가하고 변경하려는 파티션 번호를 입력한다.

kafka-consumer-groups.sh --bootstrap-server <Kafka 서버 주소> --group <컨슈머 그룹 ID> --topic <토픽 이름> --reset-offsets --to-offset <새로운 오프셋> --partition <파티션 번호> --execute

예를 들어 ‘my-consumer-group’ 이라는 컨슈머 그룹의 ‘my-topic’ 에서 2번 파이션의 offset을 1200 으로 변경하기 위해 다음 과같이 명령어를 사용하면 된다.

kafka-consumer-groups.sh --bootstrap-server localhost:9092 --group my-consumer-group --topic my-topic --reset-offsets --to-offset 1200 --partition 2 --execute

--partition 옵션을 사용하여 특정 파티션의 위치만 변경 가능 하다.

특정시간으로 옮기기 위해서는 --to-datetime 2023-05-05T13:58:23.223 옵션을 이용하여 특정 시간을 정의해주면된다