同一Kafka Stage内两个函数启动报相同State Directory错误如何解决

问题根因 #

Spring Cloud Stream for Kafka Streams 默认会自动为每一个类型为Function<KStream, KStream>的Bean单独创建独立的Kafka Streams运行实例。你同时注册了normaliseTypeaObject、correlation、correlationStageFunction三个符合自动绑定规则的Function Bean，相当于同一个应用内启动了三个Streams实例，共用默认的状态目录，就会抛出你遇到的冲突错误。就算提前清空状态目录，启动时三个实例同时抢占目录写入权限，依然会报错。

可行解决方案 #

方案1：取消独立子Function的Bean注册 #

只保留最终组合后的correlationStageFunction作为公开的Bean，另外两个仅作为内部方法调用即可，避免Spring为它们创建独立Streams上下文：

// 去掉@Bean注解，仅作为内部方法调用
public Function<KStream<?, TypeA>, KStream<?,TypeB>> normaliseTypeaObject() {
    log.info("Starting Normalize");
    return l ->
        l
            .peek((k,v)-> log.info(v))
            .mapValues(this::normalise);
}

// 去掉@Bean注解，仅作为内部方法调用
public Function<KStream<?, TypeB>, KStream<String, TypeC>> correlation() {
    log.info("Starting Correlate");
    return keyGenerator()
        .andThen(correlate());
}

// 仅保留这个组合后的Function作为Bean注册
@Bean
public Function<KStream<?, TypeA>, KStream<String, TypeC>> correlationStageFunction(){
    log.info("Correlate Stage");
    return normaliseTypeaObject()
        .andThen(correlation());
}

同时在配置文件中指定绑定的唯一函数：

spring:
  cloud:
    function:
      definition: correlationStageFunction

方案2：显式配置要激活的Function #

如果一定要保留两个子Function的@Bean注解，就显式指定Spring Cloud Stream仅激活组合后的那一个函数，忽略另外两个，直接在配置中添加即可：

spring:
  cloud:
    function:
      definition: correlationStageFunction

Spring就不会为另外两个未激活的Function自动创建Streams实例。

方案3：多实例场景单独配置状态目录 #

如果后续你确实需要在同一个应用内运行多个独立的Streams函数，为每个函数单独指定状态目录即可避免冲突：

spring:
  cloud:
    stream:
      kafka:
        streams:
          bindings:
            normaliseTypeaObject-in-0:
              stateStore: /tmp/kafka-streams/normalise
            correlation-in-0:
              stateStore: /tmp/kafka-streams/correlate
            correlationStageFunction-in-0:
              stateStore: /tmp/kafka-streams/combined

你的场景仅需要单个处理链路，优先选择前两个方案即可。

验证步骤 #

1. 删除原有状态目录下的所有残留文件
2. 启动应用，确认仅初始化了一个Kafka Streams实例，无目录冲突报错

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