transactional outbox & idempotent consumer

Two-Phase Commit (2PC) hay XA Transaction giữa database và message broker vừa khó scale, vừa kéo tính sẵn sàng của cả hệ xuống bằng thành phần yếu nhất, mà Kafka thì hiện cũng chưa tham gia XA được. Cách thực tế và bền hơn để có Eventual Consistency giữa database quan hệ và message broker là ghép Transactional Outbox Pattern ở phía producer với Idempotent Processing ở phía consumer.

Làm microservice hay Event-Driven Architecture, yêu cầu gặp nhiều nhất là: vừa lưu dữ liệu nghiệp vụ vào database, vừa bắn event sang Apache Kafka hoặc RabbitMQ.

Nhiều người viết theo phản xạ: mở @Transactional, gọi repository.save(), rồi gọi kafkaTemplate.send(). Đoạn code trông vô hại đó lại là nguồn gốc của mấy sự cố rất khó chịu: event ma (Phantom Events) khi DB rollback mà message đã bay đi rồi, hoặc mất event (Event Loss) khi DB commit xong mà broker lại lỗi mạng.

Dưới đây là gốc của bài toán Dual-Write, cách Transactional Outbox với Debezium CDC hoạt động, và cách thiết kế Idempotent Consumer cho hệ thống thật.

# gốc của vấn đề Dual-Write

Hai hệ thống lưu trữ độc lập (RDBMS và Apache Kafka) không dùng chung một Transaction Coordinator:

# sao không dùng 2PC?

Nhiều người đề xuất XA / 2PC. Có điều trên cloud bây giờ:

1. Blocking Protocol: coordinator treo ở pha prepare là tài nguyên database bị lock cứng, kéo theo cascading failure cả hệ thống.
2. Kafka không tham gia XA Transaction xuyên hệ thống: Kafka Transactions chỉ đảm bảo atomic trong nội bộ Kafka (vòng Read-Process-Write giữa các topic, kèm commit offset). Bạn có thể mở Kafka transaction lồng với DB transaction, nhưng đó chỉ là hai lần commit nối nhau: commit DB xong mà commit Kafka lỗi (hoặc ngược lại) thì vẫn lệch như cũ. KIP-939 (cho Kafka tham gia 2PC) vẫn đang được phát triển, chưa có trong bản phát hành chính thức tính tới 10/2026.
3. Hiệu năng tệ: mỗi transaction tốn thêm ít nhất một vòng prepare qua mạng và thêm lần fsync log ở cả coordinator lẫn các bên tham gia, nên latency và throughput đều xấu đi rõ rệt so với transaction cục bộ.

# giải pháp tận gốc: transactional outbox pattern

Ý tưởng chính của Transactional Outbox Pattern là kéo bài toán phân tán về lại bài toán ACID cục bộ (Local ACID Transaction) trên một database duy nhất.

Thay vì bắn thẳng sang Kafka:

• Bản ghi nghiệp vụ (orders) và bản ghi event (outbox_events) được lưu trong cùng một transaction của database.
• Transaction thành công thì cả hai chắc chắn đã xuống đĩa (Write-Ahead Log - WAL). Thất bại thì cả hai cùng bị huỷ.
• Một tiến trình riêng (Message Relay Engine) lo việc chuyển các event từ bảng outbox_events sang Kafka Broker, đảm bảo At-Least-Once Delivery.

# thiết kế bảng outbox cho hệ thống thật

Bảng outbox không chỉ chứa payload mà phải có đủ thông tin để định tuyến động (Dynamic Routing) và giữ thứ tự:

CREATE TABLE outbox_events (
    id UUID PRIMARY KEY,
    aggregate_type VARCHAR(100) NOT NULL, -- Tên aggregate: 'ORDER', 'CUSTOMER'
    aggregate_id VARCHAR(100) NOT NULL,   -- Khóa phân vùng Kafka (Partition Key): order_id
    event_type VARCHAR(150) NOT NULL,     -- Loại sự kiện: 'ORDER_CREATED', 'ORDER_CANCELLED'
    payload JSONB NOT NULL,               -- Dữ liệu nghiệp vụ chi tiết
    tracing_context JSONB,                -- W3C TraceContext (traceparent, baggage) để phân tán tracing
    created_at TIMESTAMP WITH TIME ZONE DEFAULT CURRENT_TIMESTAMP,
    processed BOOLEAN DEFAULT FALSE       -- Cờ hiệu nếu dùng Polling Publisher
);
 
-- Index phục vụ Polling Publisher nếu không dùng CDC
CREATE INDEX idx_outbox_unprocessed ON outbox_events (created_at) WHERE processed = FALSE;

Cột processed chỉ dùng cho Polling Publisher. Với Debezium thì bảng outbox được coi như hàng đợi chỉ có INSERT: Outbox Event Router không chấp nhận UPDATE trên bảng này và tự lọc bỏ các bản ghi DELETE.

# so sánh cách relay: Polling vs Debezium CDC

Đưa dữ liệu từ bảng outbox_events sang Kafka bằng cách nào? Có hai hướng chính:

# so sánh chi tiết

Cái giá của CDC nằm ở vận hành. Debezium đọc WAL qua một replication slot, và Postgres sẽ giữ lại mọi đoạn WAL mà slot chưa xác nhận đã đọc. Connector chết vài ngày mà không ai để ý thì WAL dồn lại tới đầy đĩa của chính database chính. Nên theo dõi độ trễ của slot (pg_replication_slots) và đặt giới hạn max_slot_wal_keep_size (Postgres 13+).

# polling cho đúng: tránh lock bảng bằng SKIP LOCKED

Nếu chưa có Kafka Connect và buộc phải dùng Polling Publisher, đừng phân trang kiểu LIMIT/OFFSET ngây thơ. Dùng FOR UPDATE SKIP LOCKED của PostgreSQL/MySQL 8+ để nhiều worker poll song song mà không giành nhau:

-- Worker lấy ra 100 bản ghi chưa xử lý và khóa lại, bỏ qua các bản ghi đang bị worker khác khóa
SELECT id, aggregate_id, event_type, payload
FROM outbox_events
WHERE processed = FALSE
ORDER BY created_at ASC
LIMIT 100
FOR UPDATE SKIP LOCKED;

Lock chỉ giữ trong phạm vi transaction, nên worker phải gửi Kafka, chờ ACK rồi UPDATE processed = TRUE (hoặc DELETE) trong cùng transaction đó. Gửi Kafka xong mà commit DB lỗi thì lần poll sau gửi lại, tức vẫn là at-least-once, phía consumer vẫn phải dedup. Và như bảng trên đã nói, nhiều worker cùng lấy các lô khác nhau thì thứ tự giữa các lô không còn được đảm bảo; cần thứ tự theo từng aggregate thì phải chia việc theo aggregate_id hoặc chỉ chạy một worker.

# Debezium Outbox Event Router (SMT)

Ở môi trường enterprise, Debezium đi cùng Kafka Connect gần như là lựa chọn mặc định. Với plugin Outbox Event Router (SMT), Debezium tự biến bản ghi trong outbox_events thành Kafka record đúng chuẩn:

// Cấu hình Debezium PostgreSQL Connector (bỏ các dòng comment // khi gửi lên Kafka Connect REST API, JSON thật không có comment)
{
  "name": "order-outbox-connector",
  "config": {
    "connector.class": "io.debezium.connector.postgresql.PostgresConnector",
    "plugin.name": "pgoutput",
    "database.hostname": "postgres-cluster",
    "database.port": "5432",
    "database.user": "debezium",
    "database.password": "${file:/secrets/db.properties:password}",
    "database.dbname": "order_db",
    "topic.prefix": "orderdb", // bắt buộc từ Debezium 2.0 (thay cho database.server.name)
    "table.include.list": "public.outbox_events",
 
    // Kích hoạt Outbox Event Router SMT
    "transforms": "outbox",
    "transforms.outbox.type": "io.debezium.transforms.outbox.EventRouter",
 
    // Định tuyến động: giá trị cột aggregate_type xác định topic đích (vd: 'ORDER' -> enterprise.events.ORDER)
    "transforms.outbox.route.by.field": "aggregate_type",
    "transforms.outbox.route.topic.replacement": "enterprise.events.${routedByValue}",
 
    // Sử dụng aggregate_id làm Kafka Message Key để bảo toàn thứ tự Partition
    "transforms.outbox.table.field.event.key": "aggregate_id",
    "transforms.outbox.table.field.event.payload": "payload",
    "transforms.outbox.table.expand.json.payload": "true", // payload JSONB tới dạng chuỗi, bật để đẩy ra JSON thật
 
    // Đưa thêm cột event_type và id vào header của Kafka record
    "transforms.outbox.table.fields.additional.placement": "event_type:header:eventType,id:header:eventId"
  }
}

Debezium không tự xoá bản ghi trong bảng outbox. Một cách làm phổ biến (được gợi ý trong bài viết của Debezium về outbox) là INSERT rồi DELETE ngay bản ghi outbox trong cùng transaction: WAL vẫn chứa lệnh INSERT để Debezium đọc, SMT bỏ qua lệnh DELETE, còn bảng thì gần như luôn trống. Nếu muốn giữ lại event để tra cứu thì dùng cách chia partition ở phần dưới.

# phía consumer: idempotent consumer

Debezium và Kafka chạy theo kiểu At-Least-Once Delivery, nên mấy tình huống sau sẽ sinh ra message trùng (Duplicate Messages):

1. Debezium connector bị restart hoặc crash sau khi đã gửi event nhưng chưa kịp lưu offset (vị trí đã đọc trong WAL) -> khởi động lại thì đọc và gửi lại các event đó. (Lỗi mất ACK rồi producer gửi lại thì idempotent producer của Kafka, mặc định bật từ Kafka 3.0, đã chặn được phần lớn.)
2. Consumer chạy xong nghiệp vụ, đang chuẩn bị commit Kafka offset thì pod Kubernetes bị kill -> pod mới kéo lại đúng message đó.

Vì vậy consumer bắt buộc phải idempotent: xử lý 1 lần hay 100 lần thì trạng thái cuối của hệ thống vẫn y như nhau.

# làm idempotent consumer với ON CONFLICT DO NOTHING

Cách an toàn nhất để tránh race condition giữa 2 consumer chạy song song là dựa vào unique constraint của database:

CREATE TABLE processed_events (
    event_id UUID NOT NULL,
    consumer_group VARCHAR(100) NOT NULL,
    processed_at TIMESTAMP WITH TIME ZONE DEFAULT CURRENT_TIMESTAMP,
    status VARCHAR(30) NOT NULL, -- 'PROCESSING', 'COMPLETED'
    PRIMARY KEY (event_id, consumer_group) -- cùng một event, mỗi consumer group xử lý một lần
);

Khoá chính gồm cả consumer_group vì cùng một event có thể được nhiều service (nhiều consumer group) xử lý, nếu dùng chung bảng mà khoá chỉ có event_id thì service thứ hai sẽ tưởng event đã xử lý rồi.

Vì sao INSERT ... ON CONFLICT DO NOTHING chống được race condition? Hai consumer cùng nhận một event, cùng INSERT: consumer đến sau sẽ bị Postgres bắt chờ cho tới khi transaction của consumer đầu commit hoặc rollback. Commit thì INSERT thứ hai trả về 0 dòng, rollback thì nó insert được và xử lý lại. Không có khe hở "cả hai cùng thấy chưa có".

Code Spring Kafka consumer:

@Component
public class PaymentOrderConsumer {
 
    private static final Logger log = LoggerFactory.getLogger(PaymentOrderConsumer.class);
    private final JdbcTemplate jdbcTemplate;
    private final PaymentProcessingService paymentService;
 
    public PaymentOrderConsumer(JdbcTemplate jdbcTemplate, PaymentProcessingService paymentService) {
        this.jdbcTemplate = jdbcTemplate;
        this.paymentService = paymentService;
    }
 
    // AckMode.MANUAL: acknowledge() chỉ xếp hàng, container commit offset SAU khi method return,
    // tức là sau khi @Transactional đã commit DB. Đừng dùng MANUAL_IMMEDIATE ở đây: nó commit offset
    // ngay trong method, trước khi DB commit, DB commit lỗi là mất message.
    @KafkaListener(topics = "enterprise.events.ORDER", groupId = "payment-service-group")
    @Transactional
    public void consumeOrderEvent(
            @Payload String payload,
            @Header("eventId") String eventId,
            @Header("eventType") String eventType,
            Acknowledgment acknowledgment) {
 
        log.info("Tiếp nhận event: id={}, type={}", eventId, eventType);
 
        // 1. Kỹ thuật Atomic Dedup Insert với PostgreSQL
        String sql = """
            INSERT INTO processed_events (event_id, consumer_group, status)
            VALUES (?, 'payment-service-group', 'PROCESSING')
            ON CONFLICT (event_id, consumer_group) DO NOTHING
        """;
 
        int rowsInserted = jdbcTemplate.update(sql, UUID.fromString(eventId));
 
        if (rowsInserted == 0) {
            log.warn("Phát hiện message trùng lặp (Duplicate)! Bỏ qua xử lý: eventId={}", eventId);
            acknowledgment.acknowledge(); // Commit offset ngay để không nhận lại
            return;
        }
 
        try {
            // 2. Thực thi nghiệp vụ cốt lõi
            paymentService.processPaymentForOrder(payload);
 
            // 3. Cập nhật trạng thái hoàn tất
            jdbcTemplate.update(
                "UPDATE processed_events SET status = 'COMPLETED' WHERE event_id = ?",
                UUID.fromString(eventId)
            );
 
            // 4. Đánh dấu ACK; với AckMode.MANUAL, offset thực sự được commit sau khi method return và DB đã commit
            acknowledgment.acknowledge();
 
        } catch (Exception ex) {
            log.error("Lỗi khi xử lý đơn hàng, rollback transaction: eventId={}", eventId, ex);
            throw ex; // Ném lỗi để kích hoạt Spring Kafka ErrorHandler / Retry Topic
        }
    }
}

Một giới hạn cần nói rõ: cách này chỉ idempotent trọn vẹn khi nghiệp vụ trong processPaymentForOrder ghi vào cùng database và cùng transaction với bảng processed_events. Nếu nghiệp vụ gọi ra ngoài (cổng thanh toán, email, service khác), transaction DB không rollback được lời gọi đó. Khi ấy phải truyền một idempotency key (vd chính eventId) sang bên kia để họ tự dedup, hoặc tách lời gọi ra ngoài thành một outbox event khác.

# dọn dẹp bảng outbox

Hệ thống tạo 50 triệu event mỗi tháng thì bảng outbox_events sẽ phình rất nhanh, ghi chậm đi và quét index cũng chậm theo.

# cách làm

1. Chia partition theo ngày (Table Partitioning by Day): Thay vì chạy DELETE FROM outbox_events WHERE created_at < NOW() - INTERVAL '7 days' (tốn I/O khủng khiếp và gây table bloat trong PostgreSQL vì MVCC), hãy chia partition theo ngày:

   CREATE TABLE outbox_events (
       ...
   ) PARTITION BY RANGE (created_at);

   Lưu ý trên bảng partition của PostgreSQL, PRIMARY KEY phải chứa cột partition, nên khoá chính phải đổi thành (id, created_at). Nếu dùng Debezium với bảng partition, publication cần tạo với publish_via_partition_root = true để event mang tên bảng cha outbox_events thay vì tên từng partition con.
2. Xoá partition bằng lệnh DDL, rất nhanh: Đêm nào cũng cho cron-job gỡ partition cũ: ALTER TABLE outbox_events DETACH PARTITION outbox_events_y2026m09d18 CONCURRENTLY; (Postgres 14+) rồi DROP TABLE outbox_events_y2026m09d18;. Xoá cả một file bảng thay vì xoá từng dòng nên trả lại dung lượng đĩa gần như ngay lập tức, không sinh dead tuple, không cần VACUUM. DROP trực tiếp một partition vẫn được, nhưng nó phải lấy lock ACCESS EXCLUSIVE trên cả bảng cha trong chốc lát; DETACH ... CONCURRENTLY giảm thời gian chặn các transaction đang ghi.

# checklist trước khi lên production

• 1. Khớp Partition Key: cột aggregate_id trong bảng outbox có map đúng vào Kafka Message Key để giữ thứ tự (Ordering) trên partition chưa?
• 2. Manual Acknowledgment: Kafka consumer đã cấu hình AckMode.MANUAL (hoặc để mặc định BATCH/RECORD, không gọi ack thủ công) để offset chỉ được commit sau khi database transaction commit xong chưa? Tránh MANUAL_IMMEDIATE khi gọi acknowledge() bên trong method @Transactional.
• 3. Bảng dedup có TTL / partition: bảng processed_events đã có cơ chế dọn ID cũ hơn 30 ngày để khỏi đầy đĩa chưa?
• 4. Dead-Letter Queue (DLQ): khi consumer gặp lỗi nghiệp vụ không retry được (payload sai), message đã được đẩy sang DLQ topic thay vì kẹt mãi (Poison Pill) chưa?
• 5. Lan truyền distributed tracing: Event Router có lấy traceparent gắn vào header của Kafka record để theo dõi trọn luồng từ Order Service sang Payment Service trên OpenTelemetry/Jaeger không?

# tổng kết

Transactional Outbox và Idempotent Consumer không phải món "có thì tốt". Với hệ microservices và event-driven làm nghiêm túc, đây là phần bắt buộc phải có. Outbox lo cho event không bị mất và không bị bắn ra khi DB rollback, idempotent consumer lo cho chuyện nhận trùng không gây hậu quả. Ghép hai thứ lại, bạn có at-least-once ở đường truyền nhưng kết quả cuối cùng như thể mỗi event chỉ được xử lý một lần, và hệ thống tự hồi phục được khi một mắt xích nào đó chết giữa chừng.

Chỉ là những ghi chép cá nhân với hy vọng mang lại chút giá trị. Nếu thấy hữu ích, đừng ngại chia sẻ cho bạn bè & đồng nghiệp nhé!

Happy coding 😎 👍🏻 🚀 🔥.