Managing Offsets

1. Understanding Offset Commits

Property	Description	Default Value	Notes
enable.auto.commit	Enables automatic offset commits	`true`	When set to `false`, manual commits are required
auto.commit.interval.ms	Interval in milliseconds to commit offsets automatically	`5000`	Only applies when `enable.auto.commit` is `true`
auto.offset.reset	What to do when there is no initial offset or the offset is out of range	`latest`	Options: `earliest`, `latest`, `none`

Example: Manual Offset Commit

const { Kafka } = require('kafkajs');

const kafka = new Kafka({
  clientId: 'my-app',
  brokers: ['localhost:9092']
});

const consumer = kafka.consumer({ groupId: 'test-group' });

await consumer.connect();
await consumer.subscribe({ topic: 'test-topic', partitions: [0] });

await consumer.run({
  eachMessage: async ({ topic, partition, message }) => {
    console.log(message.value.toString());
    // Manually commit offset after processing
    await consumer.commitOffsets([
      { topic, partition, offset: message.offset }
    ]);
  }
});

Note: Kafka consumers can either auto-commit offsets or manually commit them. Manual commits provide more control and are recommended for applications requiring exactly-once processing semantics.

2. Enabling Auto Commit

Property	Default Value	Description
enable.auto.commit	`true`	Enables or disables automatic offset commits to Kafka.
auto.commit.interval.ms	`5000`	Frequency (in milliseconds) at which offsets are committed when `enable.auto.commit` is true.

Example: Configuring Auto Commit

{
  "enable.auto.commit": true,
  "auto.commit.interval.ms": 1000
}

Note: Auto committing can lead to duplicate processing if the consumer crashes between commits. Consider manual offset management for exactly-once semantics.

3. Setting Auto Commit Interval

Property	Default Value	Description
enable.auto.commit	`true`	Enables or disables automatic offset commits to Kafka.
auto.commit.interval.ms	`5000`	Interval in milliseconds at which to commit offsets when `enable.auto.commit` is set to `true`.

Example: Configuring Auto Commit Interval

{
  "enable.auto.commit": true,
  "auto.commit.interval.ms": 10000
}

Note: Setting auto.commit.interval.ms to a lower value increases the frequency of offset commits, which can improve consumer reliability but may increase load on Kafka.

4. Committing Synchronously

Method	Description	Parameters	Notes
commitSync()	Commits offsets synchronously	`Map<TopicPartition, OffsetAndMetadata>`	Blocks until commit completes or throws exception
commitSync(TopicPartition, OffsetAndMetadata)	Commits offset for a specific topic partition	`TopicPartition`, `OffsetAndMetadata`	Useful for manual offset control
commitSync(Collection<TopicPartition>, Duration)	Commits offsets with timeout	`Collection<TopicPartition>`, `Duration`	Allows specifying a timeout for the commit operation

Example: Committing Offsets Synchronously

Map<TopicPartition, OffsetAndMetadata> offsets = new HashMap<>();
offsets.put(new TopicPartition("my-topic", 0), new OffsetAndMetadata(100));
consumer.commitSync(offsets);

Note: Synchronous commits can block consumer processing, so use with caution in high-throughput scenarios.

5. Committing Asynchronously

Method	Description	Parameters	Notes
commitAsync()	Commits offsets asynchronously	`Map<TopicPartition, OffsetAndMetadata> offsets`	Non-blocking call, returns immediately
commitAsync(OffsetCommitCallback callback)	Commits offsets asynchronously with a callback	`Map<TopicPartition, OffsetAndMetadata> offsets`, `OffsetCommitCallback callback`	Allows handling commit results or errors

Example: Asynchronous Commit with Callback

consumer.commitAsync((offsets, exception) -> {
    if (exception != null) {
        System.err.println("Commit failed for offsets " + offsets);
    } else {
        System.out.println("Successfully committed offsets: " + offsets);
    }
});

Note: Asynchronous commits do not block the consumer thread, but they may not be immediately acknowledged by the broker.

6. Committing Specific Offsets

Method	Description	Parameters	Notes
commitSync()	Synchronously commits the specified offsets	`Map<TopicPartition, OffsetAndMetadata> offsets`	Blocks until commit completes or throws exception
commitAsync()	Asynchronously commits the specified offsets	`Map<TopicPartition, OffsetAndMetadata> offsets`	Does not block; callback for success/failure
commitSync(Collection<TopicPartition> partitions)	Synchronously commits offsets for given partitions	`Collection<TopicPartition> partitions`	Commits offsets for all specified partitions

Example: Committing Specific Offsets

Map<TopicPartition, OffsetAndMetadata> offsets = new HashMap<>();
offsets.put(new TopicPartition("my-topic", 0), new OffsetAndMetadata(100));
offsets.put(new TopicPartition("my-topic", 1), new OffsetAndMetadata(200));

consumer.commitSync(offsets);

Note: When committing specific offsets, ensure that the offset values are valid and within the topic's range to avoid data loss or reprocessing issues.

7. Storing Offsets Externally

Storage Type	Description	Use Case	Notes
Database	Store offsets in a relational or NoSQL database	High availability, persistence across restarts	Requires custom implementation or use of libraries like Kafka Offset Store
ZooKeeper	Use ZooKeeper to store consumer offsets	Legacy approach, Kafka versions < 0.10	DEPRECATED in favor of internal topic storage
Kafka Internal Topic	Offsets stored in Kafka’s internal __consumer_offsets topic	Default and recommended method for modern Kafka	Automatic management, supports committing offsets with transactions

Example: Manual Offset Commit to Database

import org.apache.kafka.clients.consumer.KafkaConsumer;
import java.util.*;

KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
consumer.subscribe(Arrays.asList("my-topic"));

while (true) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000));
    for (ConsumerRecord<String, String> record : records) {
        // Process record
        System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value());
    }

    // Commit offset manually to database
    Map<TopicPartition, OffsetAndMetadata> currentOffsets = new HashMap<>();
    for (TopicPartition tp : consumer.assignment()) {
        currentOffsets.put(tp, new OffsetAndMetadata(consumer.position(tp)));
    }
    // Save to DB logic here
    consumer.commitSync(currentOffsets);
}

Note: When storing offsets externally, ensure atomicity and consistency to avoid data loss or duplicate processing.

8. Resetting to Earliest

Option	Values	Description
--to-earliest	`true`	Reset consumer offset to the earliest available message in the topic
--to-latest	`true`	Reset consumer offset to the latest available message in the topic
--to-datetime	`2025-04-01T10:00:00Z`	Reset consumer offset to the earliest message at or after the specified datetime
--to-offset	`12345`	Reset consumer offset to a specific message offset

Example: Resetting Consumer Group to Earliest Offset

kafka-consumer-groups.sh --bootstrap-server localhost:9092 --group my-group --reset-offsets --to-earliest --execute --topic my-topic

Note: The --execute flag is required to actually perform the reset. Without it, only a dry-run will be performed.

9. Resetting to Latest

Resetting consumer offsets to the latest available offset is useful when you want to start processing messages from the most recent point in the topic, effectively skipping all previously consumed messages.

Option	Description	Notes
--to-latest	Reset offset to the latest available offset in the partition.	This option resets all consumers in the consumer group to the latest offset.

Example: Resetting Offsets to Latest

kafka-consumer-groups.sh --bootstrap-server localhost:9092 --group my-group --topic my-topic --reset-offsets --to-latest

Note: This command resets the consumer group's offsets to the latest available offset in the topic. It does not reprocess any messages that were previously consumed.

10. Resetting to Specific Offset

Option	Values	Description
--reset-offsets	`--to-datetime`	Reset to a specific datetime
--reset-offsets	`--to-earliest`	Reset to earliest offset
--reset-offsets	`--to-latest`	Reset to latest offset
--reset-offsets	`--to-current`	Reset to current offset
--reset-offsets	`--to-offset`	Reset to a specific offset value

Example: Reset to Specific Offset

kafka-run-class.sh kafka.tools.ResetOffsets --bootstrap-server localhost:9092 \
  --topic my-topic \
  --group my-group \
  --to-offset 100

Example: Reset to Earliest Offset

kafka-run-class.sh kafka.tools.ResetOffsets --bootstrap-server localhost:9092 \
  --topic my-topic \
  --group my-group \
  --to-earliest

11. Resetting by Timestamp

Option	Values	Description
--time	`timestamp`	Reset offset to the earliest message at or before this timestamp
--to-datetime	`YYYY-MM-DDTHH:mm:ss`	Reset offset to the earliest message at or before this datetime
--to-earliest	`true`	Reset offset to the earliest available message in the partition
--to-latest	`true`	Reset offset to the latest available message in the partition

Example: Resetting Offsets by Timestamp

kafka-consumer-groups.sh --bootstrap-server localhost:9092 --group my-group --reset-offsets --to-datetime 2025-04-01T10:00:00 --topic my-topic

Note: The --to-datetime option requires the timestamp to be in ISO format and must be within the retention period of the topic.