Understanding Message Delivery Semantics
1. Understanding Delivery Guarantees
| Semantic | Meaning | Tradeoff |
|---|---|---|
| At-most-once | May lose, never duplicate | Commit before process |
| At-least-once | Never lose, may duplicate | Commit after process |
| Exactly-once | No loss, no dup | Transactions |
Example: At-least-once
for (var r : records) process(r);
consumer.commitSync(); // commit AFTER processing2. Understanding At-Most-Once Delivery
| Aspect | Description | Detail |
|---|---|---|
| Order | Commit then process | Loss on crash |
| No Dup | Never reprocesses | Guaranteed |
| Use Case | Metrics where loss is OK | Lossy tolerable |
Example: At-most-once
consumer.commitSync(); // commit FIRST
for (var r : records) process(r); // crash here = data lost3. Understanding At-Least-Once Delivery
| Aspect | Description | Detail |
|---|---|---|
| Order | Process then commit | Default pattern |
| Duplicates | Reprocess on crash | Need idempotent ops |
| acks=all | Producer durability | No loss |
Example: At-least-once with idempotent sink
for (var r : records) upsert(r.key(), r.value()); // idempotent
consumer.commitSync();4. Implementing Exactly-Once Semantics
| Mechanism | Role | Detail |
|---|---|---|
| Idempotent Producer | No dup on retry | Per partition |
| Transactions | Atomic multi-partition write | Consume-transform-produce |
| read_committed | Consumer skips aborts | End-to-end EOS |
5. Configuring Producer Idempotence
| Config | Description | Detail |
|---|---|---|
| enable.idempotence | true | Default 3.0+ |
| acks | Forced to all | Required |
| PID + Epoch | Broker dedup key | Per session |
6. Using Transactional Producer
| Element | Description | Detail |
|---|---|---|
| transactional.id | Stable producer identity | Fencing |
| initTransactions | Register and recover | Once at start |
| commit/abort | Atomic batch outcome | All or nothing |
Example: Transactional send
props.put("transactional.id", "tx-1");
producer.initTransactions();
producer.beginTransaction();
producer.send(record);
producer.commitTransaction();7. Setting Consumer Isolation Level
| Level | Behavior | Use Case |
|---|---|---|
| read_uncommitted | See aborted records | Default |
| read_committed | Only committed transactions | EOS |
| LSO bound | Reads stop at last stable offset | Tx-aware |
8. Understanding Duplicate Messages
| Cause | Description | Fix |
|---|---|---|
| Producer Retry | Ack lost, resend | Idempotence |
| Reprocessing | Crash before commit | Idempotent consumer |
| Rebalance | Uncommitted offsets replayed | Commit on revoke |
Example: Dedup by key+offset
String id = r.topic() + "-" + r.partition() + "-" + r.offset();
if (seen.add(id)) process(r); // skip duplicates9. Handling Message Ordering
| Aspect | Description | Detail |
|---|---|---|
| Idempotence | Preserves order on retry | Per partition |
| EOS | Order + no dup | Transactions |
| Key Routing | Same key stays ordered | Co-located |
Example: Ordered EOS pipeline
props.put("enable.idempotence", "true");
props.put("transactional.id", "ordered-tx");10. Implementing Idempotent Consumer
| Strategy | Description | Detail |
|---|---|---|
| Upsert | Write keyed, overwrite-safe | DB merge |
| Dedup Store | Track processed IDs | Redis/DB |
| Atomic Commit | Persist offset with data | Same tx |
Example: Atomic DB upsert + offset
tx.begin();
db.upsert(r.key(), r.value());
db.saveOffset(r.partition(), r.offset() + 1);
tx.commit(); // exactly-once into DB11. Understanding Message Loss Scenarios
| Scenario | Cause | Prevention |
|---|---|---|
| acks=0/1 | Leader fails before replicate | acks=all |
| Unclean Election | Stale replica promoted | Disable unclean |
| Early Commit | Commit before process | Commit after |