Kafka Offset Management: Manual Commits, __consumer_offsets & Message Replay

Suppose you are running a financial billing consumer service.

Your consumer reads a batch of 100 payment records from Kafka, processes them, and charges customer credit cards.

Halfway through processing message #50, the underlying cloud server instance experiences a physical hardware failure and crashes.

When your container orchestrator restarts the consumer on a new server, where does the new consumer start reading?

If it starts reading at message #1, it re-processes payments #1 through #49, charging customer credit cards twice.
If it skips ahead to message #101, it misses payments #50 through #100 entirely.

How your consumer manages its committed read offset determines your system’s delivery semantics.

The Internal `__consumer_offsets` Topic

Kafka consumers do not store their read positions on local server disk.

Instead, consumers publish their committed offset coordinates to a special internal Kafka topic named __consumer_offsets.

__consumer_offsets RECORD FORMAT:
Key: [ GroupID: "billing-service", Topic: "payments", Partition: 0 ]
Value: [ CommittedOffset: 4500, Metadata: "", Timestamp: 1725819000 ]

Because __consumer_offsets is a standard compact Kafka topic replicated across brokers, if a consumer instance crashes, any replacement consumer instance assigned to that partition queries __consumer_offsets to discover where to resume processing.

Automatic vs Manual Offset Commits

Kafka provides two strategies for committing offsets:

1. Automatic Commits (`enable.auto.commit = true`)

By default, the consumer client commits its current read position automatically every 5 seconds (auto.commit.interval.ms = 5000) during poll() calls.

// Automatic commit configuration
props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "true");
props.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "5000");

The Automatic Commit Danger:

Suppose poll() fetches records 100 to 200. At second 3, auto-commit writes offset 200 to __consumer_offsets. At second 4, your application crashes while processing record 140.

When the application restarts, it reads offset 200. Records 140 to 199 were never processed, resulting in silent data loss.

2. Manual Offset Commits (`enable.auto.commit = false`)

To prevent data loss, production applications disable auto-commit and control offset commits explicitly in code.

props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");

while (running) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
    for (ConsumerRecord<String, String> record : records) {
        processRecord(record); // Process record first!
    }
    // Commit offsets manually AFTER processing completes!
    consumer.commitSync(); 
}

Delivery Semantics Matrix

The sequence of when you process a record versus when you commit its offset establishes your delivery guarantees:

Semantic Guarantee	Processing vs Commit Order	Failure Behavior
At-Most-Once	Commit offset FIRST $\rightarrow$ Process record SECOND	Messages can be lost, but never duplicated.
At-Least-Once	Process record FIRST $\rightarrow$ Commit offset SECOND	Messages are never lost, but can be duplicated.
Exactly-Once	Process + Commit within a single Transaction	Messages processed once and only once.

1. At-Most-Once Semantics

Commit offset to Kafka $\rightarrow$ process record. If processing crashes, the offset is already committed. The message is skipped upon restart.

2. At-Least-Once Semantics (Production Standard)

Process record $\rightarrow$ commit offset to Kafka. If processing crashes before committing, the message is re-read upon restart. Downstream consumer systems must handle duplicate messages using idempotent writes (e.g. INSERT INTO table ON CONFLICT DO NOTHING).

Synchronous (`commitSync`) vs Asynchronous (`commitAsync`)

When committing manually, you can choose between blocking and non-blocking calls:

// Option A: Synchronous Commit (Blocks until broker acks offset write)
try {
    consumer.commitSync();
} catch (CommitFailedException e) {
    logger.error("Failed to commit offset", e);
}

// Option B: Asynchronous Commit (Non-blocking performance)
consumer.commitAsync((offsets, exception) -> {
    if (exception != null) {
        logger.error("Async commit failed for offsets: " + offsets, exception);
    }
});

The Recommended Hybrid Pattern:

Use commitAsync() inside the main processing loop for high throughput, and execute a final commitSync() inside a finally block during application shutdown:

try {
    while (running) {
        ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
        for (ConsumerRecord<String, String> record : records) {
            processRecord(record);
        }
        consumer.commitAsync(); // Non-blocking commit during loop
    }
} finally {
    try {
        consumer.commitSync(); // Final blocking commit on shutdown!
    } finally {
        consumer.close();
    }
}

Replaying Historical Messages via `seek()`

Because Kafka retains log data durably on disk, you can reset a consumer’s position to any point in historical time.

Suppose a bug in your code corrupted calculation logic yesterday. After deploying a bug fix, you can instruct your consumer to seek back to yesterday’s offset:

// Reset consumer position to specific offset
TopicPartition partition = new TopicPartition("orders", 0);
consumer.assign(Collections.singletonList(partition));

// Option 1: Seek to absolute offset
consumer.seek(partition, 142000L);

// Option 2: Seek to beginning of partition log
consumer.seekToBeginning(Collections.singletonList(partition));

// Option 3: Seek by timestamp (Find offset at yesterday 9:00 AM)
long yesterdayMs = System.currentTimeMillis() - (24 * 60 * 60 * 1000);
Map<TopicPartition, Long> timestamps = Collections.singletonMap(partition, yesterdayMs);
Map<TopicPartition, OffsetAndTimestamp> offsets = consumer.offsetsForTimes(timestamps);

OffsetAndTimestamp target = offsets.get(partition);
if (target != null) {
    consumer.seek(partition, target.offset());
}

Quick Summary

Read positions are stored in the internal Kafka topic __consumer_offsets.
Automatic commits (enable.auto.commit = true) risk data loss if consumer processes crash mid-batch.
Manual commits after processing deliver At-Least-Once guarantees; downstream consumers handle duplicates via idempotency.
consumer.seek() allows applications to rewind read pointers to any historical offset or timestamp.

References & Further Reading

Apache Software Foundation. Apache Kafka Documentation: Log Compaction Mechanics & Cleaning Threads Architecture. Apache Kafka Docs.
Kreps, J. (2013). Log Compaction in Apache Kafka. LinkedIn Engineering Blog.
Shapira, G., et al. (2021). Kafka: The Definitive Guide (2nd Edition) — Chapter 5. O’Reilly Media.