Kafka: How to limit the number of messages per partition

Introduction

Apache Kafka is a distributed streaming platform enabling thousands of companies worldwide to process and analyze streams of data in real-time. One of Kafka’s key features is the ability to organize messages into partitions for scalability and parallelism. However, managing the flow of messages ensures that no partition is overloaded with messages becomes essential. This article explores different methods on how to limit the number of messages per partition in Kafka.

Understanding Partitions

Before diving into limiting messages per partition, it is important to understand what partitions are in the context of Kafka. A Kafka topic is divided into multiple partitions, which allows for parallel processing. Each partition is an ordered, immutable sequence of messages that is continually appended. Kafka ensures message order within a partition, not across different partitions.

Setting Partition Size

One way to indirectly limit the number of messages in a Kafka partition is by setting the maximum size of a partition. Kafka can dynamically allocate the size of log segments.

log.segment.bytes=1073741824

This configuration sets the maximum size of a single log segment to 1GB. When the size is reached, a new log segment is created. Although this setting limits the size, not the exact number of messages, it can provide a predictable maximum capacity.

Using the Retention Policy

Kafka allows you to set retention policies that specify how long messages should be retained inside a partition:

log.retention.hours=168
log.retention.bytes=-1

The above configuration retains messages for a maximum of 168 hours (7 days). The second line indicates that there is no limit on the partition size — historical data will be determined by time. By manipulating these settings, you can control how much data, and implicitly, how many messages, are stored in each partition.

Monitoring and Balancing

Kafka does not have built-in configuration to limit the number of messages directly per partition. However, you can monitor partition sizes using Kafka’s metrics and rebalance messages among partitions if necessary.

// Retrieve broker metrics from Kafka
KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(props);
consumer.subscribe(Arrays.asList("my-topic"));

// Logic to analyze partitions and rebalance
// ...

Rebalance logic can be implemented using custom code or third-party tools like LinkedIn’s Cruise Control for Kafka.

Custom Producer Logic

An advanced method of controlling the message throughput is to incorporate logic into the producer application that limits the number of messages sent to a partition based on your custom rules.

// Producer pseudocode
int messagesSent = 0;
final int MAX_MESSAGES_PER_PARTITION = 1000;

while(messagesSent < MAX_MESSAGES_PER_PARTITION) {
    // ... produce messages ...
    messagesSent++;
}

This simplistic method can limit messages, but it requires coordination with all producers writing to the same partitions.

AdminClient API for Inspection

Another method is to use the Kafka AdminClient API to periodically inspect and control the partition sizes.

Properties props = new Properties();
// Set Kafka client properties
// ...
AdminClient adminClient = AdminClient.create(props);
DescribeTopicsResult result = adminClient.describeTopics(Collections.singletonList("my-topic"));
// Analyze the topic and manage partitions

Combine this with broker configuration and custom logic to stop sending messages when a partition reaches a certain size.

Kafka Streams for Smart Data Routing

Kafka Streams is Kafka’s stream processing library. It can be used for smarter routing of the messages to control the population of partitions.

// Kafka Streams pseudo code for data routing
StreamsBuilder builder = new StreamsBuilder();
KStream<String, String> kStream = builder.stream("source-topic");

kStream.filter((key, value) -> {
    // Implement your logic to decide whether to forward the message to a specific
    // partition based on its current state
})
.to("destination-topic");
StreamsConfig config = new StreamsConfig(getProperties());
KafkaStreams streams = new KafkaStreams(builder.build(), config);
streams.start();

This strategy can spread the load evenly across partitions if the business logic permits such distribution.

Useful Configurations

To aid in intelligently controlling the distribution of messages, several broker configurations can be tweaked:

• message.max.bytes – maximum size of a message.
• replica.fetch.max.bytes – controls the number of byte of messages a replica can process from the leader for a given partition.
• max.partition.fetch.bytes – maximum data returned by Kafka in a single fetch request.

These settings, adjusted in harmony, can help ensure that partitions do not get overloaded with large messages or batches of messages.

Conclusion

In conclusion, while Kafka does not provide a direct configuration option to limit the number of messages per partition, there are multiple strategies that can be adopted to manage partition sizes. This ensures optimal performance and reliability of your Kafka system. By implementing appropriate retention policies, monitoring tools, and custom producer logic, you can maintain a balanced Kafka ecosystem.