1146 Initial fix for eager initialization in StreamListenerAnnotationBPP

- Removed dependency and callbacks to 'BeanFactoryUtils.beansOfTypeIncludingAncestors' in
    StreamListenerAnnotationBeanPostProcessor in favor of more appropriate and lazy DI
    mechanism provided by Spring.
  - Polishing

polishing as part of the cherry-picking

spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/StreamListenerAnnotationBeanPostProcessor.java

@@ -26,9 +26,7 @@ import org.springframework.aop.support.AopUtils;
 import org.springframework.beans.BeansException;
 import org.springframework.beans.factory.BeanFactory;
 import org.springframework.beans.factory.BeanFactoryAware;
-import org.springframework.beans.factory.BeanFactoryUtils;
 import org.springframework.beans.factory.BeanInitializationException;
-import org.springframework.beans.factory.InitializingBean;
 import org.springframework.beans.factory.NoSuchBeanDefinitionException;
 import org.springframework.beans.factory.SmartInitializingSingleton;
 import org.springframework.beans.factory.annotation.Autowired;
@@ -73,16 +71,19 @@ import org.springframework.util.StringUtils;
  * @author Soby Chacko
  */
 public class StreamListenerAnnotationBeanPostProcessor
-		implements BeanPostProcessor, ApplicationContextAware, BeanFactoryAware, SmartInitializingSingleton,
-		InitializingBean {
+		implements BeanPostProcessor, ApplicationContextAware, BeanFactoryAware, SmartInitializingSingleton {
 
 	private static final SpelExpressionParser SPEL_EXPRESSION_PARSER = new SpelExpressionParser();
 
 	private final MultiValueMap<String, StreamListenerHandlerMethodMapping> mappedListenerMethods = new LinkedMultiValueMap<>();
 
-	private final List<StreamListenerParameterAdapter<?, Object>> streamListenerParameterAdapters = new ArrayList<>();
+	@Autowired(required=false)
+	@Lazy
+	private List<StreamListenerParameterAdapter<?,?>> streamListenerParameterAdapters;
 
-	private final List<StreamListenerResultAdapter<?, ?>> streamListenerResultAdapters = new ArrayList<>();
+	@Autowired(required=false)
+	@Lazy
+	private List<StreamListenerResultAdapter<?, ?>> streamListenerResultAdapters;
 
 	@Autowired
 	@Lazy
@@ -119,24 +120,8 @@ public class StreamListenerAnnotationBeanPostProcessor
 		}
 	}
 
-	@SuppressWarnings({ "rawtypes", "unchecked" })
 	@Override
-	public void afterPropertiesSet() throws Exception {
-		Map<String, StreamListenerParameterAdapter> parameterAdapterMap = BeanFactoryUtils
-				.beansOfTypeIncludingAncestors(this.applicationContext, StreamListenerParameterAdapter.class);
-		for (StreamListenerParameterAdapter parameterAdapter : parameterAdapterMap.values()) {
-			this.streamListenerParameterAdapters.add(parameterAdapter);
-		}
-		Map<String, StreamListenerResultAdapter> resultAdapterMap = BeanFactoryUtils
-				.beansOfTypeIncludingAncestors(this.applicationContext, StreamListenerResultAdapter.class);
-		this.streamListenerResultAdapters.add(new MessageChannelStreamListenerResultAdapter());
-		for (StreamListenerResultAdapter resultAdapter : resultAdapterMap.values()) {
-			this.streamListenerResultAdapters.add(resultAdapter);
-		}
-	}
-
-	@Override
-	public final Object postProcessBeforeInitialization(Object bean, String beanName) throws BeansException {
+	public Object postProcessBeforeInitialization(Object bean, String beanName) throws BeansException {
 		return bean;
 	}
 
@@ -242,7 +227,7 @@ public class StreamListenerAnnotationBeanPostProcessor
 		if (!this.streamListenerParameterAdapters.isEmpty()) {
 			try {
 				Object targetBean = this.applicationContext.getBean(targetBeanName);
-				for (StreamListenerParameterAdapter<?, Object> streamListenerParameterAdapter : this.streamListenerParameterAdapters) {
+				for (StreamListenerParameterAdapter<?, ?> streamListenerParameterAdapter : this.streamListenerParameterAdapters) {
 					if (streamListenerParameterAdapter.supports(targetBean.getClass(), methodParameter)) {
 						return true;
 					}
@@ -276,7 +261,7 @@ public class StreamListenerAnnotationBeanPostProcessor
 				Assert.isInstanceOf(String.class, targetReferenceValue, "Annotation value must be a String");
 				Object targetBean = this.applicationContext.getBean((String) targetReferenceValue);
 				// Iterate existing parameter adapters first
-				for (StreamListenerParameterAdapter<?, Object> streamListenerParameterAdapter : this.streamListenerParameterAdapters) {
+				for (StreamListenerParameterAdapter streamListenerParameterAdapter : this.streamListenerParameterAdapters) {
 					if (streamListenerParameterAdapter.supports(targetBean.getClass(), methodParameter)) {
 						arguments[parameterIndex] = streamListenerParameterAdapter.adapt(targetBean, methodParameter);
 						break;
@@ -520,5 +505,4 @@ public class StreamListenerAnnotationBeanPostProcessor
 		}
 
 	}
-
 }

spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BinderFactoryConfiguration.java

@@ -30,6 +30,7 @@ import java.util.Properties;
 import org.springframework.beans.factory.BeanCreationException;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.annotation.Value;
+import org.springframework.beans.factory.config.BeanDefinition;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
 import org.springframework.cloud.stream.binder.BinderConfiguration;
 import org.springframework.cloud.stream.binder.BinderFactory;
@@ -40,6 +41,7 @@ import org.springframework.cloud.stream.binder.DefaultBinderTypeRegistry;
 import org.springframework.context.ConfigurableApplicationContext;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
+import org.springframework.context.annotation.Role;
 import org.springframework.core.io.Resource;
 import org.springframework.core.io.UrlResource;
 import org.springframework.core.io.support.PropertiesLoaderUtils;
@@ -52,6 +54,7 @@ import org.springframework.util.StringUtils;
  * @author Ilayaperumal Gopinathan
  */
 @Configuration
+@Role(BeanDefinition.ROLE_INFRASTRUCTURE)
 public class BinderFactoryConfiguration {
 
 	private static final String SPRING_CLOUD_STREAM_INTERNAL_PREFIX = "spring.cloud.stream.internal";

spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BindingServiceConfiguration.java

@@ -29,6 +29,7 @@ import org.springframework.beans.BeansException;
 import org.springframework.beans.factory.BeanFactoryUtils;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.annotation.Qualifier;
+import org.springframework.beans.factory.config.BeanDefinition;
 import org.springframework.beans.factory.config.BeanPostProcessor;
 import org.springframework.beans.factory.config.ConfigurableListableBeanFactory;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
@@ -44,6 +45,7 @@ import org.springframework.cloud.stream.binding.ContextStartAfterRefreshListener
 import org.springframework.cloud.stream.binding.DynamicDestinationsBindable;
 import org.springframework.cloud.stream.binding.InputBindingLifecycle;
 import org.springframework.cloud.stream.binding.MessageChannelConfigurer;
+import org.springframework.cloud.stream.binding.MessageChannelStreamListenerResultAdapter;
 import org.springframework.cloud.stream.binding.MessageConverterConfigurer;
 import org.springframework.cloud.stream.binding.OutputBindingLifecycle;
 import org.springframework.cloud.stream.binding.SingleBindingTargetBindable;
@@ -53,6 +55,7 @@ import org.springframework.cloud.stream.converter.CompositeMessageConverterFacto
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 import org.springframework.context.annotation.DependsOn;
+import org.springframework.context.annotation.Role;
 import org.springframework.expression.PropertyAccessor;
 import org.springframework.integration.channel.DirectChannel;
 import org.springframework.integration.channel.PublishSubscribeChannel;
@@ -82,6 +85,7 @@ import org.springframework.util.CollectionUtils;
  */
 @Configuration
 @EnableConfigurationProperties({ BindingServiceProperties.class, SpringIntegrationProperties.class })
+@Role(BeanDefinition.ROLE_INFRASTRUCTURE)
 public class BindingServiceConfiguration {
 
 	public static final String STREAM_LISTENER_ANNOTATION_BEAN_POST_PROCESSOR_NAME = "streamListenerAnnotationBeanPostProcessor";
@@ -99,6 +103,11 @@ public class BindingServiceConfiguration {
 	@Autowired(required = false)
 	private List<MessageConverter> customMessageConverters;
 
+	@Bean
+	public MessageChannelStreamListenerResultAdapter messageChannelStreamListenerResultAdapter() {
+		return new MessageChannelStreamListenerResultAdapter();
+	}
+
 	@Bean
 	public static MessageHandlerMethodFactory messageHandlerMethodFactory(
 			CompositeMessageConverterFactory compositeMessageConverterFactory) {

spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/SpelExpressionConverterConfiguration.java

@@ -18,10 +18,12 @@ package org.springframework.cloud.stream.config;
 
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.annotation.Qualifier;
+import org.springframework.beans.factory.config.BeanDefinition;
 import org.springframework.boot.context.properties.ConfigurationPropertiesBinding;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 import org.springframework.context.annotation.Lazy;
+import org.springframework.context.annotation.Role;
 import org.springframework.core.convert.converter.Converter;
 import org.springframework.expression.EvaluationContext;
 import org.springframework.expression.Expression;
@@ -37,6 +39,7 @@ import org.springframework.integration.context.IntegrationContextUtils;
  * @author Eric Bottard
  */
 @Configuration
+@Role(BeanDefinition.ROLE_INFRASTRUCTURE)
 public class SpelExpressionConverterConfiguration {
 
 	@Bean