[Kafka 기초 학습] 카프카 기초 개념 6 - 컨슈머 이해 및 실습 진행
지난 글에서는 프로듀서에 대해 이해하고 실습을 진행해 보았다.
사실 이전 글에서 프로듀서와 컨슈머를 다 정리하려고 했는데 프로듀서 관련해서 알아볼 내용이 많았기 때문에 컨슈머 관련 글을 따로 분리했다.
하여간 이번 글에서는 컨슈머에 대해 이해하고 컨슈머 사용 실습을 진행해볼 예정이다.
목표
1. 컨슈머의 구조에 대해 이해한다.
2. 컨슈머의 다양한 설정을 통해 실습을 진행한다.
0. 들어가기 전에
컨슈머(consumer)란 무엇인지 정리해보자.
An Apache Kafka® Consumer is a client application that subscribes to (reads and processes) events.
출처 : https://docs.confluent.io/platform/current/clients/consumer.html
Apache Kafka Consumer는 이벤트를 읽거나 처리하기위해 구독하는 클라이언트 어플리케이션이다.
이렇게 보면 전통적인 pub-sub 구조와 다르지 않아 보이지만 Kafka는 분산 이벤트 데이터 스트리밍 플랫폼으로 명명하는것이 낫다.
컨슈머는 소비하려는 파티션의 상위 브로커에(leading partition) fetch 요청을 보낸다.
컨슈머가 요청을 보낼 때 로그 오프셋을 지정하고, 지정된 오프셋 위치부터 시작하는 로그 청크를 수신합니다. 이를 통해 컨슈머는 소비할 데이터를 제어할 수 있으며, 필요한 경우 오프셋을 지정하여 데이터를 다시 소비할 수도 있다.
여기서 chunk란 덩어리를 의미하며 로그 청크는 말 그대로 log의 덩어리를 말한다.
이해를 돕기 위해 필자가 Consumer의 그림을 그려보았다.
정리한 글을 읽고 해당 구조에 대해서 이해하도록 하는 것이 내 목표이다.
어떻게 해서 결국 컨슈머는 고정된 파티션에서 메시지를 pull해올 수 있는가?
1. Consumer의 Pull 디자인의 장점(Push vs Pull)
Apache Kafka 의 Consumer는 전통적인 메시지 시스템과 다르게 Consumer라 broker로 부터 메시지를 당겨온다.(pull)
이것을 정리하면 pull-based system이라고 하며 이의 장점은 다음과 같다.
1.1) 컨슈머가 프로듀서보다 뒤처져도 따라잡을수 있다.
전통적인 메시지 큐 브로커는 연결된 컨슈머에게 데이터를 일방적으로 밀어넣는다.(push)
실제 메시지를 처리하는 컨슈머를 고려하지 않고 감당 못할 정도로 메시지를 밀어넣게 되면 컨슈머에 과부하가 걸리거나 시스템이 뻗어버리게 된다.
하지만 pull 방식을 사용한다며 컨슈머 자체가 처리할 수 있는 만큼만 브로커에게서 데이터를 직접 요청해서 가져온다.
이렇게 컨슈머는 자신의 페이스를 유지하며 작업 흐름을 제어한다.
그래서 컨슈머가 프로듀서보다 뒤처져도 꾸준이 작업하다보면 언젠가는 쌓인 데이터를 다 처리하고 따라잡을 수 있다는 말이다.
1.2) 전송되는 데이터를 적극적으로 일괄처리(batching) 할 수 있다.
단건으로 네트워크 통신을 하며 데이터를 가져오는 것 보다 한 번의 요청으로 대량의 데이터를 묶어 가져온 뒤 DB에 한번에 저장는 것이 좋다. 이것을 다르게 말하면 '적극적인 일괄 처리' 가 가능하다고 할 수 있다.
이렇게 불팔요한 latency를 최소화함으로 성능 향상을 도모할 수 있다.
정리하면 컨슈머 스스로 읽기 비율을 조절하고 backpressure를 자연스럽게 다룰 수 있다는 말이다.
고로, Push 하는 방식보다 Pull 하는 방식이 효율적이다.
물론 pull하는 방식의 단점도 있다. 컨슈머가 계속 물어봐야 한다는 점이다.
push 방식은 데이터가 생기면 브로커가 알아서 보내주니까 컨슈머는 가만히 받기만 하면 된다. 하지만 pull 방식은 데이터가 있든 없든 컨슈머가 "데이터 있어?" 하고 계속 물어봐야 한다.
문제는 토픽에 새 메시지가 안 들어올 때다. 데이터가 없는데도 컨슈머가 쉴 새 없이 요청을 날리면, 빈 응답만 주고받으며 CPU와 네트워크를 낭비하게 된다.
데이터가 없는데도 컨슈머가 쉴 새 없이 요청을 날리면, 빈 응답만 주고 받으며 CPU와 네트워크를 낭비하게 된다.
그럼 카프카는 이 문제를 어떻게 해결했을까?
바로 롱 폴링(long-polling) 방식이다. 컨슈머가 poll()로 데이터를 요청하면, 브로커는 가져올 데이터가 없을 경우 바로 빈 응답을 주는 게 아니라 데이터가 쌓일 때까지 잠시 기다렸다가 응답을 준다.
이걸 조절하는 설정이 두 가지 있다.
- fetch.min.bytes: "최소 이만큼 데이터가 모이면 보내줘" (기본 1바이트)
- fetch.max.wait.ms: "근데 데이터가 안 모여도 이 시간까지만 기다렸다 보내줘" (기본 500ms)
즉, 데이터가 fetch.min.bytes만큼 쌓이거나, fetch.max.wait.ms 시간이 지나거나 둘 중 하나라도 만족하면 브로커가 응답을 보낸다.
이렇게 해서 카프카는 pull 방식의 단점인 불필요한 요청 낭비를 줄이면서, 동시에 1.2)에서 말한 일괄 처리(batching)의 장점까지 함께 챙긴 것이다.
2. 컨슈머의 순차적 읽기와 Offset 관리
컨슈머는 파티션 내의 오프셋에 기반해 레코드를 순차적으로 읽는다.
각각의 파티션은 자신만의 offset순서를 독립적으로 유지한다.
오프셋 커밋(Offset Commit)
컨슈머가 어디까지 메시지를 완벽하게 읽었는지 카프카 브로커에게 보고하는 과정이다.
카프카는 이 책갈피 기록을 __consumer_offsets이라는 내부 비밀 토픽에 저장한다.
장애 복구(Failover Recovery)
만약 컨슈머가 작업을 하다가 서버가 터지면?
새로 띄워진 컨슈머는 __consumer_offsets를 뒤져서 어디까지 읽었는지 확인한 후 그 지점부터 다시 처리를 이어간다.
커밋을 하는 방법은 두 가지가 있다.
자동 커밋(automatic / enable.auto.commit=true)
타이머를 맞추고 주기적으로 컨슈머가 알아서 카프카에게 책갈피를 꽂아주는 방식.
이 방식은 중복 처리를 발생시킬 수 있다.
수동 커밋(manual)
어프리케이션이 성공적으로 메시지 처리를 한 다음 직접 커밋하는 방식.
다음은 conductor 사이트에서 KafkaConsumer 관련 설명 코드를 가져온 것이다. (출처 : conduktor의 consumer 설명)
예시 코드 - 주석 추가
Properties props = new Properties();
// 1. 카프카 클러스터 연결 정보
// 컨슈머가 처음 연결을 맺을 카프카 브로커의 주소입니다.
props.put("bootstrap.servers", "localhost:9092");
// 2. 컨슈머 그룹 ID (핵심 설정 ⭐)
// 이 컨슈머가 속한 그룹의 이름입니다. 같은 이름을 가진 컨슈머들끼리 파티션을 공평하게 나눠 가집니다.
props.put("group.id", "order-processing-group");
// 3. 역직렬화 (Deserializer) 설정
// 프로듀서가 데이터를 전송할 때 네트워크를 타기 위해 바이트(Byte) 배열로 잘게 쪼개서(직렬화) 보냈습니다.
// 컨슈머가 이것을 읽으려면 다시 우리가 아는 문자열(String) 형태로 조립(역직렬화)해야 합니다.
props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
// --- [실무형(Modern production) 고급 설정] ---
// 4. 자동 커밋 끄기 (수동 커밋 사용)
// 타이머에 맞춰 알아서 오프셋(책갈피)을 기록하는 기능을 끕니다. 데이터 유실이나 중복을 막기 위해 개발자가 직접 통제하겠다는 뜻입니다.
props.put("enable.auto.commit", "false");
// 5. 능동적인 배치 처리 조절 (Pull 방식의 장점)
// 한 번 poll() 할 때 브로커로부터 당겨올 최대 레코드 수를 500개로 제한합니다.
// 데이터가 폭주해도 내 메모리가 터지지 않게, 딱 내가 감당할 수 있는 만큼만 가져옵니다.
props.put("max.poll.records", 500);
// 6. 파티션 재할당(리밸런싱) 최적화 전략
// 컨슈머가 새로 들어오거나 죽었을 때 파티션을 다시 나눠주는 작업(리밸런싱)을 할 때,
// 기존에는 하던 일을 다 멈추고(Stop-the-world) 다시 나눴지만, 이 설정(Cooperative)을 쓰면 꼭 필요한 파티션만 부드럽게 교체하여 중단 시간을 최소화합니다.
props.put("partition.assignment.strategy", "org.apache.kafka.clients.consumer.CooperativeStickyAssignor");
// 위에서 세팅한 설정값들을 넣어서 실제 컨슈머 객체를 생성합니다.
KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
// 7. 토픽 구독 (Subscribe)
// "orders"라는 토픽을 구독합니다. (내가 몇 번 파티션을 맡을지는 카프카 코디네이터가 알아서 배정해 줍니다.)
consumer.subscribe(Arrays.asList("orders"));
// 8. 무한 폴링(Polling) 루프
// 컨슈머는 프로그램이 종료될 때까지 무한 반복하며 브로커에게 "데이터 있어?" 하고 능동적으로 물어봅니다.
while (true) {
// 9. 데이터 당겨오기 (Pull)
// 브로커에 데이터가 없으면 최대 100ms 동안 기다려봅니다.
// 100ms 안에 데이터가 들어오거나, 이미 쌓여있는 데이터가 있다면 통째로 가져옵니다. (위에서 설정한 최대 500개 이내로 묶어서 가져옴)
ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
// 10. 가져온 데이터(배치) 반복 처리
for (ConsumerRecord<String, String> record : records) {
// 실제 비즈니스 로직 수행 (예: DB에 주문 정보 저장, 결제 서버 호출 등)
processOrder(record.value());
}
// 11. 수동 커밋 (Sync 방식)
// 위 for문에서 500개(또는 그 이하)의 처리가 '안전하게 모두 끝난 직후'에
// 카프카에게 "나 여기까지 에러 없이 완벽하게 다 처리했어!"라고 오프셋 책갈피를 꽂습니다.
// Sync(동기) 방식이므로, 커밋이 카프카 서버에 확실히 기록될 때까지 다음 줄로 넘어가지 않고 기다립니다.
consumer.commitSync();
}
3. Consumer group에 대해
컨슈머 그룹은 하나 혹은 더 많은 토픽으로부터 메시지를 소비하거나 처리하기 위해 함께 일하는 컨슈머들의 한 세트를 말한다.
토픽은 여러개의 파티션으로 나뉘는데, 각각의 파티션은 적어도 하나의 컨슈머에 의해 소비된다.
카프카는 컨슈머를 그룹으로 묶어서 일을 나눠주고(병렬 처리), 절대 두 명이 같은 일을 하게 놔두지 않으며(독점권), 누가 죽으면 남은 애들한테 일을 다시 나눠준다(리밸런싱)
이게 무슨말이냐고? 자세히 알아보도록 하자.
독점적인 파티션 접근(Exclusive Partition Access)
각각의 파티션을 그룹 내에 하나의 컨슈머에 할당한다. 이것은 그룹 내에 중복 처리를 방지하고 오직 하나의 컨슈머에 의해서만 처리되는걸 보장한다.
리밸런싱(Rebalancing)
컨슈머 인스턴스가 그룹을 떠나거나 그룹에 참여하면 혹은 토픽의 파티션이 변한다면, 그룹 리밸런스가 발생한다.
파티션들은 활동하는 멤버들 사이에 재분배되고 고가용성을 보장한다.
그러니가 리밸런싱은 컨슈머가 추가되거나 제거되면 파티션을 골고루 나눠주는 것이다.
병럴 처리(Parallel Procecssing)
여러 명의 컨슈머가 같은 시간에 동시에(Parallel) 각자 할당받은 파티션의 데이터를 가져와 처리하는 것을 의미
위의 3가지를 비유를 통해 설명하면 다음과 같다.
- 병렬 처리: "우리 다 같이 동시에 일해서 속도 3배로 올리자!"
- 독점권: "대신 똑같은 손님 물건을 두 번 계산하면(중복 처리) 안 되니까, 각자 딱 한 줄씩만 전담해서 계산해!"
- 리밸런싱: "근데 만약 계산원 한 명이 갑자기 화장실 가면(장애 발생), 남은 두 명이 그 줄까지 합쳐서 다시 나눠 맡자!"
4. Consumer 실습 진행
너무 이론적인 내용만 읽게 되면 재미없다.
고로, 실습과 함께 실제로 필자가 원하는대로 나눠서 잘 동작하는지 알아보도록 하겠다.
docker-compose-consumers.yml
version: '3.8'
networks:
# =========================================================================
# [네트워크 설정]
# 이미 켜져 있는 카프카 브로커 서버들(kafka-1, 2, 3)과 통신하기 위해
# 그들이 만들어둔 'kafka-local-network'에 무단 침입(external: true) 합니다.
# =========================================================================
kafka-network:
external: true
name: kafka-local-network
services:
# =========================================================================
# 🚨 [Consumer Group 1 (g1)] - 컨슈머 2대 (인원 부족 상태)
# - 파티션은 3개인데 일할 사람은 2명(1번, 2번)뿐입니다.
# - 결과: 1명은 파티션 1개, 나머지 1명은 파티션 2개를 맡아서 '독박'을 쓰게 됩니다.
# =========================================================================
consumer-g1-1:
image: apache/kafka:latest
container_name: consumer-g1-1
networks:
- kafka-network
# [명령어 해부]
# --property 대신 최신 버전 권장 옵션인 --formatter-property를 사용했습니다!
command: |
/opt/kafka/bin/kafka-console-consumer.sh \
--bootstrap-server kafka-1:9092 \
--topic cluster-test \
--group g1 \
--formatter-property print.key=true \
--formatter-property print.value=true \
--formatter-property print.partition=true
consumer-g1-2:
image: apache/kafka:latest
container_name: consumer-g1-2
networks:
- kafka-network
command: |
/opt/kafka/bin/kafka-console-consumer.sh \
--bootstrap-server kafka-1:9092 \
--topic cluster-test \
--group g1 \
--formatter-property print.key=true \
--formatter-property print.value=true \
--formatter-property print.partition=true
# =========================================================================
# ✨ [Consumer Group 2 (g2)] - 컨슈머 3대 (완벽한 1:1 분업 상태)
# - 파티션 3개, 일할 사람도 3명(3번, 4번, 5번)입니다.
# - 결과: 3명이 각자 파티션을 1개씩 전담하여 최고의 병렬 처리 속도를 냅니다.
# - 독립성: 위의 g1 그룹이 데이터를 어떻게 빼가든 상관없이,
# g2 그룹은 토픽의 모든 데이터를 100% 똑같이 싹 다 가져옵니다. (Pub/Sub 구조)
# =========================================================================
consumer-g2-3:
image: apache/kafka:latest
container_name: consumer-g2-3
networks:
- kafka-network
command: |
/opt/kafka/bin/kafka-console-consumer.sh \
--bootstrap-server kafka-1:9092 \
--topic cluster-test \
--group g2 \
--formatter-property print.key=true \
--formatter-property print.value=true \
--formatter-property print.partition=true
consumer-g2-4:
image: apache/kafka:latest
container_name: consumer-g2-4
networks:
- kafka-network
command: |
/opt/kafka/bin/kafka-console-consumer.sh \
--bootstrap-server kafka-1:9092 \
--topic cluster-test \
--group g2 \
--formatter-property print.key=true \
--formatter-property print.value=true \
--formatter-property print.partition=true
consumer-g2-5:
image: apache/kafka:latest
container_name: consumer-g2-5
networks:
- kafka-network
command: |
/opt/kafka/bin/kafka-console-consumer.sh \
--bootstrap-server kafka-1:9092 \
--topic cluster-test \
--group g2 \
--formatter-property print.key=true \
--formatter-property print.value=true \
--formatter-property print.partition=true
다음 을 보고 이제 실행하자.
docker compose -f docker-compose-consumers.yml up -d
그리고 Mini PC 환경에서 필자가 메시지를 kafka-console-producer.sh를 사용해서 메시지를 전송햇다.
~/kafka-consumer$ docker exec -it kafka-1 /opt/kafka/bin/kafka-console-producer.sh --bootstrap-server localhost:9092 --topic cluster-test --property "parse.key=true" --property "key.separator=:"
Warning: --property is deprecated and will be removed in a future version. Use --reader-property instead.
>UserA:hello test
>UserB:hi test
>UserC:greetings test
>USerD:apple test
>UserE:banana test
>UserF:carrot test
그리고 다음 명령어를 실행했다.
docker compose -f docker-compose-consumers.yml logs | grep "test"
그럼 과연... 메시지가 consumer에 고루고루 분배되어서 들어갓는지 보도록 하겠다.
~/kafka-consumer$ docker compose -f docker-compose-consumers.yml logs | grep "test"
WARN[0000] /home/thelovemsg/kafka-consumer/docker-compose-consumers.yml: the attribute `version` is obsolete, it will be ignored, please remove it to avoid potential confusion
consumer-g2-4 | Partition:0 UserA hello test
consumer-g2-4 | Partition:0 USerD apple test
consumer-g2-4 | Partition:0 UserF carrot test
consumer-g2-5 | Partition:1 UserB hi test
consumer-g2-5 | Partition:1 UserC greetings test
consumer-g2-3 | Partition:2 UserE banana test
consumer-g1-2 | Partition:2 UserE banana test
consumer-g1-1 | Partition:0 UserA hello test
consumer-g1-1 | Partition:1 UserB hi test
consumer-g1-1 | Partition:1 UserC greetings test
consumer-g1-1 | Partition:0 USerD apple test
consumer-g1-1 | Partition:0 UserF carrot test
consumer group 1 을 보면 consumer-g1-1이 partition0과 partition1 데이터를 모두 처리하고 있다.
그에 반해 consuger group 2를 보면 consumer-g2-3은 partition 3데이터를, consumer-g2-4는 partition 0을, consumer-g2-5는partition 1을 처리한다.
우선 g1 과 g2의 group 정보를 차례대로 출력해 보았다.
:~$ docker exec -it kafka-2 /opt/kafka/bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 --describe --group g1
GROUP TOPIC PARTITION CURRENT-OFFSET LOG-END-OFFSET LAG CONSUMER-ID HOST CLIENT-ID
g1 cluster-test 0 21 21 0 console-consumer-1f72c83f-387e-4eef-9e5c-2470e5e4c044 /172.18.0.9 console-consumer
g1 cluster-test 1 24 24 0 console-consumer-1f72c83f-387e-4eef-9e5c-2470e5e4c044 /172.18.0.9 console-consumer
g1 cluster-test 2 13 13 0 console-consumer-fb5d34a8-a6f8-4930-90d5-0169bcfb5e0c /172.18.0.5 console-consumer
~$ docker exec -it kafka-2 /opt/kafka/bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 --describe --group g2
GROUP TOPIC PARTITION CURRENT-OFFSET LOG-END-OFFSET LAG CONSUMER-ID HOST CLIENT-ID
g2 cluster-test 2 13 13 0 console-consumer-822fe054-a97e-4785-8427-2337eabbda02 /172.18.0.6 console-consumer
g2 cluster-test 1 24 24 0 console-consumer-70bc48bc-c473-4d86-b2c5-0a6ab9ab4917 /172.18.0.7 console-consumer
g2 cluster-test 0 21 21 0 console-consumer-127efbb6-cb13-44c2-a961-39456fac5475 /172.18.0.8 console-consumer
컨슈머는 오직 리더만 바라본다.
리더는 읽기와 쓰기를 다 담당하는데, 팔로워는 단지 문제가 생겻을 때 백업용인 것이다
PARTITION 0, 1, 2가 표시된 것은, 해당 파티션들의 '리더'를 담당하고 있는 브로커와 g2 그룹의 컨슈머들이 현재 연결되어 있다는 뜻이다.
우선 이렇게 봐서는 음... 잘 나뉘어서 들어가는거 같은데? producer에서 우리가 key로 UserA ~ UserF 까지 hash화가 잘 되어서 나뉘어서 들어갔으니 consumer에서 어떻게 갔는지 확인을 한 것이다.
g2(컨슈머 3 = 파티션 3) 는 1:1로 딱 떨어져서 '병렬 처리'가 완벽하게 작동했고, g1(컨슈머 2 = 파티션 3) 은 consumer-g1-1이 파티션 0·1을 혼자 떠안았다.
위 명령어에서 나온 위의 Column 명들에 대해서 정리하고 가자.
GROUP
Consumer Group 이름. 같은 토픽을 여러 Consumer가 나눠서 소비할 때 묶는 단위다.
TOPIC / PARTITION
어떤 토픽의 몇 번 파티션을 이 그룹이 소비하고 있는지를 나타낸다.
CURRENT-OFFSET
Consumer가 지금까지 읽은 위치 (커밋된 오프셋). "나 여기까지 읽었어" 라고 Kafka에 기록된 값이다.
LOG-END-OFFSET
해당 파티션에 실제로 쌓인 메시지의 마지막 위치. Producer가 보낸 메시지가 여기까지 존재한다.
LAG
LOG-END-OFFSET - CURRENT-OFFSET = 아직 안 읽은 메시지 수.
LOG-END-OFFSET(155) - CURRENT-OFFSET(150) = LAG(5)
LAG이 계속 커지면 Consumer가 Producer 속도를 못 따라가고 있다는 신호다.
CONSUMER-ID
현재 해당 파티션을 소비 중인 Consumer 인스턴스 식별자. 값이 - 이면 현재 활성화된 Consumer가 없다는 뜻이다.
HOST
Consumer가 실행 중인 서버 IP.
CLIENT-ID
Consumer 코드에서 설정한 애플리케이션 레벨 식별자. Properties에서 client.id로 지정하는 값이다.
여기서 뭐 다른건 그렇게 치더라고 LAG는 심각하게 우리가 고려해야할 사항이다.
5. 리밸런싱의 원리와 RangeAssignor 알아보기
위에서 리밸런싱과 그룹에 따른 분배가 되는 것을 들었을텐데, 정확히 어떻게 동작하는지를 설명하지 않았다.
이에 대해 공부하고 가고자 한다.
리밸런싱 관련해서 confluent사이트에서 어떻게 동작하는지 자세히 설명해주고 있다. (Kafka Rebalancing Explained)
리밸런싱 발생 이유
리밸런싱은 멤버의 변화, 세션 타임아웃 그리고 최대 poll 간격의 초과 혹은 토픽 메타데이터 변화때문에 일어난다.
리밸런싱은 어떻게 작동하는가?
카프카 리밸런싱은 한 컨슈머 그룹에서 작동하는 컨슈머들 사이에 토픽 파티션을 재분배하는 조정 과정이다.
이 과정은 Group Coordinator라는 특별한 Kafka broker로 관리하며 컨슈머가 그룹에 참여하거나 떠날 때 혹은 토픽 메타데이터가 변할 때 발생한다.
카프카 리밸런싱은 다음과 같은 과정을 거친다.
- 트리거 및 알림: 컨슈머 장애 같은 이벤트가 발생하면, Group Coordinator가 이를 감지하고 영향받는 컨슈머들에게 알립니다. 컨슈머들은 주기적인 하트비트 또는 요청 시 수신되는 에러(예: REBALANCE_IN_PROGRESS)를 통해 새로운 리밸런싱을 인지하고, 그룹에 재가입하도록 유도됩니다.
- 일시 중지 및 파티션 반납: 그룹 내 모든 컨슈머는 메시지 처리를 일시적으로 중단하고, 현재 오프셋을 커밋한 후, 기존에 할당받았던 파티션의 소유권을 반납합니다.
- 재가입 및 리더 선출: 각 컨슈머가 Group Coordinator에게 JoinGroup 요청을 전송합니다. 코디네이터는 이 중 하나를 이번 리밸런싱을 위한 임시 그룹 리더로 선출합니다.
- 파티션 할당: Group Coordinator는 리더에게 현재 활성화된 전체 컨슈머 목록과 할당이 필요한 전체 파티션 목록을 제공합니다. 리더는 정의된 할당 전략(예: CooperativeStickyAssignor)을 사용하여 각 파티션을 특정 컨슈머에 매핑하는 새로운 할당 계획을 수립합니다.
- 동기화 및 배포: 리더는 새로운 할당 계획을 Group Coordinator에게 전달합니다. 코디네이터는 이 계획을 배포하며, 각 컨슈머에게는 자신이 담당할 파티션 정보만을 전송합니다.
- 처리 재개: 컨슈머들은 새로운 할당을 수신하면, 새로 배정된 파티션에서 메시지 페치 및 처리를 재개합니다.
잠깐, 여기서 '리더'가 두 개라 헷갈릴 수 있다.
4번에서 "컨슈머는 리더만 바라본다"고 했을 때의 리더는 파티션 리더 브로커다. 즉 읽기/쓰기를 담당하는 서버를 말한다.
그런데 지금 5번에서 말하는 "그룹 리더"는 컨슈머 중 한 명이다. 파티션을 누구한테 나눠줄지 할당 계획을 짜는 역할일 뿐, 데이터를 읽는 서버와는 전혀 상관없다.
정리하면, 파티션 리더 = 브로커(서버), 그룹 리더 = 컨슈머. 이름만 같지 완전히 다른 놈들이다.
작동 원리를 읽어보면 뭔가 JVM의 Stop The World와 같은 현상이 여기 카프카 리밸런싱에도 작동한다.
그룹 내 모든 컨슈머 메시지 처리를 일시 중단한다고?
결국 컨슈머들이 새로운 할당을 수신할 때 까지 처리가 멈춰버린다.
리밸런싱 중 컨슈머가 데이터 처리를 일시적으로 중단하기 때문에 운영 중단 및 서비스 지연이 생길 수 있고
리밸런싱은 CPU, 메모리, 네트워크 사용량의 급격한 증가를 유발할 수 있으므로 리소스 사용량 급증 및 클라우드 비용 증가할 뿐더러
잦은 리밸런싱은 수동 개입과 트러블슈팅을 필요로 하는 경우가 많아, 개발자 생산성을 저하시켜 운영 부담 증가 및 디버깅 사이클 증가시키니 굉장히 민감한 사항이다.
위에 첨부한 Kafka Rebalancing Explained 를 잘 읽어보면 좋을 것 같다.
마지막에는 FAQ로 다음과 같이 정리한다. 요약 부분이니 이것만 정리하고 가겠다.
FAQ
1) Kafka 컨슈머 그룹 리밸런싱은 무엇이 트리거하나요?
멤버십 변경(컨슈머 참여/이탈/타임아웃), 파티션 수 변경, 또는 코디네이터 변경이 트리거가 됩니다. 리밸런싱은 각 활성 컨슈머가 작업을 할당받을 수 있도록 파티션을 재분배합니다.
2) 리밸런싱이 왜 지연이나 백로그를 유발하나요?
파티션이 반납되고 재할당되는 동안 컨슈머가 잠시 처리를 중단하기 때문입니다. 이로 인해 컨슈머 랙이 급증하고, 그룹이 안정 상태로 돌아올 때까지 하위 애플리케이션의 처리가 느려질 수 있습니다.
3) 리밸런싱으로 인한 영향을 어떻게 줄일 수 있나요?
group.instance.id를 사용하는 정적 멤버십(Static Membership)을 적용하면 짧은 재시작에도 전체 리밸런싱이 트리거되지 않습니다. 또한 파티션 이동을 최소화하기 위해 Cooperative/Sticky Assignor를 선호하고, 실제 처리 시간에 맞게 타임아웃 값을 설정하는 것이 좋습니다.
4) 문제를 사전에 파악하려면 무엇을 모니터링해야 하나요?
리밸런싱 빈도, 컨슈머 랙 SLO, 리밸런싱 이후 안정 상태까지의 소요 시간을 추적하세요. 이 지표들이 지속적으로 상승하는 추세라면 보통 컨슈머 사이징, 롤아웃 정책, 또는 타임아웃 설정에 문제가 있다는 신호입니다.
자, 이제 처음 질문으로 돌아가보자.
"어떻게 컨슈머는 고정된 파티션에서 메시지를 pull해올 수 있는가?"
지금까지 본 내용을 엮으면 답이 나온다.
컨슈머가 그룹에 들어가면, 코디네이터가 "너는 몇 번 파티션 담당해"라고 파티션을 배정해준다(3·5번).
이 배정은 리밸런싱이 일어나기 전까지 고정된다.
컨슈머는 배정받은 파티션에서 오프셋을 기준으로 어디까지 읽었는지 기억하면서(2번), 자기가 감당할 수 있는 만큼만 능동적으로 poll해서 당겨온다(1번).
즉 '고정'은 코디네이터의 파티션 배정이 만들어주고, 'pull'은 컨슈머의 poll 루프가 담당하는 것이다. 이 둘이 합쳐져서 처음 질문의 답이 된다.
6. Consumer Lag 의 정의
해당 섹션은 다음의 글을 참고했다.(https://www.conduktor.io/glossary/consumer-lag-monitoring)
Kafka Cousumer Lag는 컨슈머 그룹이 파티션에서 아직 읽어야 할 메시지 수를 나타낸다.
우리가 게임을 할 때 느리면 랙걸린다고 하지 않나?
consumer에도 메시지 처리가 늦어지면 Consumer Lag가 쌓인다.
이 Lag는 Kafka 컨슈머가 토픽의 처리 속도를 따라가고 있는지 여부를 판단하는 데 가장 유용한 지표이며, 대부분의 팀에서 가장 먼저 알림을 설정하는 항목이다.
왜 Lag가 많이 쌓이면 문제일까?
Consumer lag은 스트리밍 아키텍처의 여러 핵심 요소에 직접적인 영향을 미친다.
- 데이터 신선도: Lag이 높으면 하위 시스템이 오래된 데이터를 처리하게 된다. 실시간 대시보드나 사기 탐지 시스템을 구축하는 경우, 과도한 Lag은 인사이트를 무의미하게 만들거나 사기 거래를 놓치게 할 수 있다.
- SLA 준수: 많은 조직이 데이터 처리 지연에 대해 엄격한 SLA를 가지고 있다. Consumer가 뒤처지면 이 계약을 위반하게 되어 계약상 패널티가 발생하거나 고객 신뢰를 잃을 수 있다.
- 시스템 상태 지표: Lag이 갑자기 증가하면 과부하된 Consumer, 네트워크 문제, 데이터베이스 병목, 리소스 부족 등 근본적인 문제가 있다는 신호다. Lag은 시스템 성능 저하의 첫 번째 증상으로 나타나는 경우가 많다.
- 용량 계획: 시간에 따른 Lag 메트릭 추세를 보면 Consumer 인프라를 언제 확장해야 할지 파악할 수 있다. Lag이 지속적으로 증가한다면 현재 용량이 들어오는 메시지 속도를 따라가지 못하고 있다는 뜻이다.
Consumer Lag의 근본 원인
Lag의 원인을 이해하는 것은 효과적인 해결을 위해 필수적이다.
- 처리 병목: 가장 흔한 원인은 단순히 메시지 처리 시간이 너무 오래 걸리는 것이다. 무거운 연산, 느린 DB 쿼리, 외부 API 호출, 비효율적인 코드 등이 원인이 될 수 있다.
- Consumer 리소스 부족: CPU, 메모리, 네트워크 대역폭이 충분하지 않으면 메시지를 빠르게 처리할 수 없다. Consumer 인스턴스의 높은 CPU 사용률이나 메모리 압박으로 나타나는 경우가 많다.
- 리밸런싱 이벤트: Consumer가 그룹에 참여하거나 이탈할 때 Kafka는 리밸런싱을 트리거하고, 이 동안 처리가 일시 중단된다. 비정상적인 Consumer, 네트워크 문제, 잘못 설정된 타임아웃으로 인해 리밸런싱이 자주 발생하면 주기적인 Lag 스파이크가 생긴다.
- 하위 시스템의 백프레셔: Consumer는 DB에 쓰거나, HTTP 요청을 보내거나, 다른 시스템에 메시지를 발행하는 경우가 많다. 이런 하위 의존 시스템이 느려지거나 사용 불가 상태가 되면 Consumer가 응답을 기다리며 블로킹된다.
- 메시지 쏠림: 파티션 간 메시지 분배가 고르지 않거나 특정 파티션에 트래픽이 집중되는 "핫 파티션"이 생기면 일부 Consumer는 뒤처지고 나머지는 유휴 상태가 된다.
이번 시간에 이렇게 Consumer에 대해 좀 더 이해하고 원리를 알아보았다.
consumer관련해서 정보를 찾아보려 하니 정말로 많은 자세한 공식문서와 또 이 내용에만 국한되지 않은 여러 내용들이 같이 딸려 나왔다.
그래서 이걸 정리하기 생각보다 힘들었고,
이전에 producer관련해서는 내가 읽어보고 했는데 이번 컨슈머의 경우네는 코드레벨까지 볼 엄두가 나질 않았다.
(너무 뭐가 많아! ㅠㅠ)
그래서 참고한 내용에는 최대한 많은 링크를 달아두었으니 시간이 날 때 정독해야겠다는 생각이 든다.
물론 가장 좋은 것은 정말 사용하고 세팅할 때 알음알음 알아가면서 써야 좋은것이지 당장 열심히 정리해도 기억도 못하는데
너무 깊이 파는은 과한게 아닌가 하는 생각도 든다.
참고한 내용
https://www.youtube.com/watch?v=mHaVGVLyfB4&list=PLf38f5LhQtheK16nwnCYFqH23WUUvZfSb&index=7
https://www.conduktor.io/glossary/kafka-producers-and-consumers
https://d2.naver.com/helloworld/9581727
https://d2.naver.com/helloworld/0974525
https://www.conduktor.io/glossary/consumer-lag-monitoring
https://www.confluent.io/learn/kafka-rebalancing/