Kafka retry와 dead letter topic (Spring Kafka)

1. 개요

Kafka consumer는 틀림없이 메시지 처리 오류를 만나게 됩니다. 일시적인 오류도 있고(예: 결제 게이트웨이 타임아웃), 영구적인 오류도 있습니다(예: 잘못된 데이터 형식). 제대로 처리하지 않으면 메시지 손실이나 consumer가 절대 성공하지 않을 메시지에서 영구적으로 멈추는 두 가지 상황 중 하나에 직면하게 됩니다.

이 글은 두 가지 유형으로 Spring Kafka에서 retry를 구성하는 방법을 안내합니다: DefaultErrorHandler를 사용한 차단 retry와 @RetryableTopic을 사용한 비차단 retry. 재시도 횟수를 초과한 오류 메시지는 dead letter topic(DLT)으로 전송됩니다. 읽은 후 어떤 유형을 언제 선택할지 알 수 있고, 동작을 검증하는 테스트를 작성할 수 있습니다.

2. 셋업

Spring Kafka dependency를 추가하세요. Spring Boot 3.x는 버전을 자동으로 관리하므로 명시할 필요가 없습니다:

<dependency>
    <groupId>org.springframework.kafka</groupId>
    <artifactId>spring-kafka</artifactId>
</dependency>
<dependency>
    <groupId>org.springframework.kafka</groupId>
    <artifactId>spring-kafka-test</artifactId>
    <scope>test</scope>
</dependency>

application.properties에서 최소 구성:

spring.kafka.bootstrap-servers=localhost:9092
spring.kafka.consumer.group-id=order-service
spring.kafka.consumer.auto-offset-reset=earliest

Boot는 기본값으로 StringSerializer와 StringDeserializer를 사용합니다. 예제를 간단히 하기 위해 메시지의 value는 String 형식의 주문 코드입니다.

3. 오류를 발생시킬 수 있는 Consumer

다음은 결제를 처리하는 서비스입니다. 잘못된 형식의 주문 코드는 영구적인 오류입니다. 결제 게이트웨이 호출 오류는 일시적인 오류입니다:

public class PaymentGatewayException extends RuntimeException {
    public PaymentGatewayException(String message) {
        super(message);
    }
}

@Service
public class PaymentService {

    public void charge(String orderId) {
        if (orderId == null || !orderId.startsWith("ORD-")) {
            throw new IllegalArgumentException("유효하지 않은 주문 코드: " + orderId);
        }
        // 결제 게이트웨이 호출; 타임아웃 시 PaymentGatewayException을 발생시킬 수 있음
    }
}

Listener는 단순히 서비스를 호출합니다:

@Component
public class OrderListener {

    private static final Logger log = LoggerFactory.getLogger(OrderListener.class);
    private final PaymentService paymentService;

    public OrderListener(PaymentService paymentService) {
        this.paymentService = paymentService;
    }

    @KafkaListener(topics = "orders")
    public void onOrder(String orderId) {
        log.info("주문 {} 처리 중", orderId);
        paymentService.charge(orderId);
    }
}

아무것도 설정하지 않으면 Spring Kafka는 DefaultErrorHandler를 FixedBackOff(0, 9)로 사용합니다. 오류 메시지는 9번 재시도되어 총 10번 시도됩니다. 그 후 handler는 로그만 남기고 무시합니다. 메시지는 손실된 것입니다. 이것이 다시 구성해야 하는 이유입니다.

4. DefaultErrorHandler를 사용한 차단 retry

4.1. Backoff 및 DLT 설정

Listener가 exception을 발생시키면 DefaultErrorHandler는 오류 메시지의 offset을 다시 구성하여 다음 poll에서 다시 읽습니다. 재시도 횟수가 초과되면 handler는 ConsumerRecordRecoverer를 호출합니다. 여기서 recoverer는 DeadLetterPublishingRecoverer입니다:

@Configuration
public class KafkaErrorConfig {

    @Bean
    public DefaultErrorHandler errorHandler(KafkaTemplate<Object, Object> template) {
        var recoverer = new DeadLetterPublishingRecoverer(template);

        var backOff = new ExponentialBackOffWithMaxRetries(3);
        backOff.setInitialInterval(1_000L);
        backOff.setMultiplier(2.0);
        backOff.setMaxInterval(10_000L);

        var handler = new DefaultErrorHandler(recoverer, backOff);
        handler.addNotRetryableExceptions(IllegalArgumentException.class);
        return handler;
    }
}

각 부분 설명:

• DeadLetterPublishingRecoverer(template)은 오류 메시지를 <토픽명>.DLT 토픽으로 전송하며, 원본 메시지와 같은 파티션에 저장됩니다. orders 토픽의 경우 DLT는 orders.DLT입니다.
• ExponentialBackOffWithMaxRetries(3)은 최대 3번 재시도합니다. 대기 시간은 각각 1초, 2초, 4초이며 10초를 초과하지 않습니다.
• addNotRetryableExceptions는 재시도가 필요 없는 exception을 선언합니다. IllegalArgumentException은 DLT로 직접 가는데, 아무리 재시도해도 잘못된 데이터는 계속 잘못되었기 때문입니다.

Spring Boot는 자동으로 bean CommonErrorHandler를 기본 container factory에 연결합니다. ConcurrentKafkaListenerContainerFactory를 직접 만들 필요가 없습니다.

ORD-1을 전송하고 결제 게이트웨이가 계속 타임아웃되면 로그는 다음과 같이 보일 것입니다:

10:00:00.010 INFO  주문 ORD-1 처리 중
10:00:01.020 INFO  주문 ORD-1 처리 중
10:00:03.030 INFO  주문 ORD-1 처리 중
10:00:07.040 INFO  주문 ORD-1 처리 중
10:00:07.060 ERROR Backoff ... exhausted for orders-0@42

총 4번의 처리: 1번 초기 시도와 3번 재시도. 그 후 메시지는 orders.DLT에 있습니다.

4.2. DLT의 Header

DLT의 메시지는 원래의 key와 value를 유지합니다. Recoverer는 원인 조사를 위해 header를 추가합니다:

kafka_dlt-original-topic      = orders
kafka_dlt-original-partition  = 0
kafka_dlt-original-offset     = 42
kafka_dlt-exception-fqcn      = org.springframework.kafka.listener.ListenerExecutionFailedException
kafka_dlt-exception-cause-fqcn = com.example.PaymentGatewayException
kafka_dlt-exception-message   = Listener method ... threw exception
kafka_dlt-exception-stacktrace = ...

원본 exception은 kafka_dlt-exception-cause-fqcn에 있습니다. 코드로 header를 읽어야 하면 KafkaHeaders의 상수를 사용하세요. 예: KafkaHeaders.DLT_ORIGINAL_TOPIC.

주의: Recoverer가 기본값으로 같은 파티션으로 전송하므로 orders.DLT는 최소한 orders와 같은 수의 파티션을 가져야 합니다. 그렇지 않으면 publish가 실패합니다. bean NewTopic으로 DLT를 만들 때 파티션 수를 같게 설정할 수 있습니다. 다른 방법은 new TopicPartition(r.topic() + ".DLT", -1)을 반환하는 destination resolver를 전달하여 Kafka가 파티션을 자동 선택하도록 하는 것입니다.

4.3. 차단 retry가 적절한 경우

차단 retry는 메시지 순서를 유지합니다. 이후 메시지는 이전 메시지가 성공하거나 DLT로 이동한 후에만 처리됩니다. 이는 순서가 중요한 경우에 올바른 선택입니다. 예: 같은 주문의 OrderCreated → OrderPaid → OrderShipped 이벤트.

대가로 전체 파티션이 대기 중에 차단됩니다. Spring Kafka 2.9 이상에서는 handler가 backoff를 기다리는 동안 max.poll.interval.ms를 초과하지 않으므로 seek 및 pause 메커니즘 덕분입니다. 그럼에도 backoff 합계가 몇 분이면 lag이 급증할 것입니다.

5. @RetryableTopic을 사용한 비차단 retry

5.1. Annotation을 사용한 설정

비차단 retry에서는 오류 메시지가 별도의 retry topic으로 이동합니다. 덕분에 consumer는 다음 메시지 처리를 즉시 계속합니다. 다른 consumer는 정의된 backoff 후 retry topic을 읽습니다:

@Component
public class OrderRetryListener {

    private static final Logger log = LoggerFactory.getLogger(OrderRetryListener.class);
    private final PaymentService paymentService;

    public OrderRetryListener(PaymentService paymentService) {
        this.paymentService = paymentService;
    }

    @RetryableTopic(
            attempts = "4",
            backoff = @Backoff(delay = 1_000, multiplier = 2.0, maxDelay = 10_000),
            exclude = IllegalArgumentException.class)
    @KafkaListener(topics = "payments")
    public void onPayment(String orderId,
                          @Header(KafkaHeaders.RECEIVED_TOPIC) String topic) {
        log.info("{} 처리 중 ({}에서)", orderId, topic);
        paymentService.charge(orderId);
    }

    @DltHandler
    public void onDlt(String orderId,
                      @Header(KafkaHeaders.EXCEPTION_MESSAGE) String error) {
        log.error("주문 {}이 DLT로 이동: {}", orderId, error);
    }
}

핵심 포인트:

• attempts = "4"는 총 4번 처리를 의미합니다: 1번 원본 토픽 + 3번 retry 토픽. 값은 String이므로 "${app.retry.attempts}" 같은 placeholder를 지원합니다.
• @Backoff는 org.springframework.retry.annotation 패키지에서 옵니다. Spring Retry는 이미 spring-kafka에 포함되어 있습니다.
• exclude는 DLT로 직접 가는 exception을 나열합니다. addNotRetryableExceptions와 동일합니다.
• @DltHandler를 같은 클래스에 배치하세요. 이곳이 로그, 알림 전송 또는 수동 처리를 위한 테이블 저장 등을 하는 곳입니다.

5.2. 생성되는 Topic

점진적인 backoff로 기본값은 각 retry마다 고유한 topic이 있으며, 지연 간격으로 suffix를 붙입니다:

payments
payments-retry-1000
payments-retry-2000
payments-retry-4000
payments-dlt

Spring Kafka는 시작할 때 이 topic들을 자동으로 생성합니다 (autoCreateTopics = "true"는 기본값). 프로덕션 환경에서는 보통 broker에서 auto-create를 비활성화합니다. 이 경우 올바른 이름과 파티션 수로 미리 topic을 생성해야 합니다. 지연 대신 인덱스를 사용하려면 topicSuffixingStrategy = TopicSuffixingStrategy.SUFFIX_WITH_INDEX_VALUE를 설정하세요.

Suffix 주의: @RetryableTopic은 -dlt를 사용하고 DeadLetterPublishingRecoverer는 .DLT를 사용합니다. 모니터링 설정 시 이 차이를 기억하세요.

5.3. 비차단 retry의 대가

비차단 retry는 순서를 깹니다. ORD-1이 오류나고 retry topic으로 이동했는데 ORD-2가 즉시 성공했다면, ORD-2가 ORD-1보다 먼저 처리됩니다. 비즈니스 로직이 이벤트 순서에 의존한다면 이 방법을 사용하지 마세요.

또한 topic과 consumer 수가 증가합니다. 하지만 “문제가 있는” 하나의 메시지가 전체 consumer를 차단하지 않습니다.

6. DLT 흐름 테스트

다음 테스트는 @EmbeddedKafka를 사용합니다. 잘못된 형식의 주문 코드를 전송하고 그 메시지가 재시도 없이 orders.DLT에 나타나는지 검증합니다:

@SpringBootTest(properties = "spring.kafka.bootstrap-servers=${spring.embedded.kafka.brokers}")
@EmbeddedKafka(partitions = 1, topics = {"orders", "orders.DLT"})
class OrderDltTest {

    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;

    @Autowired
    private EmbeddedKafkaBroker broker;

    @Test
    void givenInvalidOrderId_whenConsumed_thenSentToDltWithoutRetry() {
        kafkaTemplate.send("orders", "order-key", "INVALID-99");

        Map<String, Object> props = KafkaTestUtils.consumerProps("dlt-test", "false", broker);
        try (Consumer<String, String> consumer = new DefaultKafkaConsumerFactory<>(
                props, new StringDeserializer(), new StringDeserializer()).createConsumer()) {

            broker.consumeFromAnEmbeddedTopic(consumer, "orders.DLT");
            ConsumerRecord<String, String> record =
                    KafkaTestUtils.getSingleRecord(consumer, "orders.DLT", Duration.ofSeconds(10));

            assertThat(record.value()).isEqualTo("INVALID-99");
            assertThat(record.key()).isEqualTo("order-key");

            String cause = new String(record.headers()
                    .lastHeader(KafkaHeaders.DLT_EXCEPTION_CAUSE_FQCN).value());
            assertThat(cause).isEqualTo(IllegalArgumentException.class.getName());

            String originalTopic = new String(record.headers()
                    .lastHeader(KafkaHeaders.DLT_ORIGINAL_TOPIC).value());
            assertThat(originalTopic).isEqualTo("orders");
        }
    }
}

테스트는 다음과 같이 작동합니다:

• ${spring.embedded.kafka.brokers}는 애플리케이션을 내장 broker로 가리킵니다.
• KafkaTestUtils.consumerProps는 auto.offset.reset=earliest 설정으로 consumer 설정을 생성하므로 테스트는 이전에 publish된 메시지를 놓치지 않습니다.
• IllegalArgumentException이 non-retryable이므로 메시지는 거의 즉시 DLT에 도달합니다. 테스트는 전체 backoff를 기다릴 필요가 없습니다.
• 세 가지 assertion은 다음을 검증합니다: key와 value가 유지됨, header가 올바른 원본 exception을 기록함, header가 올바른 원본 topic을 기록함.

Retry 흐름을 테스트하려면 backoff 값을 properties로 추출하세요. 테스트에서 테스트 실행 속도를 위해 수십 밀리초로 설정할 수 있습니다.

7. 일반적인 오류

• Exception을 분류하지 않기. NullPointerException이나 validation 오류를 재시도하는 것은 시간 낭비입니다. 어떤 exception을 재시도할지, 어떤 exception을 DLT로 직접 보낼지 명확히 선언하세요. DeserializationException 같은 역직렬화 오류는 기본값으로 이미 non-retryable입니다. 하지만 handler가 이런 오류를 받으려면 역직렬화 기기를 ErrorHandlingDeserializer로 감싸야 합니다.
• KafkaTemplate의 Serializer가 일치하지 않음. Recoverer는 역직렬화된 value를 다시 publish합니다. Consumer가 JSON을 object로 읽는데 template이 StringSerializer를 사용하면 DLT로 전송할 때 실패합니다. Template과 일치하는 serializer를 사용하세요. 역직렬화 오류가 있는 메시지의 경우 value는 byte[]이므로 ByteArraySerializer가 필요합니다.
• Listener가 멱등성이 없음. Retry는 같은 메시지를 여러 번 처리할 수 있음을 의미합니다. charge가 이미 돈을 뺀 후 타임아웃되면 재시도하면 다시 돈을 빼게 됩니다. idempotency key로 중복을 방지하세요. 예: 처리된 orderId를 저장합니다.
• DLT를 아무도 읽지 않음. DLT는 휴지통이 아닙니다. DLT에 새 메시지가 있을 때 경고를 해야 하고 오류 수정 후 재생할 프로세스가 필요합니다.
• 차단 retry의 backoff가 너무 김. 파티션이 그 동안 차단됩니다. 전체 대기 시간이 몇십 초를 넘으면 비차단으로 전환을 고려하세요.

주의: 엄격한 순서 요구사항이 있는 key별 흐름에 @RetryableTopic을 사용하지 마세요. 반대로 처리량이 높은 topic에 오래된 backoff를 가진 차단 retry를 사용하지 마세요. 하나의 오류 메시지가 lag을 수만까지 증가시킬 수 있습니다.

8. 결론

이 글은 Spring Kafka에서 오류 메시지를 처리하는 두 가지 방법을 제시했습니다. DefaultErrorHandler와 DeadLetterPublishingRecoverer의 조합은 차단 retry를 하고 순서를 유지하지만 파티션을 차단합니다. @RetryableTopic과 @DltHandler는 비차단 retry를 하고 consumer를 차단하지 않지만 순서를 포기합니다.

어느 방식을 선택하든 네 가지를 해야 합니다: retryable과 non-retryable exception을 분류하고, listener가 멱등성이 있는지 확인하고, 적절한 파티션 수로 DLT를 생성하고, DLT를 정기적으로 모니터링하세요. 섹션 6 같은 @EmbeddedKafka 테스트는 프로덕션 배포 전 설정이 올바르게 작동하는지 확인하는 데 도움이 됩니다.