5.1 Offsets Explained

Module Overview

Goal: Explain what offsets actually represent — the foundation of everything else in offset management.

What You'll Learn:

• What partition offsets are
• Current position vs committed offset
• Offset ordering guarantees
• Offset storage in __consumer_offsets topic
• How offsets enable resumption and tracking
• Common offset-related misconceptions

What's NOT Covered Here:

• How to commit offsets (covered in Module 5.2)
• Delivery semantics (covered in Module 5.3)
• Error recovery patterns (covered in Module 8.1)

Key Insight: Understanding offsets is fundamental to understanding Kafka consumption, commits, and delivery guarantees.

1. What is an Offset?

Definition

An offset is a unique integer that identifies a record's position within a partition.

Partition 0:
┌─────┬─────┬─────┬─────┬─────┬─────┬─────┐
│  0  │  1  │  2  │  3  │  4  │  5  │  6  │
└─────┴─────┴─────┴─────┴─────┴─────┴─────┘
  ↑
  offset

Key Properties:

• Monotonically increasing (never decreases)
• Immutable (never changes once assigned)
• Per-partition (not global)
• Sequential (no gaps in normal operation)

Think of offsets as page numbers in a book:

• Each page has a unique number
• Pages are numbered sequentially
• You bookmark your current page to resume later
• Different chapters (partitions) have independent page numbers

2. Offset Scope

Offsets are Per-Partition

Topic: "orders"

Partition 0: [0] [1] [2] [3] [4] ...
Partition 1: [0] [1] [2] [3] [4] ...
Partition 2: [0] [1] [2] [3] [4] ...

Critical: Offset 5 in Partition 0 is completely different from offset 5 in Partition 1!

TopicPartition("orders", 0) offset 5  ≠  TopicPartition("orders", 1) offset 5

No Global Ordering

Topic-level view (does NOT exist!):
orders: [msg1] [msg2] [msg3] ...  ← NO GLOBAL OFFSET!

Reality:
orders-0: [msg1] [msg3] [msg5]  ← offsets 0, 1, 2
orders-1: [msg2] [msg4] [msg6]  ← offsets 0, 1, 2

You cannot say: "Give me offset 100 from topic 'orders'"
You must say: "Give me offset 100 from partition 0 of topic 'orders'"

3. Two Critical Offset Concepts

Current Position vs Committed Offset

Kafka tracks two different offsets per partition for each consumer group:

1. Current Position (In-Memory)

• Where the consumer will read next
• Tracked in consumer's memory
• Updated after each poll()
• Lost if consumer crashes

Consumer position = last returned offset + 1

2. Committed Offset (Persistent)

• Where the consumer should resume after restart
• Stored in __consumer_offsets topic
• Updated explicitly via commit
• Survives consumer crashes

Committed offset = last successfully processed offset + 1

The Difference Illustrated

Partition 0:
┌─────┬─────┬─────┬─────┬─────┬─────┬─────┐
│  0  │  1  │  2  │  3  │  4  │  5  │  6  │
└─────┴─────┴─────┴─────┴─────┴─────┴─────┘
              ↑                 ↑
         Committed           Current
         offset: 3         position: 6
         (last commit)   (last poll)

What this means:

• Consumer has read up to offset 5 (position is 6)
• Consumer has committed up to offset 2 (committed is 3)
• If consumer crashes, it resumes from offset 3
• Offsets 3, 4, 5 will be reprocessed

4. How Offsets Work in Practice

The Consumption Flow

// 1. Consumer starts
consumer.subscribe(Collections.singletonList("orders"));

// 2. First poll() - Check committed offset
ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
// → Coordinator: "What's the last committed offset for orders-0?"
// → Coordinator: "Offset 1000"
// → Consumer starts reading from offset 1000

// 3. Process records
for (ConsumerRecord<String, String> record : records) {
    System.out.println("Processing offset: " + record.offset());
    // Offset 1000, 1001, 1002, ...
}

// 4. Commit (explicit)
consumer.commitSync();
// → Save committed offset = 1010 (for example)

// 5. Next poll()
records = consumer.poll(Duration.ofMillis(100));
// → Continue from offset 1010

Offset Lifecycle

1. Producer writes record
   ↓
   Broker assigns offset (e.g., 100)
   
2. Consumer polls
   ↓
   Consumer reads offset 100
   ↓
   Current position advances to 101
   
3. Consumer processes record
   ↓
   (processing logic)
   
4. Consumer commits
   ↓
   Committed offset updated to 101
   
5. Consumer crashes and restarts
   ↓
   Reads committed offset: 101
   ↓
   Resumes from offset 101

5. Offset Storage: __consumer_offsets

Where Are Committed Offsets Stored?

Kafka stores committed offsets in a special internal topic:

__consumer_offsets

Properties:

• Compacted topic (only latest offset per group/partition kept)
• 50 partitions by default
• Managed automatically by Kafka
• Can be read (but not written) by users

Offset Storage Format

Key:
{
  group: "my-consumer-group",
  topic: "orders",
  partition: 0
}

Value:
{
  offset: 12345,
  metadata: "",
  timestamp: 1674567890123,
  leaderEpoch: 5
}

How it's partitioned:

hash(group.id) % offsets.topic.num.partitions

This ensures offsets for a consumer group are stored on the same broker (the coordinator).

Reading Committed Offsets

# View offset commits
kafka-console-consumer.sh \
  --bootstrap-server localhost:9092 \
  --topic __consumer_offsets \
  --formatter "kafka.coordinator.group.GroupMetadataManager\$OffsetsMessageFormatter" \
  --from-beginning

Output:

[my-group,orders,0]::OffsetAndMetadata(offset=12345, leaderEpoch=Optional[5], metadata=, commitTimestamp=1674567890123)
[my-group,orders,1]::OffsetAndMetadata(offset=67890, leaderEpoch=Optional[5], metadata=, commitTimestamp=1674567890125)

How Commits Work Internally: The OffsetCommitRequest Protocol

Important: When you commit an offset, it does NOT use the regular producer API to write to the __consumer_offsets topic.

Instead, Kafka uses a dedicated internal request protocol for offset commits.

The Offset Commit Protocol

Consumer commits offset 1000 for orders-0
                ↓
    [Consumer sends OffsetCommitRequest]
                ↓
        Group Coordinator (broker)
                ↓
    Validates and writes to __consumer_offsets
                ↓
    [OffsetCommitResponse sent to consumer]
                ↓
        Commit acknowledged

Why Not Use Producer API?

❌ Does NOT happen:
Consumer → ProducerRecord(topic: "__consumer_offsets", ...) → Broker

✅ What actually happens:
Consumer → OffsetCommitRequest(group, topic, partition, offset) → Group Coordinator

Reasons for dedicated protocol:

1. Coordination: Group Coordinator manages all group state, not just offset storage
2. Validation: Coordinator validates consumer is still part of group before accepting commit
3. Atomicity: Multiple partition commits can be atomic in one request
4. Generation Fencing: Coordinator checks generation ID to prevent stale commits
5. Efficiency: Optimized for offset commit patterns (batching, deduplication)

OffsetCommitRequest Structure

// Simplified structure
OffsetCommitRequest {
  String groupId;                      // "my-consumer-group"
  int generationId;                    // Current rebalance generation
  String memberId;                     // Consumer's member ID
  long retentionTimeMs;                // How long to retain offset
  
  Map<TopicPartition, OffsetMetadata> offsets;
  // Example:
  // {
  //   (orders, 0) -> OffsetMetadata(offset: 1000, metadata: ""),
  //   (orders, 1) -> OffsetMetadata(offset: 2000, metadata: ""),
  //   (payments, 0) -> OffsetMetadata(offset: 500, metadata: "custom-metadata")
  // }
}

Key Fields:

Generation-Based Fencing

The generationId is critical for preventing zombie consumers from committing stale offsets:

Scenario: Consumer C1 in generation 5

1. C1 is processing slowly
2. Coordinator times out C1 (no heartbeat)
3. Rebalance → Generation 6
4. C1 wakes up and tries to commit

OffsetCommitRequest {
  groupId: "my-group",
  generationId: 5,  ← STALE!
  ...
}

Coordinator rejects:
- Current generation is 6
- C1's generation (5) is old
- Error: ILLEGAL_GENERATION

Result:

• ✅ Prevents zombie consumers from corrupting offsets
• ✅ Ensures only active generation can commit
• ✅ Protects against race conditions during rebalance

OffsetCommitResponse

// Coordinator responds with status for each partition
OffsetCommitResponse {
  Map<TopicPartition, Errors> errors;
  
  // Example success:
  // {
  //   (orders, 0) -> NO_ERROR,
  //   (orders, 1) -> NO_ERROR
  // }
  
  // Example failure:
  // {
  //   (orders, 0) -> ILLEGAL_GENERATION,
  //   (orders, 1) -> NOT_COORDINATOR_FOR_GROUP
  // }
}

Common Error Codes:

Why This Matters for Developers

1. Generation Awareness:

// This will fail if consumer is zombie/stale
consumer.commitSync();
// → Coordinator checks generation ID
// → Rejects if consumer is in old generation

2. Atomic Multi-Partition Commits:

// All partitions committed atomically in one OffsetCommitRequest
Map<TopicPartition, OffsetAndMetadata> offsets = new HashMap<>();
offsets.put(new TopicPartition("orders", 0), new OffsetAndMetadata(1000));
offsets.put(new TopicPartition("orders", 1), new OffsetAndMetadata(2000));
offsets.put(new TopicPartition("payments", 0), new OffsetAndMetadata(500));

consumer.commitSync(offsets);
// → Single OffsetCommitRequest with all 3 offsets
// → Either all succeed or all fail

3. Custom Metadata (Optional):

// You can attach custom metadata to commits
OffsetAndMetadata metadata = new OffsetAndMetadata(
    1000,                    // offset
    "processed-by-worker-3"  // custom metadata (max 4KB)
);
consumer.commitSync(Collections.singletonMap(
    new TopicPartition("orders", 0),
    metadata
));

Custom metadata use cases:

• Tracking which worker processed partition
• Storing processing timestamps
• Application-specific checkpoints

Commit Latency Implications

Typical commit latency breakdown:

Client → Coordinator:        5-10ms (network)
Coordinator validation:      < 1ms
Write to __consumer_offsets: 5-15ms (replication)
Coordinator → Client:        5-10ms (network)
────────────────────────────────────────────
Total:                       15-36ms

Why commits are "expensive":

• Network round-trip to coordinator
• Disk write to __consumer_offsets (replicated to ISR)
• Fsync for durability (if acks=-1 for offsets topic)

Optimization: Batch commits for multiple partitions in one request.

Monitoring Offset Commits

JMX Metrics:

// Consumer metrics
kafka.consumer:type=consumer-coordinator-metrics,client-id=<client-id>

- commit-latency-avg: Average commit latency
- commit-latency-max: Max commit latency
- commit-rate: Commits per second
- commit-total: Total commits

Coordinator Metrics:

// Broker metrics
kafka.coordinator.group:type=GroupMetadataManager,name=OffsetCommitsPerSec

- Count of offset commits per second across all groups

Visual Summary: Offset Commit Flow

┌──────────────────────────────────────────────────────────┐
│  CONSUMER (commitSync() or commitAsync())                │
└───────────────────┬──────────────────────────────────────┘
                    │
                    ↓ OffsetCommitRequest
            ┌───────────────────┐
            │ Group Coordinator │
            │   (Broker)        │
            └───────┬───────────┘
                    │
                    ├─ Validate generation ID
                    ├─ Validate member ID
                    ├─ Check group state
                    │
                    ↓ Write
            ┌────────────────────┐
            │ __consumer_offsets │  (compacted topic)
            │   Partition X      │
            └────────────────────┘
                    │
                    ↓ Replicate to ISR
            ┌────────────────────┐
            │   Follower Brokers │
            └────────────────────┘
                    │
                    ↓ OffsetCommitResponse
            ┌───────────────────┐
            │   Consumer        │
            │   (acknowledged)  │
            └───────────────────┘

6. Initial Offset Behavior

What Happens on First Start?

When a consumer starts for the first time (no committed offset exists):

props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); // or "latest"

auto.offset.reset = earliest

Start reading from **beginning** of partition

Partition 0:
┌─────┬─────┬─────┬─────┬─────┐
│  0  │  1  │  2  │  3  │  4  │
└─────┴─────┴─────┴─────┴─────┘
  ↑
  Start here (offset 0)

Use when: You want to process all historical data.

auto.offset.reset = latest

Start reading from **end** of partition (only new messages)

Partition 0:
┌─────┬─────┬─────┬─────┬─────┐
│  0  │  1  │  2  │  3  │  4  │
└─────┴─────┴─────┴─────┴─────┘
                            ↑
                        Start here (offset 5)

Use when: You only want new messages going forward.

auto.offset.reset = none

props.put("auto.offset.reset", "none");

Behavior: Throw exception if no committed offset exists.

Use when: You want to fail fast rather than guess.

When Does auto.offset.reset Apply?

Applies when:
✅ Consumer group is new (no committed offset)
✅ Committed offset is out of range (deleted by retention)

Does NOT apply when:
❌ Committed offset exists and is valid

7. Offsets and Retention

Log Retention vs Offset Validity

Topic: "orders"
Log retention: 7 days

Time  Partition 0              Committed Offset
────┼─────────────────────────┼──────────────────
Day 1│ [0][1][2][3][4][5]...  │ Consumer: offset 3
Day 8│ (offsets 0-1000 deleted)│ Consumer: offset 3 (INVALID!)

What happens:

• Offsets 0-1000 were deleted by retention
• Consumer's committed offset (3) no longer exists
• Consumer uses auto.offset.reset to decide what to do

Out-of-Range Offsets

Scenario:
1. Consumer commits offset 100
2. Consumer offline for 10 days
3. Retention deletes offsets 0-5000
4. Consumer restarts

Result:
- Committed offset 100 is out of range
- auto.offset.reset = "latest" → Skip to current end
- auto.offset.reset = "earliest" → Start from offset 5000
- auto.offset.reset = "none" → Throw exception

8. Offset Ordering Guarantees

Within a Partition

Partition 0:
[A] [B] [C] [D] [E]
 0   1   2   3   4

Guarantee: If A is at offset 0 and E is at offset 4,
           then A was produced before E

Kafka guarantees:

• ✅ Offsets increase sequentially
• ✅ Lower offset = earlier message
• ✅ Order preserved within partition

Across Partitions

Partition 0: [A] [B]  ← offsets 0, 1
Partition 1: [C] [D]  ← offsets 0, 1

NO guarantee: A before C, or C before A

Kafka does NOT guarantee:

• ❌ Cross-partition ordering
• ❌ Topic-level ordering

9. Consumer Position Tracking

Checking Current Position

// Get current position (next offset to read)
long position = consumer.position(new TopicPartition("orders", 0));
System.out.println("Next offset to read: " + position);

Checking Committed Offset

// Get last committed offset
OffsetAndMetadata offsetMeta = consumer.committed(new TopicPartition("orders", 0));
if (offsetMeta != null) {
    System.out.println("Last committed offset: " + offsetMeta.offset());
} else {
    System.out.println("No committed offset");
}

Seeking to Specific Offset

// Manually set position to specific offset
TopicPartition partition = new TopicPartition("orders", 0);
consumer.seek(partition, 1000); // Jump to offset 1000

// Next poll() will read from offset 1000
ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));

Use cases:

• Replay from specific point
• Skip problematic messages
• Time-based replay (with offsetsForTimes())

10. Common Offset Misconceptions

❌ Misconception 1: "Offsets are timestamps"

Reality: Offsets are sequential integers, not timestamps. Records do have timestamps, but offsets are separate.

❌ Misconception 2: "Offsets are global across the topic"

Reality: Offsets are per-partition. Offset 100 in partition 0 ≠ offset 100 in partition 1.

❌ Misconception 3: "Committed offset is where I am now"

Reality: Committed offset is where you last saved your progress, not where you are now.

❌ Misconception 4: "If I don't commit, I won't lose my place"

Reality: Without committing, your place is lost on restart. Current position is in-memory only.

❌ Misconception 5: "Offset 10 means 10 messages"

Reality: Offset 10 means position 10 in that partition. Could be 10 messages (if starting from 0) or the 3rd message (if log started at offset 8 due to retention).

11. Mental Model Summary

Key Concepts

OFFSET = Position in a partition

Types:
├─ Current position (in-memory, volatile)
└─ Committed offset (persistent, stored in __consumer_offsets)

Properties:
├─ Per-partition (not global)
├─ Sequential (monotonically increasing)
├─ Immutable (never changes once assigned)
└─ Enables resumption

Storage:
└─ __consumer_offsets topic (compacted)

The Offset Gap

Current Position: Where you are now
Committed Offset: Where you last saved

Gap = Risk of reprocessing on crash

Questions You Should Be Able to Answer

✅ What is an offset?
✅ What's the difference between current position and committed offset?
✅ Are offsets global or per-partition?
✅ Where are committed offsets stored?
✅ What is auto.offset.reset?
✅ What happens if my committed offset is deleted by retention?
✅ Can I seek to a specific offset?

Questions We Haven't Answered Yet

❌ How do I commit offsets? → Module 5.2
❌ What are the different commit strategies? → Module 5.2
❌ What delivery semantics do commits provide? → Module 5.3

12. What's Next?

Now that you understand what offsets are, the next modules will cover:

• Module 5.2: How to commit offsets (strategies and trade-offs)
• Module 5.3: How commits affect delivery semantics
• Module 6.1: Poll loop internals

Summary

You learned:

• ✅ Offsets are sequential integers that identify record position in a partition
• ✅ Current position (in-memory) vs committed offset (persistent)
• ✅ Offsets are per-partition, not global
• ✅ Committed offsets stored in __consumer_offsets topic
• ✅ auto.offset.reset controls behavior when no offset exists
• ✅ Offsets guarantee ordering within partition, not across partitions

Next up: Commit Strategies (Module 5.2) — Learn how and when to commit offsets safely.