이 문서를 통해 아래 질문들에 답할 수 있게 됩니다.
- Idempotent Consumer 코드는 listener, transaction handler, processed table, ack를 어떻게 배치해야 하는가?
- duplicate branch, source position logging, business upsert, metric은 코드에서 어디에 드러나야 하는가?
- Spring Kafka 예시에서 어떤 부분이 누락과 중복을 가르는 핵심 지점인가?
- 같은 event를 두 번 넣는 테스트와 ack 전 crash 테스트를 어떻게 구성해야 하는가?
개요
idempotent consumer 코드는 추상 개념이 아니라 순서를 보여줘야 한다.
message를 받고, source position을 기록하고, processed table에 insert하고, business update를 commit한 뒤 ack한다.
중복 message는 exception이 아니라 duplicate branch로 빠진다.
코드 예시는 이 흐름을 숨기지 않아야 한다.
Table
CREATE TABLE processed_messages (
consumer_name varchar(100) NOT NULL,
message_id varchar(120) NOT NULL,
source_topic varchar(120) NOT NULL,
source_partition int NOT NULL,
source_offset bigint NOT NULL,
processed_at timestamp NOT NULL DEFAULT now(),
PRIMARY KEY (consumer_name, message_id)
);
CREATE UNIQUE INDEX uk_settlement_payment
ON settlements (payment_id);processed table은 duplicate branch를 만든다.
business unique index는 마지막 방어선이다.
Event
public record OrderPaidEvent(
String eventId,
String orderId,
String paymentId,
long amount,
Instant occurredAt
) {
}eventId는 idempotency key다.
paymentId는 settlement의 business unique key다.
Kafka offset은 source position으로만 기록한다.
Listener
@KafkaListener(topics = "order.events", groupId = "settlement-service")
public void listen(
ConsumerRecord<String, OrderPaidEvent> record,
Acknowledgment ack
) {
settlementConsumer.handle(record.value(), SourcePosition.from(record));
ack.acknowledge();
}listener는 ack를 handler 성공 이후에 호출한다.
handler가 예외를 던지면 ack하지 않는다.
Spring Kafka error handler가 retry 또는 DLT로 보낸다.
Transaction Handler
@Transactional
public void handle(OrderPaidEvent event, SourcePosition source) {
boolean first = processedMessages.insertIfAbsent(
"settlement-service",
event.eventId(),
source
);
if (!first) {
metrics.duplicateSkipped("settlement-service", event.eventId());
return;
}
settlements.upsert(event.paymentId(), event.orderId(), event.amount());
}처리 이력 insert와 settlement upsert는 같은 transaction이다.
duplicate branch는 정상 return이다.
ack는 이 transaction이 commit된 뒤 listener에서 수행된다.
Repository 구현
INSERT INTO processed_messages (
consumer_name,
message_id,
source_topic,
source_partition,
source_offset
)
VALUES (?, ?, ?, ?, ?)
ON CONFLICT (consumer_name, message_id) DO NOTHING;영향 row 수가 1이면 첫 처리다.
0이면 중복이다.
동시성은 DB unique constraint가 해결한다.
나쁜 예시
public void listen(OrderPaidEvent event, Acknowledgment ack) {
ack.acknowledge();
settlements.addAmount(event.orderId(), event.amount());
}ack가 먼저라 누락 위험이 있다.
addAmount는 재처리 때 금액을 누적할 수 있다.
event id와 처리 이력도 없다.
이 코드는 정상 상황에서는 통과하지만 장애 후 데이터가 틀어진다.
Error Handler
validation 오류와 일시 오류를 구분한다.
@Bean
DefaultErrorHandler kafkaErrorHandler(KafkaTemplate<Object, Object> template) {
var recoverer = new DeadLetterPublishingRecoverer(template);
var handler = new DefaultErrorHandler(recoverer, new FixedBackOff(1_000L, 3L));
handler.addNotRetryableExceptions(InvalidEventException.class);
return handler;
}retry를 늘리는 것만으로 멱등성이 생기지는 않는다.
retry가 많을수록 duplicate branch가 더 중요해진다.
Metric
코드는 metric을 남겨야 한다.
- processed count
- duplicate skipped count
- handler failure count
- ack failure count
- DLT publish count
- business upsert count
중복 skip이 증가하면 producer retry나 consumer restart를 확인한다.
business duplicate가 증가하면 멱등성이 깨진 것이다.
테스트
테스트는 정상 처리보다 장애 재현이 중요하다.
1. 같은 eventId를 두 번 넣는다.
2. settlement row가 하나인지 확인한다.
3. duplicateSkipped metric이 1 증가했는지 확인한다.
4. DB commit 후 ack 전 crash를 재현한다.
5. 재시작 후 duplicate branch로 빠지는지 확인한다.Testcontainers Kafka나 embedded Kafka를 사용할 수 있다.
핵심은 Kafka API 성공이 아니라 DB 결과다.
개인 프로젝트 기준
개인 프로젝트에서는 최소한 세 파일이 있으면 좋다.
- processed table migration
- listener와 transactional handler
- duplicate/crash 테스트
README에는 topic, group id, idempotency key, ack mode, duplicate metric을 적는다.
이 정도가 있어야 idempotent consumer 코드 예시라고 부를 수 있다.
위험 신호!
- code sample에 ack 위치가 없다.
- duplicate branch가 exception 처리로 되어 있다.
- processed table insert와 business update가 다른 transaction이다.
- source offset을 idempotency key로 쓴다.
- 테스트가 정상 처리 한 번만 검증한다.
확인 질문
확인 질문
- listener에서 ack는 transaction handler 성공 이후에 호출되는가?
- 코드 순서로 확인할 수 있어야 한다.
- 같은 event id를 두 번 넣으면 어떤 branch를 타는가?
- duplicate branch와 metric이 있어야 한다.
- DB 결과가 한 번만 남는지 어떤 constraint가 보장하는가?
- processed table primary key와 business unique key가 함께 필요하다.