We have a Spring Boot app configured with @EnableKafka and multiple instances of ConcurrentKafkaListenerContainerFactory to consume from different topics that require different deserializers.
With this configuration application startup fails with message
Error creating bean with name 'kafkaListenerContainerFactory' defined in class path resource [org/springframework/boot/autoconfigure/kafka/KafkaAnnotationDrivenConfiguration.class]: Unsatisfied dependency expressed through method 'kafkaListenerContainerFactory' parameter 1; nested exception is org.springframework.beans.factory.NoSuchBeanDefinitionException: No qualifying bean of type 'org.springframework.kafka.core.ConsumerFactory<java.lang.Object, java.lang.Object>' available: expected at least 1 bean which qualifies as autowire candidate. Dependency annotations: {}
KafkaAutoConfiguration attempts to create a ConcurrentKafkaListenerContainerFactory bean due to the condition @ConditionalOnMissingBean(name = "kafkaListenerContainerFactory"). Our configuration does not have a bean with this name hence the error.
Should the condition not check a bean exists with type ConcurrentKafkaListenerContainerFactory rather than a specific bean name? It would then backoff in a scenario like ours where we have multiple beans of this type defined?
Comment From: wilkinsona
Thanks for the issue report. Can you please share a minimal example that reproduces the failure? Looking at the auto-configuration and your description, I cannot see why there's no bean of type ConsumerFactory<Object, Object>. It should have been created by KafkaAutoConfiguration:
https://github.com/spring-projects/spring-boot/blob/2c9cf6511efd8453b39e1c31d66996e8da9b1e33/spring-boot-project/spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/kafka/KafkaAutoConfiguration.java#L82-L86
Comment From: manderson23
Please see the test case at https://gist.github.com/manderson23/bed77497cd556f24bf77d758ae781455
If fails with the error message I mentioned. If I remove the // to exclude KafkaAutoConfiguration the test runs.
Comment From: wilkinsona
Thank you. The definition of your own ConsumerFactory beans was the missing piece. The auto-configured ConsumerFactory backs off if there's any user-provided ConsumerFactory bean in the context but the definition for kafkaListenerContainerFactory expects to consume one that's specifically a ConsumerFactory<Object, Object>. Yours are typed as ConsumerFactory<String, Integer> and ConsumerFactory<Integer, String> so the definition backs off and the consumption then fails.
Comment From: snicoll
I am not entirely sure I agree with that although there is clearly an issue. The current back-off arrangement is meant to make sure a default factory can be created automatically even if the user provides additional factories. Backing off when a bean of that type is present will remove that feature so we shouldn't do that.
The main problem I think is that we expect a ConsumerFactory to be found with that specific signature and there is nothing that actually validates that. @ConditionalOnBean doesn't allow us to specify the generic signature so one option would be to add a custom condition that does the relevant check.
Flagging for team attention to double check if I've missed something from Andy's comment.
Comment From: snicoll
@manderson23 renaming one of your two factories there to kafkaListenerContainerFactory should be enough to workaround the problem for now. I guess none of those are really a "default" for you so it's a bit annoying.
Comment From: manderson23
@snicoll renaming if fine as a workaround but the surprising part to me was that a specific bean name for the factory was required for the auto-configuration to back-off. Requiring a specific bean name doesn't make sense when there are multiple factories and auto-configurations back off more elegantly.
Comment From: snicoll
The issue you’ve reported is legit and we’re going to fix it.
The point I was trying to make was that this feature uses a default factory with a specific bean and its absence from the context may lead to a failure unless you specify a custom factory for each endpoint.
The workaround consists at electing one of the two you have to be the default.
In this particular case, the bean name is a feature of the underlying library, not spring boot so it makes sense to require it.
Comment From: wilkinsona
I am not entirely sure I agree with that
Was there anything in particular that you didn't agree with? My comment was attempting to describe the current behaviour rather than suggesting a solution.
Comment From: snicoll
Was there anything in particular that you didn't agree with? My comment was attempting to describe the current behaviour rather than suggesting a solution.
Yes, I read your comment that way. I think I was disagreeing with "The definition of your own ConsumerFactory beans was the missing piece" but I don't see why now so please ignore me :-)
Comment From: wilkinsona
:-) That was just the missing piece for me to understand the reported behaviour.
Comment From: snicoll
We've discussed this and we think that a condition on the generic type can be avoided by creating the ConsumerFactory<Object,Object> if none exists rather than failing the way we do know. We should double check that this particular ConsumerFactory isn't used anywhere else though.