如何限制Kafka Consumer的poll返回结果字节数？

我的环境包含3个Kafka Broker、十几个各含15个分区的Topic。Java应用启动15个线程，每个线程中的Consumer订阅所有Topic，每个Consumer被分配到不同Topic的3个分区。Topic中的记录为String类型，每条大小均小于1Kb。Consumer配置如下：

"key.deserializer" = "org.apache.kafka.common.serialization.StringDeserializer"
"value.deserializer" = "org.apache.kafka.common.serialization.StringDeserializer"
"bootstrap.servers" = "kafka-1:9092,kafka-2:9092,kafka-3:9092"
"group.id" = "my-group"
"enable.auto.commit" = "false"
"fetch.max.bytes" = "60000"
"max.poll.records" = "10000000"
"auto.offset.reset" = "earliest"

我将fetch.max.bytes设置为60000字节，目的是限制consumer.poll返回的整体数据大小，方便后续调用单请求限制为64Kb的API。但实际中consumer.poll返回的数据远超过该值（比如150Kb），导致API调用失败。

请问我在配置和应用设计中遗漏了什么？如何严格限制consumer.poll的结果字节数？

官方文档提到：

Note that the consumer performs multiple fetches in parallel.

这句话是什么意思？我需要限制并行度吗？

1. fetch.max.bytes的作用误区 #

你误解了fetch.max.bytes的作用范围：它不是限制poll()返回的总数据量，而是限制单个Broker返回的单批次拉取数据大小。你的每个Consumer分配了3个分区，这些分区大概率分布在不同的Broker上，Consumer会同时向多个Broker发起拉取请求，每个请求都能拉取到最多60000字节的数据——比如3个分区分属3个Broker，总数据量就可能达到3*60000=180Kb，这就是实际返回数据超标的原因。

另外，Kafka有个隐含规则：如果单条记录大小超过fetch.max.bytes，为了保证数据能被拉取，这个限制会被突破，但你的每条记录都小于1Kb，所以这个情况可以排除。

2. 官方文档那句话的含义 #

"consumer performs multiple fetches in parallel"的意思是：当Consumer分配的分区分布在多个Broker上时，它会同时向这些Broker发起独立的拉取请求，而不是逐个Broker依次拉取。这种设计是为了提升拉取效率，但也会导致多个Broker返回的数据叠加，总大小突破单个fetch.max.bytes的限制。

你不需要限制这个并行度（Kafka也没有直接配置项可以限制），而是要通过其他方式控制总数据量。

3. 严格限制poll()返回数据量的方案 #

方案一：调整max.poll.records #

既然每条记录小于1Kb，要控制总数据在64Kb以内，可以把max.poll.records设置为60左右（留少量冗余空间）。这个配置是限制poll()返回的最大记录数，配合单条记录的大小，就能间接控制总数据量。

注意：如果后续记录大小有变化，需要同步调整这个值。

方案二：业务代码中拆分处理 #

在poll()获取到数据后，不要直接批量调用API，而是在代码里对拉取到的记录进行分组，每组的总大小控制在64Kb以内，再分别调用API。这种方式更灵活，不受记录大小变化的影响。

方案三：调整分区分配策略 #

可以尝试让Consumer分配的分区尽量集中在同一个Broker上（比如使用自定义分区分配策略），这样fetch.max.bytes的限制就能更接近总数据量的上限。不过这种方式依赖集群的分区分布，灵活性较差。

4. 配置和设计中的遗漏点 #

• 错误理解了fetch.max.bytes的作用范围，将单Broker单批次限制当成了总数据量限制
• max.poll.records设置得过大（10000000），完全没有起到限制作用，反而让poll()能返回大量记录

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