如何解决Spring Cloud Stream Kafka Streams绑定器多函数绑定同一主题报错？

正确配置方式 #

调整application.yaml配置 #

移除每个函数单独的application-id配置，改为全局配置Kafka Streams应用的application-id，让两个函数共享同一个Streams拓扑：

spring:
  cloud:
    stream:
      function:
        bindings:
          sqrt-in-0: numbers
          sqrt-out-0: sqrt-numbers
          log-in-0: sqrt-numbers
        # 显式指定要启用的函数（可选，默认会扫描所有@Bean声明的函数）
        definition: sqrt;log
      kafka:
        streams:
          # 全局配置统一的application-id
          application-id: combined-streams-application
          # 可添加其他全局配置，如bootstrap-servers等

代码保持不变 #

原有的两个函数Bean无需修改：

@Bean
public Function<KStream<String, Double>, KStream<String, Double>> sqrt() {
    return numbers -> numbers.mapValues(Math::sqrt);
}

@Bean
public Consumer<KStream<String, Double>> log() {
    return sqrtStream -> sqrtStream.foreach((key, value) -> log.info("{}: {}", key, value));
}

错误原因解析 #

你遇到的bean 'sqrt-numbers' could not be registered错误，根源在于：

• 每个application-id对应一个独立的Kafka Streams实例，Spring Cloud Stream会为每个实例用到的Topic创建绑定Bean。
• 原配置中给sqrt和log函数分别配置了独立的application-id，相当于启动两个独立的Streams应用，而它们都依赖sqrt-numbers这个中间Topic，导致Spring尝试两次注册该Topic的绑定Bean，触发重复定义冲突。

将两个函数纳入同一个Streams拓扑（共用一个application-id）后，它们会在同一个Kafka Streams实例中运行，中间Topic的绑定Bean只会被注册一次，解决冲突问题。

内容的提问来源于stack exchange，提问作者Cédric Schaller