Consumer - 1

💡 이번 포스팅에서는 Consumer에 대해 정리해 보도록 하겠습니다.

 

Consumer(이하 컨슈머)는 프로듀서가 브로커에 전송한 메시지를 읽는 역할을 수행합니다.

컨슈머는 반드시 컨슈머 그룹에 속해야 하며 컨슈머 그룹 내에서 여러 개의 컨슈머들은 토픽 파티션 별로 분배됩니다.

출처: https://ooeunz.tistory.com/136

 

 

아파치와 컨플루언트 공식 문서에서는 아래와 같이 설명하고 있습니다.

 

출처: https://kafka.apache.org/documentation/#theconsumer

출처: https://docs.confluent.io/platform/current/clients/consumer.html#ak-consumer

컨슈머는 오프셋 위치에서 시작하는 로그를 받는다고 합니다. 또한 오프셋 위치를 제어할 수 있기 때문에 이미 가져왔던 데이터를 다시 가져오는 것 또한 가능합니다.

이때 오프셋 위치는 __consumer_offset 토픽에 저장합니다.

 

📢 앞서 컨슈머가 프로듀서가 브로커에 전송한 메시지를 읽어온다고 했는데, 정확히 얘기하면 컨슈머가 subscribe(구독)하고 있는 토픽의 파티션 중 리더 파티션에서 메시지를 읽어옵니다.

아래 컨플루언트 공식 문서의 consumer groups에 보다 자세하게 설명되어 있습니다.

 

consumer groups

출처: https://docs.confluent.io/platform/current/clients/consumer.html#consumer-groups

 

 

컨슈머 그룹은 일부 토픽의 데이터를 consume 하기 위해 협력하는 컨슈머 집합입니다. 모든 토픽의 파티션은 그룹의 컨슈머 간에 분할됩니다. 새 그룹 구성원이 도착하고 이전 구성원이 떠나면 각 구성원이 파티션의 비례적 몫을 받도록 파티션이 다시 할당됩니다. 이를 Group Rebalancing이라고 합니다.

이전 버전의 컨슈머와 새로운 컨슈머의 주요 차이점은 전자는 그룹 관리를 위해 ZooKeeper에 의존하는 반면 후자는 Kafka 자체에 내장된 그룹 프로토콜을 사용한다는 것입니다. 이 프로토콜에서 브로커 중 한 명이 그룹의 코디네이터로 지정되고 그룹 구성원 및 파티션 할당 관리를 담당합니다.

각 그룹의 코디네이터는 커밋된 오프셋을 저장하는 데 사용되는 내부 오프셋 항목 __consumer_offsets의 리더에서 선택됩니다. 기본적으로 그룹의 ID는 이 토픽에 대한 파티션 중 하나에 해시되고 해당 파티션의 리더가 코디네이터로 선택됩니다. 이러한 방식으로 컨슈머 그룹의 관리는 클러스터의 모든 브로커에 걸쳐 대략 균등하게 분할되며, 이는 브로커 수를 늘려 그룹 수를 확장할 수 있도록 합니다.

컨슈머가 실행되면 해당 그룹의 코디네이터를 찾고 그룹 가입 요청을 보냅니다. 그런 다음 코디네이터는 그룹 재조정을 시작하여 새 구성원이 그룹 파티션의 공평한 몫을 할당받도록 합니다. 모든 재조정은 그룹의 새로운 세대로 이어집니다.

그룹의 각 구성원은 그룹 구성원으로 남아 있으려면 코디네이터에게 heartbeat를 보내야 합니다. 구성된 session timeout이 만료되기 전에 heartbeat가 수신되지 않으면 코디네이터는 해당 구성원을 그룹에서 추방하고 해당 파티션을 다른 구성원에게 재할당합니다.

 

컨슈머 그룹에서 언급된 내용 중  Group Coordinator와 Heartbeat 부분에 대해 더 자세하게 알아보겠습니다.

 

Group Coordinator

group coordinator는 새로운 컨슈머가 추가되거나 기존 컨슈머가 종료될 때, 또는 토픽에 새로운 파티션이 추가될 때 컨슈머 그룹 내의 컨슈머들에게 파티션을 재 할당하는 Rebalancing을 수행하도록 지시합니다.

출처: https://seonkyukim.github.io/kafka-rebalancing/

 

컨슈머 그룹, group coordinator와 컨슈머의 동작 순서는 아래와 같습니다.

1. 컨슈머 그룹 내의 컨슈머가 브로커에 최초 접속 요청 시 Group Coordinator 생성
2. 동일 group.id로 여러 개의 컨슈머로 브로커의 Group Coordinator로 접속
3. 가장 빨리 그룹에 join 요청을 한 컨슈머에게 컨슈머 그룹 내의 리더 컨슈머로 지정
4. 리더로 지정된 컨슈머는 파티션 할당 전략에 따라 컨슈머들에게 파티션 할당
5. 리더 컨슈머는 최종 할당된 파티션 정보를 Group Coordinator로 전달
6. 정보 전달 성공을 공유한 뒤 개별 컨슈머들은 할당된 파티션에서 메시지 읽음

정리하면 Group Coordinator는 각 컨슈머 그룹 별로 생성되고, 속해 있는 컨슈머 그룹의 리더 컨슈머를 선별하는 역할을 하며 각 그룹 내의 컨슈머들이 생성/종료됨에 따라 Rebalancing을 수행하도록 지시하는 역할을 담당합니다.

이때 컨슈머 그룹은 상황에 따라 상태를 갖는데 이를 Consumer Group State라고 합니다.

 

Consumer Group State

출처: https://chrzaszcz.dev/2019/06/kafka-rebalancing/

모든 컨슈머 그룹은 하나의 특정 상태를 갖습니다. 허용되는 상태는 GroupMetatda 클래스에 정의되어 있으며 아래와 같이 5개의 상태가 있습니다.

• Empty: 컨슈머 그룹은 존재하지만 더 이상 컨슈머가 없습니다.
• Stable: Rebalance가 종료되고 안정적으로 컨슈머가 운영되고 있습니다.
• Rebalance
  • PreparingRebalance: 무언가 변경되었으며 파티션 재할당이 필요하므로 카프카가 재조정 중입니다.
  • CompletingRebalance: 카프가 재조정에 성공한 상태입니다.
  • Dead: 그룹이 곧 카프카 노드에서 제거될 예정입니다. 활동이 없거나 그룹이 다른 Group Coordinator로 마이그레이션 되고 있기 때문일 수 있습니다.

Rebalance(재조정)에 관련해서는 위 이미지와 같이 크게 Empty, Stable, Rebalance 3개의 상태로 표현할 수 있는데, Rebalance를 다시 PreparingRebalance, CompletingRebalance, Dead 상태로 세분화할 수 있습니다.

 

Heartbeat

위에 Group Coordinator가 파티션을 재조정해야 하는 상황에서 Heartbeat failure detected(heartbeat 실패 감지)가 있습니다.

heartbeat는 직역하면 '심장 박동' 인데, 컨슈머에서 heartbeat란 각 컨슈머들이 session을 유지하기 위해 주기적으로 Group Coordinator에게 heartbeat 요청을 보냅니다.

컨슈머가 Group Coordinator에게 join, leave등의 요청을 보내면 정상적으로 처리되지만, 컨슈머 장애로 인해 leave 요청을 보내지 못할 경우 Group Coordinator가 이를 감지할 방법이 없기 때문에 Group Coordinator는 이런 컨슈머 장애 감지를 위해 Heartbeat를 확인합니다.

 

heartbeat.interval.ms 설정을 통해 얼마의 주기로 Group Coordinator로 heartbeat 요청을 보낼지 설정할 수 있는데 만약 설정한 시간 동안 heartbeat 요청이 오지 않는다면 Group Coordinator는 해당 컨슈머가 이상이 있다고 판단하여 Rebalancing을 지시하여 해당 컨슈머를 컨슈머 그룹에서 제외하도록 합니다.

 

heartbeat 요청은 heartbeat thread가 보내게 되며 heartbeat와 관련된 설정은 아래와 같습니다.

파라미터	기본값(ms)	설명
heartbeat.interval.ms	3000	Heartbeat Thread가 Heartbeat를 보내는 간격. session.timeout.ms 보다 낮게 설정되어야 합니다. session.timeout.ms의 1/3보다 낮게 설정을 권장합니다.
session.timeout.ms	45000	브로커가 컨슈머로 Hearbeat를 기다리는 최대 시간입니다. 브로커는 이 시간 동안 Heartbeat를 컨슈머로부터 받지 못하면 해당 컨슈머를 그룹에서 제외하도록 Rebalancing을 지시합니다.

 

Consumer - 2

Consumer - 3

 

참고 자료

• https://seonkyukim.github.io/kafka-rebalancing/
• https://chrzaszcz.dev/2019/06/kafka-rebalancing/
• https://kafka.apache.org/documentation/#theconsumer
• https://docs.confluent.io/platform/current/clients/consumer.html
• <실전 카프카 개발부터 운영까지> - 고승범 지음
• <인프런 카프카 완벽 가이드 - 코어편> - 권철민