10.1.3 – Consumer Isolation Levels

Module: M10 – Advanced Topics & Architecture
Section: 10.1 – Transactions & Exactly-Once Semantics
Goal: Understand consumer isolation levels and how consumers interact with transactional producers

Introduction

When producers use transactions, consumers have two options for how they read data:

1. read_uncommitted: Read all messages immediately (ignore transactions)
2. read_committed: Read only committed messages (wait for transaction markers)

This module explains how each isolation level works, when to use them, and their performance implications.

Isolation Levels Overview

isolation.level Configuration

Consumer Property:

isolation.level=read_committed  # or read_uncommitted (default)

Comparison Table

read_uncommitted Isolation Level

How It Works

Behavior: Consumer reads all messages immediately, regardless of transaction state.

Partition Log:

Offset 100: message (pid=12345, uncommitted)
Offset 101: message (pid=12345, uncommitted)
Offset 102: COMMIT marker (pid=12345)
Offset 103: message (pid=67890, uncommitted)
Offset 104: ABORT marker (pid=67890)

read_uncommitted Consumer:

Reads offsets 100, 101, 102, 103, 104
→ Sees all messages, including aborted ones (offset 103)
→ Sees control markers (but typically filters them out)

Configuration

Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("group.id", "my-consumer-group");
props.put("isolation.level", "read_uncommitted");  // Default before Kafka 2.5
props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");

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

When to Use

Good For:

• Logging and metrics: Data loss is acceptable, low latency is critical
• Monitoring and alerting: Need real-time visibility, duplicates OK
• Non-critical analytics: Approximate counts, trending data

Example:

// Log aggregation consumer (duplicates are acceptable)
consumer.subscribe(Arrays.asList("application-logs"));
while (true) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
    for (ConsumerRecord<String, String> record : records) {
        logStorage.append(record.value());  // Duplicates don't break anything
    }
}

Risks

1. Seeing Aborted Messages

Producer starts transaction
  → Writes message A, B, C
  → Aborts transaction

read_uncommitted consumer:
  → Sees messages A, B, C (even though transaction aborted)
  → Processes them
  → Later realizes transaction aborted (if checking markers)

Impact: Consumer may process messages that were never supposed to be committed.

2. Inconsistent State

Producer writes to topics "orders" and "payments" transactionally
  → order-1 → orders topic
  → payment-1 → payments topic
  → Transaction aborts

read_uncommitted consumer 1 (reading "orders"):
  → Sees order-1 ❌

read_uncommitted consumer 2 (reading "payments"):
  → Sees payment-1 ❌

Result: Inconsistent view (order exists but no payment)

read_committed Isolation Level

How It Works

Behavior: Consumer waits for transaction markers before reading uncommitted messages.

Partition Log:

Offset 100: message (pid=12345, uncommitted)
Offset 101: message (pid=12345, uncommitted)
Offset 102: COMMIT marker (pid=12345)
Offset 103: message (pid=67890, uncommitted)
Offset 104: ABORT marker (pid=67890)

read_committed Consumer:

Reads offsets 100, 101 (after seeing COMMIT marker at 102)
→ Skips offset 102 (control marker)
→ Blocks on offset 103 (waits for marker)
→ Sees ABORT marker at 104
→ Skips offset 103 (aborted message)
→ Continues to next message

Configuration

Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("group.id", "payment-processor");
props.put("isolation.level", "read_committed");  // Exactly-once semantics
props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");

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

When to Use

Good For:

• Financial transactions: No duplicates or missed messages
• Critical data pipelines: Exactly-once guarantees required
• State management: Consistent view of data
• Read-process-write pipelines: Ensures atomicity

Example:

// Payment processor (duplicates would cause double charges)
consumer.subscribe(Arrays.asList("payment-requests"));
while (true) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
    for (ConsumerRecord<String, String> record : records) {
        Payment payment = parsePayment(record.value());
        processPayment(payment);  // Must process exactly once
    }
    consumer.commitSync();
}

Transaction Marker Handling

How Consumers Use Markers:

Consumer polls partition log
  ↓
Broker filters out aborted messages
  ↓
Broker sends only committed messages to consumer
  ↓
Consumer never sees aborted data

Key Point: The broker filters aborted messages, not the consumer. The consumer only receives committed data.

LSO (Last Stable Offset)

What Is LSO?

Last Stable Offset (LSO) = The offset of the first message in an incomplete transaction.

Purpose: Prevents read_committed consumers from reading uncommitted messages.

LSO Example:

Partition Log:
  Offset 100: message (committed, pid=12345)
  Offset 101: message (committed, pid=12345)
  Offset 102: COMMIT marker
  Offset 103: message (uncommitted, pid=67890)  ← Transaction started but not finished
  Offset 104: message (uncommitted, pid=67890)
  Offset 105: message (committed, pid=88888)    ← Different producer
  Offset 106: COMMIT marker

LSO = 103 (first uncommitted message)

read_committed consumer:
  → Can read up to offset 102
  → Blocks on offset 103 (waits for COMMIT or ABORT marker for pid=67890)
  → Cannot read offset 105 (even though it's committed, blocked by LSO)

Impact: Slow or stuck transactions can block consumers from reading newer committed messages.

Latency Implications

Latency Overhead:

• read_uncommitted: ~0ms additional latency (reads immediately)
• read_committed: ~5-10ms additional latency (waits for transaction markers)

Factors Affecting Latency:

1. Transaction size: Larger transactions take longer to commit
2. Transaction timeout: Slow producers delay marker writes
3. Broker load: High load increases marker write time

Tuning for Lower Latency:

Producer Side:

# Reduce transaction timeout (faster abort of stuck transactions)
transaction.timeout.ms=30000  # 30 seconds

# Smaller batch sizes (faster transaction commits)
batch.size=16384  # 16 KB

Consumer Side:

# Increase poll timeout (wait longer for markers)
max.poll.interval.ms=300000  # 5 minutes

Read-Process-Write Pipeline with read_committed

Full Example

// Consumer config
Properties consumerProps = new Properties();
consumerProps.put("bootstrap.servers", "localhost:9092");
consumerProps.put("group.id", "etl-pipeline");
consumerProps.put("isolation.level", "read_committed");  // Only read committed messages
consumerProps.put("enable.auto.commit", "false");  // Manual commit
consumerProps.put("key.deserializer", StringDeserializer.class.getName());
consumerProps.put("value.deserializer", StringDeserializer.class.getName());

// Producer config
Properties producerProps = new Properties();
producerProps.put("bootstrap.servers", "localhost:9092");
producerProps.put("transactional.id", "etl-pipeline-1");
producerProps.put("enable.idempotence", "true");
producerProps.put("key.serializer", StringSerializer.class.getName());
producerProps.put("value.serializer", StringSerializer.class.getName());

KafkaConsumer<String, String> consumer = new KafkaConsumer<>(consumerProps);
KafkaProducer<String, String> producer = new KafkaProducer<>(producerProps);

consumer.subscribe(Arrays.asList("input-topic"));
producer.initTransactions();

while (true) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
    
    if (!records.isEmpty()) {
        try {
            producer.beginTransaction();
            
            for (ConsumerRecord<String, String> record : records) {
                // Transform message
                String transformed = transform(record.value());
                
                // Produce to output topic
                producer.send(new ProducerRecord<>("output-topic", record.key(), transformed));
            }
            
            // Commit offsets as part of transaction
            Map<TopicPartition, OffsetAndMetadata> offsets = new HashMap<>();
            for (TopicPartition partition : records.partitions()) {
                List<ConsumerRecord<String, String>> partitionRecords = records.records(partition);
                long lastOffset = partitionRecords.get(partitionRecords.size() - 1).offset();
                offsets.put(partition, new OffsetAndMetadata(lastOffset + 1));
            }
            producer.sendOffsetsToTransaction(offsets, consumer.groupMetadata());
            
            // Commit transaction
            producer.commitTransaction();
            
        } catch (ProducerFencedException | OutOfOrderSequenceException | AuthorizationException e) {
            log.error("Fatal error, closing", e);
            producer.close();
            consumer.close();
            break;
        } catch (KafkaException e) {
            log.warn("Transaction failed, aborting", e);
            producer.abortTransaction();
        }
    }
}

Guarantees:

• Consumer reads only committed messages (read_committed)
• Producer writes and commits offsets atomically
• If producer crashes, consumer resumes from last committed offset
• Result: Exactly-once semantics end-to-end

Isolation Level and Lag

Lag Calculation with read_committed

Consumer Lag = Log-End-Offset (LEO) - Current Offset

BUT with read_committed, lag is based on LSO (Last Stable Offset), not LEO.

Example:

Partition Log:
  LEO (Log-End-Offset) = 110
  LSO (Last Stable Offset) = 103 (first uncommitted message)
  Current Offset (consumer position) = 100

Lag calculation:
  read_uncommitted: Lag = LEO - Current = 110 - 100 = 10
  read_committed: Lag = LSO - Current = 103 - 100 = 3

Key Point: read_committed consumers show lower apparent lag because they can't read past LSO.

Monitoring LSO

Metrics to Track:

• kafka.log:type=Log,name=LogEndOffset,topic=*,partition=*: Log-end-offset (LEO)
• kafka.log:type=Log,name=LogStartOffset,topic=*,partition=*: Log-start-offset
• LSO: Not directly exposed, but can be inferred from consumer lag behavior

Alerting:

If (LEO - LSO) > threshold:
  Alert: "Transaction blocking consumers (stuck transaction)"

Best Practices

1. Match Isolation Level to Use Case

2. Use read_committed with Transactional Producers

Best Practice: If producer uses transactions, consumer should use read_committed.

Why: Ensures consistent view of data (no partial transaction reads).

3. Monitor Transaction Duration

Metric: kafka.producer:type=producer-metrics,client-id=*,attribute=txn-commit-time-avg

Alert If: Average transaction duration > 10 seconds (indicates slow producers blocking consumers).

4. Tune Transaction Timeout

# Producer
transaction.timeout.ms=60000  # 1 minute (fast transactions)

# Increase for slow batch jobs
transaction.timeout.ms=300000  # 5 minutes

Trade-off: Longer timeout = more tolerance for slow processing, but longer consumer blocking if transaction fails.

Common Issues

Issue 1: Consumer Stuck Waiting for Marker

Symptom: Consumer stops reading, lag increases.

Cause: Producer started transaction but crashed before committing/aborting.

Solution:

• Wait for transaction.timeout.ms (coordinator will auto-abort)
• Reduce transaction.timeout.ms for faster recovery

Issue 2: High Latency with read_committed

Symptom: p99 latency increases by 50-100ms.

Cause: Waiting for transaction markers.

Solutions:

• Reduce transaction size (smaller batches)
• Reduce transaction.timeout.ms
• Increase max.poll.interval.ms to tolerate longer waits

Issue 3: Inconsistent Lag Metrics

Symptom: Lag metrics show lower lag for read_committed consumers than expected.

Cause: Lag calculated from LSO, not LEO.

Solution: Monitor both LEO and LSO, calculate "blocked lag" = LEO - LSO.

Key Takeaways

1. read_uncommitted: Reads all messages immediately, including aborted ones. Use for non-critical data.

2. read_committed: Waits for transaction markers, reads only committed messages. Use for critical data.

3. LSO (Last Stable Offset): First uncommitted message in partition. read_committed consumers cannot read past LSO.

4. Latency: read_committed adds ~5-10ms latency (waiting for markers).

5. Lag Calculation: read_committed lag based on LSO, not LEO.

6. Best Practice: Match isolation level to producer (transactional producer → read_committed consumer).

7. Monitoring: Track transaction duration and LSO to detect blocked consumers.

What's Next?

Next Module: 10.1.4 – End-to-End EOS Pipelines

What We Covered:

• ✅ read_uncommitted vs read_committed isolation levels
• ✅ Transaction markers and LSO
• ✅ Read-process-write pipeline example
• ✅ Latency and lag implications

What's Coming:

• ❌ Designing end-to-end exactly-once pipelines
• ❌ Multi-stage transformations with EOS
• ❌ EOS anti-patterns and debugging

References:

• Transaction protocol → 10.1.1
• Producer transactions → 10.1.2
• Consumer configs → M4: Consumers