将 StateRestoreListener 与 Spring Cloud Kafka Streams binder 一起使用

Question

我打算将 StateRestoreListener 与 Spring Cloud Kafka Streams binder 一起使用。 我需要监控我的应用程序的容错状态存储的恢复进度。 confluent https://docs.confluent.io/current/streams/monitoring.html#streams-monitoring-runtime-status 中有示例。

为了观察您提供的所有状态商店的恢复情况 您的应用程序的一个实例 org.apache.kafka.streams.processor.StateRestoreListener 接口。你 通过设置 org.apache.kafka.streams.processor.StateRestoreListener 调用 KafkaStreams#setGlobalStateRestoreListener 方法。

第一个问题是从应用程序获取 Kafka Streams。我使用

解决了这个问题

StreamsBuilderFactoryBean streamsBuilderFactoryBean = context.getBean("&stream-builder-process", StreamsBuilderFactoryBean.class);
KafkaStreams kafkaStreams = streamsBuilderFactoryBean.getKafkaStreams();

第二个问题是将StateRestoreListener设置为KafkaStreams，因为我得到错误

java.lang.IllegalStateException: 只能设置 GlobalStateRestoreListener 处于已创建状态。当前状态为：正在运行

是否可以在 Spring Cloud Kafka Streams binder 中使用 StateRestoreListener？ 谢谢

Answer 1

您可以通过使用StreamsBuilderFactoryBeanCustomizer 来实现这一点，该StreamsBuilderFactoryBeanCustomizer 允许您访问底层KafkaStreams 对象。如果您使用的是 3.0 或更高版本的活页夹，这是推荐的方法。例如，您可以在应用程序中提供以下 bean 并使用 GlobalStateRestoreListener 对其进行自定义。

@Bean
public StreamsBuilderFactoryBeanCustomizer streamsBuilderFactoryBeanCustomizer() {
    return factoryBean -> {
        factoryBean.setKafkaStreamsCustomizer(new KafkaStreamsCustomizer() {
            @Override
            public void customize(KafkaStreams kafkaStreams) {
                kafkaStreams.setGlobalStateRestoreListener(...);
            }
        });
    };
}

This blog 有更多关于这个策略的细节。