Make ConcurrentKafkaListenerContainerFactoryConfigurer Generic

[lipiridi]

Summary

This PR updates ConcurrentKafkaListenerContainerFactoryConfigurer to be generic, allowing it to configure any container factory instead of being limited to <Object, Object>.

Motivation

• Previously, the class was restricted to <Object, Object>, which prevented reuse for other container factories with different key/value types. Making it generic improves flexibility and reusability across various Kafka container configurations.
• This change also enables configuring programmatically created container factories using Kafka listener properties from application.yml, providing greater control and consistency in Kafka consumer setup.

[wilkinsona]

When we're reviewing this we should keep in mind #40174 and #19221. Both issues are related to Kafka auto-configuration and generics.

[snicoll]

Unfortunately, I am not sure that we should be making this change. The configurer is mostly for applying auto-configuration to custom container that are to be exposed for management of the ApplicationContext. I also wonder if the additional complexity is worth it given the primary goal of this class.

This PR doesn't introduce any tests and isn't actionable as such. Before you amend that, I'd like to better understand what you actually mean in the two motivations. Can you provide examples of both and show how the current class is relevant?

[lipiridi]

@snicoll thanks for your response!
We are currently creating container factories manually, but we want to configure them using spring.kafka.listener properties in a more convenient way, instead of writing boilerplate code.
Another risk is that the listener properties might be extended in the future, and we could miss these changes, failing to update the property injection code accordingly.

Here is the a real example of usage I'm trying to achieve:

@Configuration
@RequiredArgsConstructor
public class KafkaConfiguration {

    private final KafkaProperties kafkaProperties;

    @Bean
    public ConcurrentKafkaListenerContainerFactory<String, GenericRecord> buildFactory() {
        var buildProperties = this.kafkaProperties.buildConsumerProperties(null);

        DefaultKafkaConsumerFactory<String, Object> consumerFactory =
                new DefaultKafkaConsumerFactory<>(buildProperties, new StringDeserializer(), new ErrorHandlingDeserializer<>(new KafkaAvroDeserializer()));

        ConcurrentKafkaListenerContainerFactory<String, GenericRecord> factory =
                new ConcurrentKafkaListenerContainerFactory<>();

        ConcurrentKafkaListenerContainerFactoryConfigurer<String, GenericRecord> configurer = new ConcurrentKafkaListenerContainerFactoryConfigurer<>();
        configurer.setKafkaProperties(this.kafkaProperties);
        configurer.configure(factory, consumerFactory);

        return factory;
    }
}

[snicoll]

@lipiridi we won't make the setters public, for sure as this is meant to be injected in user's code and they can't mutate what's been configured for them. Rather, they should apply the configurer and adapt whatever code is necessary.

I had a look to your sample, and we'll take it to the team. This part of the API always was a bit strange to me, perhaps @artembilan has an opinion about this change?

[artembilan]

I think this is wrong:

 ConcurrentKafkaListenerContainerFactoryConfigurer<String, GenericRecord> configurer = new ConcurrentKafkaListenerContainerFactoryConfigurer<>();

There is no need in creating this object manually, rather do this:

 @Bean
    public ConcurrentKafkaListenerContainerFactory<String, GenericRecord> buildFactory(ConcurrentKafkaListenerContainerFactoryConfigurer configurer) {
        var buildProperties = this.kafkaProperties.buildConsumerProperties(null);

        DefaultKafkaConsumerFactory<String, Object> consumerFactory =
                new DefaultKafkaConsumerFactory<>(buildProperties, new StringDeserializer(), new ErrorHandlingDeserializer<>(new KafkaAvroDeserializer()));

        ConcurrentKafkaListenerContainerFactory<String, GenericRecord> factory =
                new ConcurrentKafkaListenerContainerFactory<>();

        configurer.configure(factory, consumerFactory);

        return factory;
    }

I agree with @snicoll , that those generics for key/value on all the Apache Kafka API is painful and useless.

Tell us, please, why current ConcurrentKafkaListenerContainerFactoryConfigurer API is out of use for you?

[lipiridi]

@artembilan Hello!
I cannot use it because the configure method expects <Object, Object> listener and consumer factories, whereas my factory is typified with GenericRecord as the value.

[artembilan]

Hm. OK. My bad.
So, you have to cast it into expectations of that method.
However, let's see now how those generics on your container factory bean are useful afterwards?
Cannot it just be <Object, Object> as well?

[lipiridi]

@artembilan the question is how is it supposed to cast the typified StringDeserializer to Object?
Anyway, I believe erasing types is not the way we want to write code.

I'm okay with not making all methods public, especially since I customize the configurer only by setting my own error handler, and the setCommonErrorHandler method is already public. I just thought it could be useful for many others.

[snicoll]

It's unfortunate that the Kafka API is done this way, but I don't think this is the right place to fix the problem. Introducing the generics would complicate the code for no good reason.

The current situation is far from ideal but I'd rather open this against the Kafka project.

[nosan]

the question is how is it supposed to cast the typified StringDeserializer to Object?

as a workaround, you can use the following:

	@SuppressWarnings({ "unchecked", "rawtypes" })
	private ConcurrentKafkaListenerContainerFactory<Object, Object> buildFactory(
			ConcurrentKafkaListenerContainerFactoryConfigurer configurer) {
		Map<String, Object> kafkaProperties = this.properties.buildConsumerProperties();
		ErrorHandlingDeserializer<Object> avroDeserializer = new ErrorHandlingDeserializer<>(
				new KafkaAvroDeserializer());
		ConcurrentKafkaListenerContainerFactory<Object, Object> factory = new ConcurrentKafkaListenerContainerFactory<>();
		configurer.configure(factory,
				new DefaultKafkaConsumerFactory(kafkaProperties, new StringDeserializer(), avroDeserializer));
		return factory;
	}
	

[artembilan]

You can do just like this instead:

	@Bean
	public ConcurrentKafkaListenerContainerFactory<Object, Object> buildFactory(
			ConcurrentKafkaListenerContainerFactoryConfigurer configurer) {
		
		var buildProperties = this.kafkaProperties.buildConsumerProperties(null);

		DefaultKafkaConsumerFactory<Object, Object> consumerFactory =
				new DefaultKafkaConsumerFactory<>(buildProperties, null, new JsonDeserializer<>());

		ConcurrentKafkaListenerContainerFactory<Object, Object> factory =
				new ConcurrentKafkaListenerContainerFactory<>();

		configurer.configure(factory, consumerFactory);

		return factory;
	}

That StringDeserializer is a default value for the ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG.

I believe erasing types is not the way we want to write code.

I understand your pain and really don't see reason in those generics on the ListenerContainerFactory and ConsumerFactory.
Same applies for the producer side.
We really even don't need those generics in Kafka Client itself.
But the cat is out of the bag and we have to live with whatever is there so far.

Perhaps with the current 4.0.0 generation in the work we may reconsider these generics in Spring Kafka and remove them altogether to avoid confusion like this.
It would be huge effort for Spring Kafka itself.
Then this Spring Boot auto-configuration has to be adjusted as well.
And so on for all the projects using Spring Kafka.
I cannot predict all the impact such a breaking change would cause.
Plus right now I don't see an easy migration path from this generics-aware API.