99 Ngày Spring — Ngày 78: Kafka với Spring

Kafka ghi sự kiện vào topic phân vùng; producer gửi, consumer group đọc theo offset. Spring Kafka giảm phần cấu hình Java nhưng không thay đổi hợp đồng phân tán: message có thể được giao lại, schema phải tiến hóa và lỗi cần đường xử lý rõ. Ví dụ dùng Spring Boot 3.x và Java 21.

1. Producer và key phân vùng

KafkaTemplate gửi bất đồng bộ và trả CompletableFuture. Dùng orderId làm key để các sự kiện của cùng đơn thường vào một partition và giữ thứ tự trong partition đó. Chờ future hoặc xử lý callback để biết broker xác nhận hay lỗi; gọi send() rồi bỏ qua kết quả không bảo đảm sự kiện đã được ghi.

record OrderPlaced(String eventId, String orderId) {}
CompletableFuture<SendResult<String, OrderPlaced>> sent =
    template.send("orders.placed", event.orderId(), event);
sent.whenComplete((result, error) -> {
    if (error != null) log.error("publish failed", error);
});

2. Consumer, group và idempotency

@KafkaListener đọc record trong một consumer group; mỗi partition được một member của group xử lý tại một thời điểm. Xử lý thành công rồi mới commit offset theo cấu hình container. Nếu DB ghi xong nhưng offset chưa commit, record có thể đến lại. Lưu eventId trong bảng có unique constraint cùng transaction nghiệp vụ để xử lý trùng an toàn.

@KafkaListener(topics = "orders.placed", groupId = "billing")
void onOrder(OrderPlaced event) {
    billingService.applyOnce(event.eventId(), event.orderId());
}

3. Serialization là một hợp đồng

Producer và consumer phải thống nhất kiểu dữ liệu, field và phiên bản. JSON đủ cho demo nhưng cần kiểm soát type headers hoặc cấu hình trusted packages khi đọc từ nguồn khác; không deserialise class tùy ý. Trong hệ thống lớn, cân nhắc schema registry và quy tắc tương thích. Đừng đổi tên/xóa field mà chưa nghĩ tới consumer cũ.

spring.kafka.producer.value-serializer=org.springframework.kafka.support.serializer.JsonSerializer
spring.kafka.consumer.value-deserializer=org.springframework.kafka.support.serializer.JsonDeserializer
spring.kafka.consumer.properties.spring.json.value.default.type=com.example.OrderPlaced
spring.kafka.consumer.properties.spring.json.trusted.packages=com.example

4. Retry và dead-letter topic

Lỗi tạm thời có thể retry với backoff. Record hỏng vĩnh viễn cần đi vào DLT kèm ngữ cảnh để sửa và phát lại có kiểm soát. DefaultErrorHandler với DeadLetterPublishingRecoverer là một cách; @RetryableTopic dùng các topic retry riêng. Với batch listener, non-blocking retries không hỗ trợ như record listener. Theo dõi DLT và lag; DLT không phải nơi quên lỗi.

@Bean
DefaultErrorHandler errorHandler(KafkaTemplate<Object, Object> template) {
    var recoverer = new DeadLetterPublishingRecoverer(template);
    return new DefaultErrorHandler(recoverer, new FixedBackOff(1000L, 2L));
}

Kafka cho luồng sự kiện bền vững, nhưng không tự làm nghiệp vụ exactly once. Ngày 79 so sánh RabbitMQ với Kafka và dựng luồng hàng đợi theo routing key.

Tài liệu đối chiếu

• https://docs.spring.io/spring-kafka/reference/kafka/sending-messages.html

• https://docs.spring.io/spring-kafka/reference/kafka/annotation-error-handling.html