9.1 Exactly-Once Processing in Kafka Streams

Module Overview

Goal: Master exactly-once semantics (EOS) in Kafka Streams.

What You'll Learn:

What exactly-once means
How Kafka Streams achieves EOS
EOS v1 vs EOS v2 (ALOS vs EOS)
Transactional processing
Producer transactions and consumer offsets
Performance implications
Configuration and best practices

What's NOT Covered Here:

General delivery semantics (at-most-once, at-least-once) — conceptually covered, but deep producer/consumer specifics are in their respective modules
State stores (covered in Module 6.1)
Changelog topics (covered in Module 6.2)

Key Insight: EOS is the authoritative source — all concepts related to transactions, commits, idempotency, and exactly-once guarantees live here.

1. Understanding Delivery Semantics

The Three Guarantees

AT-MOST-ONCE:
├─ Messages delivered 0 or 1 times
├─ May lose messages
└─ No duplicates

AT-LEAST-ONCE:
├─ Messages delivered 1 or more times
├─ No data loss
└─ May have duplicates

EXACTLY-ONCE:
├─ Messages delivered exactly 1 time
├─ No data loss
└─ No duplicates

Why Exactly-Once is Hard

Distributed system challenges:

1. Network failures
   ├─ Producer sends message
   ├─ Broker receives it
   └─ Ack lost → producer retries → duplicate

2. Multiple operations
   ├─ Read from Kafka
   ├─ Process data
   ├─ Write to Kafka
   ├─ Commit offset
   └─ Any failure → inconsistency

3. State updates
   ├─ Update state store
   ├─ Write to changelog
   └─ Failure → state mismatch

Exactly-once requires atomicity across ALL these

2. Kafka Streams Exactly-Once Approach

The Solution: Transactions

Kafka Streams uses transactions to make operations atomic:

1. Read-process-write as atomic unit
   ├─ Read records
   ├─ Process (update state, write output)
   ├─ Commit offsets
   └─ ALL or NOTHING

2. Idempotent producer
   └─ Prevents duplicates from retries

3. Transactional reads
   └─ Consumers only see committed data

Together: Exactly-once semantics

High-Level Flow

Transaction boundary (commit.interval.ms):

BEGIN TRANSACTION
├─ Read records from input topic
├─ Process records
├─ Update state store
├─ Write to changelog topic
├─ Write to output topic
├─ Commit consumer offsets
COMMIT TRANSACTION

All succeed → visible to downstream
Any failure → rollback → retry

3. Enabling Exactly-Once Semantics

Configuration

Properties props = new Properties();
props.put(StreamsConfig.APPLICATION_ID_CONFIG, "my-app");
props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");

// Enable exactly-once semantics (v2, recommended)
props.put(StreamsConfig.PROCESSING_GUARANTEE_CONFIG, 
    StreamsConfig.EXACTLY_ONCE_V2);

// Alternatives:
// StreamsConfig.AT_LEAST_ONCE (default, no EOS)
// StreamsConfig.EXACTLY_ONCE (v1, older, slower)

EOS v1 vs EOS v2

EOS V1 (Kafka < 2.5):
├─ Uses transactional producer per task
├─ Zombie fencing via transaction.id
├─ Slower (more transactions)
└─ Higher overhead

EOS V2 (Kafka 2.5+):
├─ Uses single transactional producer per thread
├─ Zombie fencing via group.instance.id
├─ Faster (fewer transactions)
├─ Lower overhead
└─ Preferred for production

Use EXACTLY_ONCE_V2 unless on older Kafka

4. How Exactly-Once Works

Components

1. IDEMPOTENT PRODUCER:
   ├─ Enabled automatically with EOS
   ├─ Producer ID + sequence number
   └─ Broker deduplicates retries

2. TRANSACTIONS:
   ├─ Atomic writes across partitions
   ├─ Transaction coordinator tracks state
   └─ All-or-nothing commit

3. TRANSACTIONAL READS:
   ├─ Consumer reads only committed data
   ├─ isolation.level=read_committed
   └─ Uncommitted data invisible

4. OFFSET COMMIT IN TRANSACTION:
   ├─ Offsets committed as part of transaction
   ├─ Read and write atomic
   └─ No offset/output mismatch

Visual Flow

┌────────────────────────────────────────────────┐
│  KAFKA STREAMS INSTANCE                        │
│                                                │
│  BEGIN TRANSACTION                             │
│  ├─ Read: offset 100-105 (input-topic)        │
│  ├─ Process: aggregate, transform              │
│  ├─ Write: state store (local)                 │
│  ├─ Write: changelog topic (transactional)     │
│  ├─ Write: output topic (transactional)        │
│  └─ Commit: offset 106 (transactional)         │
│  COMMIT TRANSACTION                            │
│                                                │
│  Result: All visible atomically or none        │
└────────────────────────────────────────────────┘

5. Idempotent Producer

What is Idempotence?

Idempotence = same operation multiple times → same result

Producer idempotence:
├─ Producer ID (PID) assigned by broker
├─ Sequence number per partition
└─ Broker tracks (PID, sequence) → deduplicates

Example:
├─ Producer sends record (PID=5, seq=10)
├─ Broker receives, ack lost
├─ Producer retries (PID=5, seq=10)
└─ Broker: "Already received seq=10, ignore" → no duplicate

Configuration

// Idempotent producer enabled automatically with EOS
props.put(StreamsConfig.PROCESSING_GUARANTEE_CONFIG, 
    StreamsConfig.EXACTLY_ONCE_V2);

// Internally sets:
// enable.idempotence=true
// acks=all
// max.in.flight.requests.per.connection=5 (default, can be higher)
// retries=Integer.MAX_VALUE

Limitations

Idempotence guarantees:
├─ Within a single producer session
├─ No duplicates from retries
└─ Order preserved per partition

Does NOT guarantee:
├─ Across producer restarts (PID changes)
├─ Multiple producers
└─ Exactly-once by itself (needs transactions)

Use transactions for true exactly-once

6. Transactions

What is a Transaction?

Transaction = atomic unit of work across multiple partitions

Properties (ACID):
├─ Atomicity: All writes succeed or none
├─ Consistency: Valid state transitions
├─ Isolation: Uncommitted data invisible
└─ Durability: Committed data persists

Kafka Streams transaction includes:
├─ Output topic writes
├─ Changelog topic writes
├─ Consumer offset commits

Transaction Lifecycle

1. BEGIN:
   └─ Producer sends BeginTransaction

2. WRITE:
   ├─ Write to output topics (uncommitted)
   ├─ Write to changelog topics (uncommitted)
   └─ All writes marked with transaction ID

3. COMMIT:
   ├─ Producer sends CommitTransaction
   ├─ Transaction coordinator marks committed
   └─ Data visible to consumers

4. ABORT (on failure):
   ├─ Producer sends AbortTransaction
   ├─ Transaction coordinator marks aborted
   └─ Data never visible to consumers

Transaction Coordinator

Transaction Coordinator (broker-side):
├─ Tracks transaction state
├─ Manages transaction log (__transaction_state)
├─ Coordinates commits/aborts
└─ Ensures atomicity

Transaction flow:
Producer → Transaction Coordinator → Brokers

7. Transactional Reads

Read Committed Isolation

// Automatically set with EOS
props.put(StreamsConfig.PROCESSING_GUARANTEE_CONFIG, 
    StreamsConfig.EXACTLY_ONCE_V2);

// Internally sets consumer config:
// isolation.level=read_committed

// Consumer only reads committed data
// Uncommitted data buffered until commit/abort

How It Works

Timeline:

1. Producer begins transaction
2. Writes records to topic (uncommitted)
3. Consumer polls
   └─ Records buffered (not returned)
4. Producer commits transaction
5. Consumer polls
   └─ Records now visible and returned

Aborted transaction:
└─ Records discarded, never visible

Performance Impact

WITH read_committed:
├─ Consumer waits for transaction commit
├─ Higher latency (buffering)
└─ Guaranteed no uncommitted data

WITHOUT read_committed (read_uncommitted):
├─ Consumer reads immediately
├─ Lower latency
└─ May read data from aborted transactions

Trade-off: Correctness vs latency

8. Commit Interval

What is Commit Interval?

commit.interval.ms = how often to commit transactions

Default: 30 seconds (30000 ms)

Behavior:
├─ Process records for commit.interval.ms
├─ Commit transaction (output + offsets + changelog)
└─ Repeat

Smaller interval:
├─ More frequent commits
├─ Lower latency
├─ Higher overhead

Larger interval:
├─ Fewer commits
├─ Higher latency (data visible less often)
├─ Lower overhead

Configuration

// Commit every 10 seconds
props.put(StreamsConfig.COMMIT_INTERVAL_MS_CONFIG, 10000);

// Commit every minute (low overhead)
props.put(StreamsConfig.COMMIT_INTERVAL_MS_CONFIG, 60000);

// Commit every 5 seconds (low latency)
props.put(StreamsConfig.COMMIT_INTERVAL_MS_CONFIG, 5000);

Trade-offs

SHORT COMMIT INTERVAL (e.g., 5 seconds):
✅ Lower end-to-end latency
✅ Smaller transactions
✅ Faster recovery (less replay)
❌ Higher overhead (more commits)
❌ Lower throughput

LONG COMMIT INTERVAL (e.g., 60 seconds):
✅ Higher throughput
✅ Lower overhead
❌ Higher latency (data visible less often)
❌ Slower recovery (more replay)

9. Exactly-Once Across Application Boundaries

Internal vs External

WITHIN KAFKA STREAMS (INTERNAL):
✅ Exactly-once guaranteed
├─ Kafka → Kafka Streams → Kafka
└─ All transactional

EXTERNAL SYSTEMS:
❌ Exactly-once NOT guaranteed automatically
├─ Kafka → Kafka Streams → Database/API
└─ Requires additional work

Achieving External Exactly-Once

Two approaches:

1. IDEMPOTENT WRITES:
   ├─ External system deduplicates
   ├─ Example: Upsert to database (same key overwrites)
   └─ Application-level idempotence

2. TRANSACTIONAL OUTBOX PATTERN:
   ├─ Write to Kafka + DB in single transaction
   ├─ Outbox table in DB
   ├─ Separate process reads outbox → external system
   └─ Complex but reliable

Simplest: Design external writes to be idempotent

10. Zombie Fencing

The Zombie Problem

Scenario:

1. Instance A processes task 0
2. Network partition (A appears dead)
3. Rebalance: Instance B assigned task 0
4. Network recovered: Instance A alive (zombie!)
5. Both A and B process task 0 → duplicates

Solution: Zombie fencing

How Zombie Fencing Works

EOS v1 (transaction.id)

1. Each task gets unique transaction.id
   └─ Format: {application.id}-{task.id}

2. Broker tracks transaction.id → producer epoch
   └─ Latest epoch wins

3. Zombie writes rejected
   ├─ Old epoch → ProducerFencedException
   └─ Zombie shuts down

Downside: One producer per task (high overhead)

EOS v2 (group.instance.id)

1. Static membership (group.instance.id)
   └─ Instance has fixed ID

2. Single producer per thread (not per task)
   └─ Lower overhead

3. Zombie fencing via consumer group protocol
   ├─ New instance joins → fences old
   └─ Zombie's writes rejected

Advantage: Fewer producers, better performance

Configuration (EOS v2)

// Static membership for zombie fencing
props.put(StreamsConfig.PROCESSING_GUARANTEE_CONFIG, 
    StreamsConfig.EXACTLY_ONCE_V2);

// Optionally set static group instance ID
// (Usually auto-assigned)
props.put(ConsumerConfig.GROUP_INSTANCE_ID_CONFIG, "instance-1");

// Higher session timeout for stability
props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, 45000);

11. Performance Implications

Overhead of EOS

Compared to at-least-once:

LATENCY:
├─ +10-30% higher
└─ Transaction commit overhead

THROUGHPUT:
├─ -10-20% lower
└─ Transaction coordination overhead

CPU:
├─ Slightly higher
└─ Transaction management

Memory:
├─ Similar
└─ No significant difference

Network:
├─ Slightly higher
└─ Transaction metadata

When to Use EOS

USE EXACTLY-ONCE WHEN:
✅ Duplicates unacceptable (financial, billing)
✅ Correctness critical
✅ Can tolerate latency overhead
✅ Kafka 2.5+ (EOS v2)

USE AT-LEAST-ONCE WHEN:
✅ Duplicates acceptable
✅ Idempotent downstream processing
✅ Need maximum throughput
✅ Latency sensitive

Optimizing EOS Performance

1. Use EOS v2 (not v1)
   └─ 3-5× faster than v1

2. Tune commit interval
   └─ Balance latency vs throughput

3. Increase batch size
   └─ props.put(StreamsConfig.CACHE_MAX_BYTES_BUFFERING_CONFIG, ...)

4. Use compression
   └─ Reduce network overhead

5. Ensure broker capacity
   └─ Transaction coordination needs resources

12. Failure Scenarios

Scenario 1: Producer Failure Mid-Transaction

Timeline:

1. BEGIN transaction
2. Write records (uncommitted)
3. Producer crashes (no commit)
4. Transaction timeout (transaction.timeout.ms)
5. Broker aborts transaction
6. New producer starts
7. Resumes from last committed offset
8. Reprocesses records (exactly-once maintained)

Result: No data loss, no duplicates

Scenario 2: Broker Failure

Timeline:

1. Producer writes to broker (uncommitted)
2. Broker crashes
3. Producer retries (idempotent)
4. Writes to new leader
5. Commits transaction
6. Data visible

Result: Exactly-once maintained

Scenario 3: Consumer Rebalance

Timeline:

1. Instance A processing task 0
2. Instance B joins → rebalance
3. Task 0 reassigned to B
4. A's in-flight transaction aborted
5. B starts from last committed offset
6. Reprocesses uncommitted records

Result: Exactly-once maintained

13. Monitoring EOS

Important Metrics

# Transaction state
kafka.streams:type=stream-thread-metrics,
              thread-id=*,
              name=commit-latency-avg

# Failed transactions
kafka.streams:type=stream-thread-metrics,
              thread-id=*,
              name=failed-stream-threads

# Producer metrics (transaction)
kafka.producer:type=producer-metrics,
               client-id=*,
               name=txn-commit-time-avg

# Consumer lag (should be low)
kafka.consumer:type=consumer-fetch-manager-metrics,
               client-id=*,
               name=records-lag-max

Checking Transaction State

# List transaction state topics
kafka-topics.sh --bootstrap-server localhost:9092 --list | grep transaction

# __transaction_state topic (internal)
kafka-console-consumer.sh --bootstrap-server localhost:9092 \
  --topic __transaction_state \
  --from-beginning \
  --formatter kafka.coordinator.transaction.TransactionLog\$TransactionLogMessageFormatter

14. Common EOS Issues

Issue 1: ProducerFencedException

Error:
org.apache.kafka.common.errors.ProducerFencedException

Cause:
├─ Zombie instance writing with old epoch
└─ Transaction ID fenced by newer producer

Solution:
├─ Instance should shut down (automatic)
└─ Ensure static membership configured correctly

Issue 2: TransactionCoordinatorFencedException

Error:
org.apache.kafka.common.errors.TransactionCoordinatorFencedException

Cause:
├─ Transaction coordinator changed
└─ Broker rebalanced

Solution:
├─ Retry (automatic)
└─ Ensure broker stability

Issue 3: High Commit Latency

Symptom: Slow transaction commits

Causes:
├─ High broker load
├─ Large transactions
├─ Network latency
└─ Many partitions

Solutions:
1. Scale brokers
2. Reduce commit interval (smaller transactions)
3. Optimize network
4. Reduce partition count (if excessive)

15. Best Practices

1. Use EOS v2

// Always use v2 (unless Kafka < 2.5)
props.put(StreamsConfig.PROCESSING_GUARANTEE_CONFIG, 
    StreamsConfig.EXACTLY_ONCE_V2);

2. Tune Commit Interval

// Balance latency vs throughput
// Start with 30s, adjust based on requirements
props.put(StreamsConfig.COMMIT_INTERVAL_MS_CONFIG, 30000);

// Low latency use case
props.put(StreamsConfig.COMMIT_INTERVAL_MS_CONFIG, 5000);

// High throughput use case
props.put(StreamsConfig.COMMIT_INTERVAL_MS_CONFIG, 60000);

3. Monitor Transaction Health

Track:
├─ Commit latency
├─ Failed transactions
├─ ProducerFencedExceptions
└─ Consumer lag

Alert on anomalies

4. Design Idempotent External Writes

// If writing to external system
// Ensure writes are idempotent

// BAD: Increment counter
// db.execute("UPDATE counters SET count = count + 1 WHERE id = ?");

// GOOD: Set exact value
// db.execute("UPDATE counters SET count = ? WHERE id = ?", newCount, id);

// Or use upserts, deduplication keys, etc.

5. Test Failure Scenarios

Test cases:
├─ Kill instances mid-processing
├─ Rebalance during transaction
├─ Broker failures
└─ Network partitions

Verify no duplicates, no data loss

16. Mental Model Summary

Exactly-Once in Kafka Streams

┌───────────────────────────────────────────────────┐
│  EXACTLY-ONCE SEMANTICS (EOS)                     │
│                                                   │
│  Components:                                      │
│  ├─ Idempotent producer (no duplicates)          │
│  ├─ Transactions (atomic writes)                 │
│  ├─ Transactional reads (read_committed)         │
│  └─ Offset commits in transaction                │
│                                                   │
│  Result:                                          │
│  └─ Read → Process → Write → Commit (atomic)     │
│                                                   │
│  Failure:                                         │
│  └─ Abort → Retry → Exactly-once maintained      │
└───────────────────────────────────────────────────┘

EOS v1 vs EOS v2

EOS V1:
├─ One producer per task
├─ transaction.id fencing
├─ High overhead
└─ Legacy (Kafka < 2.5)

EOS V2:
├─ One producer per thread
├─ group.instance.id fencing
├─ Lower overhead
└─ Recommended (Kafka 2.5+)

Transaction Flow

BEGIN TRANSACTION
   ↓
READ (input topics)
   ↓
PROCESS (transform, aggregate)
   ↓
WRITE (state store + changelog + output)
   ↓
COMMIT (offsets + transaction)
   ↓
VISIBLE (downstream consumers)

17. Questions You Should Be Able to Answer

After this module, you should be able to answer:

✅ What is exactly-once semantics?
✅ How does Kafka Streams achieve exactly-once?
✅ What's the difference between EOS v1 and EOS v2?
✅ What is an idempotent producer?
✅ What are transactions in Kafka?
✅ How does zombie fencing work?
✅ What is isolation.level=read_committed?
✅ What's the performance impact of EOS?
✅ When should I use exactly-once vs at-least-once?

18. What's Next?

Now that you understand exactly-once semantics, the next modules will cover:

Module 10.1: Error handling — handling failures with EOS
Module 11.1: Interactive queries — querying state with EOS
Module 12.1: Monitoring — tracking EOS metrics

Summary

You learned:

✅ Exactly-once = no data loss + no duplicates
✅ Achieved via idempotent producer + transactions + transactional reads
✅ EOS v2 preferred (faster, lower overhead, Kafka 2.5+)
✅ Transactions make read-process-write-commit atomic
✅ Zombie fencing prevents duplicate processing from old instances
✅ commit.interval.ms controls transaction frequency (latency vs throughput)
✅ Overhead: 10-30% latency, 10-20% throughput reduction
✅ Use EOS for critical applications (finance, billing, compliance)
✅ External exactly-once requires idempotent writes or outbox pattern

Next up: Module 10.1 — Error Handling & Failure Scenarios — Learn how to handle errors gracefully in Kafka Streams applications.