Kafka 消费者未加入自定义 groupId答案

【问题标题】：Kafka consumer doesn't join custom groupIdKafka 消费者未加入自定义 groupId
【发布时间】：2017-09-04 13:41:29
【问题描述】：

我根据 Spring Kafka 文档设置了 Kafka ConsumerFactory。但是 groupId 似乎没有被使用。也许我也只是把整个事情弄错了，所以我想让你知道我的经历。

这是我的配置，似乎不起作用：

@Bean
ConsumerFactory<String, KafkaEvent> kafkaEventConsumerFactory() {
    return new DefaultKafkaConsumerFactory<>(
            getConsumerProperties(),
            new StringDeserializer(),
            new JsonDeserializer<>(KafkaEvent.class));
}

Map<String, Object> getConsumerProperties() {
    Map<String, Object> props = new HashMap<>();
    props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092"); // TODO
    props.put(ConsumerConfig.GROUP_ID_CONFIG, "myGroupId");
    props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, true);


    props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 3);
    props.put(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG, 120000);

    props.put(ConsumerConfig.HEARTBEAT_INTERVAL_MS_CONFIG, 45000);
    props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, 70000);

    return props;
}

我有一个像这样配置的@KafkaEventListener，没有再次明确指定 groupId：

@KafkaListener(topics = KafkaEventPublisher.ORDER_TOPIC)
public class KafkaEventListener {

   @Autowired
   private ConsumerFactory<String, KafkaEvent> consumerFactory;

   @KafkaHandler
   public void listenTo(@Payload KafkaEvent event) {
       LOGGER.error(LogMarker.KAFKA, consumerFactory.getConfigurationProperties().toString());
   }

}

我还可以看到我的 groupId "myGroupId" 包含在上面记录的错误日志中。然而让我怀疑的是一些 ConsumerCoordinator 的 DEBUG 日志，它总是声明要加入不同的 groupId，我有点担心这看起来是正确的。

2017-09-04 15:28:13.904 (    ) INFO consumer.internals.AbstractCoordinator             - Successfully joined group org.springframework.kafka.KafkaListenerEndpointContainer#0 with generation 40
2017-09-04 15:28:13.904 (    ) INFO consumer.internals.AbstractCoordinator             - Successfully joined group org.springframework.kafka.KafkaListenerEndpointContainer#0 with generation 40
2017-09-04 15:28:13.906 (    ) INFO consumer.internals.ConsumerCoordinator             - Setting newly assigned partitions [] for group org.springframework.kafka.KafkaListenerEndpointContainer#0
2017-09-04 15:28:13.907 (    ) INFO consumer.internals.ConsumerCoordinator             - Setting newly assigned partitions [my-topic-0] for group org.springframework.kafka.KafkaListenerEndpointContainer#0

同样在 Spring 启动时输出 ConsumerConfig。我可以看到 groupId 是错误的，但是其他属性被正确接管了。

据我了解，我可以通过在 ConsumerFactory 上设置 groupId 或使用 spring.kafka.consumer.group-id 在 application.properties 中设置它来全局设置 groupId。两种变体都不起作用。

只有当我使用 @KafkaListener 注释配置 groupId 时，LOG 才会表明消费者加入了正确的组：

2017-09-04 15:38:30.787 (    ) DEBUG consumer.internals.AbstractCoordinator             - Received successful JoinGroup response for group myGroupId: org.apache.kafka.common.requests.JoinGroupResponse@4c51c449

有了这个配置：

@KafkaListener(topics = KafkaEventPublisher.ORDER_TOPIC, groupId = "myGroupId")

我们正在使用 Spring Boot 2.0.0.M3（因此，Spring Kafka 2.0.0.M3）

【问题讨论】：

标签： java spring apache-kafka spring-kafka

【解决方案1】：

这是 M3 中的一个错误； fixed on master (2.0.3.BUILD-SNAPSHOT)（在 1.3.0.M2 中）。我们预计将在本周晚些时候发布 2.0.0.RC1 候选版本（等待 Spring Framework RC4）。

【讨论】：