Kafka, RabbitMQ, retry, DLQ, message trùngKafka, RabbitMQ, retries, DLQ, duplicate messages
Order Service ghi sự kiện vào topic orders, key là order_id. Nhóm consumer payment đọc từng partition, ghi kết quả và message_id vào DB trong một transaction rồi mới commit offset. Bật các kịch bản lỗi để xem message được giao lại, bị loại trùng, và đi qua topic retry tới DLQ.The order service writes events to the orders topic, keyed by order_id. The payment consumer group reads each partition, stores the result and the message_id in one DB transaction, and only then commits the offset. Turn on the failure scenarios to see redelivery, deduplication and the path through retry topics to the DLQ.
1Đơn 1001 được tạo. Producer gửi OrderPlaced v2 với key=order-1001, chờ đủ replica xác nhận.Order 1001 is created. The producer sends OrderPlaced v2 with key=order-1001 and waits for all in-sync replicas.
Auto-commit có thể commit offset trước khi xử lý xong, crash lúc đó là mất message (at-most-once). Tắt enable.auto.commit, commit sau khi DB commit để có at-least-once.Auto-commit can commit an offset before processing finishes; a crash then loses the message (at-most-once). Disable enable.auto.commit and commit after the DB commit to get at-least-once.
Rebalance, crash hoặc producer retry đều sinh trùng. Consumer ghi message_id vào processed_messages trong cùng transaction với tác dụng phụ; khoá chính trùng thì bỏ qua. Kafka exactly-once chỉ đúng khi cả đầu vào và đầu ra đều là Kafka.Rebalances, crashes and producer retries all create duplicates. The consumer records the message_id in processed_messages in the same transaction as the side effect and skips on a key conflict. Kafka exactly-once only holds when both input and output are Kafka.
Thử lại tại chỗ mãi thì mọi message sau nó trong partition đều đứng. Chuyển sang topic retry có độ trễ tăng dần (1 phút, 10 phút), hết lần thì vào DLQ. Lỗi deserialize hoặc schema sai thì vào DLQ ngay.Retrying in place forever stalls every later message in the partition. Forward to retry topics with growing delays (1 minute, 10 minutes) and then to the DLQ. Deserialization and schema errors go straight to the DLQ.
Kafka chỉ giữ thứ tự trong một partition. Dùng order_id làm key để mọi sự kiện của một đơn vào cùng partition. Topic retry phá thứ tự này, nên với luồng cần thứ tự chặt phải dừng cả key khi lỗi.Kafka only orders messages within a partition. Use order_id as the key so one order's events share a partition. Retry topics break that order, so strictly ordered flows must pause the whole key on failure.
Lag tăng nghĩa là consumer chậm hơn producer. Theo dõi lag theo partition, scale consumer tối đa bằng số partition, giới hạn max.poll.records và pause partition khi DB downstream quá tải.Growing lag means consumers are slower than producers. Track lag per partition, scale consumers up to the partition count, cap max.poll.records and pause partitions when the downstream DB is overloaded.
Payload có trường version và đi qua Schema Registry với chế độ BACKWARD. Chỉ thêm trường có giá trị mặc định; đổi nghĩa một trường thì tạo version mới, consumer hỗ trợ cả hai trong giai đoạn chuyển.Payloads carry a version field and go through a Schema Registry in BACKWARD mode. Only add fields with defaults; changing a field's meaning needs a new version, and consumers accept both during the transition.
| BảngTable | Cột chínhKey columns | Ghi chúNotes |
|---|---|---|
processed_messages | message_id, consumer, processed_at | PK(message_id, consumer); ghi cùng transaction với tác dụng phụ; dọn bản ghi cũ hơn thời gian giữ logPK(message_id, consumer); written in the same transaction as the side effect; purge rows older than the log retention |
payment | id, order_id, amount, status, created_at | unique(order_id) là lớp chặn trùng thứ haiunique(order_id) is a second guard against duplicates |
dead_letter | id, topic, partition, offset, msg_key, payload, error, attempts, created_at, replayed_at | Bản sao DLQ để tra cứu và đánh dấu đã phát lạiA copy of the DLQ for lookup and for marking replays |
Consumer thanh toán: loại trùng bằng processed_messages, lỗi tạm thời đi topic retry, payload sai version vào thẳng DLQ.The payment consumer deduplicates with processed_messages, sends transient failures to retry topics and wrong payload versions straight to the DLQ.
@Slf4j
@Component
@RequiredArgsConstructor
public class PaymentConsumer {
private final ProcessedMessageRepository processed;
private final PaymentService payments;
@RetryableTopic(
attempts = "3",
backoff = @Backoff(delay = 60_000, multiplier = 10),
exclude = {DeserializationException.class, ValidationException.class},
dltTopicSuffix = ".dlq")
@KafkaListener(topics = "orders", groupId = "payment")
@Transactional
public void onOrderPlaced(@Payload OrderPlaced event,
@Header(KafkaHeaders.RECEIVED_KEY) String key) {
if (!processed.tryInsert(event.messageId(), "payment")) {
return; // duplicate delivery
}
switch (event.version()) {
case 1, 2 -> payments.charge(event.orderId(), event.amount());
default -> throw new ValidationException("unknown version " + event.version());
}
}
@DltHandler
public void onDeadLetter(OrderPlaced event,
@Header(KafkaHeaders.EXCEPTION_MESSAGE) String error) {
log.error("DLQ order={} error={}", event.orderId(), error);
}
}type PaymentConsumer struct {
reader *kafka.Reader // GroupID: "payment", no auto-commit
db *sql.DB
retry Forwarder // orders.retry.1m → orders.retry.10m → orders.dlq
payments PaymentService
}
func (c *PaymentConsumer) Run(ctx context.Context) error {
for {
m, err := c.reader.FetchMessage(ctx) // does not commit
if err != nil {
return err // ctx cancelled or broker unreachable
}
if err := c.handle(ctx, m); err != nil {
if err := c.retry.Forward(ctx, m, err); err != nil {
return err // cannot park it: stop without committing
}
}
if err := c.reader.CommitMessages(ctx, m); err != nil {
return err
}
}
}
func (c *PaymentConsumer) handle(ctx context.Context, m kafka.Message) error {
var e OrderPlaced
if err := json.Unmarshal(m.Value, &e); err != nil {
return Permanent(err) // poison: Forward sends it straight to the DLQ
}
tx, err := c.db.BeginTx(ctx, nil)
if err != nil {
return err
}
defer tx.Rollback()
res, err := tx.ExecContext(ctx, `INSERT INTO processed_messages(message_id, consumer)
VALUES ($1, 'payment') ON CONFLICT DO NOTHING`, e.MessageID)
if err != nil {
return err
}
if n, _ := res.RowsAffected(); n == 0 {
return nil // duplicate delivery: commit the offset and move on
}
if err := c.payments.Charge(ctx, tx, e.OrderID, e.Amount); err != nil {
return err
}
return tx.Commit()
}| Trước (at-most-once)Before (at-most-once) | Sau (at-least-once)After (at-least-once) | |
|---|---|---|
| Consumer crash giữa chừngConsumer crashes midway | Message bị mấtThe message is lost | Message được giao lạiThe message is redelivered |
| Cần thêmAlso needs | Không cần gì thêmNothing extra | Consumer idempotent (processed_messages, unique key)An idempotent consumer (processed_messages, unique keys) |
| Hợp vớiGood for | Metrics, log, dữ liệu mất vài dòng không saoMetrics, logs, data where losing a few rows is fine | Thanh toán, tồn kho, mọi tác dụng phụ quan trọngPayments, inventory, any side effect that matters |
Mỗi pattern mở bài học tương ứng trong menu Pattern.Each one opens its lesson in the Patterns menu.
Exactly-once của Kafka (idempotent producer + transaction) chỉ bao phần đọc-ghi trong Kafka. Khi tác dụng phụ là ghi DB hay gọi API ngoài, crash giữa tác dụng phụ và commit offset vẫn gây giao lại. Thực tế là at-least-once cộng xử lý idempotent, gọi là effectively-once.Kafka's exactly-once (idempotent producer plus transactions) only covers read-process-write inside Kafka. When the side effect is a DB write or an external call, a crash between it and the offset commit still causes redelivery. In practice it is at-least-once plus idempotent handling, often called effectively-once.
Xem lag theo partition: dồn vào một partition là key nóng hoặc poison message; đều ở mọi partition là thiếu năng lực. Thêm instance (tối đa bằng số partition), tối ưu batch ghi DB, hoặc tăng partition. Nếu downstream quá tải thì pause partition thay vì đẩy thêm tải.Check lag per partition: a single hot partition points to a hot key or a poison message; even lag everywhere means too little capacity. Add instances up to the partition count, batch DB writes, or add partitions. If the downstream is overloaded, pause partitions instead of pushing more load.
Kafka giao theo partition tuần tự, nên một message thử lại mãi sẽ chặn mọi message phía sau và có thể làm hết max.poll.interval, gây rebalance. Topic retry tách message lỗi ra để partition chạy tiếp.Kafka delivers a partition sequentially, so a message retried forever blocks everything behind it and can exceed max.poll.interval, triggering a rebalance. Retry topics move the failing message aside so the partition keeps flowing.
Topic phải giữ log đủ lâu. Dừng group, reset offset theo thời gian (kafka-consumer-groups --reset-offsets --to-datetime) hoặc tạo group mới. Consumer phải idempotent; nếu muốn xử lý lại thật thì dùng group mới với bảng loại trùng riêng.The topic must retain enough log. Stop the group and reset offsets by time (kafka-consumer-groups --reset-offsets --to-datetime) or start a new group. Consumers must be idempotent; to actually reprocess, use a new group with its own dedup table.
Phím 1–4 để chọn đáp án, Enter để kiểm tra.Press 1–4 to choose an answer, Enter to check.
Ôn tập ngắt quãng (spaced repetition): thẻ được lặp lại sau 1 → 3 → 7 → 14 → 30 ngày để củng cố trí nhớ dài hạn.Spaced repetition: cards are repeated after 1 → 3 → 7 → 14 → 30 days to build long-term memory.