Outbox pattern với Kafka: không mất sự kiện khi ghi database

1. Tổng quan

Một service thường phải làm hai việc liền nhau: lưu dữ liệu vào database và phát sự kiện lên Kafka cho các service khác. Hai hệ thống này không chung transaction. Nếu một bước thành công còn bước kia thất bại, hệ thống sẽ mất nhất quán. Kết quả là đơn hàng đã lưu nhưng không ai nhận được sự kiện OrderPlaced.

Bài này giải thích vấn đề dual write và cài đặt Outbox pattern với Java 21, Spring Boot 3.x và Spring Kafka. Đọc xong, bạn sẽ có một message relay đẩy sự kiện từ bảng outbox lên Kafka, kèm test và consumer idempotent.

2. Vấn đề dual write

Hãy xem cách viết “ngây thơ” sau:

@Transactional
public Order placeOrder(String customerId, BigDecimal amount) {
    Order order = orderRepository.save(new Order(customerId, amount));
    kafkaTemplate.send("orders", order.getId().toString(), toJson(order));
    return order;
}

Đoạn code này có ít nhất hai kịch bản lỗi:

• Kafka gửi thành công, transaction rollback. Ví dụ lỗi khi commit hoặc vi phạm constraint. Sự kiện “ma” đã ra ngoài cho một đơn hàng không tồn tại.
• Transaction commit, Kafka gửi thất bại. Broker có thể không truy cập được, hoặc ứng dụng crash ngay sau commit. Đơn hàng có trong database nhưng sự kiện biến mất.

Đổi thứ tự hai lệnh cũng không giải quyết được gốc rễ. Hai thao tác trên hai hệ thống khác nhau không thể atomic nếu không có distributed transaction. Phần lớn đội ngũ cũng không muốn dùng XA/2PC vì phức tạp và chậm.

3. Ý tưởng của Outbox pattern

Outbox pattern biến “ghi database + gửi Kafka” thành “ghi database + ghi database”:

1. Trong cùng một transaction, service lưu Order và chèn một dòng vào bảng outbox_event.
2. Một tiến trình riêng gọi là message relay đọc các dòng chưa gửi trong bảng outbox, gửi chúng lên Kafka rồi đánh dấu đã gửi.

Vì bước 1 chỉ đụng tới một database, nó atomic tuyệt đối: có đơn hàng thì chắc chắn có sự kiện. Bước 2 có thể thử lại bao nhiêu lần cũng được. Đổi lại, bạn nhận ngữ nghĩa at-least-once: một sự kiện có thể bị gửi trùng, không bao giờ bị mất.

4. Cài đặt với Spring Boot

4.1. Dependency

<dependencies>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-data-jpa</artifactId>
    </dependency>
    <dependency>
        <groupId>org.springframework.kafka</groupId>
        <artifactId>spring-kafka</artifactId>
    </dependency>
    <dependency>
        <groupId>org.postgresql</groupId>
        <artifactId>postgresql</artifactId>
        <scope>runtime</scope>
    </dependency>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-test</artifactId>
        <scope>test</scope>
    </dependency>
    <dependency>
        <groupId>com.h2database</groupId>
        <artifactId>h2</artifactId>
        <scope>test</scope>
    </dependency>
</dependencies>

Tiếp theo là cấu hình producer an toàn:

spring.kafka.bootstrap-servers=localhost:9092
spring.kafka.producer.key-serializer=org.apache.kafka.common.serialization.StringSerializer
spring.kafka.producer.value-serializer=org.apache.kafka.common.serialization.StringSerializer
spring.kafka.producer.acks=all
spring.kafka.producer.properties.enable.idempotence=true

acks=all buộc broker xác nhận sau khi mọi replica đồng bộ đã ghi. enable.idempotence=true giúp producer không tự tạo bản trùng khi retry ở tầng mạng.

4.2. Bảng outbox và entity

CREATE TABLE outbox_event (
    id             UUID PRIMARY KEY,
    aggregate_type VARCHAR(100) NOT NULL,
    aggregate_id   VARCHAR(100) NOT NULL,
    event_type     VARCHAR(100) NOT NULL,
    payload        TEXT         NOT NULL,
    created_at     TIMESTAMP    NOT NULL,
    published_at   TIMESTAMP
);
CREATE INDEX idx_outbox_unpublished ON outbox_event (created_at) WHERE published_at IS NULL;

Partial index chỉ chứa các dòng chưa gửi. Nhờ vậy truy vấn của relay luôn nhanh, kể cả khi bảng đã có hàng triệu dòng cũ.

@Entity
@Table(name = "outbox_event")
public class OutboxEvent {

    @Id
    private UUID id;
    private String aggregateType;
    private String aggregateId;
    private String eventType;
    @Column(columnDefinition = "text")
    private String payload;
    private Instant createdAt;
    private Instant publishedAt;

    protected OutboxEvent() {
    }

    public OutboxEvent(String aggregateType, String aggregateId, String eventType, String payload) {
        this.id = UUID.randomUUID();
        this.aggregateType = aggregateType;
        this.aggregateId = aggregateId;
        this.eventType = eventType;
        this.payload = payload;
        this.createdAt = Instant.now();
    }

    public void markPublished(Instant at) {
        this.publishedAt = at;
    }

    public UUID getId() { return id; }
    public String getAggregateId() { return aggregateId; }
    public String getEventType() { return eventType; }
    public String getPayload() { return payload; }
    public Instant getPublishedAt() { return publishedAt; }
}

id cũng đóng vai trò event id duy nhất. Consumer sẽ dùng nó để loại bản trùng (mục 6). aggregateId sẽ là key của message Kafka.

4.3. Ghi Order và sự kiện trong cùng transaction

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

@Service
public class OrderService {

    private final OrderRepository orderRepository;
    private final OutboxEventRepository outboxRepository;
    private final ObjectMapper objectMapper;

    public OrderService(OrderRepository orderRepository,
                        OutboxEventRepository outboxRepository,
                        ObjectMapper objectMapper) {
        this.orderRepository = orderRepository;
        this.outboxRepository = outboxRepository;
        this.objectMapper = objectMapper;
    }

    @Transactional
    public Order placeOrder(String customerId, BigDecimal amount) {
        Order order = orderRepository.save(new Order(customerId, amount));

        var event = new OrderPlacedEvent(order.getId(), customerId, amount);
        outboxRepository.save(new OutboxEvent(
                "Order", order.getId().toString(), "OrderPlaced", toJson(event)));

        return order;
    }

    private String toJson(Object value) {
        try {
            return objectMapper.writeValueAsString(value);
        } catch (JsonProcessingException e) {
            throw new IllegalStateException("Không serialize được sự kiện", e);
        }
    }
}

Điểm mấu chốt nằm ở @Transactional. Hai lệnh save cùng commit hoặc cùng rollback. OrderService không còn biết gì về Kafka, nên không có lệnh gọi mạng nào nằm trong transaction.

4.4. Message relay đẩy sự kiện lên Kafka

Đầu tiên, viết repository có truy vấn khóa dòng:

public interface OutboxEventRepository extends JpaRepository<OutboxEvent, UUID> {

    @Query(value = """
            SELECT * FROM outbox_event
            WHERE published_at IS NULL
            ORDER BY created_at
            LIMIT :batchSize
            FOR UPDATE SKIP LOCKED
            """, nativeQuery = true)
    List<OutboxEvent> findBatchToPublish(@Param("batchSize") int batchSize);
}

FOR UPDATE SKIP LOCKED (PostgreSQL, MySQL 8+) cho phép chạy nhiều instance của ứng dụng cùng lúc. Mỗi instance khóa một lô riêng và bỏ qua các dòng instance khác đang xử lý. Nhờ vậy hai instance không gửi cùng một sự kiện song song.

Tiếp theo là relay:

@Component
public class OutboxRelay {

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

    private final OutboxEventRepository outboxRepository;
    private final KafkaTemplate<String, String> kafkaTemplate;

    public OutboxRelay(OutboxEventRepository outboxRepository,
                       KafkaTemplate<String, String> kafkaTemplate) {
        this.outboxRepository = outboxRepository;
        this.kafkaTemplate = kafkaTemplate;
    }

    @Scheduled(fixedDelay = 1000)
    @Transactional
    public void publishPending() {
        for (OutboxEvent event : outboxRepository.findBatchToPublish(100)) {
            try {
                kafkaTemplate.send(toRecord(event)).get(5, TimeUnit.SECONDS);
                event.markPublished(Instant.now());
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                return;
            } catch (Exception e) {
                log.warn("Gửi sự kiện {} thất bại, sẽ thử lại", event.getId(), e);
                break; // dừng để giữ thứ tự, lần chạy sau thử lại
            }
        }
    }

    private ProducerRecord<String, String> toRecord(OutboxEvent event) {
        var record = new ProducerRecord<>("orders", event.getAggregateId(), event.getPayload());
        record.headers()
              .add("eventId", event.getId().toString().getBytes(StandardCharsets.UTF_8))
              .add("eventType", event.getEventType().getBytes(StandardCharsets.UTF_8));
        return record;
    }
}

Lần lượt từng phần:

• @Scheduled(fixedDelay = 1000) chạy relay một giây sau khi lần trước kết thúc. Nhớ thêm @EnableScheduling vào một class cấu hình.
• .get(5, TimeUnit.SECONDS) chờ broker xác nhận. Relay chỉ đánh dấu publishedAt khi chắc chắn Kafka đã nhận message.
• markPublished thay đổi entity đang được quản lý. JPA tự sinh lệnh UPDATE khi transaction commit, bạn không cần gọi save.
• Khi lỗi, relay break thay vì bỏ qua dòng đó. Các sự kiện của cùng một đơn hàng vì thế không bị đảo thứ tự.
• Key là aggregateId, nên mọi sự kiện của một Order rơi vào cùng partition và giữ thứ tự trên Kafka.

Log khi broker tạm ngừng sẽ trông như sau:

WARN  OutboxRelay : Gửi sự kiện 3f6c1e0a-... thất bại, sẽ thử lại
org.apache.kafka.common.errors.TimeoutException: Topic orders not present in metadata after 5000 ms.

Khi broker sống lại, lần chạy kế tiếp sẽ gửi tiếp từ đúng sự kiện đó.

Lưu ý: Nếu ứng dụng crash sau khi Kafka đã nhận message nhưng trước khi transaction commit, dòng outbox vẫn là “chưa gửi”. Lần chạy sau sẽ gửi lại. Đây chính là at-least-once, và consumer bắt buộc phải xử lý được bản trùng.

5. Kiểm thử

Test thứ nhất kiểm tra rằng mỗi đơn hàng sinh ra đúng một dòng outbox:

@DataJpaTest
@Import({OrderService.class, JacksonAutoConfiguration.class})
class OrderServiceTest {

    @Autowired
    OrderService orderService;

    @Autowired
    OutboxEventRepository outboxRepository;

    @Test
    void placeOrder_shouldWriteOutboxEventInSameTransaction() {
        Order order = orderService.placeOrder("customer-42", new BigDecimal("150000"));

        List<OutboxEvent> events = outboxRepository.findAll();

        assertThat(events).hasSize(1);
        OutboxEvent event = events.getFirst();
        assertThat(event.getAggregateId()).isEqualTo(order.getId().toString());
        assertThat(event.getEventType()).isEqualTo("OrderPlaced");
        assertThat(event.getPayload()).contains("\"customerId\":\"customer-42\"");
        assertThat(event.getPublishedAt()).isNull();
    }
}

Native query của repository không được kiểm tra khi khởi động. Vì vậy H2 vẫn chạy được test này dù không hỗ trợ đầy đủ SKIP LOCKED. Để test relay với truy vấn thật, bạn nên dùng PostgreSQL qua Testcontainers.

Test thứ hai đảm bảo relay không đánh dấu đã gửi khi Kafka lỗi:

class OutboxRelayTest {

    OutboxEventRepository repository = mock(OutboxEventRepository.class);
    KafkaTemplate<String, String> kafkaTemplate = mock(KafkaTemplate.class);
    OutboxRelay relay = new OutboxRelay(repository, kafkaTemplate);

    @Test
    void shouldKeepEventUnpublished_whenKafkaFails() {
        var event = new OutboxEvent("Order", "1", "OrderPlaced", "{}");
        when(repository.findBatchToPublish(100)).thenReturn(List.of(event));
        when(kafkaTemplate.send(any(ProducerRecord.class)))
                .thenReturn(CompletableFuture.failedFuture(new RuntimeException("broker down")));

        relay.publishPending();

        assertThat(event.getPublishedAt()).isNull();
    }

    @Test
    void shouldMarkPublished_whenKafkaAcknowledges() {
        var event = new OutboxEvent("Order", "1", "OrderPlaced", "{}");
        when(repository.findBatchToPublish(100)).thenReturn(List.of(event));
        when(kafkaTemplate.send(any(ProducerRecord.class)))
                .thenReturn(CompletableFuture.completedFuture(null));

        relay.publishPending();

        assertThat(event.getPublishedAt()).isNotNull();
    }
}

Hai test này khóa lại hành vi cốt lõi: chỉ đánh dấu đã gửi khi broker đã xác nhận.

6. Phía consumer: xử lý idempotent

Vì có thể nhận trùng, consumer cần ghi nhớ các eventId đã xử lý. Việc ghi nhớ này phải nằm trong cùng transaction với nghiệp vụ:

@Component
public class InventoryListener {

    private final ProcessedEventRepository processedEvents;
    private final InventoryService inventoryService;

    public InventoryListener(ProcessedEventRepository processedEvents,
                             InventoryService inventoryService) {
        this.processedEvents = processedEvents;
        this.inventoryService = inventoryService;
    }

    @KafkaListener(topics = "orders", groupId = "inventory-service")
    @Transactional
    public void onOrderPlaced(@Header("eventId") String eventId, String payload) {
        if (processedEvents.existsById(eventId)) {
            return; // đã xử lý, bỏ qua bản trùng
        }
        inventoryService.reserveStock(payload);
        processedEvents.save(new ProcessedEvent(eventId));
    }
}

Bảng processed_event nên đặt event_id làm primary key. Nếu hai thread cùng xử lý một bản trùng, constraint sẽ chặn một trong hai.

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

• Gọi Kafka trong OrderService “cho nhanh”. Làm vậy là quay lại dual write. Mọi sự kiện phải đi qua bảng outbox.
• Quên dọn bảng outbox. Bảng sẽ phình to rất nhanh. Hãy có job xóa các dòng có published_at cũ hơn vài ngày.
• Bỏ qua dòng lỗi rồi gửi tiếp. Cách này làm hỏng thứ tự sự kiện. Nếu một payload hỏng vĩnh viễn, hãy chuyển nó sang trạng thái FAILED để điều tra, đừng retry vô hạn.
• Polling quá dày. Chu kỳ 100 ms với nhiều instance tạo tải đáng kể lên database. Chu kỳ 0,5–2 giây thường là đủ.
• Cần độ trễ rất thấp hoặc thông lượng rất lớn. Lúc này nên thay polling bằng CDC (Change Data Capture). Ví dụ Debezium đọc WAL/binlog rồi đẩy thẳng các dòng outbox lên Kafka. Bảng outbox giữ nguyên, chỉ relay thay đổi.

Lưu ý: Đừng dùng Outbox pattern nếu sự kiện không quan trọng, ví dụ metric hay log hành vi người dùng có thể chấp nhận mất. Khi đó gửi thẳng lên Kafka đơn giản hơn nhiều. Ngược lại, Kafka transactions (transactional.id) không thay thế được Outbox: chúng chỉ đảm bảo atomic giữa các topic Kafka, không bao gồm database của bạn.

8. Kết luận

Dual write giữa database và Kafka luôn có kẽ hở làm mất hoặc sinh thừa sự kiện. Outbox pattern bịt kẽ hở đó theo ba bước. Thứ nhất, ghi sự kiện vào bảng outbox trong cùng transaction với dữ liệu nghiệp vụ. Thứ hai, dùng một relay riêng đẩy sự kiện lên Kafka và chỉ đánh dấu đã gửi khi broker xác nhận. Thứ ba, xử lý idempotent ở consumer để hấp thụ bản trùng của ngữ nghĩa at-least-once.

Bạn đã có đủ các mảnh ghép: schema có partial index, OrderService không phụ thuộc Kafka, relay dùng FOR UPDATE SKIP LOCKED để chạy được nhiều instance, test cho hành vi chính và consumer chống trùng. Khi hệ thống lớn hơn, bạn có thể thay relay polling bằng CDC mà không phải sửa code nghiệp vụ.