GH-3544: Support SpEL in @KafkaListener containerPostProcessor

Fixes: #3544

#3544

- Enhance resolveContainerPostProcessor method in KafkaListenerAnnotationBeanPostProcessor
  to evaluate SpEL expressions
- Verify containerPostProcessor property in KafkaListener annotation can be specified as a SpEL expression

Author: sobychacko

spring-kafka/src/main/java/org/springframework/kafka/annotation/KafkaListenerAnnotationBeanPostProcessor.java

@@ -745,11 +745,16 @@ private KafkaListenerContainerFactory<?> resolveContainerFactory(KafkaListener k
 
 	private void resolveContainerPostProcessor(MethodKafkaListenerEndpoint<?, ?> endpoint,
 			KafkaListener kafkaListener) {
-
-		String containerPostProcessor = kafkaListener.containerPostProcessor();
-		if (StringUtils.hasText(containerPostProcessor)) {
-			endpoint.setContainerPostProcessor(this.beanFactory.getBean(containerPostProcessor,
-					ContainerPostProcessor.class));
+		Object containerPostProcessor = resolveExpression(kafkaListener.containerPostProcessor());
+		if (containerPostProcessor instanceof ContainerPostProcessor<?, ?, ?> cpp) {
+			endpoint.setContainerPostProcessor(cpp);
+		}
+		else {
+			String containerPostProcessorBeanName = resolveExpressionAsString(kafkaListener.containerPostProcessor(), "containerPostProcessor");
+			if (StringUtils.hasText(containerPostProcessorBeanName)) {
+				endpoint.setContainerPostProcessor(
+						this.beanFactory.getBean(containerPostProcessorBeanName, ContainerPostProcessor.class));
+			}
 		}
 	}
 

spring-kafka/src/test/java/org/springframework/kafka/listener/ContainerCustomizationTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2019-2023 the original author or authors.
+ * Copyright 2019-2024 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -43,6 +43,7 @@
  * Tests for container customizations.
  *
  * @author Francois Rosiere
+ * @author Soby Chacko
  * @since 3.1
  */
 @SuppressWarnings("unused")
@@ -129,7 +130,7 @@ public void postProcessor(String foo) {
 				id = CONTAINER_CUSTOMIZER_AND_POST_PROCESSOR,
 				topics = TOPIC,
 				containerFactory = "containerFactoryWithCustomizer",
-				containerPostProcessor = "infoContainerPostProcessor")
+				containerPostProcessor = "#{__listener.infoContainerPostProcessor}")
 		public void containerCustomizerAndPostProcessor(String foo) {
 		}