Make ConcurrentKafkaListenerContainerFactoryConfigurer generic to support configuring any container factory

Signed-off-by: Dimitrii Lipiridi <dimitrii.lipiridi@delasport.com>

Author: Dimitrii Lipiridi

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

@@ -44,31 +44,32 @@
  * @author Eddú Meléndez
  * @author Thomas Kåsene
  * @author Moritz Halbritter
+ * @author Dimitrii Lipiridi
  * @since 1.5.0
  */
-public class ConcurrentKafkaListenerContainerFactoryConfigurer {
+public class ConcurrentKafkaListenerContainerFactoryConfigurer<K, V> {
 
 	private KafkaProperties properties;
 
 	private BatchMessageConverter batchMessageConverter;
 
 	private RecordMessageConverter recordMessageConverter;
 
-	private RecordFilterStrategy<Object, Object> recordFilterStrategy;
+	private RecordFilterStrategy<? super K, ? super V> recordFilterStrategy;
 
-	private KafkaTemplate<Object, Object> replyTemplate;
+	private KafkaTemplate<?, ?> replyTemplate;
 
-	private KafkaAwareTransactionManager<Object, Object> transactionManager;
+	private KafkaAwareTransactionManager<?, ?> transactionManager;
 
 	private ConsumerAwareRebalanceListener rebalanceListener;
 
 	private CommonErrorHandler commonErrorHandler;
 
-	private AfterRollbackProcessor<Object, Object> afterRollbackProcessor;
+	private AfterRollbackProcessor<? super K, ? super V> afterRollbackProcessor;
 
-	private RecordInterceptor<Object, Object> recordInterceptor;
+	private RecordInterceptor<K, V> recordInterceptor;
 
-	private BatchInterceptor<Object, Object> batchInterceptor;
+	private BatchInterceptor<K, V> batchInterceptor;
 
 	private Function<MessageListenerContainer, String> threadNameSupplier;
 
@@ -102,23 +103,23 @@ void setRecordMessageConverter(RecordMessageConverter recordMessageConverter) {
 	 * Set the {@link RecordFilterStrategy} to use to filter incoming records.
 	 * @param recordFilterStrategy the record filter strategy
 	 */
-	void setRecordFilterStrategy(RecordFilterStrategy<Object, Object> recordFilterStrategy) {
+	void setRecordFilterStrategy(RecordFilterStrategy<? super K, ? super V> recordFilterStrategy) {
 		this.recordFilterStrategy = recordFilterStrategy;
 	}
 
 	/**
 	 * Set the {@link KafkaTemplate} to use to send replies.
 	 * @param replyTemplate the reply template
 	 */
-	void setReplyTemplate(KafkaTemplate<Object, Object> replyTemplate) {
+	void setReplyTemplate(KafkaTemplate<?, ?> replyTemplate) {
 		this.replyTemplate = replyTemplate;
 	}
 
 	/**
 	 * Set the {@link KafkaAwareTransactionManager} to use.
 	 * @param transactionManager the transaction manager
 	 */
-	void setTransactionManager(KafkaAwareTransactionManager<Object, Object> transactionManager) {
+	void setTransactionManager(KafkaAwareTransactionManager<?, ?> transactionManager) {
 		this.transactionManager = transactionManager;
 	}
 
@@ -144,23 +145,23 @@ public void setCommonErrorHandler(CommonErrorHandler commonErrorHandler) {
 	 * Set the {@link AfterRollbackProcessor} to use.
 	 * @param afterRollbackProcessor the after rollback processor
 	 */
-	void setAfterRollbackProcessor(AfterRollbackProcessor<Object, Object> afterRollbackProcessor) {
+	void setAfterRollbackProcessor(AfterRollbackProcessor<? super K, ? super V> afterRollbackProcessor) {
 		this.afterRollbackProcessor = afterRollbackProcessor;
 	}
 
 	/**
 	 * Set the {@link RecordInterceptor} to use.
 	 * @param recordInterceptor the record interceptor.
 	 */
-	void setRecordInterceptor(RecordInterceptor<Object, Object> recordInterceptor) {
+	void setRecordInterceptor(RecordInterceptor<K, V> recordInterceptor) {
 		this.recordInterceptor = recordInterceptor;
 	}
 
 	/**
 	 * Set the {@link BatchInterceptor} to use.
 	 * @param batchInterceptor the batch interceptor.
 	 */
-	void setBatchInterceptor(BatchInterceptor<Object, Object> batchInterceptor) {
+	void setBatchInterceptor(BatchInterceptor<K, V> batchInterceptor) {
 		this.batchInterceptor = batchInterceptor;
 	}
 
@@ -187,14 +188,14 @@ void setListenerTaskExecutor(SimpleAsyncTaskExecutor listenerTaskExecutor) {
 	 * to configure
 	 * @param consumerFactory the {@link ConsumerFactory} to use
 	 */
-	public void configure(ConcurrentKafkaListenerContainerFactory<Object, Object> listenerFactory,
-			ConsumerFactory<Object, Object> consumerFactory) {
+	public void configure(ConcurrentKafkaListenerContainerFactory<K, V> listenerFactory,
+			ConsumerFactory<? super K, ? super V> consumerFactory) {
 		listenerFactory.setConsumerFactory(consumerFactory);
 		configureListenerFactory(listenerFactory);
 		configureContainer(listenerFactory.getContainerProperties());
 	}
 
-	private void configureListenerFactory(ConcurrentKafkaListenerContainerFactory<Object, Object> factory) {
+	private void configureListenerFactory(ConcurrentKafkaListenerContainerFactory<K, V> factory) {
 		PropertyMapper map = PropertyMapper.get().alwaysApplyingWhenNonNull();
 		Listener properties = this.properties.getListener();
 		map.from(properties::getConcurrency).to(factory::setConcurrency);

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

@@ -114,23 +114,23 @@ class KafkaAnnotationDrivenConfiguration {
 	@Bean
 	@ConditionalOnMissingBean
 	@ConditionalOnThreading(Threading.PLATFORM)
-	ConcurrentKafkaListenerContainerFactoryConfigurer kafkaListenerContainerFactoryConfigurer() {
+	ConcurrentKafkaListenerContainerFactoryConfigurer<Object, Object> kafkaListenerContainerFactoryConfigurer() {
 		return configurer();
 	}
 
 	@Bean(name = "kafkaListenerContainerFactoryConfigurer")
 	@ConditionalOnMissingBean
 	@ConditionalOnThreading(Threading.VIRTUAL)
-	ConcurrentKafkaListenerContainerFactoryConfigurer kafkaListenerContainerFactoryConfigurerVirtualThreads() {
-		ConcurrentKafkaListenerContainerFactoryConfigurer configurer = configurer();
+	ConcurrentKafkaListenerContainerFactoryConfigurer<Object, Object> kafkaListenerContainerFactoryConfigurerVirtualThreads() {
+		ConcurrentKafkaListenerContainerFactoryConfigurer<Object, Object> configurer = configurer();
 		SimpleAsyncTaskExecutor executor = new SimpleAsyncTaskExecutor("kafka-");
 		executor.setVirtualThreads(true);
 		configurer.setListenerTaskExecutor(executor);
 		return configurer;
 	}
 
-	private ConcurrentKafkaListenerContainerFactoryConfigurer configurer() {
-		ConcurrentKafkaListenerContainerFactoryConfigurer configurer = new ConcurrentKafkaListenerContainerFactoryConfigurer();
+	private ConcurrentKafkaListenerContainerFactoryConfigurer<Object, Object> configurer() {
+		ConcurrentKafkaListenerContainerFactoryConfigurer<Object, Object> configurer = new ConcurrentKafkaListenerContainerFactoryConfigurer<>();
 		configurer.setKafkaProperties(this.properties);
 		configurer.setBatchMessageConverter(this.batchMessageConverter);
 		configurer.setRecordMessageConverter(this.recordMessageConverter);
@@ -149,7 +149,7 @@ private ConcurrentKafkaListenerContainerFactoryConfigurer configurer() {
 	@Bean
 	@ConditionalOnMissingBean(name = "kafkaListenerContainerFactory")
 	ConcurrentKafkaListenerContainerFactory<?, ?> kafkaListenerContainerFactory(
-			ConcurrentKafkaListenerContainerFactoryConfigurer configurer,
+			ConcurrentKafkaListenerContainerFactoryConfigurer<Object, Object> configurer,
 			ObjectProvider<ConsumerFactory<Object, Object>> kafkaConsumerFactory,
 			ObjectProvider<ContainerCustomizer<Object, Object, ConcurrentMessageListenerContainer<Object, Object>>> kafkaContainerCustomizer) {
 		ConcurrentKafkaListenerContainerFactory<Object, Object> factory = new ConcurrentKafkaListenerContainerFactory<>();

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

@@ -39,7 +39,7 @@
  */
 class ConcurrentKafkaListenerContainerFactoryConfigurerTests {
 
-	private ConcurrentKafkaListenerContainerFactoryConfigurer configurer;
+	private ConcurrentKafkaListenerContainerFactoryConfigurer<Object, Object> configurer;
 
 	private ConcurrentKafkaListenerContainerFactory<Object, Object> factory;
 
@@ -50,7 +50,7 @@ class ConcurrentKafkaListenerContainerFactoryConfigurerTests {
 	@BeforeEach
 	@SuppressWarnings("unchecked")
 	void setUp() {
-		this.configurer = new ConcurrentKafkaListenerContainerFactoryConfigurer();
+		this.configurer = new ConcurrentKafkaListenerContainerFactoryConfigurer<>();
 		this.properties = new KafkaProperties();
 		this.configurer.setKafkaProperties(this.properties);
 		this.factory = spy(new ConcurrentKafkaListenerContainerFactory<>());