그룹

컨슈머에서 그룹을 지정하여 메시지를 polling 할 경우 다음과 같이 동작한다.

• 지정한 그룹을 추가한다.
• 메시지를 읽은 offset을 기억하여 offset 부터 읽기 시작한다.
• N개의 파티션에 기록된 메시지를 병렬적으로 가져온다(입력된 메시지 순서가 보장되지 않을 수 있다)

$ kafka-topics.bat --create --bootstrap-server localhost:9092 --partitions 3 --topic hello-kafka
$ kafka-console-producer.bat --bootstrap-server localhost:9092 --topic hello-kafka
>1
>2
>3
>4
>5
>6
>7
>8
>9
>10

$ kafka-console-consumer.bat --bootstrap-server localhost:9092 --topic hello-kafka -group testgroup --from-beginning
1
2
3
4
5
6
7
8
9
10

$ kafka-console-producer.bat --bootstrap-server localhost:9092 --topic hello-kafka
>QWEQWEQWEQWE
>!@#!@#!@#!@#
>ZXCASDQWEZXCASDQWE
>ZXCASDQWEASDZXCZXCASDQWE

$ kafka-console-consumer.bat --bootstrap-server localhost:9092 --topic hello-kafka -group testgroup --from-beginning
QWEQWEQWEQWE
!@#!@#!@#!@#
ZXCASDQWEZXCASDQWE
ZXCASDQWEASDZXCZXCASDQWE

생성된 그룹 확인

$ kafka-consumer-groups.bat --bootstrap-server localhost:9092 --list
console-consumer-58133    <-- 그룹 지정 없이 컨슈머를 사용할 경우 임시 그룹이 자동으로 생성됨
console-consumer-80593
console-consumer-23689
testgroup
...

그룹 상태 확인

> kafka-consumer-groups.bat --bootstrap-server localhost:9092 --group testgroup --describe

Consumer group 'testgroup' has no active members.

GROUP           TOPIC           PARTITION  CURRENT-OFFSET  LOG-END-OFFSET  LAG             CONSUMER-ID     HOST            CLIENT-ID
testgroup       hello-kafka     0          44              44              0               -               -               -
testgroup       hello-kafka     1          0               0               0               -               -               -
testgroup       hello-kafka     2          0               0               0               -               -               -

컨슈머 오프셋 : 컨슈머가 현재까지 읽은 오프셋
로그 엔드 오프셋 : 토픽의 마지막 오프셋
컨슈머 랙 : 컨슈머 오프셋과 로그 엔드 오프셋의 차이. 이 차이가 크다는 뜻은 컨슈머가 느리게 처리하고 있음을 의미.

컨슈머 오프셋을 리셋하여 컨슈머가 처음부터 메시지를 읽도록 처리 하고 싶을 수 도 있다.

$ kafka-consumer-groups.bat --bootstrap-server localhost:9092 --group testgroup --topic hello-kafka --reset-offsets --to-earliest --execute

GROUP                          TOPIC                          PARTITION  NEW-OFFSET
testgroup                      hello-kafka                    0          0
testgroup                      hello-kafka                    1          0
testgroup                      hello-kafka                    2          0

$ kafka-consumer-groups.bat --bootstrap-server localhost:9092 --group testgroup --describe

Consumer group 'testgroup' has no active members.

GROUP           TOPIC           PARTITION  CURRENT-OFFSET  LOG-END-OFFSET  LAG             CONSUMER-ID     HOST            CLIENT-ID
testgroup       hello-kafka     0          0               44              44              -               -               -
testgroup       hello-kafka     1          0               0               0               -               -               -
testgroup       hello-kafka     2          0               0               0               -               -               -

특정 파티션의 offset을 특정 값으로 초기화 하는 것도 가능하다.

1번 파티션의 오프셋을 10으로 초기화

$ kafka-consumer-groups.bat --bootstrap-server localhost:9092 --group testgroup --topic hello-kafka:1 --reset-offsets --to-offset 10 --execute