Kafka Offsets & Offset Management

An offset is a sequential integer that uniquely identifies each record within a partition. Kafka does not delete records after they are consumed: it retains them for a configurable retention period. Each consumer group maintains its own offset per partition, allowing multiple independent consumers to read the same data at different positions.

Kafka

How Offsets Work
Auto Commit
Manual Commit (Sync & Async)
Offset Reset Strategies
Seeking to a Specific Offset
Committed Offset vs Current Position

How Offsets Work

Every partition is an ordered, immutable log. Records are appended with ever-increasing offsets starting at 0. Kafka stores consumer offsets in an internal topic called __consumer_offsets. When a consumer restarts, it resumes from the last committed offset for each assigned partition.

Offsets are per partition, per consumer group, not global.
The committed offset is the offset of the next record to read, not the last one processed. After processing the record at offset 41, a consumer commits 42.
After a restart or a rebalance, the consumer that gets the partition starts reading at the committed offset.

Kafka retains records based on time (retention.ms) or size (retention.bytes), not based on whether they were consumed. A consumer can re-read old records by resetting its offset.

Auto Commit

With enable.auto.commit=true (the default), the consumer commits for you. There is no separate thread: each call to poll() checks whether auto.commit.interval.ms has passed and, if so, commits the offsets of the records returned by the previous poll().

The code in this article shows only the parts that matter for offsets. It goes inside a main method like the complete example in Kafka Consumers, which also lists the imports.

Java

Properties props = new Properties();
props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
props.put(ConsumerConfig.GROUP_ID_CONFIG, "auto-commit-group");
props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, true);
props.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, 3000); // commit every 3 seconds

KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
consumer.subscribe(List.of("orders"));

while (true) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
    for (ConsumerRecord<String, String> record : records) {
        process(record); // finished before the next poll(), which may commit these offsets
    }
}

Because the commit happens in the next poll(), auto-commit is safe when records are processed inside the loop, as above: a crash means some records are processed again, never that one is skipped. It becomes unsafe when processing is handed to other threads. The loop then polls again before that work is done, the offsets get committed, and records still in progress are lost if the application crashes. In that case, commit manually once the work is finished.

Manual Commit (Sync & Async)

Disable auto-commit and call commitSync() or commitAsync() after processing each batch.

Java

props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);

KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
consumer.subscribe(List.of("orders"));

while (true) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(200));

    for (ConsumerRecord<String, String> record : records) {
        process(record);
    }

    // Blocks until the broker confirms, and retries on recoverable errors
    consumer.commitSync();
}

commitAsync() does the same without waiting, which is faster but does not retry a failed commit. The usual pattern combines the two: commitAsync() in the loop and one commitSync() on shutdown, as shown in Kafka Consumers.

To commit a specific partition and offset, pass them explicitly. Committing after every record, as here, keeps repeats after a crash to a minimum, but each commit is a round trip to the broker, so it is much slower than committing once per batch:

Java

for (ConsumerRecord<String, String> record : records) {
    process(record);
    // commit offset+1 so next poll starts after this record
    Map<TopicPartition, OffsetAndMetadata> offsetMap = Map.of(
        new TopicPartition(record.topic(), record.partition()),
        new OffsetAndMetadata(record.offset() + 1)
    );
    consumer.commitSync(offsetMap);
}

Offset Reset Strategies

auto.offset.reset controls what happens when a consumer group has no committed offset for a partition, or when the committed offset no longer exists. That happens the first time a group runs, when a group has been inactive longer than the broker's offsets.retention.minutes (7 days by default) and its offsets were removed, or when the committed record has already been deleted by retention.

Value	Behaviour	Use case
`latest` (default)	Start from the newest record (skip historical data)	Real-time processing: you only care about new events
`earliest`	Start from the oldest record still retained (which is offset 0 only if nothing has been deleted yet)	Reprocessing or first-time full load
`none`	Throw an exception if no committed offset exists	When a group must never start without a known position, for example because reading from the wrong place would cause harm

auto.offset.reset only applies when there is no existing committed offset. It has no effect if the consumer group already has a committed offset for that partition.

Seeking to a Specific Offset

Use the seek API to manually reposition a consumer to any offset: useful for reprocessing or skipping bad records.

A consumer can only seek on partitions assigned to it. With subscribe(), partitions are assigned during poll(), so the reliable place to seek is a rebalance listener, which runs as soon as partitions arrive:

Java

// Read every assigned partition from the beginning
consumer.subscribe(List.of("events"), new ConsumerRebalanceListener() {
    @Override
    public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
        consumer.seekToBeginning(partitions);
    }

    @Override
    public void onPartitionsRevoked(Collection<TopicPartition> partitions) { }
});

With assign(), you choose the partitions yourself, so you can seek straight away:

Java

TopicPartition partition0 = new TopicPartition("events", 0);
consumer.assign(List.of(partition0));

// Skip to the newest record
consumer.seekToEnd(List.of(partition0));

// Seek to a specific offset
TopicPartition partition0 = new TopicPartition("events", 0);
consumer.seek(partition0, 1500L); // start reading from offset 1500

// Find the offset for a specific timestamp
Map<TopicPartition, Long> timestampMap = Map.of(partition0, Instant.now().minusHours(1).toEpochMilli());
Map<TopicPartition, OffsetAndTimestamp> offsets = consumer.offsetsForTimes(timestampMap);
offsets.forEach((tp, offsetAndTimestamp) -> {
    if (offsetAndTimestamp != null) {
        consumer.seek(tp, offsetAndTimestamp.offset());
    }
});

Committed Offset vs Current Position

Concept	API	Meaning
Current position	`consumer.position(tp)`	Next offset to be fetched on the next poll
Committed offset	`consumer.committed(Set.of(tp))`	The last offset committed to `__consumer_offsets`: where reading resumes after a restart

If the consumer crashes, it resumes from the committed offset, not the current position. Any records fetched but not yet committed will be re-delivered: this is the at-least-once guarantee.

Contents

How Offsets Work

Auto Commit

Manual Commit (Sync & Async)

Offset Reset Strategies

Seeking to a Specific Offset

Committed Offset vs Current Position

Related articles