Kafka delivery semantics: at-most, at-least, exactly-once

Có, nhưng với phạm vi rất cụ thể. Hiểu sai phạm vi đó là lỗi phổ biến nhất khi phỏng vấn. Đi từ bài toán của shop-app.

1. Bài toán: ba chỗ có thể hỏng

order-service                Kafka                 inventory-service
 save order (DB) ──(1)──▶ send OrderCreated ──(2)──▶ reserve stock (DB) ──(3)──▶ commit offset

• (1) Lưu đơn vào DB thành công nhưng send Kafka lỗi → đơn tồn tại mà không ai biết (mất event). Hoặc ngược lại.
• (2) Producer gửi, broker ghi xong nhưng ACK bị mất mạng → producer retry → record trùng trong log.
• (3) Inventory trừ kho xong, crash trước commit offset → đọc lại → trừ kho hai lần.

2. Vấn đề: không có "exactly-once" miễn phí

DB của inventory-service, email của notification-service nằm ngoài transaction Kafka. Vì vậy thực tế: at-least-once + consumer idempotent = hiệu quả "exactly-once".

3. Giải pháp từng chặng

4. Code từng bước

Bước 1: Idempotent producer (mặc định từ Kafka 3.0, ghi rõ cho chắc):

spring:
  kafka:
    producer:
      acks: all
      properties:
        enable.idempotence: true
        max.in.flight.requests.per.connection: 5   # ≤ 5 vẫn giữ thứ tự khi idempotent

Broker nhớ (producerId, partition, sequence). Retry gửi lại sequence đã có → broker bỏ qua. Chỉ chống trùng do retry của cùng một producer session, không chống việc app gọi send() hai lần.

Từ bài này OrderEvent có thêm eventId (UUID duy nhất cho mỗi event) để consumer nhận diện bản trùng:

public record OrderEvent(String eventId, String orderId, String type,
                         BigDecimal amount, Instant occurredAt) {}

Bước 2: Transactional Outbox ở order-service. Ghi đơn và event trong cùng một DB transaction, một relay đẩy sang Kafka sau:

@Entity
@Table(name = "outbox_event")
public class OutboxEvent {
    @Id private UUID id;
    private String aggregateId;     // orderId → dùng làm Kafka key
    private String type;
    @Column(columnDefinition = "jsonb") private String payload;
    private Instant createdAt;
    private Instant publishedAt;
}

@Service
@RequiredArgsConstructor
public class OrderService {
    private final OrderRepository orders;
    private final OutboxRepository outbox;
    private final ObjectMapper mapper;

    @Transactional
    public Order create(CreateOrderRequest req) throws JsonProcessingException {
        Order order = orders.save(Order.from(req));
        var event = new OrderEvent(UUID.randomUUID().toString(), order.getId(),
                "ORDER_CREATED", order.getTotal(), Instant.now());
        outbox.save(OutboxEvent.of(order.getId(), event.type(), mapper.writeValueAsString(event)));
        return order;                       // commit 1 lần: có cả order lẫn event, hoặc không có gì
    }
}

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

    @Scheduled(fixedDelay = 500)
    @Transactional
    public void publish() throws Exception {
        for (OutboxEvent e : outbox.findTop100ByPublishedAtIsNullOrderByCreatedAt()) {
            kafka.send("orders", e.getAggregateId(), e.getPayload()).get();  // chờ ACK
            e.setPublishedAt(Instant.now());
        }
    }
}

Relay có thể gửi xong rồi crash trước khi đánh dấu → gửi lại → trùng. Outbox cho at-least-once, không mất event; trùng thì consumer lo. Production hay dùng Debezium (CDC đọc WAL) thay cho polling. Nhiều instance relay cần khóa (ví dụ SELECT ... FOR UPDATE SKIP LOCKED) để không gửi song song.

Bước 3: Idempotent consumer ở inventory-service (ý tưởng ở bài 8.1):

@KafkaListener(topics = "orders", groupId = "inventory-service")
@Transactional
public void onOrderEvent(OrderEvent event) {
    // bảng processed_event(consumer, event_id) có UNIQUE constraint
    int inserted = processedRepo.insertIfAbsent("inventory-service", event.eventId());
    if (inserted == 0) {
        log.info("Duplicate eventId={}, skip", event.eventId());
        return;
    }
    inventoryService.reserve(event);   // cùng DB transaction với dòng processed_event
}

-- PostgreSQL
INSERT INTO processed_event(consumer, event_id, processed_at)
VALUES (:consumer, :eventId, now())
ON CONFLICT DO NOTHING;

Bước 4: Kafka transactions cho consume-process-produce. Ví dụ analytics-service đọc orders, ghi kết quả sang order-stats:

spring:
  kafka:
    producer:
      transaction-id-prefix: analytics-tx-
    consumer:
      isolation-level: read_committed
      enable-auto-commit: false

@KafkaListener(topics = "orders", groupId = "analytics-service")
public void onOrder(OrderEvent event) {
    // Container đã mở Kafka transaction khi có transaction-id-prefix
    kafkaTemplate.send("order-stats", event.orderId(), Stats.from(event));
    // Khi listener trả về: offset của "orders" + record "order-stats" commit nguyên tử
}

Hoặc gom tất cả vào một thao tác: sendOffsetsToTransaction được container gọi giúp. Consumer phía sau đặt read_committed để không thấy record của transaction bị abort.

Bước 5: Retry topic + DLT cho notification-service (gửi email có thể lỗi tạm thời):

@RetryableTopic(
    attempts = "4",
    backoff = @Backoff(delay = 1000, multiplier = 2.0),
    exclude = { DeserializationException.class, IllegalArgumentException.class })
@KafkaListener(topics = "orders", groupId = "notification-service")
public void onOrder(OrderEvent event) {
    emailClient.sendOrderConfirmation(event);
}

@DltHandler
public void onDlt(OrderEvent event, @Header(KafkaHeaders.EXCEPTION_MESSAGE) String error) {
    log.error("Give up orderId={} error={}", event.orderId(), error);
    alertService.notifyOps(event, error);
}

Spring Kafka tạo các topic orders-retry-1000, orders-retry-2000... và orders-dlt. Record lỗi được chuyển sang retry topic nên không chặn partition chính. Đổi lại: mất thứ tự cho record đó. Nếu thứ tự quan trọng (inventory), dùng DefaultErrorHandler retry tại chỗ (blocking) rồi đẩy DLT bằng DeadLetterPublishingRecoverer.

Bước 6: Test

# Gửi trùng một event thủ công → inventory log "Duplicate eventId=..., skip"
# Tắt SMTP mock → notification retry 1s, 2s, 4s → vào orders-dlt
kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic orders-dlt --from-beginning

5. Cách hoạt động

order-service
 ┌── DB transaction ─────────────┐
 │ INSERT orders                 │
 │ INSERT outbox_event           │
 └───────────────────────────────┘
          │ relay / Debezium  (at-least-once)
          ▼
 Kafka "orders" (idempotent producer: retry không tạo bản trùng)
          │
 ┌────────┼───────────────────┬──────────────────────────┐
 ▼        ▼                   ▼                          ▼
inventory               notification              analytics
 processed_event         @RetryableTopic           Kafka transaction:
 + reserve (1 DB tx)     → retry topics → DLT       read orders → write order-stats
 → idempotent            (email idempotent theo     + commit offset nguyên tử
                          eventId nếu có thể)       (exactly-once trong Kafka)

6. So sánh

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

• Mặc định: at-least-once + idempotent consumer + Outbox nếu cần ghi DB và phát event cùng lúc.
• Kafka transactions: khi pipeline thuần Kafka → Kafka (Kafka Streams dùng processing.guarantee=exactly_once_v2).
• At-most-once: chỉ cho dữ liệu mất cũng không sao (metrics, log tracking).
• Không nên: tin rằng bật transaction là DB của bạn cũng "exactly-once".

Vì hai cái đó chỉ bảo vệ bên trong Kafka. Consumer vẫn có thể nhận lại cùng một record:

inventory: reserve stock (DB commit ✅) → crash → offset chưa commit
rebalance → consumer khác đọc lại record đó → reserve lần 2 ❌

• Idempotent producer: chỉ chống bản trùng trong log do retry.
• Transaction: commit offset + record output nguyên tử, nhưng DB của inventory không tham gia transaction đó.
• Còn trùng từ nguồn khác: outbox relay gửi lại, người vận hành reset offset để replay.

Nên quy tắc chung: mọi consumer có side effect phải idempotent, dựa vào eventId hoặc business key (orderId + type).

Vì đó là dual write: hai hệ thống, không có transaction chung.

@Transactional
create() {
   orders.save(order);
   kafka.send(...);      // gửi thành công
}                         // commit DB lỗi (constraint, timeout) → rollback
→ Kafka có OrderCreated cho đơn không tồn tại ❌

Đảo thứ tự (send sau commit):
commit DB ✅ → app crash trước send → đơn có, event không ❌

Outbox biến việc "ghi event" thành một dòng trong cùng DB transaction. Phần gửi sang Kafka làm sau, retry đến khi thành công. Đổi lại phải có relay/CDC và consumer chịu trùng.

Có thể vỡ, vì record cũ và mới cùng nằm trong log và các service deploy lệch nhau. Đây là schema evolution. Ngắn gọn:

• JSON: consumer cấu hình bỏ qua field lạ (Jackson FAIL_ON_UNKNOWN_PROPERTIES=false, Spring Boot mặc định đã tắt). Chỉ thêm field optional, không đổi tên/xóa/đổi kiểu.
• Avro/Protobuf + Schema Registry: registry kiểm tra tương thích (BACKWARD/FORWARD/FULL) trước khi producer được dùng schema mới.
• Thay đổi phá vỡ: tạo topic mới (orders.v2) hoặc trường version trong event.

Q1. Phân biệt at-most-once, at-least-once, exactly-once.

🇻🇳 Tiếng Việt:

At-most-once là commit offset trước rồi xử lý, có thể mất nhưng không trùng. At-least-once là xử lý xong mới commit, không mất nhưng có thể trùng. Exactly-once trong Kafka dùng idempotent producer và transactions để việc đọc, ghi ra topic khác và commit offset là nguyên tử. Thực tế em dùng at-least-once kết hợp consumer idempotent.

🇬🇧 English:

At-most-once commits the offset before processing, so messages can be lost but never duplicated. At-least-once commits after processing, so nothing is lost but duplicates are possible. Exactly-once in Kafka uses the idempotent producer and transactions so that consuming, producing to another topic and committing offsets happen atomically. In practice I use at-least-once with idempotent consumers.

Q2. Exactly-once của Kafka có phủ được việc ghi database không?

🇻🇳 Tiếng Việt:

Không. Transaction của Kafka chỉ bao gồm các topic và offset trong Kafka. Ghi database, gửi email, gọi HTTP nằm ngoài nên vẫn có thể bị lặp khi consumer crash trước commit offset. Em xử lý bằng idempotent consumer: bảng processed_event có unique key theo eventId, ghi cùng transaction với nghiệp vụ.

🇬🇧 English:

No. Kafka transactions only cover Kafka topics and offsets. Database writes, emails and HTTP calls are outside, so they can repeat if the consumer crashes before committing the offset. I handle that with an idempotent consumer: a processed_event table with a unique key on eventId, written in the same transaction as the business change.

Q3. Transactional Outbox giải quyết vấn đề gì?

🇻🇳 Tiếng Việt:

Giải quyết dual write giữa database và Kafka. Service ghi dữ liệu nghiệp vụ và một dòng outbox trong cùng một transaction DB, sau đó relay hoặc CDC như Debezium đọc outbox và publish sang Kafka. Đảm bảo không mất event, nhưng là at-least-once nên consumer vẫn phải idempotent.

🇬🇧 English:

It solves the dual-write problem between the database and Kafka. The service writes the business data and an outbox row in one database transaction, then a relay or CDC tool like Debezium reads the outbox and publishes to Kafka. No event is lost, but delivery is at-least-once, so consumers must still be idempotent.

Q4. Message lỗi liên tục thì xử lý thế nào trong Spring Kafka?

🇻🇳 Tiếng Việt:

Lỗi tạm thời thì retry có backoff. Nếu không cần thứ tự, em dùng @RetryableTopic để chuyển record sang retry topic, không chặn partition chính, hết lượt thì vào DLT và có @DltHandler cảnh báo. Nếu cần thứ tự thì dùng DefaultErrorHandler retry tại chỗ rồi DeadLetterPublishingRecoverer. Lỗi không thể retry như deserialization thì đẩy thẳng DLT.

🇬🇧 English:

Transient errors get retries with backoff. If ordering does not matter, I use @RetryableTopic so the record moves to retry topics without blocking the main partition, and after the last attempt it goes to a DLT handled by @DltHandler. If ordering matters, I use DefaultErrorHandler with in-place retries and a DeadLetterPublishingRecoverer. Non-retryable errors like deserialization go straight to the DLT.

Cách nhớ ngắn gọn

• Commit trước = có thể mất; commit sau = có thể trùng.
• Exactly-once của Kafka chỉ trong Kafka (topic + offset).
• Thực tế: Outbox + at-least-once + idempotent consumer.
• Lỗi mãi: retry topic (mất thứ tự) hoặc retry tại chỗ (giữ thứ tự), cuối cùng DLT.