spring-boot - StreamsException : Unable to initialize state, 如果 Kafka Streams 的多个实例在同一状态目录中运行，则可能发生这种情况

Question

这是关于在生产中升级现有的代码库，它使用从 kafka-clients、kafka-streams、spring-kafka 2.4.0 到 2.6.x 的窗口以及将 spring-boot-starter-parent 从 2.2.2.RELEASE 升级到2.3.x 作为 2.2 与 kafka-streams 2.6 不兼容。

现有代码在旧版本(2.4.0,2.2 Spring 版本)中包含下面提到的这些 bean:

@Bean("DataCompressionCustomTopology")
public Topology customTopology(@Qualifier("CustomFactoryBean") StreamsBuilder streamsBuilder)  {
 //Your topology code
 return streamsBuilder.build();
}
    
@Bean("GenericKafkaStreams")
public KafkaStreams kStream() {
//Your kafka streams code
return kafkaStreams;
}

现在升级 kafka 流、kafka 客户端到 2.6.2 和 spring kafka 到 2.6.x 后，观察到以下异常:

2021-05-13 12:33:51.954 [Persistence-Realtime-Transformation] [main] WARN   o.s.b.w.s.c.AnnotationConfigServletWebServerApplicationContext - Exception encountered during context initialization - cancelling refresh attempt: org.springframework.context.ApplicationContextException: Failed to start bean 'CustomFactoryBean'; nested exception is org.springframework.kafka.KafkaException: Could not start stream: ; nested exception is org.apache.kafka.streams.errors.StreamsException: Unable to initialize state, this can happen if multiple instances of Kafka Streams are running in the same state directory
   

Answer 1

这里的问题是新版本的 spring-kafka 正在根据拓扑 bean 自动初始化另一个 kafka 流实例，而 generickafkaStreams 的另一个 bean 正在从现有代码库中初始化，这导致多个线程试图锁定状态目录因此错误。

即使在 Spring 启动级别禁用 KafkaAutoConfiguration 也不会禁用此行为。很难识别并浪费了很多时间。

解决方法是摆脱拓扑 bean 并拥有我们自己的自定义 kafka 流 bean，如下代码:

  protected Topology customTopology()  {

    //topology code
    return streamsBuilder.build();
    }

    /**
     * This starts kafka stream application and sets the state listener and state
     * store listener.
     * 
     * @return KafkaStreams
     */
    @Bean("GenericKafkaStreams")
    public KafkaStreams kStream() {
    KafkaStreams kafkaStreams = new KafkaStreams(customTopology(), kstreamsconfigs);
    return kafkaStreams;
    }