이 문서를 통해 아래 질문들에 답할 수 있게 됩니다.
- idempotent consumer에서 idempotency key는 어떤 값이어야 하며 offset과 왜 다른가?
- 처리 이력 테이블은 어떤 column, unique key, 보관 정책을 가져야 하는가?
- 처리 이력 insert와 business update를 같은 transaction에 넣지 않으면 어떤 race가 생기는가?
- duplicate skip metric과 table cleanup을 어떻게 운영해야 하는가?
개요
idempotency key는 같은 business 입력을 식별하는 값이다.
Kafka offset이나 Redis Stream ID는 source 위치일 뿐이다.
같은 business event가 재발행되면 offset은 달라질 수 있다.
따라서 consumer는 event id 또는 command id 같은 business key로 중복을 판단해야 한다.
처리 이력 테이블은 이 key를 저장하고 duplicate branch의 근거가 된다.
Key 선택
좋은 key는 producer가 만든 stable id다.
eventId: evt-order-paid-10042-1
consumerName: settlement-service
aggregateId: order-10042
sourcePosition: kafka:order.events:3:91200eventId는 idempotency key다.
consumerName은 같은 event를 서로 다른 consumer가 독립적으로 처리할 수 있게 한다.
sourcePosition은 장애 추적 좌표다.
offset을 idempotency key로 쓰면 같은 event가 다른 offset으로 재발행될 때 중복을 놓친다.
Table Schema
CREATE TABLE processed_messages (
consumer_name varchar(100) NOT NULL,
message_id varchar(120) NOT NULL,
aggregate_id varchar(120) NOT NULL,
source_topic varchar(120),
source_partition int,
source_offset bigint,
processed_at timestamp NOT NULL DEFAULT now(),
PRIMARY KEY (consumer_name, message_id)
);primary key는 consumer별 message id다.
같은 event라도 정산 consumer와 알림 consumer는 각각 처리해야 하므로 consumer name을 key에 포함한다.
source topic과 offset은 dedup이 아니라 추적을 위해 저장한다.
Transaction 경계
처리 이력과 business update는 같은 transaction에 있어야 한다.
@Transactional
public void handle(OrderPaidEvent event, SourcePosition source) {
if (!processedMessages.insertIfAbsent("settlement-service", event.eventId(), source)) {
metrics.duplicateSkipped();
return;
}
settlementRepository.upsert(event.orderId(), event.paymentId(), event.amount());
}처리 이력만 먼저 commit하고 business update가 실패하면 이후 재전달을 duplicate로 건너뛰어 누락된다.
business update만 먼저 commit하고 처리 이력 저장이 실패하면 이후 재전달에서 중복 처리될 수 있다.
같은 transaction에 넣어야 이 둘이 함께 성공하거나 함께 실패한다.
Insert If Absent
insertIfAbsent는 DB unique constraint를 이용한다.
INSERT INTO processed_messages (consumer_name, message_id, aggregate_id)
VALUES (:consumer_name, :message_id, :aggregate_id)
ON CONFLICT (consumer_name, message_id) DO NOTHING;영향 row 수가 1이면 처음 처리다.
영향 row 수가 0이면 이미 처리한 message다.
단순 SELECT exists 후 INSERT는 동시 처리에서 race가 생길 수 있다.
Business Unique Key
처리 이력만으로는 부족하다.
target table에도 business unique key가 있어야 한다.
CREATE UNIQUE INDEX uk_settlement_payment
ON settlements (payment_id);consumer 버그나 key 설계 오류가 있어도 DB constraint가 마지막 방어선이 된다.
business unique key는 message id와 다를 수 있다.
결제 정산에서는 payment_id, 주문 요약에서는 order_id가 될 수 있다.
보관 정책
processed table은 계속 커진다.
보관 기간은 replay 가능 기간보다 길어야 한다.
Kafka retention이 7일인데 processed table을 1일만 보관하면 3일 전 event replay에서 중복을 막지 못할 수 있다.
장기 감사가 필요하면 partition table이나 archive를 고려한다.
cleanup도 batch로 실행하고 삭제량을 지표로 남긴다.
운영 지표
- processed message insert count: 첫 처리 수다.
- duplicate skip count: 중복으로 건너뛴 수다.
- unique conflict count: DB 제약 충돌 수다.
- processed table size: table 증가량이다.
- cleanup deleted count: 정리량이다.
- replay older than retention count: 보관 기간 밖 replay 시도다.
duplicate skip은 정상 지표다.
하지만 갑자기 증가하면 producer retry, consumer crash, replay 작업을 함께 확인한다.
개인 프로젝트 기준
개인 프로젝트에서는 processed table과 unique constraint를 실제 DB에 만든다.
같은 event id를 두 번 넣고 두 번째는 duplicate branch로 빠지는지 확인한다.
서로 다른 consumer name으로 같은 event를 넣으면 각각 처리되는지도 확인한다.
cleanup 정책은 간단히 “Kafka retention보다 길게 보관”이라고 문서화해도 된다.
위험 신호!
- offset을 idempotency key로 사용한다.
- consumer name 없이 message id만 primary key로 둔다.
- 처리 이력과 business update가 다른 transaction이다.
- processed table 보관 기간이 replay 가능 기간보다 짧다.
- duplicate skip metric이 없다.
확인 질문
확인 질문
- 이 consumer의 idempotency key는 producer가 만든 stable event id인가?
- source offset이 아니라 business event id여야 한다.
- 같은 event를 다른 consumer가 읽을 때 서로 방해하지 않는가?
- primary key에 consumer name이 포함되어야 한다.
- processed table을 언제까지 보관해야 replay 중복을 막을 수 있는가?
- broker retention과 운영 replay 기간보다 길어야 한다.