이번 시간에는 Kafka 컨슈머가 처리에 실패한 메시지를 그 자리에서 재시도할 때 생기는 문제를 먼저 확인하고, 재시도 토픽과 DLQ(Dead Letter Queue, 끝내 처리하지 못한 메시지를 따로 모아 두는 곳)로 실패한 메시지를 따로 빼는 구성을 직접 만들어 보며 배워가는 시간입니다.
예를 들어서 결제 컨슈머가 주문 메시지를 받아 외부 결제 API를 호출한다고 해볼게요. 결제 API가 503을 돌려주자, 컨슈머는 2초 쉬고 다시 호출하기를 다섯 번 반복했어요.
그 사이 같은 파티션에 들어온 다른 주문들은 어떻게 됐을까요? 결제 API가 멀쩡한데도 처리되지 못하고 기다렸어요. 실패한 메시지 하나를 붙잡고 있으면, 그 뒤에 있는 메시지가 모두 멈추는 거예요.
실습 준비
이 글의 명령은 모두 실습 코드의 kafka-retry-dlq/ 디렉터리에서 실행합니다.
compose.yaml: KRaft 단일 노드 Kafkapayment.py: 메시지의fail값에 따라 일시 오류(503)나 영구 오류(400)를 내는 가짜 결제 API 모듈inplace.py: 실패한 자리에서 2초 간격으로 최대 5번 재시도하는 컨슈머와 밀린 메시지 수를 재는 감시 스레드pipeline.py: 원본·재시도·DLQ 단계를 스레드 하나씩으로 실행하는 재시도 토픽 파이프라인orders.txt: 시나리오별 주문 7건
Docker로 브로커와 컨트롤러를 한 컨테이너에서 같이 실행하는 KRaft 단일 노드 Kafka를 띄울게요.
컨테이너 안의 Kafka 명령 도구는 kafka:9092로, 호스트의 Python 프로그램은 localhost:9491로 접속해요. 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://:9491
KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka:9092,HOST://localhost:9491
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_GROUP_INITIAL_REBALANCE_DELAY_MS: 0
KAFKA_HEAP_OPTS: -Xmx512m -Xms256m
ports:
- "9491:9491"
컨테이너를 띄우고 셸 함수를 만들어 둘게요.
kafka는 Kafka 컨테이너의 명령 스크립트를 실행하고, topic은 파티션 1개짜리 토픽을 만들어요.
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 1; }
python3 -m venv venv && source venv/bin/activate
pip install confluent-kafka
결제 API는 payment.py로 흉내 내요. 메시지에 들어 있는 fail 값을 보고 일부러 실패하는 모듈이에요. 실패는 두 종류로 나눴습니다.
TransientError: 잠시 뒤 다시 하면 성공할 수 있는 오류입니다.fail이 숫자면 그 횟수만큼 503을 주고 그다음 시도에서 성공합니다.fail이down이면 결제 API가 복구될 때까지(환경 변수API_RECOVERED=1) 계속 503을 줍니다.PermanentError: 몇 번을 다시 해도 결과가 같은 오류입니다.fail이invalid면 금액이 잘못됐다는 400을 줍니다.
실습 1: 그 자리에서 재시도하면 파티션이 멈춥니다
가장 먼저 떠오르는 방법은 실패한 자리에서 몇 번 더 해 보는 거죠.
inplace.py는 최대 5번 시도하고, 실패할 때마다 2초 쉬어요. 프로듀서 스레드는 0.5초마다 주문을 하나씩 모두 16건 보내고, 그중 order-3만 결제 API가 계속 503을 줍니다.
감시 스레드는 2초마다 두 위치를 읽어요. 하나는 파티션의 끝 offset(offset은 파티션 안 메시지의 위치 번호)이고, 다른 하나는 commit된 offset(컨슈머 그룹이 처리를 마쳤다고 기록한 다음 읽을 위치)이에요. 그 차이인 lag(아직 처리하지 않은 메시지 수)를 출력합니다.
감시용 컨슈머는 subscribe하지 않고 committed()로 위치만 조회해요. 그래서 그룹 멤버로 참여하지 않습니다.
topic orders-inplace
python inplace.py
Created topic orders-inplace.
0.1s offset 0 order-1 처리, 들어온 뒤 0.1s 만에
0.6s offset 1 order-2 처리, 들어온 뒤 0.1s 만에
1.1s offset 2 order-3 1번째 시도 실패: 503 Service Unavailable
2.0s [lag] 끝 offset 4, commit 2, lag 2
3.1s offset 2 order-3 2번째 시도 실패: 503 Service Unavailable
4.0s [lag] 끝 offset 8, commit 2, lag 6
5.2s offset 2 order-3 3번째 시도 실패: 503 Service Unavailable
6.0s [lag] 끝 offset 12, commit 2, lag 10
7.2s offset 2 order-3 4번째 시도 실패: 503 Service Unavailable
8.1s [lag] 끝 offset 16, commit 2, lag 14
9.3s offset 2 order-3 5번째 시도 실패: 503 Service Unavailable
9.4s offset 3 order-4 처리, 들어온 뒤 7.8s 만에
9.4s offset 4 order-5 처리, 들어온 뒤 7.4s 만에
9.5s offset 5 order-6 처리, 들어온 뒤 7.0s 만에
9.5s offset 6 order-7 처리, 들어온 뒤 6.5s 만에
9.6s offset 7 order-8 처리, 들어온 뒤 6.1s 만에
9.7s offset 8 order-9 처리, 들어온 뒤 5.6s 만에
9.7s offset 9 order-10 처리, 들어온 뒤 5.2s 만에
9.8s offset 10 order-11 처리, 들어온 뒤 4.7s 만에
9.8s offset 11 order-12 처리, 들어온 뒤 4.3s 만에
9.9s offset 12 order-13 처리, 들어온 뒤 3.8s 만에
9.9s offset 13 order-14 처리, 들어온 뒤 3.4s 만에
10.0s offset 14 order-15 처리, 들어온 뒤 2.9s 만에
10.1s offset 15 order-16 처리, 들어온 뒤 2.5s 만에
출력을 보면 order-3이 1.1초부터 9.3초까지 다섯 번 실패하는 동안, commit 위치는 2에서 움직이지 않았어요. 그 사이 주문은 계속 들어와서 lag가 2, 6, 10, 14로 늘었습니다.
order-4부터 order-16까지는 결제 API가 정상이었는데도 order-3의 재시도가 끝날 때까지 기다렸어요. 그중 order-4는 들어온 뒤 7.8초가 지나서야 처리됐습니다.
왜 그럴까요? 컨슈머는 파티션의 메시지를 offset 순서대로 하나씩 처리해요. 그래서 한 메시지를 붙잡고 있으면 같은 파티션의 뒤 메시지가 모두 기다립니다. 결제 API가 몇 분씩 장애를 겪으면 파티션 전체가 그만큼 멈추는 거예요.
그렇다고 이 방식이 늘 나쁜 건 아니에요. 1초 안팎의 짧은 재시도 한두 번으로 대부분 회복되는 오류라면, 그 자리에서 재시도하는 방식이 가장 단순하고 순서도 지켜요. 이 실습처럼 재시도가 몇 초씩 이어질 때 파티션이 멈추는 거예요.
실습 2: 재시도 토픽으로 실패한 메시지를 따로 뺍니다
파티션을 멈추지 않으려면, 실패한 메시지를 다른 토픽으로 보내고 원본 파티션은 다음 메시지로 넘어가면 돼요. 이 실습은 다음 구성을 씁니다.
orders: 원본 토픽입니다. 실패하면 일시 오류는orders-retry-1로, 영구 오류는orders-dlq로 보냅니다.orders-retry-1: 들어오고 3초 뒤에 처리합니다. 또 실패하면orders-retry-2로 보냅니다.orders-retry-2: 들어오고 6초 뒤에 처리합니다. 또 실패하면 재시도 예산(3번)을 다 쓴 것으로 보고orders-dlq로 보냅니다.orders-dlq: 자동으로 처리하지 않고 사람이 원인을 확인합니다.
Kafka에는 재시도 토픽 기능이 따로 없어요. 토픽 이름, 단계 수, 지연 시간, 어떤 오류를 어디로 보낼지는 모두 애플리케이션 코드로 정합니다.
pipeline.py는 세 단계를 스레드 하나씩으로 실행해요. 단계마다 컨슈머 그룹이 따로 있어서, 원본 토픽의 commit 위치와 재시도 토픽의 commit 위치가 서로 영향을 주지 않습니다.
pipeline.py에서 보실 부분은 세 군데예요.
forward: 실패한 메시지를 다음 토픽으로 보낼 때 키와 값은 그대로 두고, 헤더에 원본 토픽, 파티션, offset, 지금까지의 시도 횟수, 오류 종류와 내용을 적습니다. 원본 위치는 처음 실패한 단계에서 한 번 적고 다음 단계에서는 그대로 넘깁니다.- 지연: 재시도 토픽의 메시지는 토픽에 들어간 시간(메시지 타임스탬프)에 단계별 지연을 더한 시간까지 기다렸다가 처리합니다. 기다리는 동안에는 파티션을
pause하고poll을 계속 부릅니다. - commit: 다음 토픽으로 보낸 메시지가 저장됐다는 응답을
flush로 기다린 뒤에 원본을 commit합니다.
orders.txt에는 시나리오별 주문 7건을 적었어요. | 앞이 키, 뒤가 값입니다. order-2는 결제(paid)가 한 번 실패한 뒤, 배송(shipped) 이벤트가 이어서 들어와요.
order-1|{"event":"paid"}
order-2|{"event":"paid","fail":1}
order-3|{"event":"paid","fail":2}
order-2|{"event":"shipped"}
order-4|{"event":"paid","fail":"invalid"}
order-5|{"event":"paid","fail":"down"}
order-6|{"event":"paid"}
for t in orders orders-retry-1 orders-retry-2 orders-dlq; do topic $t; done
kafka kafka-console-producer.sh --bootstrap-server kafka:9092 --topic orders \
--reader-property parse.key=true --reader-property 'key.separator=|' < orders.txt
python pipeline.py 15
Created topic orders.
Created topic orders-retry-1.
Created topic orders-retry-2.
Created topic orders-dlq.
0.3s [orders] offset 0 order-1 paid 처리 완료, 1번째 시도
0.4s [orders] offset 1 order-2 일시 오류 503 Service Unavailable -> orders-retry-1 offset 0
0.4s [orders-retry-1] offset 0 order-2 3.0s 기다림, 파티션 멈춤
0.4s [orders] offset 2 order-3 일시 오류 503 Service Unavailable -> orders-retry-1 offset 1
0.5s [orders] offset 3 order-2 shipped 처리 완료, 1번째 시도
0.5s [orders] offset 4 order-4 영구 오류 400 amount must be positive -> orders-dlq offset 0
0.6s [orders] offset 5 order-5 일시 오류 503 Service Unavailable -> orders-retry-1 offset 2
0.7s [orders] offset 6 order-6 paid 처리 완료, 1번째 시도
3.4s [orders-retry-1] offset 0 order-2 paid 처리 완료, 2번째 시도 (예정보다 0.0s 늦게 시작)
4.3s [orders-retry-1] offset 1 order-3 일시 오류 503 Service Unavailable -> orders-retry-2 offset 0 (예정보다 0.8s 늦게 시작)
4.3s [orders-retry-2] offset 0 order-3 6.0s 기다림, 파티션 멈춤
4.4s [orders-retry-1] offset 2 order-5 일시 오류 503 Service Unavailable -> orders-retry-2 offset 1 (예정보다 0.7s 늦게 시작)
10.4s [orders-retry-2] offset 0 order-3 paid 처리 완료, 3번째 시도 (예정보다 0.0s 늦게 시작)
11.3s [orders-retry-2] offset 1 order-5 일시 오류 503 Service Unavailable -> orders-dlq offset 1 (예정보다 0.9s 늦게 시작)
원본 orders는 실패한 메시지를 재시도 토픽이나 DLQ로 넘기고 바로 다음 메시지로 넘어갔어요. 그래서 7건을 0.7초 안에 모두 지나갔습니다. 앞 실습과 달리 뒤 메시지가 기다리지 않은 거죠.
실패한 메시지는 이렇게 흘러갔어요.
order-2: 첫 시도에서 503,orders-retry-1에서 두 번째 시도에 성공했습니다.order-3: 두 번 실패하고orders-retry-2에서 세 번째 시도에 성공했습니다.order-4: 영구 오류라 재시도 없이 바로orders-dlq로 갔습니다.order-5: 세 번 모두 503을 받고 재시도 예산을 다 써서orders-dlq로 갔습니다.
재시도 토픽으로 빼면 순서가 깨집니다
앞의 출력을 다시 보면, order-2의 두 이벤트가 저장된 순서와 다르게 처리됐어요.
ordersoffset 1의order-2paid는 실패해서orders-retry-1로 갔고 3.4초에 처리됐습니다.ordersoffset 3의order-2shipped는 0.5초에 바로 처리됐습니다.
배송 서비스였다면 결제가 끝나기 전에 배송 처리가 먼저 된 셈이에요.
재시도 토픽은 원본 파티션을 멈추지 않는 대신, 실패한 메시지와 같은 키를 가진 뒤 메시지가 앞질러 갈 수 있게 만들어요.
정리
범위: Kafka 4.3.1(KRaft 단일 노드), Python confluent-kafka 2.15.1, 파티션 1개 토픽, payment.py로 흉내 낸 결제 API(일시 오류 503, 영구 오류 400).
- 그 자리에서 2초 간격으로 5번 재시도하는 동안 commit 위치는 2에서 멈추고 lag는 14까지 늘었으며, 결제 API가 정상이던
order-4는 7.8초를 기다렸어요. 재시도가 몇 초씩 이어지는 오류는 그 자리에서 재시도하지 마세요. - 재시도 토픽 2단계와 DLQ로 빼자 원본 7건은 0.7초 안에 지나갔고,
order-2와order-3은 재시도 단계에서 성공했으며, 영구 오류order-4와 예산 3번을 다 쓴order-5는 DLQ로 갔어요. 일시 오류는 재시도 토픽으로, 영구 오류는 바로 DLQ로 보내세요. - 재시도 토픽의 지연은 컨슈머가 메시지 타임스탬프와
pause로 지켰고, 파티션을 멈추게 한 메시지 뒤의 메시지는 0.7~0.9초 늦게 시작했어요. 정확한 시간에 실행해야 하는 작업은 재시도 토픽이 아니라 스케줄러에 맡기세요. - 실패한
order-2의paid는 재시도 토픽에서 3.4초에, 뒤에 들어온shipped는 0.5초에 처리돼 순서가 뒤집혔어요. 같은 키의 순서가 중요하면 처리할 때 이벤트의 버전이나 발생 시간을 저장된 값과 비교해서 오래된 이벤트를 무시하게 하세요.