Kafka at-least-once에서 같은 주문 이벤트가 두 번 오면?

Q: Kafka at-least-once에서 같은 주문 이벤트가 두 번 오면?

그렇다. offset commit 전에 죽인 결제 컨슈머는 다시 읽어 주문 6건에 승인 7건을 냈고, 이벤트 ID를 Idempotency-Key로 보내자 승인 6건으로 끝났다.

이번 시간에는 Kafka 컨슈머가 같은 메시지를 다시 받을 때 생기는 중복을 Docker로 띄운 Kafka와 PostgreSQL, Python 컨슈머로 재현해 보며 배워가는 시간입니다. 중복이 실제 피해로 이어지는 경우를 만들어 보고, 멱등 처리 방법과 처리 경계를 프로세스를 강제로 죽여 가며 하나씩 확인해 볼게요.

예를 들어서 결제 컨슈머를 재시작했더니 고객 카드에서 같은 주문 금액이 두 번 빠져나갔다고 해볼게요.

Kafka 설정에는 문제가 없었고, 컨슈머도 처리를 끝낸 뒤 offset을 commit하도록 짜여 있었어요. 그런데도 왜 두 번 빠져나갔을까요?

먼저 말 두 개를 짚고 갈게요. offset은 파티션 안에서 메시지의 위치 번호이고, commit은 컨슈머 그룹이 어디까지 처리했는지 브로커에 기록하는 일입니다. 다시 시작한 컨슈머는 commit된 offset부터 읽어요.

원인은 이 "처리 후 commit" 방식 자체에 있었습니다. 처리와 commit 사이에 컨슈머가 죽으면 같은 메시지를 다시 받거든요.

처리 후 commit 방식은 메시지를 잃지 않는 대신 같은 메시지를 두 번 이상 전달할 수 있어요. 이것을 at-least-once 전달이라고 하고, Kafka 문서도 기본 동작을 at-least-once로 설명합니다.

그래서 컨슈머는 같은 메시지를 다시 받아도 결과가 한 번만 남도록 만들어 줘야 해요.

실습 준비

이 글의 명령은 모두 실습 코드의 kafka-at-least-once/ 디렉터리에서 실행합니다.

compose.yaml: KRaft 단일 노드 Kafka와 PostgreSQL
orders.txt: orders 토픽에 넣을 주문 확정 이벤트 6건
payapi.py: 승인 번호를 발급하고 멱등 키를 처리하는 가짜 결제 API 서버
pay.py: 이벤트마다 결제 API를 부르고 offset을 commit하는 결제 컨슈머
schema.sql: 포인트 적립 컨슈머의 wallets·processed_events 테이블
points.py: 주문 금액의 1%를 적립하고 처리 기록을 남기는 포인트 컨슈머

Docker로 KRaft 단일 노드 Kafka와 PostgreSQL을 함께 띄울게요. 두 서비스는 compose.yaml에 들어 있어요.

Kafka는 컨테이너 안에서는 kafka:9092, 호스트에서는 localhost:9501로 접속하고, PostgreSQL은 호스트의 5501 포트로 열린다고 보시면 됩니다.

컨테이너를 띄우면서 명령을 짧게 쓰기 위한 셸 함수 두 개도 만들어 놓을게요. kafka는 Kafka 컨테이너의 명령 스크립트를, sql은 PostgreSQL 컨테이너의 psql을 실행하는 함수예요.

컨슈머는 Python으로 작성하고, PostgreSQL 드라이버로는 psycopg를 씁니다.

아래 블록은 파일로 만들지 말고 실습 디렉터리에서 연 터미널에 그대로 붙여 넣어 주세요. 이후 명령은 모두 이 터미널에서 실행하고, 터미널을 새로 열면 다시 붙여 넣으시면 됩니다.

docker compose up -d

kafka() { docker compose exec -T kafka /opt/kafka/bin/"$@"; }

sql() { docker compose exec -T db psql -U app -d shop "$@"; }

python3 -m venv venv && source venv/bin/activate
pip install confluent-kafka "psycopg[binary]"

이제 파티션 1개짜리 orders 토픽에 주문 확정 이벤트 6건을 넣어 볼게요.

키는 주문 번호이고, 값에는 이벤트마다 고유한 event_id가 들어 있어요. 값이 JSON이라 :가 섞여 있어서 키 구분자는 |로 바꿨습니다. 아래가 orders.txt예요.

order-1001|{"event_id":"evt-1","order_id":"order-1001","user_id":"kim","amount":59000}
order-1002|{"event_id":"evt-2","order_id":"order-1002","user_id":"kim","amount":32000}
order-1003|{"event_id":"evt-3","order_id":"order-1003","user_id":"kim","amount":15000}
order-1004|{"event_id":"evt-4","order_id":"order-1004","user_id":"kim","amount":41000}
order-1005|{"event_id":"evt-5","order_id":"order-1005","user_id":"kim","amount":27000}
order-1006|{"event_id":"evt-6","order_id":"order-1006","user_id":"kim","amount":68000}

kafka kafka-topics.sh --bootstrap-server kafka:9092 --create --topic orders --partitions 1
kafka kafka-console-producer.sh --bootstrap-server kafka:9092 --topic orders \
  --reader-property parse.key=true --reader-property 'key.separator=|' < orders.txt
kafka kafka-console-consumer.sh --bootstrap-server kafka:9092 --topic orders --from-beginning --max-messages 6 \
  --formatter-property print.offset=true --formatter-property print.key=true 2> /dev/null

Created topic orders.
Offset:0	order-1001	{"event_id":"evt-1","order_id":"order-1001","user_id":"kim","amount":59000}
Offset:1	order-1002	{"event_id":"evt-2","order_id":"order-1002","user_id":"kim","amount":32000}
Offset:2	order-1003	{"event_id":"evt-3","order_id":"order-1003","user_id":"kim","amount":15000}
Offset:3	order-1004	{"event_id":"evt-4","order_id":"order-1004","user_id":"kim","amount":41000}
Offset:4	order-1005	{"event_id":"evt-5","order_id":"order-1005","user_id":"kim","amount":27000}
Offset:5	order-1006	{"event_id":"evt-6","order_id":"order-1006","user_id":"kim","amount":68000}

실습 1: 결제 API를 두 번 부르면

주문 상태를 paid로 바꾸는 처리라면 두 번 해도 결과가 같아요. 피해가 나는 건 외부 결제 API 호출처럼 한 번 일어나면 되돌릴 수 없는 처리입니다.

여기서는 외부 결제 API를 작은 HTTP 서버 payapi.py로 흉내 낼게요. 요청이 오면 승인 번호(ch_1, ch_2, …)를 새로 발급하고, GET 요청에는 승인 건수와 두 번 이상 승인된 주문을 돌려줘요. Idempotency-Key 헤더 처리는 뒤에서 씁니다.

결제 컨슈머 pay.py는 이벤트마다 결제 API를 부르고, 처리가 끝나면 offset을 commit해요.

두 번째 인자로 offset을 주면, 그 메시지의 결제 요청을 보낸 직후 commit하기 전에 자기 자신에게 SIGKILL을 보내 죽습니다. 그룹 이름이 -key로 끝나면 이벤트 ID를 멱등 키로 실어 보내고요.

pay.py에서는 session.timeout.ms를 6초로 줄여 두었어요. 강제로 죽은 컨슈머가 그룹에서 빠질 때까지 재시작한 컨슈머가 기다리는 시간을 줄이려는 거예요.

그럼 결제 서버를 띄우고, offset 2(order-1003)의 결제를 요청한 직후 죽인 다음 다시 실행해 볼게요.

python payapi.py &
python pay.py pay 2
python pay.py pay
curl -s localhost:9502/charges

[payapi] 승인 ch_1 order-1001 59000원
offset=0 order-1001 결제 요청 -> ch_1
[payapi] 승인 ch_2 order-1002 32000원
offset=1 order-1002 결제 요청 -> ch_2
[payapi] 승인 ch_3 order-1003 15000원
offset=2 order-1003 결제 요청 -> ch_3
offset commit 전에 프로세스 종료
[payapi] 승인 ch_4 order-1003 15000원
offset=2 order-1003 결제 요청 -> ch_4
[payapi] 승인 ch_5 order-1004 41000원
offset=3 order-1004 결제 요청 -> ch_5
[payapi] 승인 ch_6 order-1005 27000원
offset=4 order-1005 결제 요청 -> ch_6
[payapi] 승인 ch_7 order-1006 68000원
offset=5 order-1006 결제 요청 -> ch_7
{"charges": 7, "replayed": 0, "charged_twice": ["order-1003"]}

출력을 보면 주문은 6건인데 승인은 7건이에요. order-1003이 ch_3과 ch_4로 두 번 승인됐습니다.

재시작한 컨슈머는 commit된 위치인 offset 2부터 읽었고, 결제 API는 같은 주문의 두 번째 요청을 새 결제로 받은 거죠.

그럼 컨슈머의 DB에 고유 제약을 걸면 막을 수 있을까요? 이 중복은 그걸로도 막을 수 없어요.

결제는 결제사 시스템에서 일어나고, 내 DB 트랜잭션은 결제사의 승인을 취소하지 못하거든요. 결제 요청 전에 DB에 기록을 남기든 후에 남기든, 요청과 기록 사이에서 죽는 순간이 생깁니다.

실습 2: 외부 API에는 멱등 키를 보냅니다

외부 API가 멱등 키를 지원하면, 같은 키로 온 두 번째 요청은 새로 처리하지 않고 처음 응답을 그대로 돌려줘요. 많은 결제 API가 Idempotency-Key 같은 헤더로 이 기능을 제공합니다.

키로는 이벤트 ID를 쓸게요. 몇 번 다시 받아도 같은 이벤트라면 같은 값이 나오기 때문이에요.

결제 서버를 새로 띄워 승인 내역을 비우고, 그룹 pay-key로 같은 실험을 해 볼게요.

kill %1
python payapi.py &
python pay.py pay-key 2
python pay.py pay-key
curl -s localhost:9502/charges

[payapi] 승인 ch_1 order-1001 59000원
offset=0 order-1001 결제 요청 -> ch_1
[payapi] 승인 ch_2 order-1002 32000원
offset=1 order-1002 결제 요청 -> ch_2
[payapi] 승인 ch_3 order-1003 15000원
offset=2 order-1003 결제 요청 -> ch_3
offset commit 전에 프로세스 종료
[payapi] 같은 멱등 키 evt-3, 승인 없이 ch_3 응답 재사용
offset=2 order-1003 결제 요청 -> ch_3
[payapi] 승인 ch_4 order-1004 41000원
offset=3 order-1004 결제 요청 -> ch_4
[payapi] 승인 ch_5 order-1005 27000원
offset=4 order-1005 결제 요청 -> ch_5
[payapi] 승인 ch_6 order-1006 68000원
offset=5 order-1006 결제 요청 -> ch_6
{"charges": 6, "replayed": 1, "charged_twice": []}

컨슈머는 똑같이 offset 2를 다시 받아 결제를 다시 요청했어요. 그런데 결제 서버는 evt-3 키를 이미 본 적이 있어서 처음 승인 번호 ch_3을 돌려줬습니다.

여기서 눈여겨볼 건 승인이 6건이고 두 번 승인된 주문이 없다는 점이에요.

컨슈머 입장에서는 첫 요청이든 재요청이든 같은 응답을 받으니까, 그 응답으로 이어지는 처리도 그대로 진행할 수 있습니다.

멱등 키를 쓸 때는 다음을 확인해 주세요.

키 값: Kafka offset이 아니라 이벤트 ID나 주문 ID처럼 업무에서 정한 값을 써 주세요. 같은 이벤트가 다른 offset으로 다시 발행돼도 같은 키가 나와야 해요.
보관 기간: 외부 서비스는 멱등 키를 영원히 기억하지 않아요. 보관 기간보다 늦게 다시 처리되는 경우(예를 들어 offset을 크게 되돌린 재처리)에는 다시 승인될 수 있으니 API 문서에서 기간을 확인해 주세요.
같은 키, 다른 내용: 같은 키로 금액이 다른 요청을 보내면 거절하는 서비스가 많아요. 재시도할 때 요청 본문을 바꾸지 마세요.

실습 3: 처리 기록과 비즈니스 쓰기를 한 트랜잭션으로

포인트 적립처럼 현재 값에 더하는 쓰기는 쓰기 자체를 멱등하게 만들 방법이 없어요.

이때는 처리한 이벤트 ID를 따로 기록하는 테이블을 두고, 처음 보는 이벤트일 때만 적립합니다. 이 테이블을 inbox, processed_events 등으로 불러요.

실제 컨슈머는 한 문장으로 끝나지 않는 경우가 많아요. 적립 내역, 지갑 잔액, 처리 기록을 차례로 쓰고, 그 사이에 다른 로직이 끼기도 하죠.

그래서 여러 문장을 하나의 DB 트랜잭션으로 묶고, 그 트랜잭션이 커밋된 뒤에 offset을 commit합니다. 이 순서로 처리하는 컨슈머를 처리 도중에 죽여 보고, 결과가 한 번만 남는지 확인해 볼게요.

먼저 포인트 적립 컨슈머가 쓸 테이블이에요. 처리 기록에는 디버깅에 쓰도록 토픽, 파티션, offset도 함께 남겨요.

schema.sql은 wallets와 기본 키 (consumer, event_id)의 processed_events(topic, part, off, processed_at 포함)를 만들고 kim 지갑을 0점으로 넣습니다.

points.py는 주문 금액의 1%를 적립해요. 메시지마다 handle()이 처리 기록을 확인하고, 처음 보는 이벤트면 지갑 잔액을 올린 뒤 처리 기록을 넣습니다. 두 번째 인자로 트랜잭션 사용 여부를 고를 수 있어요.

tx: handle() 전체를 db.transaction() 블록 하나로 감싸요. 블록이 끝날 때 커밋합니다.
sep: 트랜잭션 없이 autocommit으로 실행해요. 문장 하나가 끝날 때마다 바로 커밋됩니다.

offset은 3건씩 받은 배치를 모두 처리한 뒤 한 번 commit해요. 처리량을 위해 흔히 쓰는 방식이죠.

세 번째 인자로는 프로세스를 죽일 지점을 골라요. after-write를 주면 offset 4(evt-5, 270점)를 처리하는 도중, 지갑 잔액을 올린 직후 처리 기록을 넣기 전에 SIGKILL을 보냅니다.

이벤트 6건을 모두 한 번씩 적립하면 590 + 320 + 150 + 410 + 270 + 680 = 2420점이 나와야 해요. 이 숫자를 기준으로 결과를 보시면 됩니다.

실험마다 지갑과 처리 기록을 비우고, 새 컨슈머 그룹으로 죽을 때까지 한 번, 재시작해서 한 번 실행한 뒤 결과를 볼게요. scenario는 scenario <group> <mode> <kill point>로 부릅니다.

sql -q -f - < schema.sql

scenario() {
  sql -q -c "UPDATE wallets SET points = 0; TRUNCATE processed_events;"
  echo "== $1 ($2, $3)"
  python points.py "$1" "$2" "$3"
  echo "-- 재시작"
  python points.py "$1" "$2"
  sql -At -c "SELECT 'points=' || points || ' processed=' || (SELECT count(*) FROM processed_events) FROM wallets"
}

먼저 트랜잭션 없이 실행하고, 지갑 잔액을 올린 직후에 죽여 볼게요.

scenario pt-sep-write sep after-write

== pt-sep-write (sep, after-write)
offset=0 evt-1 적립 590점
offset=1 evt-2 적립 320점
offset=2 evt-3 적립 150점
offset commit -> 3
offset=3 evt-4 적립 410점
SIGKILL: after-write, offset=4
-- 재시작
duplicate_skipped group=pt-sep-write event_id=evt-4 topic=orders partition=0 offset=3
offset=4 evt-5 적립 270점
offset=5 evt-6 적립 680점
offset commit -> 6
새 메시지 없음
points=2690 processed=6

2420점이어야 할 잔액이 2690점이 됐어요. evt-5의 270점이 두 번 들어간 거예요. 왜 그럴까요?

autocommit이라서 적립 UPDATE는 실행하자마자 커밋됐고, 처리 기록 INSERT는 실행되기 전에 프로세스가 죽었어요.

재시작한 컨슈머는 offset 3부터 다시 받았습니다. evt-4는 처리 기록이 있어서 건너뛰었지만, evt-5는 기록이 없어서 처음 보는 이벤트로 판단한 거죠.

이번엔 같은 코드를 tx로 실행해서 같은 지점에서 죽여 볼게요. 첫 실행의 앞부분 출력은 앞의 실험과 같아서 생략하고, 죽기 직전부터 옮겼어요.

scenario pt-tx-write tx after-write

SIGKILL: after-write, offset=4
-- 재시작
duplicate_skipped group=pt-tx-write event_id=evt-4 topic=orders partition=0 offset=3
offset=4 evt-5 적립 270점
offset=5 evt-6 적립 680점
offset commit -> 6
새 메시지 없음
points=2420 processed=6

출력은 거의 같지만 잔액은 2420점이에요.

프로세스가 죽으면서 DB 연결이 끊겼고, PostgreSQL은 커밋되지 않은 트랜잭션을 롤백했어요. 적립 UPDATE도 함께 사라졌으니, 재시작한 컨슈머가 evt-5를 처음 보는 이벤트로 처리한 게 맞는 결과입니다.

트랜잭션에 넣을 수 있는 건 같은 DB에 쓰는 일뿐이에요. 결제 API 호출, 메일 발송, 다른 토픽으로의 발행은 롤백되지 않으니 앞에서 본 멱등 키 같은 방법으로 따로 막아야 합니다.

정리

범위: Kafka 4.3.1(KRaft 단일 노드), Python confluent-kafka 2.15.1, PostgreSQL 17, 파티션 1개 토픽의 주문 이벤트 6건, 컨슈머를 지점마다 SIGKILL로 죽여 재시작.

offset commit 전에 죽인 결제 컨슈머는 offset 2부터 다시 읽어 주문 6건에 승인 7건을 냈고 order-1003이 두 번 승인됐습니다. 결제처럼 되돌릴 수 없는 외부 호출부터 중복 방어를 설계하세요.
이벤트 ID를 Idempotency-Key로 보내자 재요청은 처음 승인 번호 ch_3을 돌려받아 승인 6건, 재사용 1건으로 끝났습니다. 외부 API에는 offset이 아니라 이벤트 ID를 멱등 키로 보내고, 키 보관 기간을 API 문서에서 확인해 주세요.
처리 기록과 적립을 따로 커밋하고 적립 직후 죽이자 2420점이어야 할 잔액이 2690점이 됐고, 한 트랜잭션으로 묶자 같은 지점에서 죽여도 2420점이었습니다. 처리 기록과 비즈니스 쓰기는 같은 DB 트랜잭션으로 커밋한 뒤 offset을 commit하세요.