使用spring-kafka从KTable获取ReadOnlyKeyValueStore遇Bean冲突如何解决？

报错根因说明 #

Spring Boot 结合@EnableKafkaStreams注解启动自动装配时，会在KafkaStreamsDefaultConfiguration类中自动注册名为defaultKafkaStreamsBuilder的StreamsBuilderFactoryBean实例，用户自行定义同类型Bean会触发重复注册冲突。

实现步骤（完全使用Spring默认生命周期管理） #

• 第一步：保留@EnableKafkaStreams注解，不要自行声明StreamsBuilderFactoryBean实例，仅需自定义KafkaStreamsConfiguration配置Bean即可，Spring会自动读取该配置生成并管理StreamsBuilderFactoryBean的全生命周期。
  配置示例：

  @Configuration
  @EnableKafkaStreams
  public class KafkaStreamsBaseConfig {
      @Bean(name = KafkaStreamsDefaultConfiguration.DEFAULT_STREAMS_CONFIG_BEAN_NAME)
      public KafkaStreamsConfiguration streamsConfig() {
          Map<String, Object> props = new HashMap<>();
          props.put(StreamsConfig.APPLICATION_ID_CONFIG, "your-streams-app-id");
          props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "your-kafka-address:9092");
          // 其他自定义Streams配置
          return new KafkaStreamsConfiguration(props);
      }
  }
  

• 第二步：正常构建拓扑，指定GlobalKTable的存储名称
  拓扑构建示例：

  @Component
  public class CustomStreamsTopology {
      @Autowired
      public void buildTopology(StreamsBuilder streamsBuilder) {
          // 定义GlobalKTable并指定物化存储名
          GlobalKTable<String, String> userTable = streamsBuilder.globalTable("user-topic",
              Materialized.<String, String, KeyValueStore<Bytes, byte[]>>as("user_global_store")
                  .withKeySerde(Serdes.String())
                  .withValueSerde(Serdes.String()));
          // 其他流处理、分支逻辑
      }
  }
  

• 第三步：注入Spring管理的StreamsBuilderFactoryBean，监听KafkaStreams运行状态后获取存储实例
  注意：必须等KafkaStreams进入RUNNING状态后才能获取存储，否则会抛出异常
  存储获取示例：

  @Component
  public class StoreQueryService {
      private ReadOnlyKeyValueStore<String, String> userStore;
  
      // 直接注入Spring自动生成的StreamsBuilderFactoryBean
      public StoreQueryService(StreamsBuilderFactoryBean factoryBean) {
          // 监听流状态变更事件，待流启动完成后获取存储
          factoryBean.addListener(new StreamsBuilderFactoryBean.Listener() {
              @Override
              public void streamsAdded(String streamsId, KafkaStreams kafkaStreams) {
                  kafkaStreams.setStateListener((newState, oldState) -> {
                      if (newState == KafkaStreams.State.RUNNING) {
                          userStore = kafkaStreams.store(
                              StoreQueryParameters.fromNameAndType(
                                  "user_global_store",
                                  QueryableStoreTypes.keyValueStore()
                              )
                          );
                      }
                  });
              }
          });
      }
  
      // 对外提供查询能力
      public String queryUser(String userId) {
          if (userStore == null) {
              throw new RuntimeException("存储未就绪，请稍后重试");
          }
          return userStore.get(userId);
      }
  }
  

特殊场景说明 #

如果项目中存在多个Kafka Streams实例，可通过Bean名称defaultKafkaStreamsBuilder注入默认的流构建工厂实例即可。

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