定期重启 KafkaListenerEndpointRegistry 以重新加载 SSL 上下文

Posted 2023-03-22

【中文标题】定期重启 KafkaListenerEndpointRegistry 以重新加载 SSL 上下文

【英文标题】：Periodically restarting KafkaListenerEndpointRegistry to reload SSL context

【发布时间】：2022-01-20 07:24:25

【问题描述】：

我们的应用程序部署到 k8s 集群，其中 TLS 证书每小时更新一次。应用程序也连接到 Kafka，使用在 application.yml 中配置的双向 TLS 身份验证。

据我所知，密钥库和信任库都以不可变的方式加载到 SSLContext 中，并且不能轻易替换。最简单、看似可行的解决方案可能是重新启动所有侦听器，如下面的代码所示：

@Autowired
private KafkaListenerEndpointRegistry messageListenerContainer;

@PostConstruct
public void setupTlsCertificateRefresh() throws IOException 
    watch(() -> 
        log.info("Certificates have been renewed, restarting Kafka message listener container...");
        messageListenerContainer.stop();
        messageListenerContainer.start();
    );

这种方法有什么概念问题吗？是否存在因重启而丢失数据的风险，尤其是在高峰时段？

环境：

Spring Boot 2.5.6 使用@KafkaListener的处理程序方法 手动确认 批处理监听器

我知道kafka-clients 中也有 SslEngineFactory，但老实说，对于这么小的要求，要重新加载密钥库，实现起来太复杂了......

【参考方案1】：

不仅 SSL 选项是不可变的，而且我们传递给 KafkaConsumer 的所有属性都是不可变的，我们无法在运行时更改它们。因此，当我们更改这些消费者选项时，我们确实必须销毁当前消费者并启动一个新消费者。因此重新启动KafkaListenerEndpointRegistry 是正确的方法。所有MessageListenerContainer 都将被停止并从消费者那里清除。下一个start() 将为消费者开启一个新的生命周期。

当您更改属性时，您必须使用MessageListenerContainer.getContainerProperties() 来修改目标消费者的道具。原始属性被克隆到 listener 容器的 ctor 中。

【讨论】：

谢谢，@Artem Bilan！我有同样的印象，这可能是正确的方法。至于属性更改，我理解，但是在我的情况下，属性值并没有真正改变。仅更改文件内容，不更改密钥/信任库的路径。