05. Spring Kafka Producer Consumer 상세

이 문서를 통해 아래 질문들에 답할 수 있게 됩니다.

• Spring Kafka producer와 consumer 코드는 어떤 데이터 계약을 드러내야 하는가?
• KafkaTemplate, @KafkaListener, manual ack, error handler를 어떤 기준으로 설정하는가?
• DB transaction, Kafka publish, consumer idempotency는 어디서 만나는가?
• 개인 프로젝트에서 Spring Kafka를 붙였다고 말하려면 어떤 테스트가 있어야 하는가?

개요

Spring Kafka는 Kafka 사용을 쉽게 시작하게 해준다.

하지만 KafkaTemplate.send()와 @KafkaListener만으로 처리 보장이 생기지는 않는다.

producer는 어떤 topic에 어떤 key로 어떤 version의 event를 보낼지 명확히 해야 한다.

consumer는 언제 business 처리가 완료됐다고 보고 offset을 commit할지 명확히 해야 한다.

운영 품질은 코드 annotation보다 key, payload, transaction, ack, retry, DLT, metric에서 나온다.

Producer 계약

producer는 event 이름과 key를 명시적으로 다룬다.

@Service
public class OrderEventProducer {
    private final KafkaTemplate<String, OrderPaidEvent> kafkaTemplate;
 
    public void publish(OrderPaidEvent event) {
        kafkaTemplate.send("order.events", event.orderId(), event);
    }
}

여기서 event.orderId()는 partition key다.

주문별 상태 전이 순서를 지키려는 의도가 코드에 드러난다.

key를 빼고 보내면 round-robin에 가까워지고 주문별 순서를 기대하기 어렵다.

event payload도 command와 구분한다.

public record OrderPaidEvent(
    String eventId,
    String orderId,
    String paymentId,
    long amount,
    int version,
    Instant occurredAt
) {
}

eventId는 dedup key다.

version은 consumer가 schema 변경을 판단하는 기준이다.

occurredAt은 event 발생 시각이고, publish 시각과 다를 수 있다.

Transaction 경계

서비스 method 안에서 DB 저장 후 곧바로 Kafka를 발행하면 dual write 문제가 생긴다.

@Transactional
public void pay(OrderPayCommand command) {
    Order order = orderRepository.getById(command.orderId());
    order.markPaid(command.paymentId());
 
    kafkaTemplate.send("order.events", order.id(), OrderPaidEvent.from(order));
}

DB commit과 Kafka publish는 하나의 원자적 transaction으로 묶이지 않는다.

이 코드는 쉬워 보이지만 장애 순간 한쪽만 성공할 수 있다.

중요 이벤트라면 outbox pattern을 검토한다.

Kafka producer 문서에서도 이 경계를 숨기면 안 된다.

Consumer 기본형

consumer는 record key와 payload를 함께 받는 편이 좋다.

@KafkaListener(topics = "order.events", groupId = "settlement-service")
public void consume(
    ConsumerRecord<String, OrderPaidEvent> record,
    Acknowledgment ack
) {
    settlementEventHandler.handle(record.key(), record.value());
    ack.acknowledge();
}

manual ack를 쓰면 business 처리 이후 offset commit을 명시할 수 있다.

이때 listener container의 ack mode도 맞춰야 한다.

spring:
  kafka:
    listener:
      ack-mode: manual
    consumer:
      enable-auto-commit: false

auto commit을 켜둔 채 business transaction을 따로 처리하면 누락을 만들기 쉽다.

Idempotent Consumer

consumer handler는 중복 입력을 기본으로 받아들인다.

@Transactional
public void handle(String key, OrderPaidEvent event) {
    if (!processedEventRepository.insertIfAbsent(event.eventId())) {
        metrics.duplicateSkipped("order.events", "settlement-service");
        return;
    }
 
    settlementRepository.upsert(
        event.orderId(),
        event.paymentId(),
        event.amount()
    );
}

insertIfAbsent는 DB unique constraint로 구현한다.

중복을 if문과 memory cache만으로 막으면 재시작이나 병렬 처리에서 깨진다.

business update와 처리 이력 기록은 같은 transaction에 둔다.

Error Handler와 DLT

listener에서 모든 exception을 잡아 logging만 하고 ack하면 데이터가 누락된다.

Spring Kafka error handler는 retry와 DLT 경로를 명시해야 한다.

@Bean
DefaultErrorHandler kafkaErrorHandler(KafkaTemplate<Object, Object> template) {
    var recoverer = new DeadLetterPublishingRecoverer(
        template,
        (record, ex) -> new TopicPartition(record.topic() + ".dlt", record.partition())
    );
 
    var handler = new DefaultErrorHandler(recoverer, new FixedBackOff(1_000L, 3L));
    handler.addNotRetryableExceptions(InvalidEventVersionException.class);
    return handler;
}

validation 오류는 재시도해도 성공하지 않으므로 DLT로 빠르게 보내는 편이 낫다.

DB deadlock이나 일시 timeout은 제한된 retry 대상이다.

테스트 기준

Spring Kafka 코드는 정상 발행 테스트만으로 부족하다.

다음 테스트가 있어야 처리 보장을 말할 수 있다.

• producer가 같은 orderId를 key로 보내는지 확인한다.
• consumer 처리 중 DB 예외가 나면 ack되지 않는지 확인한다.
• 같은 event가 두 번 들어와도 settlement row가 한 번만 반영되는지 확인한다.
• invalid version event가 DLT로 가는지 확인한다.
• consumer 재시작 후 committed offset부터 이어 읽는지 확인한다.

운영 지표

• producer send failure count: 발행 실패를 본다.
• record send latency: producer 지연을 본다.
• consumer lag by group: 소비 지연을 본다.
• listener processing time: handler 처리 시간을 본다.
• dlt publish count: 최종 실패 격리를 본다.
• duplicate skip count: 멱등 처리 작동을 본다.

위험 신호!

• producer key 없이 topic에만 send한다.
• auto commit과 manual business transaction을 섞는다.
• listener에서 exception을 삼키고 정상 처리로 ack한다.
• event version과 dedup key가 payload에 없다.
• DLT topic은 있지만 재처리 방법과 metric이 없다.

확인 질문

확인 질문

• 이 producer는 어떤 business key로 partition을 선택하는가?
  • 코드의 send(topic, key, payload)에서 확인할 수 있어야 한다.
• 이 consumer는 business transaction이 끝난 뒤 offset을 commit하는가?
  • ack mode, auto commit 설정, handler transaction을 함께 봐야 한다.
• 같은 event를 두 번 받았을 때 어떤 DB 제약이 중복 side effect를 막는가?
  • 처리 이력 table과 unique constraint가 답이어야 한다.

참고 문서

• Spring for Apache Kafka Reference
• Apache Kafka Documentation
• Testcontainers Kafka Module