将StateRestoreListener与Spring Cloud Kafka Streams绑定器一起使用

Question

我打算将StateRestoreListener与Spring Cloud Kafka Streams绑定器一起使用。我需要监视应用程序的容错状态存储的恢复进度。汇流中有一个例子https://docs.confluent.io/current/streams/monitoring.html#streams-监控运行时状态。

为了观察所有状态存储的恢复，您需要为应用程序提供 org.apache.kafka.streams.processor.StateRestoreListener 接口的实例。您可以通过调用 KafkaStreams#setGlobalStateRestoreListener 方法来设置 org.apache.kafka.streams.processor.StateRestoreListener。

第一个问题是从应用程序获取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绑定器中使用StateRestoreListener？谢谢

Answer 1

您可以通过使用< code > streamsbuilderfactorybean customizer 来实现这一点，该工具允许您访问基础< code > kafk streams 对象。如果您使用的是binder或更高版本，这是推荐的方法。例如，您可以在应用程序中提供以下< code>bean，并使用< code > GlobalStateRestoreListener 对其进行自定义。

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

此博客提供有关此策略的更多详细信息。