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

Question

我们的应用程序部署到 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，但老实说，对于这么小的要求，要重新加载密钥库，实现起来太复杂了......

Answer 1

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

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