[Kafka] consumer-groups

컨슈머 그룹에 대한 걸 알아보자.

컨슈머 그룹으로 생성된 컨슈머로 해당 토픽에 대한 데이터를 가져갔다.

컨슈머 그룹은 따로 생성하는 명령을 날리지 않고 컨슈머를 동작할 때 컨슈머 그룹이름을 지정하면 새로 생성된다.

또한 어떤 컨슈머 그룹이 어떤 토픽을 대상으로 레코드를 가져갔는지

그리고 상태와 오프셋 컨슈머 렉 컨슈머 ID,호스트 등을 알 수 있다.

(컨슈머 렉 - 현재 레코드의 오프셋과 마지막 레코드의 차이이다 ,즉 컨슈머가 읽지않은 데이터라고 이해하면 된다.)

 

컨슈머의 상태를 조회할때도 유용하다. 또한 컨슈머그룹의 오프셋을 리셋할 수 있다.

오프셋의 리셋은 처음부터 시작되는 것만은 아니다. 특정 오프셋으로 이동할 수 있는 것이다.

순차처리가 필요없는 경우, 최신의 데이터로 오프셋을 이동할 수 있다는 것이다.

 

 

1) 컨슈머 그룹 조회

bin/kafka-consumer-groups.sh --bootstrap-server my-kafka:9092 --list

 

2) 컨슈머 그룹 상세 내용 조회

bin/kafka-consumer-groups.sh --bootstrap-server my-kafka:9092 --group console-consumer-47788 --describe

오프셋과 파티션의 정보가 노출이되고, 해당 토픽을 볼 수 있다.

 

특정 토픽을 특정 컨슈머 그룹이 리슨할 수 있다.

bin/kafka-console-consumer.sh --bootstrap-server my-kafka:9092 --topic smitfka2 --group hello-group

bin/kafka-consumer-groups.sh --bootstrap-server my-kafka:9092 --group hello-group --describe

 

신규로 듣는 그룹이 파티션에 따라 다르고 각각의 오프셋과 log-end offset이 다르다.

프로듀서에서 해당 메시지를 보냈지만, 컨슘을 하지 않는 경우에는 아래와 같이 렉이 발생한다.

 

그 다음은 오프셋 리셋을 해보자

카프카 데이터는 retention 기간이 지나고 삭제가 된다.

bin/kafka-consumer-groups.sh --bootstrap-server my-kafka:9092 --group hello-group --topic smith.kafka2 --reset-offsets --to-earliest --execute

이런식으로 가장 일찍의 데이터로 리셋을 했다고 해보자

 

현재 오프셋이 0으로 변경되었고, 렉이 그만 큼 증가되었다.

이러면 이제, 해당 컨슈머는 다시 전체의 데이터를 렉이 사라질때까지 컨슘할 수 있게 됬다.