Apache Kafka is a distributed streaming platform that allows systems to publish and consume records. It is commonly used for building real-time data pipelines and streaming applications. When working with Kafka, a common operational task is to verify if a consumer has properly consumed a specific topic. This process not only ensures data integrity but also aids in troubleshooting consumer behavior.

Understanding Kafka Consumers and Consumer Groups

In Kafka, a Consumer reads data from a topic. Consumers are typically organized into consumer groups for scalability and fault tolerance. Each record published to a topic is delivered to one consumer instance within each subscribing consumer group. Kafka uses a simple mechanism to manage which messages have been consumed using a numeric offset to indicate the last record read by a consumer.

Checking Consumption using Offsets

Kafka maintains a partition log for each topic which consists of an ordered set of messages. Each message in a partition is assigned a sequential id number called an offset. Consumers track their position in the log via these offsets. When examining whether a topic has been consumed:

1. Consumer Offset: This is where the consumer has last read up to in the log.
2. Log End Offset: This is the position of the latest published message in the log.

To ensure that a consumer has read all current messages in a topic, the log end offset for each partition should be greater than or equal to the consumer's offset for that partition.

Tools and Commands for Checking Consumer Status

kafka-consumer-groups.sh

This is a command-line tool that comes with Kafka and can be used to view the offset details to understand consumer consumption. Here's how you can use it:

1. List all consumer groups:

   kafka-consumer-groups.sh --bootstrap-server <broker-address> --list

1. Describe a consumer group:

   kafka-consumer-groups.sh --bootstrap-server <broker-address> --describe --group <group-name>

This command provides output showing each topic and partition that the group is consuming, the current offset of the group, the log end offset (end of the log for that partition), and the lag (the difference between the two).

Example Output

Assume you have a consumer group named example-consumer-group. Using the describe command provides information like:

This table indicates the consumer’s progress. The 'Lag' column showcases how far behind each partition is in terms of messages not yet consumed.

Using Consumer APIs for Programmatic Access

For applications that require regular checks or integrations at a code level, Kafka client libraries (e.g., in Java, Python) provide APIs to fetch consumer offsets and log end offsets. Here’s a snippet in Java:

1Properties props = new Properties();
2props.put("bootstrap.servers", "localhost:9092");
3props.put("group.id", "test-group");
4props.put("enable.auto.commit", "false");
5props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
6props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
7
8KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
9consumer.subscribe(Arrays.asList("my-topic"));
10...
11Map<TopicPartition, Long> logEndOffsets = consumer.endOffsets(consumer.assignment());
12...

Conclusion

Ensuring that all messages from a Kafka topic have been consumed is vital for data consistency in streaming applications. Utilizing Kafka's command-line tools or consumer APIs makes this a manageable task, aiding in both development and troubleshooting of Kafka-based systems.