Kafka的fetch.max.bytes与max.partition.fetch.bytes配置未生效问题

问题重现 #

你设置了以下消费者参数：

• max.poll.records=1
• max.partition.fetch.bytes=1
• fetch.max.bytes=1
  但使用控制台生产者发送大消息后，消费者仍能正常接收并处理，未触发预期的异常或过滤。以下是你的测试代码：

package io.conduktor.demos.kafka;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.errors.WakeupException;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

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

public class ConsumerDemo {
    private static final Logger log = LoggerFactory.getLogger(ConsumerDemo.class);

    public static void main(String[] args) {
        log.info("I am a Kafka Consumer");

        String bootstrapServers = "127.0.0.1:9092";
        String groupId = "my-fifth-application";
        String topic = "demo_java";

        // create consumer configs
        Properties properties = new Properties();
        properties.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        properties.setProperty(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        properties.setProperty(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        properties.setProperty(ConsumerConfig.GROUP_ID_CONFIG, groupId);
        properties.setProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
        properties.setProperty(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, "1");
        properties.setProperty(ConsumerConfig.MAX_PARTITION_FETCH_BYTES_CONFIG, "1"); 
        properties.setProperty(ConsumerConfig.FETCH_MAX_BYTES_CONFIG, "1");

        // create consumer
        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(properties);

        try {
            // subscribe consumer to our topic(s)
            consumer.subscribe(Arrays.asList(topic));

            // poll for new data
            while (true) {
                ConsumerRecords<String, String> records =
                        consumer.poll(Duration.ofMillis(100));

                for (ConsumerRecord<String, String> record : records) {
                    log.info("Key: " + record.key() + ", Value: " + record.value());
                    log.info("Partition: " + record.partition() + ", Offset:" + record.offset());
                }
            }

        } catch (WakeupException e) {
            log.info("Wake up exception!");
            // we ignore this as this is an expected exception when closing a consumer
        } catch (Exception e) {
            log.error("Unexpected exception", e);
        } finally {
            consumer.close(); // this will also commit the offsets if need be.
            log.info("The consumer is now gracefully closed.");
        }

    }
}

核心原因：参数作用的误解 #

fetch.max.bytes和max.partition.fetch.bytes的核心作用是控制单次拉取的批次总字节数，而非限制单条消息的大小：

• max.partition.fetch.bytes：消费者从单个分区单次拉取的消息总字节数上限。
• fetch.max.bytes：消费者从所有订阅分区单次拉取的消息总字节数上限。

Kafka设计了兜底逻辑：如果单条消息的大小超过上述两个参数，但未超过Broker端的message.max.bytes（默认1MB）或Topic的max.message.bytes，消费者仍会拉取这条消息——目的是避免单个大消息卡住消费进度，导致消费停滞。

参数生效的场景 #

这两个参数仅在以下场景中触发限制行为：

• 单分区多小消息：当单个分区内有多条小消息，总大小超过max.partition.fetch.bytes时，消费者会截断拉取结果，只返回总字节数不超过阈值的消息集合，剩余消息留到下一次拉取。
• 多分区总大小超限：当订阅多个分区时，若所有分区的可拉取消息总大小超过fetch.max.bytes，Kafka会调整每个分区的拉取量，确保总字节数不超过该阈值。
• 消息超过Broker/Topic限制：只有当消息大小超过Broker的message.max.bytes或对应Topic的max.message.bytes时，生产者发送会被Broker拒绝，消费者无法收到该消息——这才是限制消息大小的有效方式。

如何限制消费者接收的大消息 #

Kafka消费者端没有直接限制单条消息大小的参数，需从源头控制：

1. 在Broker配置文件中设置message.max.bytes，全局限制所有Topic的消息最大尺寸。
2. 针对特定Topic，通过命令行工具设置max.message.bytes，覆盖全局配置：

   kafka-topics.sh --alter --topic demo_java --bootstrap-server 127.0.0.1:9092 --config max.message.bytes=1024
   

内容的提问来源于stack exchange，提问作者Zuckerman