이번 시간에는 주문 저장과 이벤트 전송이 어긋나는 실패를 재현하고, 그 문제를 푸는 outbox 테이블과 폴링 릴레이(outbox를 주기적으로 읽어 Kafka로 보내는 프로그램)를 직접 만들어 보며 배워가는 시간입니다.
예를 들어서 우리 주문 서비스가 주문을 DB에 저장하고, 이어서 OrderCreated 이벤트를 Kafka에 보낸다고 해볼게요. 코드로는 두 줄이에요.
그런데 DB 커밋은 성공했는데 Kafka 전송이 실패하면 어떻게 될까요? 주문은 있는데 배송·알림 서비스는 그 주문을 모르게 됩니다.
그럼 순서를 바꿔 Kafka에 먼저 보내면 될까요? 이번에는 DB 저장이 실패했을 때 존재하지 않는 주문의 이벤트가 이미 나가 있어요.
DB와 Kafka는 서로 다른 시스템이라 하나의 트랜잭션으로 묶이지 않기 때문이에요. 이 문제를 이중 쓰기(dual write) 문제라고 부릅니다.
outbox 패턴은 보낼 이벤트를 먼저 같은 DB의 테이블에 주문과 함께 커밋해 두는 방식으로 이 문제를 풀어요.
실습 준비
이 글의 명령은 모두 실습 코드의 kafka-outbox-pattern/ 디렉터리에서 실행합니다.
compose.yaml: Kafka(KRaft 단일 노드), PostgreSQL(wal_level=logical),cdc프로필의 Debezium Connectschema.sql: 상품·주문·outbox테이블dual_write.py: outbox 없이 주문 저장과 이벤트 전송을 두 가지 순서로 해 보는 스크립트place_order.sql: 주문 행, outbox 행, 재고 차감을 한 트랜잭션으로 하는 주문 SQLrelay.py:published_at이 빈 outbox 행을FOR UPDATE SKIP LOCKED로 잠가 Kafka로 보내는 폴링 릴레이events.sql: 릴레이 실습에 더 넣는 이벤트
Kafka는 KRaft 단일 노드로 띄우고, 호스트에서 localhost:9571로 접속해요. PostgreSQL은 호스트의 5571 포트로 엽니다. PostgreSQL의 wal_level=logical 설정과 cdc 프로필의 Debezium Connect는 이 글에서는 쓰지 않아요.
먼저 명령을 짧게 쓸 수 있는 셸 함수를 만들고 Python 의존성을 설치해 둘게요.
kafka는 Kafka 컨테이너의 명령 스크립트를, sql은 PostgreSQL 컨테이너의 psql을 실행하는 함수라고 보시면 됩니다. consume은 토픽을 처음부터 읽어 파티션, offset, 헤더, 키, 값을 한 줄씩 출력해 줘요.
docker compose up -d kafka db
kafka() { docker compose exec -T kafka /opt/kafka/bin/"$@"; }
sql() { docker compose exec -T db psql -U app -d shop "$@"; }
consume() {
kafka kafka-console-consumer.sh --bootstrap-server kafka:9092 --topic "$1" --from-beginning --timeout-ms 10000 \
--formatter-property print.partition=true --formatter-property print.offset=true \
--formatter-property print.key=true --formatter-property print.headers=true 2> /dev/null
}
python3 -m venv venv && source venv/bin/activate
pip install confluent-kafka "psycopg[binary]"
schema.sql은 products(stock >= 0 CHECK), orders, outbox 세 테이블과 published_at IS NULL인 행만 담는 부분 인덱스 outbox_unpublished를 만듭니다. 재고는 book 1개, pen 0개, cup 1개로 넣어 둬요.
sql -q -f - < schema.sql
kafka kafka-topics.sh --bootstrap-server kafka:9092 --create --topic order-events --partitions 3
kafka kafka-topics.sh --bootstrap-server kafka:9092 --create --topic outbox.event.order --partitions 3 2> /dev/null
Created topic order-events.
WARNING: Due to limitations in metric names, topics with a period ('.') or underscore ('_') could collide. To avoid issues it is best to use either, but not both.
Created topic outbox.event.order.
order-events는 outbox 없이 바로 보내는 실습에, outbox.event.order는 outbox 릴레이 실습에 씁니다.
두 번째 토픽 이름에 마침표가 들어가서 지표 이름 충돌 경고가 나오는데, 이 실습에서는 무시하셔도 됩니다.
실습 1: DB에 먼저 커밋하면 이벤트가 사라집니다
먼저 outbox 없이 주문을 저장하고 이벤트를 보내는 dual_write.py를 써 볼게요. 첫 번째 인자로 순서를 고릅니다.
db-first: 주문 트랜잭션을 커밋한 뒤 Kafka에 이벤트를 보냅니다.kafka-first: Kafka에 이벤트를 보내고 전송 확인을 받은 뒤 주문 트랜잭션을 실행합니다.
주문 트랜잭션은 주문 행을 넣고 재고를 하나 줄여요. 실패를 빨리 보려고 message.timeout.ms를 5초로 줄였습니다. librdkafka의 기본값은 300000ms(5분)예요.
이제 브로커를 멈춘 상태에서 db-first로 order-1을 주문해 볼게요. librdkafka가 연결 실패 로그를 계속 찍기 때문에 표준 오류는 버렸습니다.
docker compose stop kafka
python dual_write.py db-first order-1 book 12000 2> /dev/null
docker compose start kafka
sql -c "SELECT * FROM orders"
kafka kafka-get-offsets.sh --bootstrap-server kafka:9092 --topic order-events
DB 커밋 완료: order-1
실패: RuntimeError: Local: Message timed out
id | sku | amount
---------+------+--------
order-1 | book | 12000
(1 row)
order-events:0:0
order-events:1:0
order-events:2:0
출력을 보면 DB에는 order-1이 있어요. 그런데 order-events의 세 파티션은 모두 끝 offset이 0이라 메시지가 하나도 없습니다.
이 이벤트를 기다리던 서비스는 order-1이 생긴 줄 모르는 거예요.
전송 실패를 잡아서 재시도하면 되지 않을까 싶으시죠? 그런데 재시도는 프로세스가 살아 있을 때만 할 수 있어요.
DB 커밋 직후 프로세스가 죽거나 배포로 재시작되면, "이 이벤트를 아직 못 보냈다"는 사실도 메모리와 함께 사라집니다. 그래서 재시작한 프로세스는 무엇을 다시 보내야 하는지 몰라요.
실습 2: 이벤트를 주문과 같은 트랜잭션에 넣기
outbox 패턴은 Kafka에 바로 보내지 않아요. 보낼 이벤트를 같은 DB의 outbox 테이블에 행으로 넣고, 그 INSERT를 주문 INSERT와 같은 트랜잭션에서 커밋합니다.
Kafka로 보내는 일은 나중에 릴레이가 맡아요. 같은 DB 안의 두 테이블이니까 PostgreSQL 트랜잭션 하나로 묶을 수 있는 거죠.
outbox 테이블의 컬럼은 다음과 같습니다.
id: 이벤트 ID입니다. 한 번 정해지면 바뀌지 않고, 컨슈머가 중복을 거르는 기준이 됩니다. 여기서는 릴레이가 읽는 순서로도 씁니다.aggregate_type: 이벤트가 속한 도메인 객체의 종류(order)입니다. 어느 토픽으로 보낼지 정하는 데 씁니다.aggregate_id: 도메인 객체의 ID(order-3)입니다. Kafka 메시지의 키로 씁니다.event_type:OrderCreated,OrderPaid같은 이벤트 종류입니다.payload: 컨슈머에게 전달할 이벤트 본문(JSON)입니다.created_at: 이벤트를 만든 시간입니다. 릴레이 지연과 정리 기준에 씁니다.published_at: 릴레이가 Kafka 전송을 확인한 시간입니다. 비어 있으면 아직 보내지 않은 이벤트입니다.
주문 하나를 처리하는 SQL은 place_order.sql이에요. BEGIN~COMMIT 한 트랜잭션 안에서 :id 주문 행(cup, 8000원)을 넣고, 같은 주문의 OrderCreated 이벤트를 outbox 행으로 넣고, cup 재고를 1 줄입니다.
cup의 재고는 1개예요. 그러니 order-3은 성공하고, 이어지는 order-4는 재고 차감에서 실패해야 합니다.
sql -v id=order-3 -f - < place_order.sql
sql -v id=order-4 -f - < place_order.sql
sql -c "SELECT id FROM orders WHERE sku = 'cup'" -c "SELECT id, aggregate_id, event_type, payload, published_at FROM outbox"
BEGIN
INSERT 0 1
INSERT 0 1
UPDATE 1
COMMIT
BEGIN
INSERT 0 1
INSERT 0 1
psql:<stdin>:5: ERROR: new row for relation "products" violates check constraint "products_stock_check"
DETAIL: Failing row contains (cup, -1).
ROLLBACK
id
---------
order-3
(1 row)
id | aggregate_id | event_type | payload | published_at
----+--------------+--------------+-----------------------------------------+--------------
1 | order-3 | OrderCreated | {"amount": 8000, "order_id": "order-3"} |
(1 row)
order-4는 주문 INSERT와 outbox INSERT까지는 성공했지만 재고 차감에서 오류가 났어요. 트랜잭션이 실패한 상태라 마지막 COMMIT은 ROLLBACK으로 끝났습니다.
그래서 orders에도 outbox에도 order-4는 없고, 성공한 order-3은 주문과 이벤트가 함께 남았어요. 실습 1처럼 한쪽만 남는 상태가 생기지 않는 거죠.
여기서 outbox의 id가 1 다음에 2가 아니라 3부터 이어지는 게 보이실 텐데, 이건 정상이에요. 롤백된 order-4의 INSERT가 시퀀스에서 2를 가져갔고, PostgreSQL 시퀀스는 롤백돼도 되돌아가지 않습니다. 그래서 id에 빈 번호가 생길 수 있어요.
실습 3: 릴레이가 보내고 기록하기 전에 죽으면 같은 이벤트가 다시 갑니다
published_at이 비어 있는 행을 읽어 Kafka로 보내는 프로그램이 릴레이예요. relay.py는 다음 순서로 동작합니다.
- 트랜잭션을 열고 보내지 않은 행을
id순서로 최대--batch개 가져오면서FOR UPDATE SKIP LOCKED로 잠급니다. 다른 릴레이가 잠근 행은 기다리지 않고 건너뜁니다. - 행마다
outbox.event.<aggregate_type>토픽에aggregate_id를 키로 보냅니다. 이벤트 ID와 종류는 헤더에 싣습니다. flush()로 모든 전송 확인을 기다립니다. 하나라도 실패하면 예외를 던져 트랜잭션을 롤백합니다.published_at이 비어 있으니 다음 실행에서 다시 보냅니다.- 전송이 모두 확인되면
published_at을 기록하고 커밋합니다.
--crash-after-send를 주면 전송 확인을 받은 뒤, published_at을 기록하기 전에 자기 자신에게 SIGKILL을 보내요.
프로듀서에는 enable.idempotence를 켰어요. librdkafka는 이 값의 기본값이 false라서 직접 켜 주셔야 합니다. 켜면 전송 재시도로 생기는 중복과 순서 역전을 브로커가 막아 줘요.
자세한 동작이 더 궁금하면 멱등 프로듀서를 다룬 글을 참고하세요.
그럼 릴레이를 한 번 실행해 볼게요.
python relay.py
[relay] 잠금 id=[1]
%4|1790822520.038|GETPID|rdkafka#producer-1| [thrd:main]: Failed to acquire idempotence PID from broker localhost:9571/1: Broker: Coordinator load in progress: retrying
[relay] 전송 id=1 key=order-3 partition=2 offset=0
[relay] published_at 기록 id=[1]
[relay] 보낼 이벤트 없음
앞에서 커밋한 order-3의 이벤트(id 1)가 파티션 2의 offset 0으로 나갔어요.
중간에 보이는 GETPID 경고는 브로커를 다시 띄운 직후라서 멱등 프로듀서용 PID를 받는 데 한 번 실패하고 재시도했다는 뜻이에요. 전송에는 영향이 없었습니다.
이벤트 4건을 더 넣어 볼게요. order-3의 결제·배송 이벤트와 새 주문 order-5의 이벤트예요.
events.sql은 order-3 OrderPaid, order-5 OrderCreated, order-3 OrderShipped, order-5 OrderPaid를 이 순서로 outbox에 넣습니다.
이번에는 릴레이가 전송 확인까지 받은 뒤, published_at을 기록하기 전에 죽게 만들어 볼게요.
sql -f - < events.sql
python relay.py --crash-after-send
sql -c "SELECT id, aggregate_id, event_type, published_at IS NOT NULL AS published FROM outbox ORDER BY id"
INSERT 0 4
[relay] 잠금 id=[3, 4, 5, 6]
[relay] 전송 id=4 key=order-5 partition=1 offset=0
[relay] 전송 id=6 key=order-5 partition=1 offset=1
[relay] 전송 id=3 key=order-3 partition=2 offset=1
[relay] 전송 id=5 key=order-3 partition=2 offset=2
[relay] published_at 기록 전에 프로세스 종료
id | aggregate_id | event_type | published
----+--------------+--------------+-----------
1 | order-3 | OrderCreated | t
3 | order-3 | OrderPaid | f
4 | order-5 | OrderCreated | f
5 | order-3 | OrderShipped | f
6 | order-5 | OrderPaid | f
(5 rows)
4건 모두 Kafka에 저장됐어요. 그런데 프로세스가 죽으면서 DB 연결이 끊겨 published_at을 기록하지 못했고, 트랜잭션은 롤백됐습니다.
그러니 DB 입장에서는 4건 모두 아직 보내지 않은 이벤트예요. 릴레이를 다시 실행하고 토픽을 읽어 볼게요.
python relay.py
consume outbox.event.order
[relay] 잠금 id=[3, 4, 5, 6]
[relay] 전송 id=4 key=order-5 partition=1 offset=2
[relay] 전송 id=6 key=order-5 partition=1 offset=3
[relay] 전송 id=3 key=order-3 partition=2 offset=3
[relay] 전송 id=5 key=order-3 partition=2 offset=4
[relay] published_at 기록 id=[3, 4, 5, 6]
[relay] 보낼 이벤트 없음
Partition:2 Offset:0 id:1,eventType:OrderCreated order-3 {"amount": 8000, "order_id": "order-3"}
Partition:2 Offset:1 id:3,eventType:OrderPaid order-3 {"order_id": "order-3"}
Partition:2 Offset:2 id:5,eventType:OrderShipped order-3 {"order_id": "order-3"}
Partition:2 Offset:3 id:3,eventType:OrderPaid order-3 {"order_id": "order-3"}
Partition:2 Offset:4 id:5,eventType:OrderShipped order-3 {"order_id": "order-3"}
Partition:1 Offset:0 id:4,eventType:OrderCreated order-5 {"amount": 21000, "order_id": "order-5"}
Partition:1 Offset:1 id:6,eventType:OrderPaid order-5 {"order_id": "order-5"}
Partition:1 Offset:2 id:4,eventType:OrderCreated order-5 {"amount": 21000, "order_id": "order-5"}
Partition:1 Offset:3 id:6,eventType:OrderPaid order-5 {"order_id": "order-5"}
출력을 보면 id 3, 4, 5, 6이 각각 두 번씩, 서로 다른 offset에 들어갔어요.
프로듀서 멱등성은 왜 이 중복을 막지 못했을까요? 재시작한 릴레이는 새 프로듀서라서 새 PID를 받고, 브로커는 두 번째 전송을 새 메시지로 받기 때문이에요.
전송과 published_at 기록도 Kafka와 DB에 각각 쓰는 일이라서 둘 사이에서 죽을 수 있습니다. 기록 전에 죽은 이벤트는 다음 실행에서 다시 나가니까 잃어버리지는 않아요. 대신 중복될 수 있죠. 그래서 릴레이는 at-least-once로 전달한다고 보시면 됩니다.
결국 컨슈머는 중복을 받아도 결과가 한 번만 남게 만들어야 해요.
출력에서 눈여겨볼 건, 중복된 메시지도 헤더의 id는 같다는 점이에요. offset은 다르지만 이벤트 ID는 같으니까, 컨슈머는 이 값으로 처리 기록을 남기거나 고유 제약을 걸어서 두 번째 메시지를 건너뛰면 됩니다.
PostgreSQL에서 멱등하게 쓰는 방법과 처리 기록을 비즈니스 쓰기와 한 트랜잭션으로 묶는 방법이 더 궁금하면 at-least-once 처리를 다룬 글을 참고하세요.
정리
범위: Kafka 4.3.1 단일 브로커, PostgreSQL 17, Python confluent-kafka 2.15.1.
- 브로커를 멈춘 채 DB를 먼저 커밋한
order-1은 DB에만 남고order-events에는 메시지가 0개였어요. 전송 실패를 재시도하더라도 커밋 직후 프로세스가 죽으면 보내야 할 이벤트를 잃습니다. - 주문 INSERT, outbox INSERT, 재고 차감을 한 트랜잭션에 넣자 재고 부족으로 실패한
order-4는 주문과 이벤트가 함께 롤백됐어요. 보낼 이벤트는 비즈니스 쓰기와 같은 트랜잭션으로 outbox 테이블에 커밋하세요. - 전송 확인 뒤
published_at기록 전에 릴레이를 죽이자 재실행에서 id 3~6이 각각 두 번씩 다른 offset에 들어갔고, 재시작한 릴레이는 새 PID를 받아 멱등 프로듀서도 막지 못했어요. 컨슈머는 헤더의 이벤트 ID로 중복을 걸러 주세요.