Comprehensive Guide to Retrieving Message Count in Apache Kafka Topics

Introduction

Apache Kafka, as a distributed event streaming platform, plays a crucial role in modern data processing architectures. In practical application scenarios, developers frequently need to understand the current number of messages stored in specific topics, which is essential for monitoring system status, evaluating processing progress, and capacity planning.

Analysis of Kafka Message Stream Characteristics

Obtaining message counts from a consumer perspective has inherent limitations. Kafka is fundamentally designed as an infinite stream processing system, where messages are treated as continuous data flows rather than discrete data collections. This design philosophy determines the complexity of directly acquiring precise message counts.

Kafka brokers expose the total number of messages received since startup through JMX counters, but this approach cannot reflect messages that have already been purged. Under message retention policies, old messages are periodically cleaned, so JMX counters often display values higher than the actual stored message count.

Challenges in Distributed Environments

In Kafka cluster environments, topic messages are distributed across multiple broker nodes, with each node storing only portions of the topic's partitions. This distributed nature makes obtaining global message counts more complex, requiring aggregation of message counts from all partitions to derive accurate totals.

Java Implementation Solution

Using Kafka's AdminClient API enables efficient retrieval of message counts for topic partitions. The core principle is based on offset calculation: for each partition, obtain the beginning offset and latest offset, with the difference representing the partition's message count.

import org.apache.kafka.clients.admin.*;
import org.apache.kafka.common.TopicPartition;
import java.util.*;
import java.util.concurrent.ExecutionException;

public class KafkaMessageCounter {
    public static Map<TopicPartition, Long> getMessageCounts(AdminClient adminClient, String topic) 
        throws ExecutionException, InterruptedException {
        
        Map<TopicPartition, Long> messageCounts = new HashMap<>();
        
        // Retrieve topic partition list
        List<TopicPartition> partitions = getPartitions(adminClient, topic);
        
        // Get beginning offsets
        Map<TopicPartition, OffsetSpec> earliestOffsets = new HashMap<>();
        for (TopicPartition partition : partitions) {
            earliestOffsets.put(partition, OffsetSpec.earliest());
        }
        Map<TopicPartition, ListOffsetsResult.ListOffsetsResultInfo> beginningOffsets = 
            adminClient.listOffsets(earliestOffsets).all().get();
        
        // Get latest offsets
        Map<TopicPartition, OffsetSpec> latestOffsets = new HashMap<>();
        for (TopicPartition partition : partitions) {
            latestOffsets.put(partition, OffsetSpec.latest());
        }
        Map<TopicPartition, ListOffsetsResult.ListOffsetsResultInfo> endOffsets = 
            adminClient.listOffsets(latestOffsets).all().get();
        
        // Calculate message counts per partition
        for (TopicPartition partition : partitions) {
            long startOffset = beginningOffsets.get(partition).offset();
            long endOffset = endOffsets.get(partition).offset();
            long partitionCount = endOffset - startOffset;
            messageCounts.put(partition, partitionCount);
        }
        
        return messageCounts;
    }
    
    private static List<TopicPartition> getPartitions(AdminClient adminClient, String topic) 
        throws ExecutionException, InterruptedException {
        
        DescribeTopicsResult describeResult = adminClient.describeTopics(Collections.singletonList(topic));
        TopicDescription topicDesc = describeResult.all().get().get(topic);
        List<TopicPartition> partitions = new ArrayList<>();
        
        for (TopicPartitionInfo partitionInfo : topicDesc.partitions()) {
            partitions.add(new TopicPartition(topic, partitionInfo.partition()));
        }
        
        return partitions;
    }
}

Configuration and Dependency Management

In Maven projects, the Kafka client dependency needs to be added:

<dependency>
    <groupId>org.apache.kafka</groupId>
    <artifactId>kafka-clients</artifactId>
    <version>3.7.0</version>
</dependency>

Configure AdminClient connection properties:

Properties props = new Properties();
props.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
AdminClient adminClient = AdminClient.create(props);

Command Line Tool Alternatives

Besides Java API, Kafka's built-in command line tools can also be used to obtain message counts:

./bin/kafka-run-class.sh kafka.tools.GetOffsetShell \
  --broker-list <broker>:<port> \
  --topic <topic-name> \
  | awk -F  ":" '{sum += $3} END {print sum}'

This method calculates total message count by retrieving the latest offsets of each partition and summing them, suitable for quick inspection in operational scenarios.

Considerations and Best Practices

In practical applications, the following points require attention:

• Message counts calculated from offsets include messages that have been committed but not yet acknowledged by consumers
• In scenarios with frequent message production and consumption, obtained message counts represent instantaneous values
• Consider the impact of network latency and cluster status on counting accuracy
• For large-scale clusters, asynchronous offset retrieval is recommended to improve performance

Conclusion

Retrieving message counts in Kafka topics is a common but carefully handled requirement. Through the offset query functionality provided by AdminClient API, relatively accurate calculation of partition message counts can be achieved. However, developers need to fully understand Kafka's streaming characteristics and avoid using message counts as precise business metrics. In practical applications, appropriate methods should be selected based on specific scenarios, considering the balance between performance, accuracy, and real-time requirements.