Kafka consumer batch control: optimize poll() behavior based on byte size

In the Kafka consumer, if you need to limit the message batch pulled by a single `poll()` operation based on the total byte size of the message rather than the number of records, you should first configure `FETCH_MAX_BYTES_CONFIG`. This parameter directly affects the underlying data crawling behavior. Together with setting `MAX_POLL_RECORDS_CONFIG` to a large enough value, byte-based batch control can be effectively implemented, thereby optimizing consumer performance and resource utilization.

Kafka consumers pull messages from topic partitions through the poll() method. By default, the MAX_POLL_RECORDS_CONFIG parameter in the consumer configuration limits the maximum number of records returned in a single poll() call, and its default value is 500. However, in practical applications, especially when message sizes vary greatly, simply limiting the number of records may not satisfy the need for precise control over resource consumption or processing efficiency. For example, if the message body is very small, 500 messages may be far from reaching the ideal batch processing capacity; and if the message body is very large, 500 messages may exhaust memory in an instant or cause processing delays. At this point, a more reasonable strategy is to limit the batch based on the total byte size of the message.

Understanding MAX_POLL_RECORDS_CONFIG and FETCH_MAX_BYTES_CONFIG

Kafka provides two key parameters to control the size of the consumer pull batch:

1. MAX_POLL_RECORDS_CONFIG (default: 500)

   • This parameter defines the maximum number of records that will be returned to the application in a single call to the poll() method.
   • It is a client-level restriction that is used primarily to control the granularity with which the application processes batches.

2. FETCH_MAX_BYTES_CONFIG (default: 52428800 bytes, or 50 MB)

   • This parameter defines the maximum number of bytes that the Kafka consumer client can pull from each partition when fetching data from the broker in a single time.
   • It is a lower-level restriction that directly affects the fetch request sent by the consumer to the broker. When the poll() method is called, it triggers one or more fetch requests, and FETCH_MAX_BYTES_CONFIG determines the maximum response size for these requests.

Implement byte-based dynamic batch control

To control poll() batches based on the total message byte size, the best practice is to utilize FETCH_MAX_BYTES_CONFIG.

Core idea: Set FETCH_MAX_BYTES_CONFIG to the desired byte limit, and set MAX_POLL_RECORDS_CONFIG to a large enough value (for example, a theoretically unreachable upper limit) to ensure that the byte limit becomes the main batch control factor.

Configuration example:

Suppose we hope that the total number of bytes of messages pulled by a single poll() operation does not exceed 1MB.

 import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.common.serialization.StringDeserializer;

import java.time.Duration;
import java.util.Properties;
import java.util.Collections;

public class ByteBasedKafkaConsumer {

    public static void main(String[] args) {
        Properties props = new Properties();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "my_byte_based_group");
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());

        // Set the maximum number of bytes to pull to 1MB (1 * 1024 * 1024 bytes)
        props.put(ConsumerConfig.FETCH_MAX_BYTES_CONFIG, 1 * 1024 * 1024); // 1MB

        // Set MAX_POLL_RECORDS_CONFIG to a very large value so that it is not a major limit // For example, Integer.MAX_VALUE or an upper limit that far exceeds the number of records expected to be processed at a time props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, Integer.MAX_VALUE); 

        // You can also set FETCH_MIN_BYTES_CONFIG to control the minimum pull amount to avoid frequent pulls in small batches // props.put(ConsumerConfig.FETCH_MIN_BYTES_CONFIG, 1024); // For example, pull at least 1KB

        KafkaConsumer<string string> consumer = new KafkaConsumer(props);
        consumer.subscribe(Collections.singletonList("my_topic"));

        try {
            while (true) {
                ConsumerRecords<string string> records = consumer.poll(Duration.ofMillis(100));
                if (!records.isEmpty()) {
                    System.out.println("Pulled " records.count() " records.");
                    long totalBytes = 0;
                    for (ConsumerRecord<string string> record : records) {
                        // Note: record.serializedValueSize() and record.serializedKeySize() 
                        // The estimate is the serialized byte size. Actual network transmission may include additional overhead totalBytes = record.serializedValueSize() record.serializedKeySize();
                        // Process record...
                    }
                    System.out.println("Total bytes in this poll: " totalBytes " bytes.");
                    consumer.commitSync(); // Commit offset}
            }
        } finally {
            consumer.close();
        }
    }
}</string></string></string>

In the above example, FETCH_MAX_BYTES_CONFIG is set to 1MB. Since MAX_POLL_RECORDS_CONFIG is set to Integer.MAX_VALUE, the batch of records returned by the poll() method will mainly be subject to the 1MB byte limit. If all messages are very small, poll() may return hundreds or even thousands of records until the total number of bytes approaches 1MB; if the messages are very large, poll() may return only a few records before reaching the 1MB limit.

Things to note and best practices

1. Scope of influence of FETCH_MAX_BYTES_CONFIG:

   • Unlike MAX_POLL_RECORDS_CONFIG, FETCH_MAX_BYTES_CONFIG not only affects the return value of poll(), but more importantly, it limits the maximum response size of the underlying data acquisition request initiated by the consumer to the broker. This means that even if MAX_POLL_RECORDS_CONFIG is set high, the broker will not return more than FETCH_MAX_BYTES_CONFIG bytes of data in a single fetch request.
   • This is critical for network bandwidth and broker load management.

2. Reserved meaning of MAX_POLL_RECORDS_CONFIG:

   • Even though bytes are the primary limit, MAX_POLL_RECORDS_CONFIG still exists as a secondary limit. In extreme cases, if FETCH_MAX_BYTES_CONFIG is set to a very large value (such as the default 50MB) and the message is very small, MAX_POLL_RECORDS_CONFIG can still prevent a single poll() from returning too many records and causing memory overflow.
   • Therefore, it is recommended to set it to a value that is much larger than what is actually expected, but does not completely lose the protective effect.

3. FETCH_MIN_BYTES_CONFIG:

   • To avoid consumers frequently pulling very small batches (especially when traffic is low), you can set FETCH_MIN_BYTES_CONFIG. This parameter defines the minimum number of bytes that the broker should accumulate before responding when a fetch request is made to the broker.
   • Use FETCH_MIN_BYTES_CONFIG in conjunction with FETCH_MAX_BYTES_CONFIG to provide more granular control over the size and frequency of pull batches.

4. Dynamic adjustment:

   • Kafka consumer configuration is usually determined at startup. If "on the fly" adjustment is required, this usually means restarting the consumer instance and loading the new configuration. Kafka itself does not provide an API to dynamically modify these client configurations at runtime.
   • The original requirement of "dynamic setting based on message size" is actually indirectly implemented through FETCH_MAX_BYTES_CONFIG, that is, the value of FETCH_MAX_BYTES_CONFIG is calculated based on the expected average message size and target batch size.

Summarize

When a Kafka consumer needs to limit the batch of messages pulled by a single poll() operation based on the total byte size of the message, it should use FETCH_MAX_BYTES_CONFIG as the main control parameter and set it to the desired byte limit. Also, set MAX_POLL_RECORDS_CONFIG to a value large enough that it does not interfere with the byte limit. This configuration method not only manages the memory and processing power of the consumer more effectively, but also optimizes the data transmission efficiency with the Kafka broker. Understanding the different roles of these two parameters and their collaborative working mechanism is the key to building an efficient and robust Kafka consumer.

The above is the detailed content of Kafka consumer batch control: optimize poll() behavior based on byte size. For more information, please follow other related articles on the PHP Chinese website!