Gửi và nhận message Kafka trong Spring Boot

1. Tổng quan

Apache Kafka là nền tảng event streaming phổ biến để các service giao tiếp bất đồng bộ. Spring Boot hỗ trợ Kafka qua thư viện Spring Kafka, giúp bạn không phải tự quản lý KafkaProducer và vòng lặp poll().

Trong bài này, bạn sẽ xây một ví dụ nhỏ: service đặt hàng gửi sự kiện OrderCreatedEvent, một consumer nhận sự kiện đó để giữ hàng trong kho. Đọc xong, bạn sẽ biết cách cấu hình, gửi và nhận message dạng JSON, xử lý lỗi bằng Dead Letter Topic và kiểm thử bằng Embedded Kafka.

2. Chuẩn bị

2.1. Dependency Maven

Bạn cần spring-kafka. Spring Boot đã quản lý version nên không cần ghi version:

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

<!-- Dùng cho test -->
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-test</artifactId>
    <scope>test</scope>
</dependency>
<dependency>
    <groupId>org.springframework.kafka</groupId>
    <artifactId>spring-kafka-test</artifactId>
    <scope>test</scope>
</dependency>
<dependency>
    <groupId>org.awaitility</groupId>
    <artifactId>awaitility</artifactId>
    <scope>test</scope>
</dependency>

spring-kafka-test cung cấp Kafka broker chạy trong bộ nhớ cho test. Awaitility giúp chờ xử lý bất đồng bộ mà không phải dùng Thread.sleep().

2.2. Chạy Kafka local

Cách nhanh nhất là dùng image chính thức của Apache Kafka ở chế độ KRaft, tức là không cần ZooKeeper:

docker run -d --name kafka -p 9092:9092 apache/kafka:latest

Broker sẽ lắng nghe tại localhost:9092.

3. Mô hình dữ liệu và cấu hình

Sự kiện được gửi đi là một record đơn giản:

package com.nonla.shop.order;

import java.math.BigDecimal;

public record OrderCreatedEvent(String orderId, String customerId, BigDecimal amount) {
}

Tiếp theo là cấu hình trong application.yml:

spring:
  kafka:
    bootstrap-servers: localhost:9092
    producer:
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.springframework.kafka.support.serializer.JsonSerializer
      acks: all
    consumer:
      group-id: inventory-service
      auto-offset-reset: earliest
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.springframework.kafka.support.serializer.JsonDeserializer
      properties:
        spring.json.trusted.packages: "com.nonla.shop.order"

Ý nghĩa từng phần:

• bootstrap-servers là địa chỉ broker.
• Producer dùng StringSerializer cho key và JsonSerializer cho value. JsonSerializer dùng Jackson để chuyển object thành JSON.
• acks: all buộc broker xác nhận chỉ sau khi mọi replica đồng bộ đã ghi xong. Cách này an toàn hơn nhưng chậm hơn một chút.
• group-id xác định consumer group. Các instance cùng group sẽ chia nhau các partition.
• auto-offset-reset: earliest cho consumer đọc từ đầu topic khi group chưa có offset nào được commit.
• spring.json.trusted.packages khai báo những package mà JsonDeserializer được phép tạo object. Nếu thiếu dòng này, bạn sẽ gặp lỗi “The class … is not in the trusted packages”.

3.1. Tạo topic tự động

Spring Boot tự cấu hình một KafkaAdmin. Bean này sẽ tạo mọi NewTopic được khai báo trong context:

package com.nonla.shop.config;

import org.apache.kafka.clients.admin.NewTopic;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.config.TopicBuilder;

@Configuration
public class KafkaTopicConfig {

    public static final String ORDERS_TOPIC = "orders";

    @Bean
    public NewTopic ordersTopic() {
        return TopicBuilder.name(ORDERS_TOPIC)
                .partitions(3)
                .replicas(1)
                .build();
    }
}

Khi ứng dụng khởi động, topic orders với 3 partition sẽ được tạo nếu chưa tồn tại. Nếu topic đã có, KafkaAdmin sẽ bỏ qua và không ghi đè cấu hình hiện có.

4. Gửi message với KafkaTemplate

Spring Boot tự tạo bean KafkaTemplate dựa trên cấu hình producer. Bạn chỉ cần inject vào:

package com.nonla.shop.order;

import com.nonla.shop.config.KafkaTopicConfig;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Service;

import java.util.concurrent.CompletableFuture;

@Service
public class OrderEventPublisher {

    private static final Logger log = LoggerFactory.getLogger(OrderEventPublisher.class);

    private final KafkaTemplate<String, OrderCreatedEvent> kafkaTemplate;

    public OrderEventPublisher(KafkaTemplate<String, OrderCreatedEvent> kafkaTemplate) {
        this.kafkaTemplate = kafkaTemplate;
    }

    public CompletableFuture<SendResult<String, OrderCreatedEvent>> publish(OrderCreatedEvent event) {
        return kafkaTemplate.send(KafkaTopicConfig.ORDERS_TOPIC, event.orderId(), event)
                .whenComplete((result, ex) -> {
                    if (ex != null) {
                        log.error("Gửi order {} thất bại", event.orderId(), ex);
                        return;
                    }
                    var metadata = result.getRecordMetadata();
                    log.info("Đã gửi order {} tới partition {} offset {}",
                            event.orderId(), metadata.partition(), metadata.offset());
                });
    }
}

Những điểm cần chú ý:

• send(topic, key, value) dùng orderId làm key. Kafka băm key để chọn partition, nên mọi message của cùng một đơn hàng sẽ vào cùng partition. Nhờ vậy thứ tự giữa chúng được giữ nguyên.
• Từ Spring Kafka 3.0, send() trả về CompletableFuture<SendResult> thay cho ListenableFuture đã bị loại bỏ.
• send() là bất đồng bộ. Lời gọi trả về ngay, còn callback trong whenComplete chạy khi broker phản hồi.

Khi gửi thành công, log sẽ có dạng:

INFO  c.n.s.order.OrderEventPublisher : Đã gửi order ORD-1001 tới partition 2 offset 0

Nếu cần chặn thread cho tới khi gửi xong, chẳng hạn trong một batch job, bạn có thể gọi publish(event).get(5, TimeUnit.SECONDS). Đừng làm vậy trong luồng xử lý request có tải cao, vì mỗi lời gọi sẽ giữ một thread chờ broker.

Lưu ý: Ghi vào database rồi gửi Kafka trong cùng một method không phải là một giao dịch nguyên tử. Nếu database commit thành công nhưng gửi Kafka thất bại, dữ liệu sẽ lệch nhau. Với dữ liệu quan trọng, bạn nên cân nhắc Transactional Outbox pattern.

5. Nhận message với @KafkaListener

Phía consumer chỉ cần một method có annotation @KafkaListener:

package com.nonla.shop.inventory;

import com.nonla.shop.config.KafkaTopicConfig;
import com.nonla.shop.order.OrderCreatedEvent;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.KafkaHeaders;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.messaging.handler.annotation.Payload;
import org.springframework.stereotype.Component;

@Component
public class OrderCreatedListener {

    private static final Logger log = LoggerFactory.getLogger(OrderCreatedListener.class);

    private final InventoryService inventoryService;

    public OrderCreatedListener(InventoryService inventoryService) {
        this.inventoryService = inventoryService;
    }

    @KafkaListener(topics = KafkaTopicConfig.ORDERS_TOPIC, concurrency = "3")
    public void onOrderCreated(@Payload OrderCreatedEvent event,
                               @Header(KafkaHeaders.RECEIVED_PARTITION) int partition,
                               @Header(KafkaHeaders.OFFSET) long offset) {
        log.info("Nhận order {} từ partition {} offset {}", event.orderId(), partition, offset);
        inventoryService.reserve(event);
    }
}

InventoryService chỉ là một service bình thường:

package com.nonla.shop.inventory;

import com.nonla.shop.order.OrderCreatedEvent;
import org.springframework.stereotype.Service;

@Service
public class InventoryService {

    public void reserve(OrderCreatedEvent event) {
        if (event.amount().signum() <= 0) {
            throw new IllegalArgumentException("Số tiền không hợp lệ: " + event.amount());
        }
        // Giữ hàng trong kho...
    }
}

Giải thích:

• topics chỉ định topic cần lắng nghe. groupId lấy từ spring.kafka.consumer.group-id, hoặc bạn có thể ghi đè ngay trong annotation.
• concurrency = "3" tạo 3 consumer thread, mỗi thread phụ trách một partition. Giá trị lớn hơn số partition là vô ích, vì các thread thừa sẽ không nhận được gì.
• @Payload nhận object đã được deserialize. @Header lấy metadata như partition và offset.
• Mặc định, offset được commit sau khi method chạy xong mà không ném exception (AckMode.BATCH). Vì vậy, nếu ứng dụng chết giữa chừng, message sẽ được xử lý lại. Đây là ngữ nghĩa at-least-once, nên logic của bạn cần idempotent: xử lý cùng một orderId hai lần không được giữ hàng hai lần.

Nếu cần cả key, timestamp hay header tùy ý, bạn có thể nhận nguyên ConsumerRecord<String, OrderCreatedEvent> làm tham số.

6. Xử lý lỗi và Dead Letter Topic

Khi listener ném exception, DefaultErrorHandler mặc định sẽ thử lại 9 lần liên tiếp rồi ghi log và bỏ qua message. Trong thực tế, bạn thường muốn chuyển message lỗi sang một topic riêng để kiểm tra sau:

package com.nonla.shop.config;

import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.listener.DeadLetterPublishingRecoverer;
import org.springframework.kafka.listener.DefaultErrorHandler;
import org.springframework.util.backoff.FixedBackOff;

@Configuration
public class KafkaErrorConfig {

    @Bean
    public DefaultErrorHandler errorHandler(KafkaTemplate<Object, Object> kafkaTemplate) {
        var recoverer = new DeadLetterPublishingRecoverer(kafkaTemplate);
        var handler = new DefaultErrorHandler(recoverer, new FixedBackOff(1000L, 2));
        handler.addNotRetryableExceptions(IllegalArgumentException.class);
        return handler;
    }
}

Spring Boot tự động gắn bean CommonErrorHandler này vào container factory mặc định. Cấu hình trên hoạt động như sau:

• FixedBackOff(1000L, 2) thử lại tối đa 2 lần, mỗi lần cách nhau 1 giây.
• Khi hết lượt thử, DeadLetterPublishingRecoverer gửi message sang topic orders.DLT. Tên này là tên gốc cộng hậu tố .DLT, và message giữ nguyên partition. Topic DLT vì thế cần có ít nhất số partition bằng topic gốc.
• IllegalArgumentException là lỗi dữ liệu nên có thử lại cũng vô ích. Handler sẽ chuyển thẳng message sang DLT.

Một message có amount = -5 sẽ cho log tương tự:

ERROR o.s.k.l.DefaultErrorHandler : Backoff none exhausted for orders-1@4
INFO  c.n.s.inventory.DltListener  : Message lỗi trong DLT: ORD-1002

Lưu ý: Lỗi xảy ra trong lúc deserialize, ví dụ JSON hỏng, sẽ không vào tới listener. Nếu không xử lý, consumer có thể kẹt mãi ở message đó, còn gọi là “poison pill”. Hãy bọc deserializer bằng ErrorHandlingDeserializer:

spring:
  kafka:
    consumer:
      value-deserializer: org.springframework.kafka.support.serializer.ErrorHandlingDeserializer
      properties:
        spring.deserializer.value.delegate.class: org.springframework.kafka.support.serializer.JsonDeserializer

Với cấu hình này, lỗi deserialize được chuyển cho DefaultErrorHandler và message lỗi cũng đi vào DLT.

7. Kiểm thử với Embedded Kafka

@EmbeddedKafka khởi động một broker trong bộ nhớ, nên bạn không cần Docker khi chạy test. Ví dụ dưới đây kiểm tra trọn vòng: gửi qua OrderEventPublisher, rồi xác nhận InventoryService nhận đúng sự kiện.

package com.nonla.shop.order;

import com.nonla.shop.inventory.InventoryService;
import org.junit.jupiter.api.Test;
import org.mockito.ArgumentCaptor;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.kafka.test.context.EmbeddedKafka;
import org.springframework.test.context.bean.override.mockito.MockitoBean;

import java.math.BigDecimal;
import java.time.Duration;

import static org.assertj.core.api.Assertions.assertThat;
import static org.awaitility.Awaitility.await;
import static org.mockito.Mockito.verify;

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

    @Autowired
    private OrderEventPublisher publisher;

    @MockitoBean
    private InventoryService inventoryService;

    @Test
    void whenOrderPublished_thenListenerReservesInventory() throws Exception {
        var event = new OrderCreatedEvent("ORD-1001", "CUS-42", new BigDecimal("250000"));

        var result = publisher.publish(event).get();
        assertThat(result.getRecordMetadata().topic()).isEqualTo("orders");

        var captor = ArgumentCaptor.forClass(OrderCreatedEvent.class);
        await().atMost(Duration.ofSeconds(10))
                .untilAsserted(() -> verify(inventoryService).reserve(captor.capture()));

        assertThat(captor.getValue()).isEqualTo(event);
    }
}

Từng phần làm gì:

• ${spring.embedded.kafka.brokers} là property do @EmbeddedKafka cung cấp. Nó chứa địa chỉ broker nhúng, và ta dùng nó để ghi đè bootstrap-servers.
• @MockitoBean (có từ Spring Boot 3.4) thay InventoryService thật bằng mock để dễ kiểm tra. Với phiên bản cũ hơn, bạn dùng @MockBean.
• publish(event).get() chờ broker xác nhận, nhờ vậy test chắc chắn message đã được ghi.
• Listener chạy trên thread khác, nên await().untilAsserted(...) sẽ thử lại phép verify trong tối đa 10 giây.
• Vì OrderCreatedEvent là record, equals() so sánh theo giá trị. assertThat(...).isEqualTo(event) vì thế xác nhận JSON đã được serialize và deserialize đúng.

Chính auto-offset-reset: earliest trong cấu hình giúp test ổn định. Nếu consumer kịp join group sau khi message đã được gửi, nó vẫn đọc được message đó.

8. Lỗi hay gặp và khi nào không nên dùng

Một số lỗi bạn dễ gặp:

• Thiếu trusted.packages: JsonDeserializer từ chối tạo class lạ. Hãy khai báo đúng package. Chỉ dùng "*" khi bạn kiểm soát hoàn toàn mọi producer.
• Producer và consumer khác package: JsonSerializer mặc định ghi tên class đầy đủ vào header __TypeId__. Nếu service consumer có class cùng tên nhưng nằm ở package khác, deserialize sẽ thất bại. Khi đó, đặt spring.json.use.type.headers: false và spring.json.value.default.type: com.nonla.inventory.OrderCreatedEvent ở phía consumer.
• Listener chạy lâu: nếu một lần poll() xử lý quá max.poll.interval.ms (mặc định 5 phút), broker coi consumer đã chết và rebalance. Kết quả là message bị xử lý lặp. Hãy giảm max.poll.records hoặc tối ưu logic.
• concurrency lớn hơn số partition: thread thừa sẽ ngồi không. Muốn tăng thông lượng, bạn phải tăng số partition trước.
• Gửi null key: message được rải đều giữa các partition, nên bạn mất đảm bảo thứ tự theo thực thể.

Kafka cũng không phải lựa chọn tốt trong mọi trường hợp:

• Bạn cần request/response đồng bộ và có kết quả ngay. Khi đó hãy dùng REST hoặc gRPC.
• Hệ thống nhỏ, lưu lượng thấp và không cần lưu lại lịch sử event. Vận hành một cluster Kafka khi đó là chi phí thừa, và RabbitMQ hoặc một bảng queue trong database có thể là đủ.
• Bạn cần định tuyến phức tạp theo từng message, hoặc cần độ trễ hay độ ưu tiên riêng cho từng message. Đây là thế mạnh của message broker truyền thống.

9. Kết luận

Trong bài này, bạn đã cấu hình Spring Kafka với JSON serializer và tạo topic bằng NewTopic. Bạn đã gửi message qua KafkaTemplate với key để giữ thứ tự, và nhận message bằng @KafkaListener có concurrency. Bạn cũng đã xử lý lỗi bằng DefaultErrorHandler kết hợp Dead Letter Topic, rồi viết integration test với @EmbeddedKafka và Awaitility.

Khi đưa vào production, hãy nhớ ba điểm: listener phải idempotent, phải xử lý lỗi deserialize bằng ErrorHandlingDeserializer, và việc gửi event phải nhất quán với dữ liệu trong database.