이번 시간에는 consumer lag가 0인데 처리가 멈춘 경우를 직접 만들어 보고, 그룹 상태와 멤버 수, commit 증가량으로 이를 구분하는 법을 배워가는 시간입니다.
예를 들어서 새벽에 주문 처리 알림이 조용하다고 해볼게요. 모니터링 화면의 consumer lag는 0이고, 브로커도 모두 살아 있어요. 그런데 아침에 보니 밤사이 들어온 주문 일부가 처리되지 않았습니다. 어떻게 된 걸까요?
lag가 0이라는 것은 그룹이 commit한 위치가 로그의 끝과 같다는 뜻일 뿐이에요. 컨슈머가 살아 있는지, commit한 메시지를 정말 처리했는지는 말해 주지 않습니다.
lag를 계산하는 법, 추세와 파티션별 편차를 읽는 법이 더 궁금하면 consumer lag를 다룬 글을, 처리 비용이 한 파티션에 쏠리는 경우가 궁금하면 핫 파티션을 다룬 글을 참고하세요.
실습 준비
이 글의 명령은 모두 실습 코드의 kafka-monitoring/ 디렉터리에서 실행합니다.
compose.yaml: Kafka 4.3 단일 노드(JMX, 보관·스레드 설정 변경)produce.py: 정해진 속도로 정해진 시간 동안 메시지를 보내는 프로듀서consume.py: 자동 commit 컨슈머(--crash,--hang-at으로 장애 흉내)groupwatch.py: 그룹 상태, 멤버 수, 로그 끝, commit 합계를 주기적으로 찍는 스크립트
compose.yaml은 apache/kafka:4.3.1 컨테이너 하나를 띄워요. 컨테이너 안에서는 kafka:9092로, 호스트의 Python 스크립트는 localhost:9591로 접속합니다. 기본 설정과 다르게 둔 값 중 이 글에서 쓰는 것은 이래요.
KAFKA_OFFSETS_RETENTION_MINUTES: 그룹이 비고 나서 commit 기록을 얼마나 남길지 정해요. 기본값은 10080분(7일)인데, 그룹이 사라지는 과정을 보려고 1분으로 줄였습니다.KAFKA_OFFSETS_RETENTION_CHECK_INTERVAL_MS: 만료된 commit 기록을 찾는 주기예요. 같은 이유로 5초로 줄였습니다.
JMX와 스레드 수 설정은 이 글에서는 쓰지 않아요.
명령을 짧게 쓸 수 있도록 셸 함수를 만들어 놓을게요. 아래 블록은 파일로 만들지 말고 실습 디렉터리에서 연 터미널에 그대로 붙여 넣어 주세요. 이후 명령은 모두 이 터미널에서 실행하고, 터미널을 새로 열면 다시 붙여 넣으시면 됩니다.
docker compose -p kmon up -d
kafka() { docker compose -p kmon exec -T kafka /opt/kafka/bin/"$@"; }
topic() { kafka kafka-topics.sh --bootstrap-server kafka:9092 --create --topic "$@"; }
kcg() { kafka kafka-consumer-groups.sh --bootstrap-server kafka:9092 "$@"; }
python3 -m venv venv && source venv/bin/activate
pip install confluent-kafka
kcg는 kafka-consumer-groups.sh를 불러요. 그룹의 lag, 상태, 멤버를 볼 때 씁니다.
produce.py는 정해진 속도로 정해진 시간 동안 메시지를 보내고 끝나요(python produce.py 토픽 초당개수 초).
컨슈머 consume.py는 consume()으로 한 번에 최대 100개씩 받아 처리해요. --crash를 주면 close() 없이 프로세스를 끝내고, --hang-at을 주면 그 offset을 처리하다가 응답 없는 외부 API를 부른 것처럼 멈춥니다.
commit은 기본 설정인 자동 commit을 쓰고, 주기만 1초로 줄였어요. 세션 타임아웃과 poll 간격 제한도 실습 시간을 줄이려고 낮췄습니다.
groupwatch.py는 몇 초마다 그룹 상태를 한 줄로 찍어 줘요. Admin API로 그룹 상태와 멤버 수, 토픽의 로그 끝 offset 합계, 그룹의 committed offset 합계를 가져오고, 직전 측정과 비교해 초당 생산량과 초당 commit 증가량을 계산합니다. commit 기록이 없으면 -로 찍어요.
실습 1: 컨슈머가 모두 죽고 생산도 멈추면
파티션 3개짜리 orders 토픽을 billing 그룹의 컨슈머 c1, c2가 읽도록 할게요.
프로듀서는 초당 30개씩 10초 동안 300개를 보내고 멈춰요. 두 컨슈머는 25초 뒤에 close() 없이 죽습니다. 그동안 groupwatch.py로 5초마다 그룹을 봐요.
topic orders --partitions 3
python consume.py billing orders --name c1 --seconds 25 --crash > c1.txt &
python consume.py billing orders --name c2 --seconds 25 --crash > c2.txt &
sleep 5
python produce.py orders 30 10 &
python groupwatch.py billing orders 5 9
wait
cat c1.txt c2.txt
Created topic orders.
13:28:42 STABLE 멤버 2 끝 3 commit - lag -
13:28:47 STABLE 멤버 2 끝 153 commit 109 lag 44 생산 30/s
13:28:52 orders: 300개 보내고 종료
13:28:52 STABLE 멤버 2 끝 300 commit 260 lag 40 생산 29/s commit 30/s
13:28:57 STABLE 멤버 2 끝 300 commit 300 lag 0 생산 0/s commit 8/s
13:29:02 STABLE 멤버 2 끝 300 commit 300 lag 0 생산 0/s commit 0/s
13:29:07 STABLE 멤버 2 끝 300 commit 300 lag 0 생산 0/s commit 0/s
13:29:12 EMPTY 멤버 0 끝 300 commit 300 lag 0 생산 0/s commit 0/s
13:29:17 EMPTY 멤버 0 끝 300 commit 300 lag 0 생산 0/s commit 0/s
13:29:22 EMPTY 멤버 0 끝 300 commit 300 lag 0 생산 0/s commit 0/s
13:28:40 c1 할당 [0, 1]
13:29:02 c1 처리 208개
13:29:02 c1 비정상 종료
13:28:40 c2 할당 [2]
13:29:02 c2 처리 92개
13:29:02 c2 비정상 종료
출력을 보면 13:28:57부터 lag는 0이에요. 13:29:02에 컨슈머가 둘 다 죽은 뒤에도 lag는 계속 0이고, 바뀐 것은 그룹 상태뿐입니다. 왜 그럴까요?
컨슈머가 죽으면 그룹 코디네이터는 heartbeat가 session.timeout.ms(실습 10초) 동안 끊긴 뒤에 멤버를 빼요. 그래서 13:29:12에야 EMPTY, 멤버 0이 됐습니다.
kafka-consumer-groups.sh로 봐도 같아요.
kcg --describe --group billing
kcg --describe --group billing --state
Consumer group 'billing' has no active members.
GROUP TOPIC PARTITION CURRENT-OFFSET LOG-END-OFFSET LAG CONSUMER-ID HOST CLIENT-ID
billing orders 0 108 108 0 - - -
billing orders 1 100 100 0 - - -
billing orders 2 92 92 0 - - -
Consumer group 'billing' has no active members.
GROUP COORDINATOR (ID) ASSIGNMENT-STRATEGY STATE #MEMBERS
billing kafka:9092 (1) - Empty 0
LAG 열은 전부 0이고, 위에 "no active members"라는 한 줄이 붙어 있을 뿐이에요. 모니터링 시스템이 LAG 숫자만 가져간다면 이 한 줄은 어디에도 남지 않습니다.
지금 이 그룹이 처리하는 메시지는 하나도 없고, 다음 주문이 들어오는 순간부터 lag가 쌓이기 시작해요. 컨슈머가 죽은 시간부터 첫 주문이 들어오는 시간까지는 아무 경보도 울리지 않는 거죠.
이 상태를 정상적인 한가한 시간과 구분하려면 세 가지를 같이 보시면 됩니다.
- 그룹 상태와 멤버 수: 평소 멤버가 2인 그룹이
Empty, 멤버 0이면 컨슈머가 없는 거예요. - 로그 끝이 움직이는지: 생산 쪽이 멈췄는지 확인합니다.
- commit 증가량: commit이 언제 멈췄는지 확인합니다.
이 실습의 confluent-kafka 컨슈머는 위치가 바뀌지 않으면 commit을 다시 보내지 않았어요. groupwatch.py 출력에서 13:29:02, 컨슈머가 아직 살아 있던 시간에 commit 증가량이 이미 0이죠.
그래서 commit이 멈췄다는 것만으로는 컨슈머가 죽었는지, 처리할 메시지가 없어 한가한지 알 수 없어요. 로그 끝이 움직이는데 commit이 멈췄다면 컨슈머 쪽 문제이고, 둘 다 멈췄다면 그룹 상태와 멤버 수로 컨슈머가 살아 있는지를 확인해야 합니다.
실습 2: 그룹이 사라지면 lag도 사라집니다
그룹이 비면 offsets.retention.minutes가 지난 뒤 commit 기록이 지워져요. 실습에서는 1분으로 줄여 두었으니 billing 그룹을 10초마다 지켜봅니다.
python groupwatch.py billing orders 10 8
kcg --list
kcg --describe --group billing 2>&1 | grep -m1 Error
13:30:01 EMPTY 멤버 0 끝 300 commit 300 lag 0
13:30:11 EMPTY 멤버 0 끝 300 commit 300 lag 0 생산 0/s commit 0/s
13:30:21 DEAD 멤버 0 끝 300 commit - lag - 생산 0/s
13:30:31 DEAD 멤버 0 끝 300 commit - lag - 생산 0/s
13:30:41 DEAD 멤버 0 끝 300 commit - lag - 생산 0/s
13:30:51 DEAD 멤버 0 끝 300 commit - lag - 생산 0/s
13:31:01 DEAD 멤버 0 끝 300 commit - lag - 생산 0/s
13:31:11 DEAD 멤버 0 끝 300 commit - lag - 생산 0/s
Error: Executing consumer group command failed due to org.apache.kafka.common.errors.GroupIdNotFoundException: Group billing not found.
그룹이 비고 나서 이 명령을 실행하기까지 걸린 시간에 따라 DEAD 앞의 EMPTY 줄 수는 달라질 수 있어요.
그룹이 EMPTY가 된 13:29:12에서 1분 남짓 지난 13:30:21에 commit 기록이 사라졌고, 그룹은 DEAD로 바뀌었습니다. kcg --list는 아무것도 출력하지 않았고, --describe는 그룹이 없다는 오류를 냈어요.
이제 lag는 숫자 자체가 없어요. 그룹 목록을 돌며 lag를 수집하는 모니터링이라면 이 그룹의 시계열이 그냥 끊깁니다. "lag가 1000을 넘으면 알림" 같은 규칙은 값이 없으니 영영 울리지 않죠.
기본값 7일이면 주말 사이에 그룹이 사라질 일은 드물지만, 테스트용 그룹이나 가끔 도는 배치 그룹에서는 실제로 일어나요.
그래서 꼭 있어야 하는 그룹은 이름을 정해 두고, 값이 없을 때도 알림이 가게 해 주세요. Prometheus라면 absent() 같은 함수로 "이 그룹의 lag 시계열이 사라짐"을 따로 검사하면 됩니다.
실습 3: 처리 중에 멈춘 컨슈머도 lag를 0으로 만듭니다
이번에는 컨슈머가 살아 있는 경우를 볼게요.
파티션 1개짜리 events 토픽에 메시지 30개를 넣고, notify 그룹의 컨슈머 n1을 띄웁니다. n1은 offset 12를 처리하다가 응답 없는 외부 API 호출에 걸려요.
topic events --partitions 1
python produce.py events 10 3
python consume.py notify events --name n1 --hang-at 12 > n1.txt 2>&1 &
sleep 8
kcg --describe --group notify
kcg --describe --group notify --members
Created topic events.
13:20:44 events: 30개 보내고 종료
GROUP TOPIC PARTITION CURRENT-OFFSET LOG-END-OFFSET LAG CONSUMER-ID HOST CLIENT-ID
notify events 0 30 30 0 n1-778ddf4c-4c42-44ca-8a07-d2e9c56852b8 /192.168.65.1 n1
GROUP CONSUMER-ID HOST CLIENT-ID #PARTITIONS
notify n1-778ddf4c-4c42-44ca-8a07-d2e9c56852b8 /192.168.65.1 n1 1
n1은 offset 12에서 멈춰 있는데 CURRENT-OFFSET은 30, LAG는 0이에요. 멤버도 1이고 파티션도 하나 맡고 있습니다. 숫자만 보면 모든 메시지를 처리한 건강한 컨슈머처럼 보이죠.
원인은 자동 commit이 저장하는 위치에 있어요. confluent-kafka(librdkafka)는 기본 설정(enable.auto.offset.store=true)에서 메시지를 애플리케이션에 넘겨주는 순간 그다음 위치를 저장해 두고, 다음 자동 commit 때 보냅니다.
consume()이 한 번에 30개를 넘겨줬으니, offset 12에서 멈췄어도 저장된 위치는 이미 30이었던 거예요. commit된 위치는 애플리케이션에 넘겨준 위치이고, 처리를 마친 위치와 다를 수 있습니다.
처리 전에 commit한 메시지가 어떻게 유실되는지 더 궁금하면 offset과 commit을 다룬 글을 참고하세요. 이 상태에서 n1을 재시작하면 offset 12부터 29까지 18개는 처리되지 않은 채 건너뜁니다.
이어서 5초마다 그룹을 볼게요.
python groupwatch.py notify events 5 7
cat n1.txt
13:21:02 STABLE 멤버 1 끝 30 commit 30 lag 0
13:21:07 STABLE 멤버 1 끝 30 commit 30 lag 0 생산 0/s commit 0/s
13:21:12 STABLE 멤버 1 끝 30 commit 30 lag 0 생산 0/s commit 0/s
13:21:17 EMPTY 멤버 0 끝 30 commit 30 lag 0 생산 0/s commit 0/s
13:21:22 EMPTY 멤버 0 끝 30 commit 30 lag 0 생산 0/s commit 0/s
13:21:27 EMPTY 멤버 0 끝 30 commit 30 lag 0 생산 0/s commit 0/s
13:21:32 EMPTY 멤버 0 끝 30 commit 30 lag 0 생산 0/s commit 0/s
13:20:44 n1 할당 [0]
13:20:45 n1 p0 offset 12 처리 중 외부 API 응답 없음
%4|1790828475.542|MAXPOLL|n1#consumer-1| [thrd:main]: Application maximum poll interval (30000ms) exceeded by 313ms (adjust max.poll.interval.ms for long-running message processing): leaving group
n1이 멈춘 13:20:45부터 30초 동안 그룹은 STABLE이었어요. heartbeat는 librdkafka의 백그라운드 스레드가 보내기 때문에, 애플리케이션 스레드가 멈춰도 멤버로 남거든요.
max.poll.interval.ms(실습 30초, 기본 300초)를 넘기자 n1이 스스로 그룹을 떠났고, 그때부터 EMPTY가 됐습니다. 그 사이 lag는 한 번도 0에서 움직이지 않았어요.
이 경우는 그룹 상태로도 최대 max.poll.interval.ms만큼 늦게 알게 돼요. 기본값이면 5분입니다. 더 빨리 잡으려면 컨슈머 안에서 마지막으로 처리를 마친 시간 같은 처리 결과를 지표로 내보내 주세요.
정리
범위: Kafka 4.3.1 단일 노드(브로커와 컨트롤러 결합), Python confluent-kafka 2.15.1, classic 그룹 프로토콜.
- 컨슈머 둘이
close()없이 죽은 뒤에도 lag는 0이었고,session.timeout.ms10초가 지나 그룹이EMPTY, 멤버 0이 된 것만 바뀌었습니다. lag와 함께 그룹 상태와 멤버 수를 수집하세요. - 이 실습의
confluent-kafka컨슈머는 위치가 바뀌지 않으면 commit하지 않아서, 컨슈머가 살아 있던 시간에 이미 commit 증가량이 0이었습니다. commit 정체만으로 판단하지 말고 로그 끝이 움직이는지 같이 보세요. - 그룹이 비고
offsets.retention.minutes(실습 1분)가 지나자 commit 기록이 지워져 그룹이DEAD가 되고 lag 값 자체가 사라졌습니다. 꼭 있어야 하는 그룹에는 lag 시계열이 없을 때 울리는 알림을 걸어 두세요. - offset 12에서 멈춘 컨슈머도 자동 commit이 넘겨준 위치 30을 commit해서,
max.poll.interval.ms(실습 30초)를 넘길 때까지 lag 0,STABLE로 보였습니다. 마지막 처리 완료 시간을 애플리케이션 지표로 따로 내보내세요.