이번 시간에는 Kafka 컨슈머 그룹이 일을 나누는 방식을 단일 브로커 Kafka와 Python 컨슈머로 직접 띄워 보며 배워가는 시간입니다.
예를 들어서 주문 이벤트를 처리하는 컨슈머가 밀리기 시작했다고 해볼게요. 컨슈머를 3개에서 5개로 늘렸는데, 처리 속도는 그대로예요. 새로 띄운 2개는 오류도 없이 떠 있기만 합니다.
왜 그럴까요? Kafka 컨슈머 그룹은 파티션 단위로 일을 나누기 때문이에요.
그래서 같은 group.id를 쓰는 컨슈머들이 파티션을 어떻게 나눠 갖는지, 컨슈머 수를 늘리면 처리 시간이 어떻게 바뀌는지를 하나씩 확인해 볼게요.
실습 준비
이 글의 명령은 모두 실습 코드의 kafka-consumer-group/ 디렉터리에서 실행합니다.
compose.yaml: 단일 브로커 Kafka 4.3 컨테이너 정의consumer.py: 이름, 그룹, 토픽을 받아 파티션 할당과 반납을 시간과 함께 로그로 남기는 컨슈머produce.py: 토픽에 메시지 N개를 파티션 0, 1, 2에 번갈아 보내는 프로듀서bench.sh: 컨슈머 수별 처리 시간 측정 스크립트common.sh:bench.sh가 불러 쓰는 셸 함수
Docker로 단일 브로커 Kafka 4.3을 띄울게요.
브로커와 컨트롤러를 한 프로세스에서 돌리는 KRaft 구성이고, 호스트의 Python 컨슈머가 localhost:9421로 접속해요. 아래가 compose.yaml입니다.
services:
kafka:
image: apache/kafka:4.3.1
environment:
KAFKA_NODE_ID: 1
KAFKA_PROCESS_ROLES: broker,controller
KAFKA_LISTENERS: PLAINTEXT://:9092,CONTROLLER://:9093,HOST://:9421
KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka:9092,HOST://localhost:9421
KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER
KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT,HOST:PLAINTEXT
KAFKA_CONTROLLER_QUORUM_VOTERS: 1@kafka:9093
KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1
KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1
KAFKA_HEAP_OPTS: -Xmx512m -Xms256m
ports:
- "9421:9421"
이제 컨테이너 안의 Kafka 명령을 짧게 부르는 셸 함수를 만들고, 파티션이 3개인 orders 토픽을 만들어 볼게요.
아래 블록은 파일로 만들지 말고 실습 디렉터리에서 연 터미널에 그대로 붙여 넣어 주세요. 이후 명령은 모두 이 터미널에서 실행하고, 터미널을 새로 열면 다시 붙여 넣으시면 됩니다.
python3 -m venv venv && . venv/bin/activate && pip install -q confluent-kafka==2.15.1
docker compose up -d
sleep 10
kafka() { docker compose exec -T kafka /opt/kafka/bin/"$@"; }
kcg() { kafka kafka-consumer-groups.sh --bootstrap-server kafka:9092 "$@"; }
kafka kafka-topics.sh --bootstrap-server kafka:9092 --create --topic orders --partitions 3 --replication-factor 1
Created topic orders.
컨슈머 consumer.py는 이름, 그룹, 토픽을 인자로 받고, 파티션을 받거나 내놓을 때마다 시간과 함께 로그를 남겨요. 메시지 하나를 처리할 때마다 지정한 초만큼 기다려서 처리 시간을 흉내 냅니다.
group.id는 같은 값을 쓰는 컨슈머끼리 한 그룹이 돼요.client.id는 그룹 조회 결과에서 어느 컨슈머인지 알아보려고 붙인 이름이에요.on_assign,on_revoke: 파티션을 받거나 내놓을 때 호출돼요.SIGTERM을 받으면 루프를 빠져나와close()를 호출해요.bench.sh가 컨슈머를 멈출 때 씁니다.
컨슈머를 백그라운드로 띄우고 로그와 프로세스 번호를 파일에 남기는 함수도 만들어 둘게요.
mkdir -p logs
start() { python consumer.py "$@" > logs/$1.log 2>&1 & echo $! > logs/$1.pid; }
실습 1: 컨슈머를 5개까지 늘려 보기
order-service 그룹에 컨슈머 5개를 띄우고, 그룹 구성원과 각자 맡은 파티션을 볼게요.
--members --verbose를 붙이면 컨슈머마다 할당된 파티션이 나와요. 컨슈머가 그룹에 들어가 파티션을 받기까지 몇 초 걸리니까, 띄운 뒤 잠시 기다렸다가 조회해 주세요.
start c1 order-service orders
start c2 order-service orders
start c3 order-service orders
start c4 order-service orders
start c5 order-service orders
kcg --describe --group order-service --members --verbose
GROUP CONSUMER-ID HOST CLIENT-ID #PARTITIONS CURRENT-EPOCH CURRENT-ASSIGNMENT TARGET-EPOCH TARGET-ASSIGNMENT
order-service c4-0a92212d-5072-4067-98fa-f9842d772557 /192.168.65.1 c4 0 - - - -
order-service c5-2e9eaedc-7acc-45bf-843f-dc516387b87a /192.168.65.1 c5 0 - - - -
order-service c1-8392707f-6a46-4483-bbb7-058319c82c2f /192.168.65.1 c1 1 - orders:0 - -
order-service c3-d6d0fc96-306b-4585-a52e-c910b9a2c9d7 /192.168.65.1 c3 1 - orders:2 - -
order-service c2-d180e5ed-07e3-4b40-b19f-8a12f5a50040 /192.168.65.1 c2 1 - orders:1 - -
c4와 c5는 그룹 구성원으로 들어왔지만 #PARTITIONS가 0이에요.
파티션이 3개뿐이라 넷째, 다섯째 컨슈머에게 줄 파티션이 없는 거예요. 같은 그룹에서 동시에 일할 수 있는 컨슈머 수는 파티션 수를 넘지 못합니다.
놀고 있는 컨슈머는 파티션을 가진 컨슈머가 빠질 때 그 파티션을 이어받아요.
구성원이 바뀌면 그룹은 파티션을 처음부터 다시 나눠요. 이걸 리밸런스라고 합니다.
실습 2: 컨슈머를 늘려도 빨라지지 않을 때
파티션이 3개인 bench 토픽에 메시지 90개를 넣고, 메시지마다 0.1초씩 걸리는 컨슈머로 모두 처리하는 데 걸린 시간을 컨슈머 수별로 비교해 볼게요.
그룹 이름은 컨슈머 수마다 새로 만들어서 매번 처음부터 읽게 했어요.
kafka kafka-topics.sh --bootstrap-server kafka:9092 --create --topic bench --partitions 3 --replication-factor 1
python produce.py bench 90
측정 스크립트 bench.sh는 컨슈머 n개를 같은 그룹으로 띄우고, 90개를 다 처리하면 멈춘 뒤 컨슈머별 처리 개수와 처음과 마지막 처리 로그를 출력해요. bench.sh는 같은 디렉터리의 common.sh에서 셸 함수(kafka, kcg, ts, start)를 불러 씁니다.
먼저 컨슈머 3개일 때예요.
chmod +x bench.sh
./bench.sh 3
bench-3-1: 30 messages
bench-3-2: 30 messages
bench-3-3: 30 messages
15:01:22.785 bench-3-3 p2 offset 0
15:01:25.792 bench-3-2 p1 offset 29
다음은 컨슈머 5개일 때예요.
./bench.sh 5
bench-5-1: 30 messages
bench-5-2: 30 messages
bench-5-3: 30 messages
bench-5-4: 0 messages
bench-5-5: 0 messages
15:01:34.982 bench-5-3 p2 offset 0
15:01:38.001 bench-5-3 p2 offset 29
1개, 2개일 때도 같은 방법으로 재고, 처음과 마지막 처리 로그의 시간 차이를 비교하면 이렇습니다.
- 컨슈머 1개: 한 컨슈머가 90개를 처리했고 9.2초 걸렸어요.
- 컨슈머 2개: 60개와 30개로 나뉘었고 6.1초 걸렸어요.
- 컨슈머 3개: 30개씩 나뉘었고 3.0초 걸렸어요.
- 컨슈머 5개: 30개씩 셋이 처리하고 둘은 0개였으며 3.0초 걸렸어요.
3개까지는 늘린 만큼 빨라졌지만, 5개는 3개와 같았어요. 늘린 2개는 파티션을 받지 못해 한 개도 처리하지 않았습니다.
2개일 때는 파티션 2개를 맡은 컨슈머가 60개를 처리하는 동안 다른 하나가 먼저 끝나고 기다려서 6.1초가 걸렸어요.
컨슈머를 늘렸는데 밀린 메시지가 줄지 않는다면 컨슈머 수 말고 다른 곳에 병목이 있을 가능성이 커요. 이런 곳을 살펴보시면 됩니다.
- 파티션 수: 컨슈머가 이미 파티션 수만큼 있다면 더 늘려도 노는 컨슈머만 생겨요.
- 처리 로직: 메시지 하나를 처리하는 시간이 길면 파티션당 처리량 자체가 낮아요.
- 외부 DB나 API: 컨슈머가 기다리는 쪽이 느리다면, 컨슈머를 늘릴수록 그쪽 부하만 커질 수 있어요.
정리
범위: Kafka 4.3.1(apache/kafka:4.3.1) 단일 브로커, Python confluent-kafka 2.15.1, 파티션 3개짜리 토픽.
- 파티션 3개에 컨슈머 5개를 띄우자 c4와 c5는 그룹 구성원으로 들어왔지만
#PARTITIONS가 0이었습니다. 같은 그룹에서 동시에 일할 수 있는 컨슈머 수는 파티션 수를 넘지 못합니다. - 메시지 90개 처리 시간은 컨슈머 1개 9.2초, 2개 6.1초, 3개 3.0초, 5개 3.0초였습니다. 컨슈머를 늘리기 전에 그룹의 컨슈머 수가 이미 파티션 수와 같은지 확인해 주세요.
- 컨슈머 2개일 때는 파티션 2개를 맡은 컨슈머가 60개를 처리하는 동안 다른 하나가 먼저 끝나고 기다려서 6.1초가 걸렸습니다. 파티션이 고르게 나뉘지 않으면 가장 많이 맡은 컨슈머가 처리 시간을 정해요.