将 ConsumerRebalanceListener 添加到 ConcurrentKafkaListenerContainerFactory

【问题标题】：Adding ConsumerRebalanceListener to the ConcurrentKafkaListenerContainerFactory将 ConsumerRebalanceListener 添加到 ConcurrentKafkaListenerContainerFactory
【发布时间】：2018-11-29 11:39:28
【问题描述】：

在 Spring Boot 应用程序中，我使用带有 @KafkaListener 注释的类作为消息侦听器。我想在我的应用程序中添加一个 ConsumerRebalanceLister 来管理重新平衡时的缓存数据。

如何将 ConsumerRebalanceListener 添加到 ConcurrentKafkaListenerContainerFactory。 documentation 表示它应该设置在 ContainerProperties 对象上。目前尚不清楚如何访问该对象以进行设置。此外，看起来ConcurrentKafkaListenerContainerFactory 丢弃了重新平衡侦听器，因为它在创建侦听器容器实例时创建了一个新的 ContainerProperties 对象。

我觉得我在这里遗漏了一些非常明显的东西，在this commit 之前，有一种方法可以直接在 ConcurrentKafkaListenerContainerFactory 上设置重新平衡侦听器。

【问题讨论】：

标签： spring-kafka

【解决方案1】：

考虑在ConcurrentKafkaListenerContainerFactory上使用这个方法：

/**
 * Obtain the properties template for this factory - set properties as needed
 * and they will be copied to a final properties instance for the endpoint.
 * @return the properties.
 */
public ContainerProperties getContainerProperties() {

您可以在此处添加您的ConsumerRebalanceListener。你 @Autowired 一个自动配置的 ConcurrentKafkaListenerContainerFactory 并执行提到的注入：

@Autowired
private ConcurrentKafkaListenerContainerFactory containerFactory;

@PostConstruct
public void init() {
    this.containerFactory.getContainerProperties()
            .setConsumerRebalanceListener(myConsumerRebalanceListener());
}

@Bean
public ConsumerRebalanceListener myConsumerRebalanceListener() {
    return new ConsumerRebalanceListener() {
        ...
    };
}

【讨论】：