Apache Kafka is a distributed streaming platform designed to handle data streams in a fault-tolerant and scalable way. One of the key concepts in Kafka is the commit offset, which plays a crucial role in message consumption. In Kafka version 0.10.x, understanding how to retrieve and manage these offsets can significantly impact the efficiency and reliability of data processing pipelines. This article details how to get the group commit offset from Kafka 0.10.x and includes technical explanations and examples.

Understanding Kafka Offsets

In Kafka, an offset is a unique identifier for each record in a partition. It denotes the position of a record within that partition. For consumers grouped together (consumer group), keeping track of which messages have been successfully processed (committed) is managed through offsets. The commit offset for a consumer group in a particular partition is the next message offset that the group is expected to read. By committing this offset, the consumer can handle failures and restarts without data loss or message duplication by resuming from the last committed offset.

How Offsets are Stored in Kafka 0.10.x

In Kafka 0.10.x, offsets can be stored in two ways:

1. Kafka internal topic (__consumer_offsets): From version 0.9 onwards, Kafka provides a built-in mechanism to store offsets within an internal Kafka topic named __consumer_offsets.
2. Zookeeper: Earlier versions largely used Zookeeper for storing offsets. However, storing offsets in Zookeeper isn’t recommended since version 0.9 because of scalability issues.

Retrieving Commit Offsets

To retrieve the committed offsets for a consumer group in Kafka 0.10.x, you can use the Kafka-consumer-groups tool. This command-line utility comes with the Kafka distribution and can list all consumer groups, describe a group, delete consumer group info, or reset consumer group offsets.

Using Kafka Consumer Groups Command

Here’s how to get the group commit offset using the kafka-consumer-groups.sh script:

./kafka-consumer-groups.sh --bootstrap-server localhost:9092 --describe --group my-consumer-group

Replace localhost:9092 with your Kafka server and my-consumer-group with the name of your consumer group. This command returns information about where each consumer in the group is in terms of offset.

Output Analysis:

The output will contain columns showing:

• TOPIC & PARTITION: The specific topic and partition.
• CURRENT-OFFSET: The latest offset that has been consumed.
• LOG-END-OFFSET: The last offset available in the log.
• LAG: The difference between the last available offset and the latest consumed one.

Example Output:

TOPIC           PARTITION  CURRENT-OFFSET  LOG-END-OFFSET  LAG  CONSUMER-ID                                       HOST            CLIENT-ID
my-topic        0          245             250             5    consumer-1-abc123ef-ghi4-5jkl-6789-mnopqr012345  /192.168.99.100 client-1

Handling Offset Commits Programmatically

In scenarios where programmatically handling offsets is necessary, consumer APIs provide methods to manually commit offsets. You can use the manual commit pattern when you need fine-grained control over offset commits.

For example, using Kafka's Consumer API in Java:

1import org.apache.kafka.clients.consumer.KafkaConsumer;
2import org.apache.kafka.clients.consumer.ConsumerRecord;
3import org.apache.kafka.clients.consumer.ConsumerRecords;
4import java.util.Collections;
5import java.util.Properties;
6
7Properties props = new Properties();
8props.put("bootstrap.servers", "localhost:9092");
9props.put("group.id", "my-consumer-group");
10props.put("enable.auto.commit", "false");
11props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
12props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
13KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
14consumer.subscribe(Collections.singletonList("my-topic"));
15
16try {
17    while (true) {
18        ConsumerRecords<String, String> records = consumer.poll(100);
19        for (ConsumerRecord<String, String> record : records) {
20            System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value());
21            consumer.commitSync(); // synchronous commit
22        }
23    }
24} finally {
25    consumer.close();
26}
27

In this example, commitSync() is called to commit the offsets in a synchronous manner.

Summary