이번 시간에는 consumer lag를 Kafka 브로커 한 대와 Python 프로듀서·컨슈머로 직접 만들어 측정하며 배워가는 시간입니다.
consumer lag는 컨슈머 그룹이 로그의 끝에서 얼마나 뒤처져 있는지를 나타내는 값이에요. 파티션에 쌓이는 메시지에는 위치 번호인 offset이 붙고, 컨슈머 그룹은 어디까지 처리했는지를 commit으로 브로커에 기록해요.
그런데 lag 200이 200초 늦었다는 뜻은 아니에요. lag가 어떻게 계산되는지, 같은 lag가 왜 서로 다른 지연을 뜻하는지, lag를 어떻게 읽어야 하는지 직접 만들어 볼게요.
실습 준비
이 글의 명령은 모두 실습 코드의 kafka-consumer-lag/ 디렉터리에서 실행합니다.
compose.yaml: 브로커와 컨트롤러를 겸하는 Kafka 4.3 한 대produce.py:초당 개수:초구간대로 메시지를 보내는 프로듀서consume.py: 메시지마다--ms만큼 처리를 흉내 내고 대기 시간을 재는 컨슈머lagwatch.py: Admin API로 파티션별 lag와 초당 생산량·commit 증가량을 재는 스크립트
컨테이너 안에서는 kafka:9092로, 호스트의 Python 스크립트는 localhost:9451로 접속해요. 아래가 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,EXTERNAL://:9094
KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka:9092,EXTERNAL://localhost:9451
KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER
KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT,EXTERNAL: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_GROUP_INITIAL_REBALANCE_DELAY_MS: 0
KAFKA_HEAP_OPTS: -Xmx512m -Xms256m
ports:
- "9451:9094"
이제 컨테이너를 띄우고 Kafka 명령을 짧게 쓰기 위한 셸 함수를 만들어 놓을게요.
kafka는 Kafka 컨테이너의 명령 스크립트를 실행하고, topic <이름> <파티션 수>는 토픽을 만드는 함수예요. describe는 kafka-consumer-groups.sh --describe 출력에서 그룹, 파티션, CURRENT-OFFSET, LOG-END-OFFSET, LAG 열만 남기는 함수예요.
docker compose up -d
kafka() { docker compose exec -T kafka /opt/kafka/bin/"$@"; }
topic() { kafka kafka-topics.sh --bootstrap-server kafka:9092 --create --topic "$1" --partitions "$2"; }
describe() { kafka kafka-consumer-groups.sh --bootstrap-server kafka:9092 --describe "$@" 2>/dev/null | awk 'NF {print $1, $3, $4, $5, $6}' | column -t; }
python3 -m venv venv && source venv/bin/activate
pip install confluent-kafka
consume.py는 메시지 하나를 처리하는 데 걸리는 시간을 --ms로 정하고, 그만큼 sleep해서 처리를 흉내 내요. commit은 기본 설정인 자동 commit을 그대로 쓰고, 주기만 --commit-ms로 바꿀 수 있게 했어요. 메시지마다 프로듀서가 보낸 시간(메시지 타임스탬프)부터 처리를 마친 시간까지를 대기 시간으로 재서, 끝날 때 첫 메시지와 마지막 메시지, 가장 오래 기다린 메시지의 대기 시간을 출력합니다.
lagwatch.py는 파티션마다 로그 끝 offset과 그룹의 committed offset을 가져와 빼고, 직전 측정과 비교해 초당 생산량과 초당 commit 증가량도 같이 출력해요.
실습 1: lag는 로그 끝에서 commit 위치를 뺀 값입니다
파티션 3개짜리 orders 토픽을 만들고, billing 그룹의 commit 위치를 0으로 잡아 둘게요. 그룹에 commit이 한 번도 없으면 describe의 LAG가 -로 나와서, 처음부터 숫자로 보려고 미리 만들어 두는 거예요. 그다음 메시지 600개를 보냅니다.
topic orders 3
kafka kafka-consumer-groups.sh --bootstrap-server kafka:9092 --group billing --topic orders --reset-offsets --to-earliest --execute
python produce.py orders 200:3
Created topic orders.
GROUP TOPIC PARTITION NEW-OFFSET
billing orders 0 0
billing orders 1 0
billing orders 2 0
orders: 600개 보냄
메시지 하나에 20ms가 걸리는 컨슈머를 12초 동안 돌리고, 그동안 lag를 1초마다 재 볼게요. 컨슈머가 멈춘 뒤 그룹 상태도 봅니다.
python consume.py billing orders --ms 20 --seconds 12 --progress > billing.txt &
python lagwatch.py orders 1 13 billing
wait
cat billing.txt
kafka kafka-consumer-groups.sh --bootstrap-server kafka:9092 --describe --group billing
lag 측정, 컨슈머 요약 줄, describe 결과를 차례로 옮겼어요(발췌).
15:22:33 billing lag 600 p0=208 p1=191 p2=201
15:22:37 billing lag 600 생산 0/s commit 0/s p0=208 p1=191 p2=201
15:22:38 billing lag 413 생산 0/s commit 185/s p0=21 p1=191 p2=201
15:22:42 billing lag 413 생산 0/s commit 0/s p0=21 p1=191 p2=201
15:22:43 billing lag 223 생산 0/s commit 190/s p0=0 p1=22 p2=201
15:22:45 billing lag 148 생산 0/s commit 75/s p0=0 p1=0 p2=148
c1 처리 452개, 대기 시간 첫 메시지 3.2초 / 마지막 14.3초 / 최대 14.3초
Consumer group 'billing' has no active members.
GROUP TOPIC PARTITION CURRENT-OFFSET LOG-END-OFFSET LAG CONSUMER-ID HOST CLIENT-ID
billing orders 0 208 208 0 - - -
billing orders 1 191 191 0 - - -
billing orders 2 53 201 148 - - -
각 열의 뜻은 이렇습니다.
- CURRENT-OFFSET: 그룹이 이 파티션에 commit한 offset이에요. 다음에 읽을 위치입니다.
- LOG-END-OFFSET: 로그의 끝이에요. 정확히는 컨슈머가 읽을 수 있는 끝인 high watermark예요. 브로커가 한 대인 이 실습에서는 둘이 같습니다.
- LAG: LOG-END-OFFSET에서 CURRENT-OFFSET을 뺀 값이에요.
파티션 2를 직접 계산하면 201 − 53 = 148이에요. CURRENT-OFFSET을 더하면 208 + 191 + 53 = 452로, 컨슈머가 처리했다고 출력한 452개와 맞아요.
lag 측정을 보면 컨슈머는 꾸준히 처리했는데, lag는 5초 동안 600에 머물다가 15:22:38에 한 번에 187이 줄고, 다시 5초 뒤인 15:22:43에 190이 줄었어요. 왜 그럴까요?
자동 commit 주기인 auto.commit.interval.ms의 기본값이 5000ms이기 때문이에요. lag는 실제로 처리한 위치가 아니라 commit된 위치로 계산해서, 처리가 진행돼도 다음 commit 전까지는 줄지 않습니다. 마지막 15:22:45의 감소는 컨슈머가 close()하면서 남은 위치를 commit한 거예요.
또 컨슈머는 하나였는데 파티션 0과 1은 끝까지 읽었고 파티션 2만 남았어요. 컨슈머 하나가 여러 파티션을 맡으면 고르게 읽는다는 보장이 없으니, lag는 합계와 함께 파티션별 값을 봐야 해요.
commit과 CURRENT-OFFSET이 왜 "다음에 읽을 위치"인지가 더 궁금하면 offset과 commit을 다룬 글을 참고하세요. 이후 실습은 변화를 촘촘하게 보려고 commit 주기를 1초(--commit-ms 1000)로 줄여서 진행할게요.
실습 2: lag 200이 200초 늦었다는 뜻은 아닙니다
파티션 1개짜리 토픽 lag-a와 lag-b를 만들고, 그룹 a와 b의 commit 위치를 0으로 잡아 둘게요. 두 토픽에 메시지를 200개씩 보내되 속도를 다르게 합니다.
lag-a: 초당 10개씩 20초 동안 보내요.lag-b: 19초 동안 쉬다가 마지막 1초에 초당 200개로 보내요.
topic lag-a 1
topic lag-b 1
kafka kafka-consumer-groups.sh --bootstrap-server kafka:9092 --group a --topic lag-a --reset-offsets --to-earliest --execute > /dev/null
kafka kafka-consumer-groups.sh --bootstrap-server kafka:9092 --group b --topic lag-b --reset-offsets --to-earliest --execute > /dev/null
python produce.py lag-a 10:20 &
python produce.py lag-b 0:19 200:1 &
wait
describe --group a --group b
Created topic lag-a.
Created topic lag-b.
lag-a: 200개 보냄
lag-b: 200개 보냄
GROUP PARTITION CURRENT-OFFSET LOG-END-OFFSET LAG
a 0 0 200 200
GROUP PARTITION CURRENT-OFFSET LOG-END-OFFSET LAG
b 0 0 200 200
두 그룹 모두 lag 200이에요. 이제 똑같이 메시지 하나에 20ms가 걸리는 컨슈머를 동시에 띄워 200개를 처리해 볼게요.
python consume.py a lag-a --ms 20 --max 200 --name A &
python consume.py b lag-b --ms 20 --max 200 --name B &
wait
두 컨슈머의 출력이 섞여 나와서 마지막 요약 줄만 옮겼어요.
B 처리 200개, 대기 시간 첫 메시지 7.0초 / 마지막 11.2초 / 최대 11.2초
A 처리 200개, 대기 시간 첫 메시지 26.0초 / 마지막 11.3초 / 최대 26.0초
같은 lag 200이었는데 A의 첫 메시지는 26.0초, B의 첫 메시지는 7.0초를 기다렸어요. 초당 10개가 들어오는 토픽에서 200개가 밀렸다면 20초어치가 밀린 거고, 초당 200개가 들어오는 토픽에서는 1초어치예요.
마지막 메시지는 양쪽 모두 11초 남짓 기다렸어요. 두 컨슈머 모두 20ms짜리 처리를 200번 해야 했기 때문이에요. 그래서 lag를 시간으로 바꾸려면 두 가지 속도를 알아야 해요.
- 들어오는 속도: 지금 쌓인 lag가 몇 초 동안 들어온 양인지 정해요. 가장 오래 기다린 메시지의 나이가 여기서 나옵니다.
- 처리 속도: 밀린 메시지를 다 처리하는 데 몇 초가 걸릴지 정해요.
실습 3: 숫자보다 추세를 봅니다
파티션 1개짜리 trend 토픽에 초당 50개씩 40초 동안 보내고, 두 그룹이 같은 토픽을 읽게 해 볼게요. 두 그룹 모두 처음 10초는 컨슈머가 없어서 lag가 쌓이고, 10초 뒤에 컨슈머를 띄웁니다.
fast: 메시지 하나에 5ms가 걸리는 컨슈머예요.slow: 메시지 하나에 30ms가 걸리는 컨슈머예요.
topic trend 1
for g in fast slow; do kafka kafka-consumer-groups.sh --bootstrap-server kafka:9092 --group $g --topic trend --reset-offsets --to-earliest --execute > /dev/null; done
python produce.py trend 50:40 &
( sleep 10; python consume.py fast trend --ms 5 --commit-ms 1000 --seconds 30 --name fast > fast.txt ) &
( sleep 10; python consume.py slow trend --ms 30 --commit-ms 1000 --seconds 30 --name slow > slow.txt ) &
python lagwatch.py trend 5 9 fast slow
wait
Created topic trend.
15:23:59 fast lag 4
15:23:59 slow lag 5
15:24:04 fast lag 254 생산 50/s commit 0/s
15:24:04 slow lag 255 생산 50/s commit 0/s
15:24:09 fast lag 504 생산 50/s commit 0/s
15:24:09 slow lag 504 생산 50/s commit 0/s
15:24:14 fast lag 39 생산 50/s commit 143/s
15:24:14 slow lag 616 생산 50/s commit 28/s
15:24:19 fast lag 1 생산 50/s commit 58/s
15:24:19 slow lag 754 생산 50/s commit 22/s
15:24:24 fast lag 49 생산 50/s commit 40/s
15:24:24 slow lag 864 생산 50/s commit 28/s
15:24:29 fast lag 47 생산 50/s commit 50/s
15:24:29 slow lag 974 생산 50/s commit 28/s
15:24:34 fast lag 45 생산 50/s commit 50/s
15:24:34 slow lag 1086 생산 50/s commit 28/s
trend: 2000개 보냄
15:24:39 fast lag 39 생산 49/s commit 51/s
15:24:39 slow lag 1168 생산 49/s commit 33/s
15:24:09에 두 그룹의 lag는 504로 같았어요. 그 뒤는 이렇게 갈렸습니다.
fast: 5초 만에 39로 줄었고, 그 뒤로는 50 안팎에서 더 늘지 않아요. 남은 50 안팎은 commit 주기 1초 동안 처리한 메시지가 아직 commit되지 않아 보이는 값이에요.slow: 5초마다 110 안팎씩 계속 늘었어요. 초당 50개가 들어오는데 28개 안팎만 처리하니 1초에 22개씩 쌓이는 거죠.
그래서 lag는 한 번 잰 숫자보다 기울기로 판단해야 해요. 줄어들고 있으면 따라잡는 중이고, 일정하게 머물면 들어오는 만큼 처리하고 있고, 계속 늘어나면 처리 속도가 들어오는 속도를 따라가지 못하고 있는 거예요.
정리
범위: Kafka 4.3.1 브로커 한 대, Python confluent-kafka 2.15.1, 자동 commit, 메시지 처리 시간은 sleep으로 흉내 낸 값.
- lag는 LOG-END-OFFSET에서 그룹이 commit한 offset을 뺀 메시지 개수입니다. 파티션 2는 201 − 53 = 148이었고, 다른 파티션은 0이었으니 그룹 합계와 함께 파티션별 lag를 보세요.
- 컨슈머가 꾸준히 처리하는 동안 lag는 600에 5초 머물다가 commit 때 187, 190씩 줄었습니다. lag에는 자동 commit 주기(기본 5000ms)만큼 이미 처리한 메시지가 섞여 있다고 보고 읽어 주세요.
- 같은 lag 200에서 초당 10개가 들어온 A의 첫 메시지는 26.0초, 초당 200개가 들어온 B의 첫 메시지는 7.0초를 기다렸습니다. lag 경보 기준을 토픽마다 같은 숫자로 두지 말고 유입 속도로 시간을 환산해 보세요.
- 15:24:09에 lag 504로 같던 두 그룹이 5초 뒤
fast는 39가 됐고 30초 뒤slow는 1168이 됐습니다. lag는 한 번 잰 숫자가 아니라 기울기로 판단하세요.