Transactional Outbox, Inbox và CDC

Đây là pattern rất thực dụng khi làm microservice + Kafka/RabbitMQ. Bài trước có một cái bẫy: Choreography Saga cần mỗi service "lưu DB rồi publish event". Nếu hai việc đó lệch nhau, Saga đứt giữa chừng.

1. Bài toán

orderRepository.save(order);
kafkaTemplate.send("order-events", event);   // ORDER_CREATED

Order được tạo → ORDER_CREATED được publish → payment-service bắt đầu charge

Nhưng hai operation nằm ở hai hệ thống khác nhau: PostgreSQL và Kafka. Không có transaction chung.

2. Vấn đề: Dual Write

BEGIN; INSERT INTO orders ...; COMMIT ✅
        ↓
Application 💥 CRASH
        ↓
kafkaTemplate.send(...) chưa chạy

Order DB:  ORDER-123 = CREATED ✅
Kafka:     ORDER_CREATED ❌
→ payment-service không bao giờ biết order này tồn tại. Order kẹt mãi ở CREATED.

Đảo thứ tự cũng hỏng: publish trước, save() fail sau → Payment nhận event cho một order không tồn tại.

"Kafka fail thì retry?" Process chết hẳn thì code retry cũng chết theo. Restart lên, app đâu biết "ORDER-123 chưa publish", trừ khi ta lưu điều đó ở đâu đó. Đó chính là ý tưởng Outbox. (Vì sao không publish thẳng trong @Transactional đã phân tích ở messaging 9.)

3. Giải pháp: Outbox table + Relay

┌──────────────────── PostgreSQL (order_db) ─────────────┐
│  orders                  outbox_event                  │
│  ORDER-123  PENDING      EVENT-999  ORDER_CREATED      │
│                          aggregate_id = ORDER-123      │
└────────────────────────────────────────────────────────┘
        cùng database, cùng transaction

• Transaction thành công → cả order và outbox có. Fail → cả hai không có. Không còn case "order có, event mất". Đây chính là chữ A trong ACID.
• Một process riêng (relay / outbox publisher) đọc event chưa gửi rồi publish lên broker.

4. Code từng bước

Bước 1: Bảng outbox (Flyway)

CREATE TABLE outbox_event (
    id             UUID PRIMARY KEY,
    aggregate_type VARCHAR(50)  NOT NULL,   -- ORDER
    aggregate_id   VARCHAR(50)  NOT NULL,   -- ORDER-123 → dùng làm Kafka key
    event_type     VARCHAR(100) NOT NULL,   -- ORDER_CREATED
    payload        JSONB        NOT NULL,
    status         VARCHAR(20)  NOT NULL DEFAULT 'PENDING',
    attempts       INT          NOT NULL DEFAULT 0,
    created_at     TIMESTAMPTZ  NOT NULL DEFAULT now()
);
CREATE INDEX idx_outbox_pending ON outbox_event (status, created_at);

Bước 2: Ghi business data + outbox trong một transaction

@Service
@RequiredArgsConstructor
public class OrderService {
    private final OrderRepository orderRepository;
    private final OutboxRepository outboxRepository;
    private final ObjectMapper objectMapper;

    @Transactional
    public Order createOrder(CreateOrderRequest req) {
        Order order = orderRepository.save(Order.pending(req));

        outboxRepository.save(OutboxEvent.builder()
            .id(UUID.randomUUID())
            .aggregateType("ORDER")
            .aggregateId(order.getId())
            .eventType("ORDER_CREATED")
            .payload(toJson(new OrderCreated(order.getId(), order.getTotal())))
            .status("PENDING")
            .build());
        return order;            // không gọi Kafka ở đây
    }

    private String toJson(Object o) {
        try { return objectMapper.writeValueAsString(o); }
        catch (JsonProcessingException e) { throw new IllegalStateException(e); }
    }
}

Bước 3: Relay bằng polling

public interface OutboxRepository extends JpaRepository<OutboxEvent, UUID> {
    // SKIP LOCKED: nhiều instance chạy relay mà không lấy trùng batch
    @Query(value = """
        SELECT * FROM outbox_event
        WHERE status = 'PENDING'
        ORDER BY created_at
        LIMIT 100
        FOR UPDATE SKIP LOCKED
        """, nativeQuery = true)
    List<OutboxEvent> lockPendingBatch();
}

@Component
@RequiredArgsConstructor
public class OutboxRelay {
    private final OutboxRepository outboxRepository;
    private final KafkaTemplate<String, String> kafkaTemplate;

    @Scheduled(fixedDelay = 500)
    @Transactional
    public void publishPending() {
        for (OutboxEvent e : outboxRepository.lockPendingBatch()) {
            try {
                kafkaTemplate.send("order-events", e.getAggregateId(), e.getPayload())
                             .get(5, TimeUnit.SECONDS);       // chờ broker ack
                e.setStatus("PUBLISHED");
            } catch (Exception ex) {
                e.setAttempts(e.getAttempts() + 1);
                break;   // dừng batch để giữ thứ tự; lần sau thử lại
            }
        }
    }
}

Ghi chú: header eventId (= outbox_event.id) nên được gửi kèm để consumer khử trùng. Dọn bảng định kỳ: xóa bản ghi PUBLISHED cũ hơn N ngày.

Bước 4: Test

# Tắt Kafka, tạo đơn: API vẫn trả 201, outbox có PENDING
docker stop kafka
curl -X POST localhost:8081/orders -H 'Content-Type: application/json' -d '{"cartId":"CART-123"}'
psql -c "select event_type,status from outbox_event"
# Bật Kafka lại: vài trăm ms sau status = PUBLISHED, payment-service nhận event
docker start kafka

5. Cách hoạt động

Client → order-service
           │ BEGIN
           ├── INSERT orders
           └── INSERT outbox_event
           │ COMMIT → return 201

Outbox Relay (độc lập)
   read PENDING → Kafka (key = orderId) → mark PUBLISHED
                    │
                    ▼
            payment-service
            inbox / idempotency
                    │
                    ▼
               payment_db

Cái bẫy: relay gửi Kafka thành công rồi crash trước khi UPDATE status='PUBLISHED'. Restart lên event vẫn PENDING → publish lần nữa. Outbox chống mất event, đổi lại có thể trùng: đây là at-least-once.

Inbox (idempotent consumer) ở phía nhận:

CREATE TABLE inbox_message (
    message_id  UUID PRIMARY KEY,          -- = eventId từ outbox
    received_at TIMESTAMPTZ NOT NULL DEFAULT now()
);

@KafkaListener(topics = "order-events", groupId = "payment-service")
@Transactional
public void onOrderCreated(@Header("eventId") String eventId, String payload) {
    int inserted = inboxRepository.insertIgnore(UUID.fromString(eventId));
    // INSERT ... ON CONFLICT (message_id) DO NOTHING
    if (inserted == 0) return;            // đã xử lý → bỏ qua
    paymentService.charge(parse(payload)); // cùng transaction với inbox insert
}

Ghi inbox và business update trong cùng local transaction: hoặc cả hai, hoặc không. At-least-once + idempotent consumer = business effect "effectively once".

Ordering theo aggregate id. ORDER_CREATED rồi ORDER_CANCELLED của cùng ORDER-123 phải đến đúng thứ tự:

• Kafka key = aggregate_id → cùng partition → giữ thứ tự trong partition.
• Relay publish theo created_at, gặp lỗi thì dừng (không nhảy qua event sau của cùng aggregate).
• Consumer có thể kiểm tra thêm version của aggregate, bỏ event cũ hơn version đã áp dụng.

6. So sánh

7. Khi nào nên dùng / không nên dùng

• Nên: bất cứ khi nào cần "update DB + publish event" và mất event là sai nghiệp vụ (Saga, đồng bộ dữ liệu sang service khác).
• Không cần: event không quan trọng (metric, log), hoặc mọi thay đổi chỉ nằm trong một DB.
• Outbox không làm hai DB khác nhau commit atomic, và không thay Saga xử lý compensation.

Bạn đúng một nửa: Outbox không bắt buộc đi cùng Saga. Nhưng cần chỉnh một điểm: nhiều việc cùng lưu một database thì không cần Outbox.

• Database transaction: nhiều thao tác trên cùng DB commit hoặc rollback cùng nhau.
• Transactional Outbox: business data và ý định publish event được lưu atomic trong cùng DB transaction.
• Saga: business transaction trải qua nhiều service, mỗi service có local transaction riêng.

@Transactional
public void createOrder() {
    orderRepository.save(order);
    paymentRepository.save(payment);   // cùng DB → @Transactional là đủ
}

Outbox chỉ cần khi một trong các "việc" là gửi ra hệ thống khác (RabbitMQ, Kafka):

@Transactional
public void createOrder() {
    orderRepository.save(order);
    rabbitTemplate.convertAndSend("order.exchange", "order.created", event);
}
// @Transactional không bao RabbitMQ:
// publish fail mà DB vẫn commit, hoặc publish xong rồi DB rollback → Dual Write

Trong Saga, Outbox giúp không mất bước chuyển trạng thái: payment-service UPDATE payment = SUCCESS + INSERT outbox PAYMENT_COMPLETED trong một transaction, inventory-service chắc chắn nhận được. Nhưng Outbox không tự xử lý compensation, timeout hay saga state.

Chính xác. Ví dụ không có Saga nào:

@Transactional
public void createUser(User user) {
    userRepository.save(user);
    outboxRepository.save(new OutboxEvent("USER_CREATED", user.getId()));
}
// Relay gửi USER_CREATED → email chào mừng, analytics, đồng bộ search index

Lưu ý: Outbox không đảm bảo DB commit và broker nhận message cùng lúc. Nó đảm bảo business data và event được lưu atomic, rồi relay publish sau đó (có độ trễ, có thể trùng).

Saga giải quyết workflow, Outbox giải quyết dual write, Idempotency giải quyết duplicate. Ba khái niệm độc lập nhưng kết hợp rất hiệu quả.

Câu hỏi đúng chỗ. Polling ổn với traffic vừa phải (có index, batch, SKIP LOCKED). Khi traffic lớn hoặc cần độ trễ thấp, dùng CDC (Change Data Capture): thay vì hỏi DB, ta đọc transaction log của DB.

order-service ── INSERT outbox_event ──► PostgreSQL
                                            │ WAL (write-ahead log)
                                            ▼
                               Debezium (Kafka Connect)
                                logical replication slot
                                            │ Outbox Event Router
                                            ▼
                                Kafka topic: order-events
                                key = aggregate_id

Cấu hình connector (rút gọn, password lấy từ secret/env):

{
  "name": "order-outbox-connector",
  "config": {
    "connector.class": "io.debezium.connector.postgresql.PostgresConnector",
    "database.hostname": "order-db",
    "database.user": "debezium",
    "database.password": "${file:/secrets/db.properties:password}",
    "database.dbname": "order_db",
    "topic.prefix": "shop",
    "table.include.list": "public.outbox_event",
    "transforms": "outbox",
    "transforms.outbox.type": "io.debezium.transforms.outbox.EventRouter",
    "transforms.outbox.table.field.event.key": "aggregate_id",
    "transforms.outbox.route.by.field": "aggregate_type"
  }
}

• Không có relay code trong service, không query định kỳ, độ trễ thấp, thứ tự theo commit log.
• Đổi lại: thêm Kafka Connect + Debezium cần vận hành; replication slot không được consume thì WAL phình to, đầy đĩa.
• Vẫn là at-least-once (connector restart có thể gửi lại) → consumer vẫn cần inbox.

Phân biệt khi phỏng vấn: Outbox là reliability pattern; CDC là cơ chế capture thay đổi từ log; CDC có thể dùng để publish outbox. CDC trực tiếp trên bảng orders (không qua outbox) thì làm lộ schema nội bộ ra ngoài, nên thường đọc bảng outbox.

Q1. Service lưu order vào DB rồi publish Kafka. App crash sau commit, trước publish thì sao?

🇻🇳 Tiếng Việt:

Đừng trả lời "retry Kafka", vì app chết thì ai retry. Em dùng Transactional Outbox: lưu business data và outbox event trong cùng một local transaction. Một publisher riêng đọc outbox và publish lên Kafka. Event không bị mất sau khi transaction commit. Vì publisher có thể gửi một event nhiều lần, consumer phải idempotent.

🇬🇧 English:

Retrying Kafka isn't enough, because if the app crashed nobody is left to retry. I would use the Transactional Outbox pattern: save the business data and an outbox event in the same local transaction, and let a separate publisher read the outbox and send to Kafka. The event can't be lost once the transaction commits. Since the publisher may send the same event more than once, consumers must be idempotent.

Q2. Saga và Outbox khác nhau thế nào?

🇻🇳 Tiếng Việt:

Saga quản lý business transaction qua nhiều service bằng local transaction và compensation. Outbox giải quyết dual write giữa DB và message broker. Chúng hay đi cùng nhau: Saga lo workflow, Outbox lo event không bị mất. Nhưng Outbox dùng độc lập được ở bất kỳ chỗ nào cần lưu DB và publish event đáng tin cậy.

🇬🇧 English:

Saga manages a business transaction across services using local transactions and compensations. Outbox solves the dual-write problem between the database and the broker. They are often used together: Saga handles the workflow, Outbox makes sure events aren't lost. But Outbox is useful on its own anywhere we need to save data and publish an event reliably.

Q3. Polling publisher và CDC khác nhau thế nào?

🇻🇳 Tiếng Việt:

Polling query bảng outbox định kỳ, đơn giản nhưng có độ trễ và tạo tải lên DB. CDC như Debezium đọc transaction log của DB, độ trễ thấp và không query, nhưng phải vận hành thêm Kafka Connect và replication slot. Cả hai đều at-least-once.

🇬🇧 English:

A polling publisher queries the outbox table periodically: simple, but it adds latency and database load. CDC tools like Debezium read the database transaction log, giving low latency without queries, but we have to operate Kafka Connect and replication slots. Both give at-least-once delivery.

Q4. Làm sao giữ thứ tự event của cùng một order?

🇻🇳 Tiếng Việt:

Dùng aggregate id làm message key để event cùng order vào cùng partition. Relay publish theo thứ tự tạo và không bỏ qua event lỗi của cùng aggregate. Consumer có thể kiểm tra version để bỏ event cũ.

🇬🇧 English:

Use the aggregate id as the message key so all events of one order land in the same partition. The relay publishes in creation order and never skips a failed event of the same aggregate. Consumers can also check a version number and drop stale events.

Cách nhớ ngắn gọn

• Dual write: DB commit OK + publish fail → event mất.
• Outbox: business row + outbox row cùng transaction → relay/CDC → broker.
• Outbox chống mất, không chống trùng → Inbox (eventId unique) ở consumer.
• Key = aggregate id để giữ thứ tự. Outbox dùng được cả khi không có Saga.