Spring Kafka 2.2.2 Consumer 只返回一条记录而不是 max-poll-records答案

【问题标题】：Spring Kafka 2.2.2 Consumer returns only one record instead of max-poll-recordsSpring Kafka 2.2.2 Consumer 只返回一条记录而不是 max-poll-records
【发布时间】：2019-06-23 14:32:35
【问题描述】：

我们正在使用 Spring Kafka 2.2.2 Release 使用 @KafkaListener 和 ConcurrentKafkaListenerContainerFactory 从 Kafka 检索记录。我们已将 max-poll-records 配置为 5，但它始终只将列表中的 1 条记录提供给消费者，而不是 5 条记录。

虽然配置相同，但它适用于 Spring Kafka 2.1.4.Release。

这是我们的 application.yml 配置：

spring:
  kafka:
    bootstrap-servers: localhost:9092
    consumer:
      enable-auto-commit: false
      max-poll-records: 5
      bootstrap-servers: localhost:9092
      group-id: group_id
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: com.gap.cascade.li.data.xx.xx.CustomDeserialiser

这是我们的 ConcurrentKafkaListenerContainerFactory：

  @Bean
  public ConcurrentKafkaListenerContainerFactory<String, String> kafkaListenerContainerFactory() {
    ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
    factory.setConsumerFactory(consumerFactory());
    factory.setBatchListener(true);

    return factory;
  }

我们是否缺少任何需要为 Spring Kafka 2.2.2 Release 完成的配置？

【问题讨论】：

显示你的@KafkaListener 方法。

标签： spring-kafka consumer

【解决方案1】：

假设你有一个听众

@KafkaListener(...)
public void listen(List<...> data) {
    ...
}

设置factory.setBatchListener(true); 应该适合您（只要准备好多个记录）。

你也可以使用启动属性

spring:
  kafka:
    listener:
      type: batch

做同样的事情；避免需要声明自己的工厂。

如果开启 DEBUG 日志记录，容器将记录轮询返回的记录数。您还可以设置fetch.min.bytes 和fetch.max.wait.ms 来影响如果只有一条立即准备好则返回多少条记录...

spring:
  kafka:
    consumer:
      auto-offset-reset: earliest
      enable-auto-commit: false
      properties:
        fetch.min.bytes: 10000
        fetch.max.wait.ms: 2000
    listener:
      type: batch

顺便说一句，当前的 2.2.x 版本是 2.2.7（启动 2.1.6）。

【讨论】：

感谢 Gary，即使在 Spring Kafka 最新版本中，它现在也可以使用。但是，我们看到一个小问题，有时它会返回 1 条或 2 条记录，而与批量大小无关。