이번 시간에는 Kafka의 offset과 commit이 어떻게 움직이는지, 그리고 commit 시점에 따라 재시작 후 무슨 일이 생기는지를 직접 확인하며 배워가는 시간입니다.
예를 들어서 결제 컨슈머를 재배포했더니 결제 한 건이 두 번 처리됐다고 해볼게요. 반대로 장애가 난 뒤 다시 띄웠더니 메시지 하나가 처리되지 않고 지나간 경우도 있어요.
둘 다 컨슈머가 "어디까지 읽었는지"를 언제 저장했느냐에서 갈립니다.
Kafka는 메시지를 읽어도 지우지 않아요. 그 대신 컨슈머 그룹이 파티션을 어디까지 읽었는지를 번호로 저장해 두고, 컨슈머가 다시 시작하면 그 번호부터 이어서 읽게 합니다. 이 번호가 offset이고, 번호를 저장하는 일이 commit이에요.
실습 준비
이 글의 명령은 모두 실습 코드의 kafka-offset/ 디렉터리에서 실행합니다.
compose.yaml: Kafka 4.3 단일 노드 컨테이너 정의read.py: 지정한 그룹으로 메시지 N개를 받고 commit 전후의 position과 committed offset을 찍는 컨슈머worker.py: 처리 후 commit(after)이나 처리 전 commit(before)으로 처리하다 지정한 offset에서 죽는 컨슈머
먼저 Docker로 브로커와 컨트롤러를 겸하는 Kafka 4.3 한 대를 띄울게요. 컨테이너 안에서는 kafka:9092로, 호스트의 Python 컨슈머는 localhost:9431로 접속합니다. 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:9431
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:
- "9431:9094"
컨테이너를 띄우고, Kafka 명령을 짧게 쓰기 위한 셸 함수를 만들어 놓을게요. 컨슈머는 Python confluent-kafka 2.15.1로 작성했어요.
아래 블록은 파일로 만들지 말고 실습 디렉터리에서 연 터미널에 그대로 붙여 넣어 주세요. 이후 명령은 모두 이 터미널에서 실행하고, 터미널을 새로 열면 다시 붙여 넣으시면 됩니다.
docker compose up -d
kafka() { docker compose exec -T kafka /opt/kafka/bin/"$@"; }
python3 -m venv venv && source venv/bin/activate
pip install confluent-kafka
실습 1: commit은 다음에 읽을 offset을 저장합니다
컨슈머에는 위치가 두 개 있어요.
- position: 컨슈머가 다음에 받을 메시지의 offset이에요. 메시지를 받을 때마다 앞으로 갑니다.
- committed offset: 컨슈머 그룹이 브로커에 저장해 둔 위치예요. 컨슈머가 죽었다가 다시 시작하면 이 위치부터 읽습니다.
두 값이 어떻게 다른지 보기 위해 파티션 1개짜리 payments 토픽에 결제 메시지 6개를 넣고, read.py로 billing 그룹이 3개만 받아 볼게요. 키는 pay-0부터 pay-5까지이고, 파티션이 하나라서 offset도 0부터 5까지 순서대로 붙어요.
read.py는 지정한 그룹으로 메시지를 몇 개 받고, commit 전과 후의 position과 committed offset을 출력해요. 자동 commit은 끄고(enable.auto.commit: False), commit은 코드에서 직접 합니다.
kafka kafka-topics.sh --bootstrap-server kafka:9092 --create --topic payments --partitions 1
seq 0 5 | awk '{print "pay-" $1 ":" ($1+1)*1000 "원"}' | kafka kafka-console-producer.sh --bootstrap-server kafka:9092 --topic payments --reader-property parse.key=true --reader-property key.separator=:
python read.py billing 3
Created topic payments.
받음 offset=0 key=pay-0
받음 offset=1 key=pay-1
받음 offset=2 key=pay-2
commit 전 position: 3 committed: -1001
commit 후 position: 3 committed: 3
메시지 3개를 받은 뒤 position은 3이 됐지만, committed offset은 -1001이에요.
-1001은 confluent-kafka가 "값 없음"을 나타내는 특수값(OFFSET_INVALID)으로, 이 그룹이 아직 한 번도 commit하지 않았다는 뜻입니다. 이 상태에서 컨슈머가 죽으면 그룹에 저장된 위치가 없으니 auto.offset.reset: earliest에 따라 처음부터 다시 읽어요.
commit을 한 뒤에야 committed offset이 position과 같아졌습니다.
그런데 컨슈머가 마지막으로 받은 메시지는 offset 2인데, commit으로 저장된 값은 3이에요. 왜 그럴까요?
commit은 "다음에는 3부터 읽으면 된다"는 뜻이기 때문이에요. Kafka 문서도 committed offset을 애플리케이션이 다음에 소비할 메시지의 offset이라고 설명합니다.
브로커 쪽에서도 확인해 볼게요.
kafka kafka-consumer-groups.sh --bootstrap-server kafka:9092 --describe --group billing
Consumer group 'billing' has no active members.
GROUP TOPIC PARTITION CURRENT-OFFSET LOG-END-OFFSET LAG CONSUMER-ID HOST CLIENT-ID
billing payments 0 3 6 3 - - -
- CURRENT-OFFSET: 그룹이 commit한 offset이에요. 3입니다.
- LOG-END-OFFSET: 파티션에 다음 메시지가 들어갈 자리예요. 메시지가 0부터 5까지 있으니 6입니다.
- LAG: 두 값의 차이로, 아직 처리하지 않은 메시지 수예요. lag를 어떻게 읽고 언제 알릴지 더 궁금하면 consumer lag를 다룬 글을 참고하세요.
직접 commit할 때도 이 규칙을 따라야 해요. confluent-kafka의 commit(message=msg)는 받은 메시지의 offset에 1을 더한 값을 commit합니다.
offset을 숫자로 직접 넘기는 API를 쓴다면, 처리한 메시지의 offset이 아니라 그다음 번호를 넘겨 주세요. 처리한 번호를 그대로 저장하면 재시작할 때마다 마지막 메시지를 한 번 더 받게 됩니다.
앞으로 describe 출력은 필요한 열만 볼게요.
describe() { kafka kafka-consumer-groups.sh --bootstrap-server kafka:9092 --describe "$@" 2>/dev/null | awk 'NF {print $1, $3, $4, $5, $6}' | column -t; }
실습 2: 처리한 뒤 commit 전에 죽으면 다시 처리합니다
이번에는 실제 처리를 흉내 내는 worker.py를 써 볼게요. 메시지를 처리했다는 표시로 processed.log에 한 줄씩 남기고, 메시지마다 commit해요. 첫 번째 인자로 commit 순서를 고릅니다.
after: 처리한 뒤 commit해요.before: commit한 뒤 처리합니다.
두 번째 인자로 offset을 주면 그 메시지에서 os._exit(1)로 프로세스를 바로 끝내요. 예외 처리도, close()도 거치지 않으니 프로세스가 갑자기 죽은 상황과 같습니다. 10초 동안 새 메시지가 없으면 정상 종료해요.
billing 그룹은 앞에서 3을 commit했어요. 처리 후 commit하는 방식으로 실행하되, offset 4를 처리한 직후 commit하기 전에 죽게 해 볼게요.
python worker.py after 4
describe --group billing
처리 완료 offset=3
처리 완료 offset=4
offset=4 commit 전에 장애
GROUP PARTITION CURRENT-OFFSET LOG-END-OFFSET LAG
billing 0 4 6 2
컨슈머는 offset 0이 아니라 3부터 시작했어요. 새로 뜬 컨슈머는 자기 메모리에 아무 위치도 없으니까, 그룹이 마지막으로 commit한 위치에서 이어서 읽는 거예요.
offset 3을 처리한 뒤 commit한 4가 그룹에 저장된 마지막 위치예요. 이제 컨슈머를 다시 띄워 봅니다.
python worker.py after
cat processed.log
처리 완료 offset=4
처리 완료 offset=5
offset=3 key=pay-3
offset=4 key=pay-4
offset=4 key=pay-4
offset=5 key=pay-5
재시작한 컨슈머는 commit된 4부터 읽었고, pay-4는 두 번 처리됐어요.
Kafka 입장에서는 4를 다음에 읽을 위치로 기억하고 있을 뿐, 그 메시지를 이미 처리했는지는 알 방법이 없거든요. 처리 후 commit하는 방식은 메시지를 잃지 않는 대신, 처리와 commit 사이에 죽으면 같은 메시지를 다시 받습니다. 그래서 이 방식을 쓸 때는 중복을 견디게 만들어야 해요.
실습 3: 처리 전에 commit하면 건너뜁니다
이번에는 결제 메시지 3개(offset 6, 7, 8)를 더 보내고, commit한 뒤 처리하는 방식으로 실행해 볼게요. offset 7은 commit까지 마친 뒤 처리하는 도중에 죽습니다.
seq 6 8 | awk '{print "pay-" $1 ":" ($1+1)*1000 "원"}' | kafka kafka-console-producer.sh --bootstrap-server kafka:9092 --topic payments --reader-property parse.key=true --reader-property key.separator=:
python worker.py before 7
처리 완료 offset=6
offset=7 처리 중 장애
다시 띄워 봅니다.
python worker.py before
cat processed.log
describe --group billing
처리 완료 offset=8
offset=3 key=pay-3
offset=4 key=pay-4
offset=4 key=pay-4
offset=5 key=pay-5
offset=6 key=pay-6
offset=8 key=pay-8
GROUP PARTITION CURRENT-OFFSET LOG-END-OFFSET LAG
billing 0 9 9 0
재시작한 컨슈머는 8부터 시작했고, processed.log에 pay-7이 없어요. offset 7을 받자마자 8을 commit했기 때문에, Kafka는 7을 이미 끝난 메시지로 여기는 거예요.
여기서 눈여겨볼 건 LAG도 0이라는 점이에요. 모니터링에서도 문제가 보이지 않습니다.
메시지 자체는 토픽에 남아 있지만, 어느 메시지가 빠졌는지 알아채고 offset을 되돌리기 전까지는 아무도 다시 읽지 않아요.
두 방식을 비교하면 이래요.
- 처리 후 commit: 처리와 commit 사이에 죽으면 같은 메시지를 다시 처리해요(at-least-once). 중복은 처리 쪽에서 막을 수 있습니다. 방법이 더 궁금하면 at-least-once와 멱등 처리를 다룬 글을 참고하세요.
- 처리 전 commit: 처리를 끝내기 전에 죽으면 그 메시지를 건너뛰어요(at-most-once). 건너뛴 메시지는 로그만 봐서는 찾기 어렵습니다.
그래서 결제나 주문처럼 빠지면 안 되는 메시지는 처리를 끝낸 뒤 commit해 주세요.
정리
범위: Kafka 4.3.1(apache/kafka:4.3.1) 브로커 1대, Python confluent-kafka 2.15.1, 파티션 1개짜리 payments 토픽에 결제 메시지 9개.
- 메시지 3개(offset 0~2)를 받은 뒤 position은 3이었고, committed offset은 commit 전 -1001, commit 후 3이었습니다. offset을 숫자로 직접 commit할 때는 처리한 offset이 아니라 그다음 번호를 넘겨 주세요.
- 처리 후 commit하는 컨슈머를 offset 4 처리 직후 죽이자 재시작한 컨슈머가 4부터 읽어
pay-4가 두 번 처리됐습니다. 처리 후 commit을 쓴다면 중복을 견디게 만들어 주세요. - commit 후 처리하는 컨슈머를 offset 7 처리 중에 죽이자 재시작한 컨슈머가 8부터 읽어
pay-7이 빠졌고, LAG는 0이었습니다. 결제나 주문처럼 빠지면 안 되는 메시지는 처리를 끝낸 뒤 commit하세요.