API & Backend
Kafka consumer offsets: test commits, replay, and recovery
A Kafka consumer resumes from committed progress, so a record read without a later offset commit can be delivered again. The commit stores where consumption should resume. Producer idempotence does not prevent this replay or deduplicate a consumer’s external business operation.
The Kafka 4.1.0 example publishes two records to one partition, reads the first, and closes without committing. A second consumer receives that record again, commits offset one, and closes. A third then receives the second record. Manual partition assignment keeps this test focused on offset recovery; automatic assignment and rebalancing need separate checks. The Java client and its output make each transition visible before you add a database or worker pool.
The Kafka example and test runner are available with the recorded results.
Run the Kafka offset recovery example
Before you start:
- Install Python 3.11 or newer, Git, and Docker. Confirm the Docker engine is running with
docker info. - Run the commands from the repository root after cloning. The examples use pinned container versions and download them on the first run.
- Use a local development machine with room for disposable containers. The harness removes the resources it creates.
The harness creates a disposable broker with one topic and one partition. It copies that broker’s client libraries into a temporary directory and runs the Java example in a separate JDK container. Kafka’s broker image supplies a runtime; the separate JDK supplies the compiler needed for Java source-file execution.
git clone https://github.com/emminextechdocs/developer-examples.git
cd developer-examples
python3 deep-dives/walkthrough.py kafkaThe test removes its containers and temporary libraries afterward. It records assertions in deep-dives/evidence/kafka.json. The official quickstart provides the corresponding broker and topic-command context. Our fixture intentionally avoids installing a broker or Java toolchain directly on your workstation.
walkthrough.py runs the lab and prints selected commands and their responses alongside the assertions. Open the full-resolution capture.
Inspect the result before changing the example
The runner writes a fresh JSON record and exits after cleanup. To print only its assertions, run:
python3 - <<'PYTHON'
import json
from pathlib import Path
report = json.loads(Path("deep-dives/evidence/kafka.json").read_text())
assert report["passed"] is True
for event in report["events"]:
if "assertion" in event:
print(event["assertion"], "=>", event["actual"])
PYTHONThis prints each assertion and its observed value from your latest run. If a check fails, fix its setup or implementation and rerun the lab. The following sections explain what the checks establish.
Fetched records, consumer position, and committed offsets
The broker log contains records at partition offsets. The running consumer has a position describing where it will fetch next. The group has committed progress used for recovery. Those values can differ even while every component is functioning normally.
Reading a record does not mean the business effect has committed. Advancing the local position does not mean the group’s progress is durable. The KafkaConsumer API distinguishes current position and committed position and documents the behavior of assignment, polling, and committing.
Name these positions explicitly in logs and dashboards. “Last offset” is ambiguous: it could mean the last fetched record, the last completed business operation, or the next committed resume position. Use separate field names for fetched, completed, and committed offsets.
Use manual assignment to isolate offset behavior
The Java client calls assign with one known TopicPartition. This gives the example manual partition ownership. It still configures a group ID so that offset commits have a stable group identity, but it does not use subscription-based group assignment.
Manual assignment removes coordinator-driven partition moves from this replay test. A production consumer that uses subscribe has additional ownership and rebalance responsibilities; those require separate tests.
Manual assignment also means the application must prevent unintended concurrent processing of the same assigned partition when that matters to its design. A group ID does not make manually assigned consumers participate in automatic assignment. Do not copy this fixture into a scalable consumer deployment and assume it inherits the usual subscription-based partition distribution.
Publish two records and await acknowledgement
The producer sends ord-1 and ord-2 explicitly to partition zero. The code waits on each send result. That keeps the consumer experiment from racing an unfinished publish and makes a producer failure visible before any offset assertions run.
The producer sets acks=all and enables idempotence. Those settings control broker acknowledgements and duplicate prevention for producer retries. They do not declare that a consumer’s database update will execute once. The producer configuration reference documents their requirements and relationships.
One broker and replication factor one provide no replica redundancy. In this fixture, waiting for the configured acknowledgement is useful for sequencing the test, but it cannot demonstrate multi-broker durability. Keep that limitation attached to the example when discussing acks=all; the setting is not independent of topic replication and in-sync replica policy.
Read one record without committing
The consumer disables automatic commits, sets max.poll.records to one, and uses earliest for the case where no valid initial committed offset exists. It reads the first record and closes without calling commitSync.
The complete client is deep-dives/kafka/Replay.java. The runner supplies its JDK and Kafka client libraries, then executes this file:
import java.time.Duration;
import java.util.*;
import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.clients.producer.*;
import org.apache.kafka.common.TopicPartition;
class Replay {
static final TopicPartition PARTITION = new TopicPartition("orders", 0);
static Properties consumerProperties() {
Properties p = new Properties();
p.put("bootstrap.servers", "localhost:9092");
p.put("group.id", "recovery-lab");
p.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
p.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
p.put("enable.auto.commit", "false");
p.put("auto.offset.reset", "earliest");
p.put("max.poll.records", "1");
return p;
}
static ConsumerRecord<String,String> next(KafkaConsumer<String,String> consumer) {
long deadline = System.nanoTime() + Duration.ofSeconds(30).toNanos();
while (System.nanoTime() < deadline) {
var records = consumer.poll(Duration.ofMillis(500));
if (!records.isEmpty()) return records.iterator().next();
}
throw new AssertionError("No record before deadline");
}
static void require(boolean condition, String message) {
if (!condition) throw new AssertionError(message);
}
public static void main(String[] args) throws Exception {
Properties p = new Properties();
p.put("bootstrap.servers", "localhost:9092");
p.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
p.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
p.put("acks", "all");
p.put("enable.idempotence", "true");
try (var producer = new KafkaProducer<String,String>(p)) {
producer.send(new ProducerRecord<>("orders", 0, "ord-1", "created")).get();
producer.send(new ProducerRecord<>("orders", 0, "ord-2", "created")).get();
}
long first;
try (var c = new KafkaConsumer<String,String>(consumerProperties())) {
c.assign(List.of(PARTITION));
var record = next(c);
first = record.offset();
System.out.println("READ without commit: offset=" + first + " key=" + record.key());
}
try (var c = new KafkaConsumer<String,String>(consumerProperties())) {
c.assign(List.of(PARTITION));
var replay = next(c);
require(replay.offset() == first, "Uncommitted record must replay");
System.out.println("REPLAY after consumer close: offset=" + replay.offset());
// This marks progress only. Business effects must already be durable.
c.commitSync(Map.of(PARTITION, new OffsetAndMetadata(replay.offset() + 1)));
System.out.println("COMMIT next offset=" + (replay.offset() + 1));
}
try (var c = new KafkaConsumer<String,String>(consumerProperties())) {
c.assign(List.of(PARTITION));
var next = next(c);
require(next.offset() == first + 1, "Committed position must be the next record");
System.out.println("RESUME next record: offset=" + next.offset() + " key=" + next.key());
}
}
}Read the three consumer blocks as separate process lifetimes:
- Read
ord-1and close without a commit. - Read the same offset, verify replay, and commit the next position.
- Start again and verify that
ord-2is now the first returned record.
The actual client output is:
READ without commit: offset=0 key=ord-1
REPLAY after consumer close: offset=0
COMMIT next offset=1
RESUME next record: offset=1 key=ord-2If the first two offsets differ, inspect the group ID and automatic-commit setting before investigating producer behavior. If the final consumer returns offset zero, check whether the commit succeeded. This fixture creates an empty topic each run, so offsets zero and one are deliberate assertions, not assumptions about an existing topic.
The polling helper has a deadline and throws if no record arrives. It does not treat an empty poll as end-of-topic, because an empty poll can occur while the consumer is waiting for data or establishing its position. The deadline limits the total wait across polls.
auto.offset.reset=earliest is a fallback when an initial offset is missing or invalid; it is not an instruction to replay the topic on every restart. Once our commit exists, the next consumer resumes from that progress. The consumer configuration reference defines the reset condition.
Check that the uncommitted record replays
The second consumer has the same group ID and manually assigned partition. Its first record must have the same offset as the first consumer’s record. The test compares those values directly and prints REPLAY after consumer close: offset=0.
Closing after the read reproduces an uncommitted consumer restart. Process kills, broker outages, and network partitions would need their own tests.
The same replay can affect business processing. If the first consumer applied a side effect before closing, the second consumer could apply it again. The fixture does not perform an external mutation, so it does not claim to prove duplicate business effects. Add a durable effect counter or a destination-specific assertion when adapting it to your application.
Commit the next resume offset
After the replay, the example calls commitSync with replay.offset() + 1. The committed value identifies where the consumer should resume, not the last record it just processed. The third consumer must therefore receive offset one and key ord-2.
The explicit map limits the commit to the partition and progress the example intends to acknowledge. In a multi-partition application, calculate progress separately for each partition. The OffsetAndMetadata API describes the object carried by that map.
Offset numbers need not be consecutive, especially with transactions or compaction. Offset-plus-one is sufficient for this two-record fixture; it is not a reason to invent missing records in a production completion tracker. The current 4.1 API documentation recommends the next record’s offset or ConsumerRecords.nextOffsets() when the batch is exhausted, with leader-epoch metadata. Track delivered work in partition order and preserve the client’s resume metadata when extending the example.
Do not commit the largest fetched offset when lower offsets in the same partition are still unfinished. With concurrent processing, the safe progress boundary is the completed contiguous prefix that your application can recover past. Tracking only the fastest finished task can skip slower work after a restart. That rule follows from committing a resume position, regardless of the worker-pool implementation.
Choose how to handle the effect-to-commit gap
Commit before applying the business effect, and a failure can make the group resume after work that never completed. Apply the effect before committing, and a failure can make completed work replay. Moving two independent operations next to each other in code does not make them atomic.
Kafka’s design documentation explains delivery semantics and the relationship between consumer position and output. Use that model to state your application contract. Manual commits control progress. A claim that replay cannot apply an order twice also needs a destination-side test.
For a database destination, consider storing a durable event identity with the business change in one database transaction. For Kafka-to-Kafka processing, evaluate Kafka transactions and the corresponding consumer isolation requirements. Neither design is implemented by this minimal fixture. Our idempotent webhook guide provides a related example of defending a database effect against repeated delivery.
Where producer idempotence applies
A successful idempotent producer send does not suppress a later application decision to publish another record with the same business key. Likewise, it does not recognize that two consumers applied the same external operation. Keep the producer protocol’s duplicate protection separate from business-level deduplication.
The KafkaProducer API documents send completion, producer idempotence, and transactional behavior. In this lab, the producer is closed after both sends finish. The consumer’s replay occurs afterward, without publishing a third record.
The result explains how duplicated processing can occur while the log still contains exactly the original two application records. Before changing producer settings during an incident, inspect whether the duplication happened in the log, in consumer delivery, or at the destination. Different boundaries require different fixes.
Test rebalances before switching to subscribe
Switching from assign to subscribe changes the experiment’s scope. Partitions can move among consumers, and work already handed to an application thread can outlive its consumer’s ownership. Your processing model must account for that transition instead of treating partition assignment as permanent.
Test a consumer joining, a consumer leaving, and processing that takes longer than your polling policy allows. Record which records completed, which offsets were committed, and what the replacement consumer receives. Also test commit failures rather than assuming every synchronous commit succeeds.
Keep Kafka client calls on the appropriate owning thread; the consumer API is not generally thread-safe. A worker pool should return processing results to the consumer loop, which remains responsible for Kafka calls. Avoid sharing the consumer object across processing threads.
Inspect a consumer group before resetting offsets
During an incident, identify the group, topic, partition, committed position, and retained log range before resetting anything. An offset reset changes where processing resumes; it does not reverse effects already committed to another system. Kafka’s operations guide documents the administrative tooling.
Use a separate test group when investigating replay mechanics. Resetting a shared group’s offsets to “see what happens” can create an application recovery event. Before a production reset, identify the records to replay and verify how the destination handles their previous effects.
For observability, distinguish fetched records from completed operations and committed offsets. Our OpenTelemetry guide shows how to connect operation-level evidence across services. Include topic, partition, and offset in investigation context while controlling metric cardinality.
Track completed work in partition order
Consider a partition where offsets ten, eleven, and twelve are handed to three application tasks. If twelve finishes first, committing thirteen would tell a replacement consumer to resume after all three. That would move recovery past unfinished work at ten and eleven. The broker cannot infer your task completion order from the fact that the records were fetched together.
Maintain a per-partition record of admitted and completed work. Advance the commit boundary only when every earlier admitted record in that partition has reached the application’s durable completion boundary. Keep failures and cancellations in that accounting; removing a failed task from memory must not accidentally make the prefix appear complete.
The fixture avoids this complexity by returning one record per poll and processing the sequence synchronously. The expected resume offset can therefore be checked directly. A worker-pool adaptation needs additional state and tests, not merely a larger max.poll.records value.
Test out-of-order completion deliberately. Hold the first task, finish a later task, interrupt the consumer, and inspect the replacement’s first record. Then allow the held task to finish and verify how the commit boundary moves. This checks whether the completion tracker handles tasks finishing out of order. Record the committed offset at each step so reviewers can check whether recovery skips unfinished work.
Frequently asked questions
Which Kafka offset should a consumer commit?
Commit the position from which the next consumer should resume after durably completed work. This fixture commits one after processing offset zero. Offsets can have gaps, so a production tracker should follow delivered records and the client’s resume metadata. The current 4.1 API recommends the next record’s offset or ConsumerRecords.nextOffsets() for an exhausted batch, including leader-epoch metadata.
Does auto.offset.reset=earliest replay the topic on every restart?
No. earliest applies when there is no valid initial offset to use. When the group has a valid committed position, recovery uses that progress. In the lab, the third consumer reads offset one despite retaining earliest in its configuration. Check the group ID and its committed offsets before diagnosing an unexpected replay.
Does committing an offset delete Kafka records?
No. A commit records consumer progress; retention and compaction control the log separately. Committing offset one does not delete offset zero, and another group can have different progress. Similarly, resetting one group’s position changes what that group reads next without reversing any database changes or other effects from its previous processing attempts.
Can manually assigned consumers share the same group ID?
They can use the same group ID for committed offsets, but assign does not invoke automatic partition distribution or rebalancing. The application must coordinate assignments itself. Two such consumers can process the same partition and interfere with its committed progress. The fixture avoids this by closing each consumer before starting its replacement.