Spring Cloud Kafka Streams绑定器中使用StateRestoreListener的问题

当然可以！你遇到的IllegalStateException是因为设置GlobalStateRestoreListener必须在KafkaStreams实例处于CREATED状态时完成，而你之前的代码是在实例已经RUNNING之后才尝试设置，所以触发了错误。下面是具体的解决方法：

核心问题分析 #

Kafka Streams的setGlobalStateRestoreListener方法有严格的状态限制：只能在实例创建完成但还未启动（CREATED状态）时调用。Spring Cloud Kafka Streams默认会自动启动KafkaStreams实例，所以如果你在应用运行后通过getKafkaStreams()获取实例再设置监听器，此时实例已经处于RUNNING状态，自然会报错。

正确实现步骤 #

1. 自定义StateRestoreListener实现类 #

先写一个你自己的监听器，实现org.apache.kafka.streams.processor.StateRestoreListener接口，用于监控恢复进度：

import org.apache.kafka.streams.processor.StateRestoreListener;
import org.apache.kafka.common.TopicPartition;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class CustomStateRestoreListener implements StateRestoreListener {

    private static final Logger log = LoggerFactory.getLogger(CustomStateRestoreListener.class);

    @Override
    public void onRestoreStart(TopicPartition topicPartition, String storeName, long startingOffset, long endingOffset) {
        log.info("状态存储恢复开始：存储名={}, 分区={}, 起始偏移量={}, 结束偏移量={}",
                storeName, topicPartition, startingOffset, endingOffset);
    }

    @Override
    public void onBatchRestored(TopicPartition topicPartition, String storeName, long batchEndOffset, long numRestored) {
        log.info("状态存储批量恢复：存储名={}, 分区={}, 已恢复记录数={}, 当前批次结束偏移量={}",
                storeName, topicPartition, numRestored, batchEndOffset);
    }

    @Override
    public void onRestoreComplete(TopicPartition topicPartition, String storeName, long totalRestored) {
        log.info("状态存储恢复完成：存储名={}, 分区={}, 总恢复记录数={}",
                storeName, topicPartition, totalRestored);
    }
}

2. 在KafkaStreams实例创建后、启动前设置监听器 #

利用Spring的生命周期机制，在StreamsBuilderFactoryBean创建KafkaStreams实例之后、启动之前注入监听器。这里推荐使用StreamsBuilderFactoryBean的内置监听器机制，更贴合Spring Cloud Kafka Streams的生命周期：

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.cloud.stream.binder.kafka.streams.StreamsBuilderFactoryBean;
import org.springframework.context.annotation.Configuration;

import javax.annotation.PostConstruct;

@Configuration
public class KafkaStreamsConfig {

    @Autowired
    @Qualifier("&stream-builder-process")
    private StreamsBuilderFactoryBean streamBuilderFactoryBean;

    @PostConstruct
    public void configureStateRestoreListener() {
        streamBuilderFactoryBean.addListener(new StreamsBuilderFactoryBean.Listener() {
            @Override
            public void onCreate(org.apache.kafka.streams.KafkaStreams streams) {
                // 此时KafkaStreams处于CREATED状态，可安全设置监听器
                streams.setGlobalStateRestoreListener(new CustomStateRestoreListener());
            }

            @Override
            public void onDestroy(org.apache.kafka.streams.KafkaStreams streams) {
                // 可选：实例销毁时的清理逻辑
            }
        });
    }
}

验证效果 #

启动应用后，当状态存储需要恢复时（比如应用重启、新增实例做负载均衡），你就能在日志中看到自定义监听器输出的恢复进度信息了。

内容的提问来源于stack exchange，提问作者Chernilin