EmbeddedKafka kafka streams 用SpringBootTest测试发现两个StreamsBuilderFactoryBeans

Question

在 advice here 之后，我正在尝试使用嵌入式 Kafka 来测试我的 Spring Boot Streams 应用程序。

但是，只需创建给定的配置

@Configuration
@EnableKafkaStreams
public class StreamsTestConfiguration {

    @Value("${" + EmbeddedKafkaBroker.SPRING_EMBEDDED_KAFKA_BROKERS + "}")
    private String brokerAddresses;

    @Bean(name = KafkaStreamsDefaultConfiguration.DEFAULT_STREAMS_CONFIG_BEAN_NAME)
    public KafkaStreamsConfiguration kStreamsConfigs() {
        Map<String, Object> props = new HashMap<>();
        props.put(StreamsConfig.APPLICATION_ID_CONFIG, "testStreams");
        props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, this.brokerAddresses);
        return new KafkaStreamsConfiguration(props);
    }
}

和一个简单的测试

@RunWith(SpringRunner.class)
@SpringBootTest
@EmbeddedKafka(topics = { "topic" })
public class EmbeddedKafkaTest {
    @Autowired
private MyBean tested;

    @Autowired
    private EmbeddedKafkaBroker kafkaBroker;

    @Test
    public void loaded() {}
}

未能达到运行：

Parameter 0 of method kafkaStreamsFactoryBeanConfigurer in org.springframework.boot.autoconfigure.kafka.KafkaStreamsAnnotationDrivenConfiguration required a single bean, but 2 were found:
    - &defaultKafkaStreamsBuilder: defined by method 'defaultKafkaStreamsBuilder' in class path resource [org/springframework/kafka/annotation/KafkaStreamsDefaultConfiguration.class]
    - &stream-builder-process: defined in null
 [...]
Caused by: org.springframework.context.ApplicationContextException: Failed to start bean 'outputBindingLifecycle'; nested exception is org.springframework.beans.factory.UnsatisfiedDependencyException: Error creating bean with name 'kafkaStreamsFactoryBeanConfigurer' defined in org.springframework.boot.autoconfigure.kafka.KafkaStreamsAnnotationDrivenConfiguration: Unsatisfied dependency expressed through method 'kafkaStreamsFactoryBeanConfigurer' parameter 0; nested exception is org.springframework.beans.factory.NoUniqueBeanDefinitionException: No qualifying bean of type 'org.springframework.kafka.config.StreamsBuilderFactoryBean' available: expected single matching bean but found 2: &defaultKafkaStreamsBuilder,&stream-builder-process

如果我从测试 class 中删除 @SpringBootTest，问题就会消失，但实际的被测 bean 无法自动装配。

我自己没有定义StreamBuilderFactoryBean，他们从哪里来？

另外：这个设置是否值得测试用于提供稍后被查询的 KTable 的流？这不像我可以 "use a different topic for each test" 因为流将始终使用相同的主题。我希望我可以通过适当的测试用例设计来解决这个问题，或者我会撞到我还看不到的墙吗？

Answer 1

根据您的堆栈跟踪，您还使用了 Spring Cloud Stream 和 Kafka Streams Binder。请添加适当的标签。

考虑删除显式 @EnableKafkaStreams，因为 Binder 会为您处理基础设施。

EmbeddedKafka kafka streams 用SpringBootTest测试发现两个StreamsBuilderFactoryBeans

EmbeddedKafka kafka streams test with SpringBootTest finds two StreamsBuilderFactoryBeans

spring-boot

spring-cloud

spring-cloud-stream

apache-kafka-streams

spring-kafka