Kafka retry và dead letter topic với Spring Kafka

1. Tổng quan

Một consumer Kafka chắc chắn sẽ gặp message xử lý lỗi. Có lỗi tạm thời, ví dụ cổng thanh toán timeout. Có lỗi vĩnh viễn, ví dụ dữ liệu sai định dạng. Nếu không xử lý đúng, bạn sẽ gặp một trong hai tình huống: mất message, hoặc consumer kẹt mãi ở một message không bao giờ thành công.

Bài này hướng dẫn cách cấu hình retry trong Spring Kafka theo hai kiểu: blocking với DefaultErrorHandler và non-blocking với @RetryableTopic. Message vẫn lỗi sau khi hết số lần retry sẽ được đẩy sang dead letter topic (DLT). Đọc xong, bạn biết khi nào chọn kiểu nào và viết được test để kiểm chứng hành vi.

2. Thiết lập

Thêm dependency Spring Kafka. Spring Boot 3.x tự quản lý version, nên bạn không cần khai báo:

<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>

Cấu hình tối thiểu trong application.properties:

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

Boot mặc định dùng StringSerializer và StringDeserializer. Để ví dụ gọn, value của message là mã đơn hàng dạng String.

3. Consumer có thể lỗi

Đây là service xử lý thanh toán. Mã đơn sai định dạng là lỗi vĩnh viễn. Lỗi gọi cổng thanh toán là lỗi tạm thời:

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("Mã đơn không hợp lệ: " + orderId);
        }
        // Gọi cổng thanh toán; có thể ném PaymentGatewayException khi timeout
    }
}

Listener chỉ gọi service:

@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("Xử lý đơn {}", orderId);
        paymentService.charge(orderId);
    }
}

Nếu bạn không cấu hình gì, Spring Kafka dùng DefaultErrorHandler với FixedBackOff(0, 9). Message lỗi được thử lại 9 lần liên tiếp, tổng cộng 10 lần. Sau đó handler chỉ ghi log và bỏ qua. Message coi như mất. Đó là lý do bạn cần cấu hình lại.

4. Retry blocking với DefaultErrorHandler

4.1. Cấu hình backoff và DLT

Khi listener ném exception, DefaultErrorHandler seek lại offset của message lỗi để lần poll sau đọc lại nó. Hết số lần retry, handler gọi một ConsumerRecordRecoverer. Ở đây, recoverer là 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;
    }
}

Giải thích từng phần:

• DeadLetterPublishingRecoverer(template) gửi message lỗi sang topic <tên topic>.DLT, ở cùng partition với message gốc. Với topic orders, DLT là orders.DLT.
• ExponentialBackOffWithMaxRetries(3) thử lại tối đa 3 lần. Khoảng chờ lần lượt là 1s, 2s, 4s và không vượt quá 10s.
• addNotRetryableExceptions khai báo các exception không cần retry. IllegalArgumentException đi thẳng sang DLT, vì có thử lại bao nhiêu lần thì dữ liệu sai vẫn sai.

Spring Boot tự gắn bean CommonErrorHandler vào container factory mặc định. Bạn không cần tự tạo ConcurrentKafkaListenerContainerFactory.

Khi gửi ORD-1 và cổng thanh toán liên tục timeout, log sẽ trông như sau:

10:00:00.010 INFO  Xử lý đơn ORD-1
10:00:01.020 INFO  Xử lý đơn ORD-1
10:00:03.030 INFO  Xử lý đơn ORD-1
10:00:07.040 INFO  Xử lý đơn ORD-1
10:00:07.060 ERROR Backoff ... exhausted for orders-0@42

Tổng cộng có 4 lần xử lý: 1 lần đầu và 3 lần retry. Sau đó message nằm trong orders.DLT.

4.2. Header trong DLT

Message trong DLT giữ nguyên key và value. Recoverer thêm các header để bạn điều tra nguyên nhân:

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 gốc của bạn nằm ở kafka_dlt-exception-cause-fqcn. Nếu cần đọc header bằng code, bạn dùng các hằng số trong KafkaHeaders, ví dụ KafkaHeaders.DLT_ORIGINAL_TOPIC.

Lưu ý: Vì recoverer mặc định gửi vào cùng partition, orders.DLT phải có ít nhất số partition bằng orders. Nếu không, việc publish sẽ lỗi. Bạn có thể tạo DLT bằng bean NewTopic có cùng số partition. Một cách khác là truyền destination resolver trả về new TopicPartition(r.topic() + ".DLT", -1) để Kafka tự chọn partition.

4.3. Khi nào retry blocking là hợp lý

Retry blocking giữ đúng thứ tự message trong partition. Message sau chỉ được xử lý khi message trước thành công hoặc đã sang DLT. Đây là lựa chọn đúng khi thứ tự quan trọng, ví dụ các sự kiện OrderCreated → OrderPaid → OrderShipped của cùng một đơn hàng.

Cái giá phải trả là cả partition bị chặn trong lúc chờ. Từ Spring Kafka 2.9 trở đi, handler chờ backoff mà không vượt max.poll.interval.ms, nhờ cơ chế seek và pause. Dù vậy, nếu bạn đặt backoff tổng cộng vài phút, lag sẽ tăng vọt.

5. Retry non-blocking với @RetryableTopic

5.1. Cấu hình bằng annotation

Với non-blocking retry, message lỗi được chuyển sang một retry topic riêng. Nhờ đó, consumer tiếp tục xử lý message kế tiếp ngay. Một consumer khác đọc retry topic sau khoảng backoff đã định:

@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("Xử lý {} từ {}", orderId, topic);
        paymentService.charge(orderId);
    }

    @DltHandler
    public void onDlt(String orderId,
                      @Header(KafkaHeaders.EXCEPTION_MESSAGE) String error) {
        log.error("Đơn {} vào DLT: {}", orderId, error);
    }
}

Các điểm chính:

• attempts = "4" nghĩa là tổng cộng 4 lần xử lý: 1 lần ở topic gốc và 3 lần ở retry topic. Giá trị là String để hỗ trợ placeholder như "${app.retry.attempts}".
• @Backoff lấy từ package org.springframework.retry.annotation. Spring Retry đã đi kèm spring-kafka.
• exclude liệt kê các exception đi thẳng sang DLT, tương đương addNotRetryableExceptions.
• @DltHandler đặt trong cùng class. Đây là nơi bạn ghi log, gửi cảnh báo hoặc lưu vào bảng để xử lý thủ công.

5.2. Các topic được tạo

Với backoff tăng dần, mặc định mỗi lần retry có một topic riêng, đặt hậu tố theo khoảng delay:

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

Spring Kafka tự tạo các topic này khi khởi động (autoCreateTopics = "true" là mặc định). Môi trường production thường tắt auto-create trên broker. Khi đó, bạn cần tạo trước các topic với đúng tên và số partition. Nếu muốn dùng chỉ số thay cho delay, hãy đặt topicSuffixingStrategy = TopicSuffixingStrategy.SUFFIX_WITH_INDEX_VALUE.

Lưu ý hậu tố: @RetryableTopic dùng -dlt, còn DeadLetterPublishingRecoverer dùng .DLT. Hãy nhớ điểm này khi cấu hình monitoring.

5.3. Đánh đổi của non-blocking

Non-blocking retry phá vỡ thứ tự. Giả sử ORD-1 lỗi và chuyển sang retry topic, còn ORD-2 thành công ngay. Khi đó ORD-2 được xử lý trước ORD-1. Nếu luồng nghiệp vụ của bạn phụ thuộc thứ tự sự kiện, đừng dùng cách này.

Ngoài ra, số topic tăng lên và số consumer cũng nhiều hơn. Bù lại, một message “độc” không làm tắc cả partition.

6. Kiểm thử luồng DLT

Test dưới đây dùng @EmbeddedKafka. Test gửi một mã đơn sai định dạng và kiểm tra message đó xuất hiện trong orders.DLT mà không bị retry:

@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");
        }
    }
}

Test hoạt động như sau:

• ${spring.embedded.kafka.brokers} trỏ ứng dụng tới broker nhúng.
• KafkaTestUtils.consumerProps tạo cấu hình consumer với auto.offset.reset=earliest, nên test không bỏ lỡ message đã được publish trước khi consumer subscribe.
• Vì IllegalArgumentException là non-retryable, message đến DLT gần như ngay lập tức. Test không phải chờ hết chuỗi backoff.
• Các assertion kiểm tra ba điều: key và value được giữ nguyên, header ghi đúng exception gốc, và header ghi đúng topic nguồn.

Nếu muốn test luồng retry, hãy đưa giá trị backoff ra properties. Trong test, bạn đặt khoảng chờ vài chục mili giây để test chạy nhanh.

7. Lỗi hay gặp

• Không phân loại exception. Retry NullPointerException hay lỗi validate chỉ tốn thời gian. Hãy khai báo rõ exception nào được retry và exception nào đi thẳng sang DLT. Lỗi deserialize như DeserializationException vốn đã là non-retryable mặc định. Nhưng để handler nhận được lỗi này, bạn cần bọc deserializer bằng ErrorHandlingDeserializer.
• Serializer của KafkaTemplate không khớp. Recoverer publish lại value đã deserialize. Nếu consumer đọc JSON thành object mà template dùng StringSerializer, việc gửi sang DLT sẽ thất bại. Hãy dùng template có serializer phù hợp. Với message lỗi deserialize, value là byte[], nên cần ByteArraySerializer.
• Listener không idempotent. Retry nghĩa là cùng một message có thể được xử lý nhiều lần. Nếu charge đã trừ tiền rồi mới timeout, retry sẽ trừ tiền lần nữa. Hãy chống trùng bằng idempotency key, ví dụ lưu orderId đã xử lý.
• Không ai đọc DLT. DLT không phải thùng rác. Bạn cần cảnh báo khi DLT có message mới, và cần một quy trình replay sau khi sửa lỗi.
• Backoff tổng quá dài với retry blocking. Partition bị chặn suốt thời gian đó. Nếu tổng thời gian chờ vượt vài chục giây, hãy cân nhắc chuyển sang non-blocking.

Lưu ý: Không nên dùng @RetryableTopic cho luồng yêu cầu thứ tự chặt chẽ theo key. Ngược lại, đừng dùng retry blocking với backoff dài cho topic có throughput cao. Một message lỗi có thể làm lag tăng hàng chục nghìn.

8. Kết luận

Bài viết đã trình bày hai cách xử lý message lỗi trong Spring Kafka. DefaultErrorHandler kết hợp DeadLetterPublishingRecoverer cho retry blocking, giữ được thứ tự nhưng chặn partition. @RetryableTopic cùng @DltHandler cho retry non-blocking, không làm tắc consumer nhưng hy sinh thứ tự.

Dù chọn cách nào, bạn cũng nên làm bốn việc: phân loại exception retryable và non-retryable, đảm bảo listener idempotent, tạo DLT với số partition phù hợp, và giám sát DLT thường xuyên. Một test với @EmbeddedKafka như ở mục 6 giúp bạn chắc chắn cấu hình hoạt động đúng trước khi lên production.