GH-3467: Fix KafkaListenerAnnotationBeanPostProcessor for early BF …

…access (#3468)

Fixes: #3467

#3467

The `KafkaListenerAnnotationBeanPostProcessor.afterPropertiesSet()` gets access to `BeanFactory`
for other beans which causes a warning into logs like:
```
is not eligible for getting processed by all BeanPostProcessors (for example: not eligible for auto-proxying). Is this bean getting eagerly injected into a currently created BeanPostProcessor
```

The rule of thumb is not to have any bean factory access from `BeanPostProcessor` initialization phase.

* Fix `KafkaListenerAnnotationBeanPostProcessor` to `buildEnhancer()` in lazy manner.
Therefore remove an `afterPropertiesSet()` altogether.

(cherry picked from commit 4a135a1)

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

@@ -34,6 +34,7 @@
 import java.util.Properties;
 import java.util.Set;
 import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.function.BiFunction;
 import java.util.regex.Pattern;
@@ -46,7 +47,6 @@
 import org.springframework.beans.BeansException;
 import org.springframework.beans.factory.BeanFactory;
 import org.springframework.beans.factory.BeanInitializationException;
-import org.springframework.beans.factory.InitializingBean;
 import org.springframework.beans.factory.ListableBeanFactory;
 import org.springframework.beans.factory.NoSuchBeanDefinitionException;
 import org.springframework.beans.factory.ObjectFactory;
@@ -154,7 +154,7 @@
  * @see MethodKafkaListenerEndpoint
  */
 public class KafkaListenerAnnotationBeanPostProcessor<K, V>
-		implements BeanPostProcessor, Ordered, ApplicationContextAware, InitializingBean, SmartInitializingSingleton {
+		implements BeanPostProcessor, Ordered, ApplicationContextAware, SmartInitializingSingleton {
 
 	private static final String UNCHECKED = "unchecked";
 
@@ -184,6 +184,8 @@ public class KafkaListenerAnnotationBeanPostProcessor<K, V>
 
 	private final AtomicInteger counter = new AtomicInteger();
 
+	private final AtomicBoolean enhancerIsBuilt = new AtomicBoolean();
+
 	private KafkaListenerEndpointRegistry endpointRegistry;
 
 	private String defaultContainerFactoryBeanName = DEFAULT_KAFKA_LISTENER_CONTAINER_FACTORY_BEAN_NAME;
@@ -297,11 +299,6 @@ public void setCharset(Charset charset) {
 		this.charset = charset;
 	}
 
-	@Override
-	public void afterPropertiesSet() throws Exception {
-		buildEnhancer();
-	}
-
 	@Override
 	public void afterSingletonsInstantiated() {
 		this.registrar.setBeanFactory(this.beanFactory);
@@ -348,7 +345,7 @@ public void afterSingletonsInstantiated() {
 	}
 
 	private void buildEnhancer() {
-		if (this.applicationContext != null) {
+		if (this.applicationContext != null && this.enhancerIsBuilt.compareAndSet(false, true)) {
 			Map<String, AnnotationEnhancer> enhancersMap =
 					this.applicationContext.getBeansOfType(AnnotationEnhancer.class, false, false);
 			if (enhancersMap.size() > 0) {
@@ -374,6 +371,7 @@ public Object postProcessBeforeInitialization(Object bean, String beanName) thro
 
 	@Override
 	public Object postProcessAfterInitialization(final Object bean, final String beanName) throws BeansException {
+		buildEnhancer();
 		if (!this.nonAnnotatedClasses.contains(bean.getClass())) {
 			Class<?> targetClass = AopUtils.getTargetClass(bean);
 			Collection<KafkaListener> classLevelListeners = findListenerAnnotations(targetClass);