Skip to content

Commit f5761bd

Browse files
committed
Auto-configure listener container factory without consumer factory
Previously, the presence of a `ConsumerFactory` bean would make the auto-configured one to back off, leading to a failure down the line if no available instance matches the generics criterion. This commit improves the auto-configuration to create a `ConsumerFactory<?,?>` behind the scenes if none is available. Closes gh-19221
1 parent 5e61f0d commit f5761bd

File tree

2 files changed

+29
-4
lines changed

2 files changed

+29
-4
lines changed

spring-boot-project/spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/kafka/KafkaAnnotationDrivenConfiguration.java

Lines changed: 5 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,5 @@
11
/*
2-
* Copyright 2012-2019 the original author or authors.
2+
* Copyright 2012-2020 the original author or authors.
33
*
44
* Licensed under the Apache License, Version 2.0 (the "License");
55
* you may not use this file except in compliance with the License.
@@ -25,6 +25,7 @@
2525
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
2626
import org.springframework.kafka.config.KafkaListenerConfigUtils;
2727
import org.springframework.kafka.core.ConsumerFactory;
28+
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
2829
import org.springframework.kafka.core.KafkaTemplate;
2930
import org.springframework.kafka.listener.AfterRollbackProcessor;
3031
import org.springframework.kafka.listener.ErrorHandler;
@@ -84,9 +85,10 @@ public ConcurrentKafkaListenerContainerFactoryConfigurer kafkaListenerContainerF
8485
@ConditionalOnMissingBean(name = "kafkaListenerContainerFactory")
8586
public ConcurrentKafkaListenerContainerFactory<?, ?> kafkaListenerContainerFactory(
8687
ConcurrentKafkaListenerContainerFactoryConfigurer configurer,
87-
ConsumerFactory<Object, Object> kafkaConsumerFactory) {
88+
ObjectProvider<ConsumerFactory<Object, Object>> kafkaConsumerFactory) {
8889
ConcurrentKafkaListenerContainerFactory<Object, Object> factory = new ConcurrentKafkaListenerContainerFactory<>();
89-
configurer.configure(factory, kafkaConsumerFactory);
90+
configurer.configure(factory, kafkaConsumerFactory
91+
.getIfAvailable(() -> new DefaultKafkaConsumerFactory<>(this.properties.buildConsumerProperties())));
9092
return factory;
9193
}
9294

spring-boot-project/spring-boot-autoconfigure/src/test/java/org/springframework/boot/autoconfigure/kafka/KafkaAutoConfigurationTests.java

Lines changed: 24 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,5 @@
11
/*
2-
* Copyright 2012-2019 the original author or authors.
2+
* Copyright 2012-2020 the original author or authors.
33
*
44
* Licensed under the Apache License, Version 2.0 (the "License");
55
* you may not use this file except in compliance with the License.
@@ -48,6 +48,7 @@
4848
import org.springframework.kafka.config.KafkaListenerContainerFactory;
4949
import org.springframework.kafka.config.KafkaStreamsConfiguration;
5050
import org.springframework.kafka.config.StreamsBuilderFactoryBean;
51+
import org.springframework.kafka.core.ConsumerFactory;
5152
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
5253
import org.springframework.kafka.core.DefaultKafkaProducerFactory;
5354
import org.springframework.kafka.core.KafkaAdmin;
@@ -476,6 +477,16 @@ public void testConcurrentKafkaListenerContainerFactoryWithKafkaTemplate() {
476477
});
477478
}
478479

480+
@Test
481+
public void testConcurrentKafkaListenerContainerFactoryWithCustomConsumerFactory() {
482+
this.contextRunner.withUserConfiguration(ConsumerFactoryConfiguration.class).run((context) -> {
483+
ConcurrentKafkaListenerContainerFactory<?, ?> kafkaListenerContainerFactory = context
484+
.getBean(ConcurrentKafkaListenerContainerFactory.class);
485+
assertThat(kafkaListenerContainerFactory.getConsumerFactory())
486+
.isNotSameAs(context.getBean(ConsumerFactoryConfiguration.class).consumerFactory);
487+
});
488+
}
489+
479490
@Configuration
480491
protected static class MessageConverterConfiguration {
481492

@@ -520,6 +531,18 @@ public AfterRollbackProcessor<Object, Object> afterRollbackProcessor() {
520531

521532
}
522533

534+
@Configuration
535+
protected static class ConsumerFactoryConfiguration {
536+
537+
private final ConsumerFactory<String, Object> consumerFactory = mock(ConsumerFactory.class);
538+
539+
@Bean
540+
public ConsumerFactory<String, Object> myConsumerFactory() {
541+
return this.consumerFactory;
542+
}
543+
544+
}
545+
523546
@Configuration
524547
@EnableKafkaStreams
525548
protected static class EnableKafkaStreamsConfiguration {

0 commit comments

Comments
 (0)