5.3 Consumer Delivery Semantics

Module Overview

Goal: Provide the ONE authoritative mapping between offset management and delivery guarantees.

What You'll Learn:

At-most-once delivery (consumer side)
At-least-once delivery (consumer side)
Exactly-once delivery challenges
How commit timing determines semantics
Failure windows and duplication scenarios
Idempotence as a solution

What's NOT Covered Here:

Producer-side semantics (covered in Producers Module 5)
End-to-end exactly-once (covered in Module 10.1)
Transactional processing (covered in Module 10.1)

Prerequisites:

Module 5.1 (Offsets Explained)
Module 5.2 (Commit Strategies)

Key Insight: Delivery semantics are determined by when you commit offsets relative to when you process messages.

1. The Three Delivery Semantics

Definitions

At-Most-Once

"Each message is delivered zero or one time. Messages may be lost, but never reprocessed."

Guarantee: No duplicates
Risk: Data loss

At-Least-Once

"Each message is delivered one or more times. Messages never lost, but may be reprocessed."

Guarantee: No data loss
Risk: Duplicates

Exactly-Once

"Each message is delivered exactly one time. No loss, no duplicates."

Guarantee: Perfect delivery
Challenge: Very hard to achieve

Visual Comparison

At-Most-Once:
[msg1] [msg2] [msg3] [msg4]
   ✓      ✓      ✗      ✓
Delivered: 1, 2, 4
Lost: 3

At-Least-Once:
[msg1] [msg2] [msg3] [msg4]
   ✓      ✓✓     ✓      ✓
Delivered: 1, 2, 2, 3, 4
Duplicate: 2

Exactly-Once:
[msg1] [msg2] [msg3] [msg4]
   ✓      ✓      ✓      ✓
Delivered: 1, 2, 3, 4
Perfect!

2. How Commit Timing Determines Semantics

Key Rule

Delivery semantics = When you commit relative to when you process

Process THEN commit → At-least-once
Commit THEN process → At-most-once
Commit AND process (transactionally) → Exactly-once

3. At-Least-Once (Kafka Default)

Implementation

props.put("enable.auto.commit", "false");

while (true) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
    
    for (ConsumerRecord<String, String> record : records) {
        processRecord(record);  // ← Process first
    }
    
    consumer.commitSync();  // ← Commit after
}

Pattern: Process → Commit

Why At-Least-Once?

Timeline:
1. Consumer reads offset 100
2. Consumer processes offset 100  ✅
3. Consumer commits offset 101  ✅
4. Continue...

Normal case: Each message processed exactly once

Failure Scenario

Timeline:
1. Consumer reads offset 100
2. Consumer processes offset 100  ✅
3. Consumer crashes! (before commit)
   
Restart:
4. Consumer reads last committed offset: 99
5. Consumer reads offset 100 again
6. Consumer processes offset 100 again  ✅✅

Result: Offset 100 processed TWICE

Key Point: Crash between process and commit = duplication.

At-Least-Once Characteristics

Guarantees:

✅ No message loss
✅ Every message processed at least once

Risks:

❌ Duplicates on failure
❌ Must be idempotent or handle duplicates

Use when:

Data loss unacceptable
Can handle duplicates (idempotent processing)
Most common pattern

At-Least-Once with Auto-Commit Risk

props.put("enable.auto.commit", "true");
props.put("auto.commit.interval.ms", "5000");

while (true) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
    
    for (ConsumerRecord<String, String> record : records) {
        processRecord(record); // Slow processing (6 seconds)
    }
}

Problem:

Time  Action
────┼──────────────────────────────────────
  0s│ poll() → offsets 0-99
  1s│ Processing 0-99...
  5s│ poll() → auto-commit offset 100 ← Committed!
  6s│ Still processing 0-99...
  7s│ CRASH!

Restart: Resume from 100
Result: Lost 0-99! (At-most-once!)

Lesson: Auto-commit can accidentally create at-most-once if processing is slow.

4. At-Most-Once

Implementation

props.put("enable.auto.commit", "false");

while (true) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
    
    consumer.commitSync();  // ← Commit FIRST
    
    for (ConsumerRecord<String, String> record : records) {
        processRecord(record);  // ← Process AFTER
    }
}

Pattern: Commit → Process

Why At-Most-Once?

Timeline:
1. Consumer reads offset 100
2. Consumer commits offset 101  ✅
3. Consumer processes offset 100  ✅
4. Continue...

Normal case: Each message processed exactly once

Failure Scenario

Timeline:
1. Consumer reads offset 100
2. Consumer commits offset 101  ✅
3. Consumer starts processing offset 100
4. Consumer crashes! (during processing)
   
Restart:
5. Consumer reads last committed offset: 101
6. Consumer skips offset 100

Result: Offset 100 NEVER processed (LOST!)

Key Point: Crash between commit and process = data loss.

At-Most-Once Characteristics

Guarantees:

✅ No duplicates
✅ Each message processed at most once

Risks:

❌ Message loss on failure
❌ Cannot recover from processing failures

Use when:

Duplicates unacceptable
Message loss tolerable
Metrics, logs, non-critical data

5. Exactly-Once (The Challenge)

Why Exactly-Once is Hard

Problem: Kafka guarantees delivery, but not processing.

Scenario:
1. Consumer reads offset 100
2. Consumer processes offset 100 (writes to database)  ✅
3. Consumer crashes before commit
4. Consumer restarts, reads offset 100 again
5. Consumer processes offset 100 again (writes to database)  ✅✅

Result: Database has duplicate!

Kafka can't control:

Your database
Your external API calls
Your file writes
Your side effects

Exactly-Once Requirements

To achieve exactly-once, you need both:

Atomic commit + processing
Offset commit and processing must succeed or fail together
Idempotent processing
Processing same message twice has same effect as once

Approach 1: Idempotent Processing

Make processing idempotent, then use at-least-once delivery.

// NOT idempotent
void processOrder(Order order) {
    inventory.decrement(order.productId);  // Run twice → wrong count!
}

// Idempotent
void processOrder(Order order) {
    if (!db.orderProcessed(order.id)) {  // Check first
        inventory.decrement(order.productId);
        db.markOrderProcessed(order.id);
    }
}

Idempotence transforms:

At-least-once → Effectively exactly-once

Approach 2: Transactional Processing

Store offset with processing results in the same transaction.

// Pseudo-code
while (true) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
    
    for (ConsumerRecord<String, String> record : records) {
        db.beginTransaction();
        try {
            // Process
            db.insert(record.value());
            
            // Store offset in same DB
            db.storeOffset(record.topic(), record.partition(), record.offset());
            
            db.commit();  // Atomic!
        } catch (Exception e) {
            db.rollback();
        }
    }
}

// On restart, read offset from DB instead of Kafka

Key: Offset and result committed atomically.

Approach 3: Kafka Transactions (Producer + Consumer)

For consume-transform-produce pipelines (Kafka Streams pattern):

// Configure transactional consumer
props.put("isolation.level", "read_committed");
props.put("enable.auto.commit", "false");

// Configure transactional producer
producerProps.put("transactional.id", "my-transactional-id");

producer.initTransactions();

while (true) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
    
    producer.beginTransaction();
    try {
        for (ConsumerRecord<String, String> record : records) {
            // Transform
            String result = transform(record.value());
            
            // Produce
            producer.send(new ProducerRecord<>("output-topic", result));
        }
        
        // Commit offsets to transaction
        producer.sendOffsetsToTransaction(
            consumer.committed(/* current offsets */),
            consumer.groupMetadata()
        );
        
        producer.commitTransaction();  // Atomic!
    } catch (Exception e) {
        producer.abortTransaction();
    }
}

Result: Consume + produce is atomic (exactly-once end-to-end).

Transactional Configuration: `isolation.level`

When working with Kafka transactions (either as a consumer reading transactional data or in consume-transform-produce pipelines), the isolation.level configuration is critical.

What is `isolation.level`?

The isolation.level configuration controls what transactional data your consumer can see.

props.put("isolation.level", "read_committed");  // or "read_uncommitted"

Two Isolation Levels

Value	Behavior	Use Case
`read_uncommitted`	Consumer sees ALL records, including those from uncommitted/aborted transactions	Default behavior, highest throughput, no transactional guarantees
`read_committed`	Consumer ONLY sees records from committed transactions. Uncommitted/aborted records are filtered out	Exactly-once processing, transactional pipelines

`read_uncommitted` (Default)

props.put("isolation.level", "read_uncommitted");

Behavior:

Producer writes to topic with transactions:
- Message 1: COMMITTED
- Message 2: UNCOMMITTED (transaction in progress)
- Message 3: ABORTED (transaction rolled back)
- Message 4: COMMITTED

Consumer with isolation.level=read_uncommitted:
├─ Sees Message 1 ✅
├─ Sees Message 2 ✅ (even though uncommitted!)
├─ Sees Message 3 ✅ (even though aborted!)
└─ Sees Message 4 ✅

Result: Consumer sees ALL messages, regardless of transaction state

Characteristics:

✅ Highest throughput (no filtering)
✅ Lowest latency (immediate visibility)
❌ No transactional guarantees (sees dirty data)
❌ Can see aborted messages (later invisible after cleanup)

Use when:

You don't care about transactional semantics
Producer is not using transactions
Maximum performance required
Reading logs, metrics, non-critical data

`read_committed` (Transactional)

props.put("isolation.level", "read_committed");

Behavior:

Producer writes to topic with transactions:
- Message 1: COMMITTED
- Message 2: UNCOMMITTED (transaction in progress)
- Message 3: ABORTED (transaction rolled back)
- Message 4: COMMITTED

Consumer with isolation.level=read_committed:
├─ Sees Message 1 ✅
├─ Does NOT see Message 2 ❌ (uncommitted, hidden until committed)
├─ Does NOT see Message 3 ❌ (aborted, filtered out)
└─ Sees Message 4 ✅

Result: Consumer ONLY sees committed messages

Characteristics:

✅ Transactional guarantees (no dirty reads)
✅ No aborted messages (clean data)
✅ Exactly-once friendly (required for EOS)
⚠️ Slightly higher latency (waits for transaction commit)
⚠️ Filtering overhead (broker filters uncommitted records)

Use when:

Producer uses transactions
Exactly-once semantics required
Consume-transform-produce pipelines
Financial systems, critical data processing

How `read_committed` Works Internally

Broker maintains transaction state for each message:
┌─────────────────────────────────────────────────┐
│ Offset │ Message │ Transaction State           │
├────────┼─────────┼─────────────────────────────┤
│   100  │  msg1   │  COMMITTED                  │
│   101  │  msg2   │  IN_PROGRESS (txn-123)      │
│   102  │  msg3   │  ABORTED (txn-124)          │
│   103  │  msg4   │  COMMITTED                  │
└─────────────────────────────────────────────────┘

Consumer with read_committed:
1. Fetch request includes isolation.level=read_committed
2. Broker filters response:
   ├─ Include: COMMITTED messages (100, 103)
   └─ Exclude: IN_PROGRESS (101), ABORTED (102)
3. Consumer receives only committed messages

Key Points:

Filtering happens broker-side (not client-side)
Consumer never sees uncommitted data
Control records (transaction markers) are hidden from application

Configuration Example

Properties props = new Properties();
props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
props.put(ConsumerConfig.GROUP_ID_CONFIG, "my-group");
props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());

// For exactly-once consumer (reading transactional data)
props.put(ConsumerConfig.ISOLATION_LEVEL_CONFIG, "read_committed");
props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");

KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);

When to Use Each Isolation Level

Use read_uncommitted (default) if:

✅ Producer is NOT using transactions
✅ You don't need transactional guarantees
✅ Maximum throughput is critical
✅ Logs, metrics, or non-critical data

Use read_committed if:

✅ Producer is using transactions (transactional.id set)
✅ You need exactly-once semantics
✅ Consume-transform-produce pipeline (Kafka Streams pattern)
✅ Financial transactions, order processing, critical data

Impact on Consumer Lag

With read_uncommitted:
├─ Consumer lag reflects all messages (committed + uncommitted)
└─ Example: 1000 messages in partition → lag shows 1000

With read_committed:
├─ Consumer lag reflects only committed messages
└─ Example: 1000 messages (800 committed, 200 uncommitted) → lag shows 800

Monitoring Consideration:

Consumer lag for read_committed consumers may appear lower
This is expected (uncommitted messages not counted)
Use records-lag metric, not records-lead

Transactional Markers (Control Records)

Producers using transactions write control records to mark transaction boundaries:

Partition:
[msg1: COMMIT_MARKER (txn-123)]  ← Transaction 123 committed
[msg2: user data]
[msg3: user data]
[msg4: ABORT_MARKER (txn-124)]   ← Transaction 124 aborted

For consumers:

read_uncommitted: Control records filtered by client (never seen by application)
read_committed: Control records filtered by broker AND client (never seen by application)

You never see control records in your application code, regardless of isolation level.

Exactly-Once with `read_committed`

// Consumer configuration (read committed data)
consumerProps.put("isolation.level", "read_committed");
consumerProps.put("enable.auto.commit", "false");

// Producer configuration (write transactionally)
producerProps.put("transactional.id", "my-txn-id-" + instanceId);
producerProps.put("enable.idempotence", "true");

// In processing loop
producer.initTransactions();

while (true) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
    
    producer.beginTransaction();
    try {
        for (ConsumerRecord<String, String> record : records) {
            // Transform and produce
            ProducerRecord<String, String> output = transform(record);
            producer.send(output);
        }
        
        // Commit consumer offsets within producer transaction
        Map<TopicPartition, OffsetAndMetadata> offsets = /* current offsets */;
        producer.sendOffsetsToTransaction(offsets, consumer.groupMetadata());
        
        producer.commitTransaction();
        // ✅ Exactly-once: Output messages + offset commits are atomic
    } catch (Exception e) {
        producer.abortTransaction();
        // ❌ Everything rolled back (output + offsets)
    }
}

Result:

✅ Consumer only sees committed input data
✅ Output and offset commits are atomic
✅ End-to-end exactly-once semantics

Performance Implications

Throughput:

Isolation Level	Relative Throughput	Latency Overhead
`read_uncommitted`	100% (baseline)	None
`read_committed`	95-98%	+5-50ms (waits for commit)

Latency Impact:

read_uncommitted:
Producer writes → Consumer sees immediately (0ms delay)

read_committed:
Producer writes → Transaction commits → Consumer sees (+5-50ms delay)

The overhead is usually acceptable for applications requiring exactly-once guarantees.

Common Misconceptions

❌ "I need read_committed to avoid duplicates"

Reality: isolation.level only affects transactional visibility. Consumer-side deduplication requires idempotent processing or transactional storage.

❌ "Using read_committed guarantees exactly-once"

Reality: It's one component of exactly-once. You also need transactional producers and atomic offset commits.

❌ "All consumers should use read_committed"

Reality: Only use if upstream producers are transactional. Otherwise, it adds overhead with no benefit.

✅ "read_committed prevents reading uncommitted data"

Correct! This is exactly what it does.

Decision Tree

Does your producer use transactions?
├─ NO  → Use read_uncommitted (default)
│        (No benefit from read_committed)
│
└─ YES → Do you need exactly-once?
         ├─ NO  → Use read_uncommitted (higher throughput)
         │        (Accept potential dirty reads)
         │
         └─ YES → Use read_committed
                  (Required for exactly-once semantics)

6. Failure Windows

At-Least-Once Failure Window

┌────────────────────────────────────┐
│         FAILURE WINDOW             │
│  (Causes duplicate processing)     │
│                                    │
│  [Process] ─────────→ [Commit]     │
│      ↑                    ↑        │
│      │                    │        │
│   Start                 End        │
│                                    │
│  Crash in this window              │
│  → Duplicate on restart            │
└────────────────────────────────────┘

Minimize window:

Commit immediately after processing
Use fast processing
Batch commits (trade-off)

At-Most-Once Failure Window

┌────────────────────────────────────┐
│         FAILURE WINDOW             │
│   (Causes message loss)            │
│                                    │
│  [Commit] ─────────→ [Process]     │
│      ↑                    ↑        │
│      │                    │        │
│   Start                 End        │
│                                    │
│  Crash in this window              │
│  → Message lost                    │
└────────────────────────────────────┘

Minimize window:

Process immediately after commit
Use fast processing
Accept that loss is possible

7. Practical Patterns

Pattern 1: At-Least-Once with Idempotence (Recommended)

props.put("enable.auto.commit", "false");

while (true) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
    
    for (ConsumerRecord<String, String> record : records) {
        processIdempotently(record);  // Idempotent logic
    }
    
    consumer.commitSync();
}

void processIdempotently(ConsumerRecord record) {
    String id = record.key();
    
    // Check if already processed (idempotence check)
    if (!cache.contains(id)) {
        database.insert(record.value());
        cache.add(id);
    }
}

Pattern 2: Store Offset with Result

while (true) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
    
    for (ConsumerRecord<String, String> record : records) {
        // Store both result and offset atomically
        database.transaction(() -> {
            database.insert(record.value());
            database.updateOffset(
                record.topic(),
                record.partition(),
                record.offset()
            );
        });
    }
    
    // Don't commit to Kafka (using DB as source of truth)
}

// On startup: Seek to offset from database
Map<TopicPartition, Long> offsets = database.getOffsets();
for (Map.Entry<TopicPartition, Long> entry : offsets.entrySet()) {
    consumer.seek(entry.getKey(), entry.getValue());
}

Pattern 3: Deduplication with Unique IDs

// Producer adds unique ID
ProducerRecord<String, String> record = new ProducerRecord<>("orders", order.toJson());
record.headers().add("message-id", UUID.randomUUID().toString().getBytes());

// Consumer deduplicates
while (true) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
    
    for (ConsumerRecord<String, String> record : records) {
        String messageId = new String(record.headers().lastHeader("message-id").value());
        
        if (!processedIds.contains(messageId)) {
            processRecord(record);
            processedIds.add(messageId);
        }
    }
    
    consumer.commitSync();
}

8. Summary Table

Semantic	Pattern	Duplicates	Loss	Complexity	Use Case
At-Least-Once	Process → Commit	Possible	Never	Low	Most applications
At-Most-Once	Commit → Process	Never	Possible	Low	Metrics, logs
Exactly-Once	Atomic / Idempotent	Never	Never	High	Critical data, transactions

9. Choosing Your Semantic

Decision Tree

Can you tolerate message loss?
├─ YES → At-most-once (commit before process)
└─ NO → Can you make processing idempotent?
         ├─ YES → At-least-once + idempotence
         └─ NO → Need exactly-once
                  ├─ Kafka-to-Kafka → Transactions
                  └─ Kafka-to-DB → Store offset with result

Recommendations

For Most Applications (Recommended)

Use: At-least-once with idempotence

// Idempotent processing
void process(Record record) {
    if (!alreadyProcessed(record.key())) {
        doWork(record);
        markProcessed(record.key());
    }
}

// Standard commit
consumer.commitSync();

For Non-Critical Data

Use: At-most-once or relaxed at-least-once

// Accept occasional loss
consumer.commitSync();
processRecord(record);

// Or use auto-commit
props.put("enable.auto.commit", "true");

For Critical Financial Data

Use: Exactly-once with transactions

// Transactional processing
db.transaction(() -> {
    processRecord(record);
    db.storeOffset(record.partition(), record.offset());
});

10. Mental Model Summary

The Fundamental Trade-off

Commit Before Process:
└─ At-most-once (loss possible, no duplicates)

Process Before Commit:
└─ At-least-once (no loss, duplicates possible)

Atomic Commit + Process:
└─ Exactly-once (no loss, no duplicates, complex)

Questions You Should Be Able to Answer

✅ What are the three delivery semantics?
✅ How does commit timing determine semantics?
✅ What is at-least-once and when should you use it?
✅ What is at-most-once and when should you use it?
✅ Why is exactly-once hard for consumers?
✅ What is idempotence and how does it help?
✅ What is the failure window?

11. What's Next?

Now that you understand delivery semantics, the next modules will cover:

Module 6.1: Poll loop internals (how consumption actually works)
Module 7.1: Performance tuning
Module 8.1: Error handling and resilience
Module 10.1: End-to-end exactly-once patterns

Summary

You learned:

✅ At-least-once: Process then commit (duplicates possible, no loss)
✅ At-most-once: Commit then process (loss possible, no duplicates)
✅ Exactly-once: Requires idempotence or transactions
✅ Delivery semantics determined by commit timing
✅ Failure windows explain duplicate/loss risk
✅ Idempotence transforms at-least-once into effectively exactly-once
✅ Most applications should use at-least-once with idempotence

Next up: Consumer Lifecycle & Poll Loop Internals (Module 6.1) — Learn how the poll loop actually works under the hood.

5.3 Consumer Delivery Semantics

Module Overview

1. The Three Delivery Semantics

Definitions

At-Most-Once

At-Least-Once

Exactly-Once

Visual Comparison

2. How Commit Timing Determines Semantics

Key Rule

3. At-Least-Once (Kafka Default)

Implementation

Why At-Least-Once?

Failure Scenario

At-Least-Once Characteristics

At-Least-Once with Auto-Commit Risk

4. At-Most-Once

Implementation

Why At-Most-Once?

Failure Scenario

At-Most-Once Characteristics

5. Exactly-Once (The Challenge)

Why Exactly-Once is Hard

Exactly-Once Requirements

Approach 1: Idempotent Processing

Approach 2: Transactional Processing

Approach 3: Kafka Transactions (Producer + Consumer)

Transactional Configuration: isolation.level

What is isolation.level?

Two Isolation Levels

read_uncommitted (Default)

read_committed (Transactional)

How read_committed Works Internally

Configuration Example

When to Use Each Isolation Level

Impact on Consumer Lag

Transactional Markers (Control Records)

Exactly-Once with read_committed

Performance Implications

Common Misconceptions

Decision Tree

6. Failure Windows

At-Least-Once Failure Window

At-Most-Once Failure Window

7. Practical Patterns

Pattern 1: At-Least-Once with Idempotence (Recommended)

Pattern 2: Store Offset with Result

Pattern 3: Deduplication with Unique IDs

8. Summary Table

9. Choosing Your Semantic

Decision Tree

Recommendations

For Most Applications (Recommended)

For Non-Critical Data

For Critical Financial Data

10. Mental Model Summary

The Fundamental Trade-off

Questions You Should Be Able to Answer

11. What's Next?

Summary

Transactional Configuration: `isolation.level`

What is `isolation.level`?

`read_uncommitted` (Default)

`read_committed` (Transactional)

How `read_committed` Works Internally

Exactly-Once with `read_committed`