Support @Beans for Partitioning Properties

Fixed support for partition properties to be Spring configured

- Fixed support for 'partitionKeyExtractor' and 'partitionSelector' to be Spring configured
- Added new producer properties 'partitionKeyExtractorName' and 'partitionSelectorName'
- Deprecated 'partitionKeyExtractorClass' and 'partitionSelectorClass' properties
- Removed InitilaizingBean from MessageConverterConfigurer
- Updated documentation
- Added additional tests
- Removed PartitionedProducerTest  as redundant
- Fixed unrelated to this effort BinderPropertiesTests due to recent Boot changes

polishing

spring-cloud-stream-binder-test/src/main/java/org/springframework/cloud/stream/binder/AbstractBinderTests.java

@@ -449,7 +449,6 @@ public abstract class AbstractBinderTests<B extends AbstractTestBinder<? extends
 				bindingServiceProperties,
 				new CompositeMessageConverterFactory(null, null));
 		messageConverterConfigurer.setBeanFactory(applicationContext.getBeanFactory());
-		messageConverterConfigurer.afterPropertiesSet();
 		if (inputChannel) {
 			messageConverterConfigurer.configureInputChannel(channel, channelName);
 		}

spring-cloud-stream-core-docs/src/main/asciidoc/spring-cloud-stream-overview.adoc

@@ -2028,7 +2028,8 @@ In a scaled-up scenario, correct configuration of these two properties is import
 
 ==== Configuring Output Bindings for Partitioning
 
-An output binding is configured to send partitioned data by setting one and only one of its `partitionKeyExpression` or `partitionKeyExtractorClass` properties, as well as its `partitionCount` property.
+An output binding is configured to send partitioned data by setting one and only one of its `partitionKeyExpression` or `partitionKeyExtractorName` (see next paragraph) properties, as well as its `partitionCount` property.
+
 For example, the following is a valid and typical configuration:
 
 ----
@@ -2042,37 +2043,35 @@ A partition key's value is calculated for each message sent to a partitioned out
 The `partitionKeyExpression` is a SpEL expression which is evaluated against the outbound message for extracting the partitioning key.
 
 
-If a SpEL expression is not sufficient for your needs, you can instead calculate the partition key value by setting the property `partitionKeyExtractorClass` to a class which implements the `org.springframework.cloud.stream.binder.PartitionKeyExtractorStrategy` interface.
-While the SpEL expression should usually suffice, more complex cases may use the custom implementation strategy.
-In that case, the property 'partitionKeyExtractorClass' can be set as follows:
+If a SpEL expression is not sufficient for your needs, you can instead calculate the partition key value by providing implementation of `org.springframework.cloud.stream.binder.PartitionKeyExtractorStrategy` and configuring it as a bean (i.e., `@Bean`). In the event you have more then one bean of type `org.springframework.cloud.stream.binder.PartitionKeyExtractorStrategy` available in the Application Context you can further filter it by specifying its name via `partitionKeyExtractorName` property:
 
 ----
-spring.cloud.stream.bindings.output.producer.partitionKeyExtractorClass=com.example.MyKeyExtractor
-spring.cloud.stream.bindings.output.producer.partitionCount=5
-----
-
-Once the message key is calculated, the partition selection process will determine the target partition as a value between `0` and `partitionCount - 1`.
-The default calculation, applicable in most scenarios, is based on the formula `key.hashCode() % partitionCount`.
-This can be customized on the binding, either by setting a SpEL expression to be evaluated against the 'key' (via the `partitionSelectorExpression` property) or by setting a `org.springframework.cloud.stream.binder.PartitionSelectorStrategy` implementation (via the `partitionSelectorClass` property).
-
-The binding level properties for 'partitionSelectorExpression' and 'partitionSelectorClass' can be specified similar to the way 'partitionKeyExpression' and 'partitionKeyExtractorClass' properties are specified in the above examples.
-Additional properties can be configured for more advanced scenarios, as described in the following section.
-
-===== Spring-managed custom `PartitionKeyExtractorClass` implementations
-
-In the example above, a custom strategy such as `MyKeyExtractor` is instantiated by the Spring Cloud Stream directly.
-In some cases, it is necessary for such a custom strategy implementation to be created as a Spring bean, for being able to be managed by Spring, so that it can perform dependency injection, property binding, etc.
-This can be done by configuring it as a @Bean in the application context and using the fully qualified class name as the bean's name, as in the following example.
-
-----
-@Bean(name="com.example.MyKeyExtractor")
-public MyKeyExtractor extractor() {
-    return new MyKeyExtractor();
+--spring.cloud.stream.bindings.output.producer.partitionKeyExtractorName=customPartitionKeyExtractor
+--spring.cloud.stream.bindings.output.producer.partitionCount=5
+. . .
+@Bean
+public CustomPartitionKeyExtractorClass customPartitionKeyExtractor() {
+	return new CustomPartitionKeyExtractorClass();
 }
 ----
 
-As a Spring bean, the custom strategy benefits from the full lifecycle of a Spring bean.
-For example, if the implementation need access to the application context directly, it can make implement 'ApplicationContextAware'.
+NOTE: In previous versions of Spring Cloud Stream you could specify the implementation of `org.springframework.cloud.stream.binder.PartitionKeyExtractorStrategy` as `spring.cloud.stream.bindings.output.producer.partitionKeyExtractorClass` property. Since version 2.0 this property is deprecated and support for it will be removed in a future version.
+
+Once the message key is calculated, the partition selection process will determine the target partition as a value between `0` and `partitionCount - 1`.
+The default calculation, applicable in most scenarios, is based on the formula `key.hashCode() % partitionCount`.
+This can be customized on the binding, either by setting a SpEL expression to be evaluated against the 'key' (via the `partitionSelectorExpression` property) or by configuring an implementation of `org.springframework.cloud.stream.binder.PartitionSelectorStrategy` as a bean (i.e., @Bean). And similarly to the `PartitionKeyExtractorStrategy` you can further filter it using `spring.cloud.stream.bindings.output.producer.partitionSelectorName` property in the event there are more then one bean of this type is available in the Application Context.
+
+----
+--spring.cloud.stream.bindings.output.producer.partitionSelectorName=customPartitionSelector
+. . .
+@Bean
+public CustomPartitionSelectorClass customPartitionSelector() {
+	return new CustomPartitionSelectorClass();
+}
+----
+
+NOTE: In previous versions of Spring Cloud Stream you could specify the implementation of `org.springframework.cloud.stream.binder.PartitionSelectorStrategy` as `spring.cloud.stream.bindings.output.producer.partitionSelectorClass` property. Since version 2.0 this property is deprecated and support for it will be removed in a future version.
+
 
 ===== Configuring Input Bindings for Partitioning
 

spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/ProducerProperties.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2017 the original author or authors.
+ * Copyright 2016-2018 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.
@@ -31,6 +31,7 @@ import org.springframework.expression.Expression;
  * @author Marius Bogoevici
  * @author Ilayaperumal Gopinathan
  * @author Gary Russell
+ * @author Oleg Zhurakousky
  */
 @JsonInclude(Include.NON_DEFAULT)
 public class ProducerProperties {
@@ -38,10 +39,34 @@ public class ProducerProperties {
 	@JsonSerialize(using = ExpressionSerializer.class)
 	private Expression partitionKeyExpression;
 
+	/**
+	 * @deprecated in favor of 'partitionKeyExtractorName'
+	 */
+	@Deprecated
 	private Class<?> partitionKeyExtractorClass;
 
+	/**
+	 * The name of the bean that implements {@link PartitionKeyExtractorStrategy}\.
+	 * Used to extract a key used to compute the partition id (see 'partitionSelector*')
+	 * <br>
+	 * Mutually exclusive with 'partitionKeyExpression'.
+	 */
+	private String partitionKeyExtractorName;
+
+	/**
+	 * @deprecated in favor of 'partitionSelectorName'
+	 */
+	@Deprecated
 	private Class<?> partitionSelectorClass;
 
+	/**
+	 * The name of the bean that implements {@link PartitionSelectorStrategy}\.
+	 * Used to determine partition id based on partition key (see 'partitionKeyExtractor*').
+	 * <br>
+	 * Mutually exclusive with 'partitionSelectorExpression'.
+	 */
+	private String partitionSelectorName;
+
 	@JsonSerialize(using = ExpressionSerializer.class)
 	private Expression partitionSelectorExpression;
 
@@ -63,22 +88,27 @@ public class ProducerProperties {
 		this.partitionKeyExpression = partitionKeyExpression;
 	}
 
+	@Deprecated
 	public Class<?> getPartitionKeyExtractorClass() {
 		return partitionKeyExtractorClass;
 	}
 
+	@Deprecated
 	public void setPartitionKeyExtractorClass(Class<?> partitionKeyExtractorClass) {
 		this.partitionKeyExtractorClass = partitionKeyExtractorClass;
 	}
 
 	public boolean isPartitioned() {
-		return this.partitionKeyExpression != null || partitionKeyExtractorClass != null;
+		return this.partitionCount > 1 || this.partitionKeyExpression != null
+				|| this.partitionKeyExtractorName != null || this.partitionKeyExtractorClass != null;
 	}
 
+	@Deprecated
 	public Class<?> getPartitionSelectorClass() {
 		return partitionSelectorClass;
 	}
 
+	@Deprecated
 	public void setPartitionSelectorClass(Class<?> partitionSelectorClass) {
 		this.partitionSelectorClass = partitionSelectorClass;
 	}
@@ -142,4 +172,20 @@ public class ProducerProperties {
 		this.errorChannelEnabled = errorChannelEnabled;
 	}
 
+	public String getPartitionKeyExtractorName() {
+		return partitionKeyExtractorName;
+	}
+
+	public void setPartitionKeyExtractorName(String partitionKeyExtractorName) {
+		this.partitionKeyExtractorName = partitionKeyExtractorName;
+	}
+
+	public String getPartitionSelectorName() {
+		return partitionSelectorName;
+	}
+
+	public void setPartitionSelectorName(String partitionSelectorName) {
+		this.partitionSelectorName = partitionSelectorName;
+	}
+
 }

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

@@ -17,11 +17,15 @@
 package org.springframework.cloud.stream.binding;
 
 import java.nio.charset.StandardCharsets;
+import java.util.Collections;
+import java.util.Map;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
 
 import org.springframework.beans.BeansException;
 import org.springframework.beans.factory.BeanFactory;
 import org.springframework.beans.factory.BeanFactoryAware;
-import org.springframework.beans.factory.InitializingBean;
 import org.springframework.beans.factory.config.ConfigurableListableBeanFactory;
 import org.springframework.cloud.stream.binder.BinderException;
 import org.springframework.cloud.stream.binder.BinderHeaders;
@@ -53,7 +57,7 @@ import org.springframework.messaging.support.ChannelInterceptorAdapter;
 import org.springframework.messaging.support.ErrorMessage;
 import org.springframework.messaging.support.MessageBuilder;
 import org.springframework.util.Assert;
-import org.springframework.util.ClassUtils;
+import org.springframework.util.CollectionUtils;
 import org.springframework.util.MimeType;
 import org.springframework.util.MimeTypeUtils;
 import org.springframework.util.ObjectUtils;
@@ -73,8 +77,9 @@ import org.springframework.util.StringUtils;
  * @author Soby Chacko
  * @author Oleg Zhurakousky
  */
-public class MessageConverterConfigurer
-		implements MessageChannelAndSourceConfigurer, BeanFactoryAware, InitializingBean {
+public class MessageConverterConfigurer implements MessageChannelAndSourceConfigurer, BeanFactoryAware {
+
+	private final Log logger = LogFactory.getLog(getClass());
 
 	private final MessageBuilderFactory messageBuilderFactory = new MutableMessageBuilderFactory();
 
@@ -84,12 +89,25 @@ public class MessageConverterConfigurer
 
 	private ConfigurableListableBeanFactory beanFactory;
 
+	private final Map<String, PartitionKeyExtractorStrategy> partitionKeyExtractors;
+
+	private final Map<String, PartitionSelectorStrategy> partitionSelectors;
+
 	public MessageConverterConfigurer(BindingServiceProperties bindingServiceProperties,
 			CompositeMessageConverterFactory compositeMessageConverterFactory) {
+		this(bindingServiceProperties, compositeMessageConverterFactory, Collections.emptyMap(), Collections.emptyMap());
+	}
+
+	public MessageConverterConfigurer(BindingServiceProperties bindingServiceProperties,
+			CompositeMessageConverterFactory compositeMessageConverterFactory,
+			Map<String, PartitionKeyExtractorStrategy> partitionKeyExtractors,
+			Map<String, PartitionSelectorStrategy> partitionSelectors) {
 		Assert.notNull(compositeMessageConverterFactory,
 				"The message converter factory cannot be null");
 		this.bindingServiceProperties = bindingServiceProperties;
 		this.compositeMessageConverterFactory = compositeMessageConverterFactory;
+		this.partitionKeyExtractors = partitionKeyExtractors == null ? Collections.emptyMap() : partitionKeyExtractors;
+		this.partitionSelectors = partitionSelectors == null ? Collections.emptyMap() : partitionSelectors;
 	}
 
 	@Override
@@ -97,11 +115,6 @@ public class MessageConverterConfigurer
 		this.beanFactory = (ConfigurableListableBeanFactory) beanFactory;
 	}
 
-	@Override
-	public void afterPropertiesSet() throws Exception {
-		Assert.notNull(this.beanFactory, "Bean factory cannot be empty");
-	}
-
 	@Override
 	public void configureInputChannel(MessageChannel messageChannel, String channelName) {
 		configureMessageChannel(messageChannel, channelName, true);
@@ -165,50 +178,67 @@ public class MessageConverterConfigurer
 		}
 	}
 
-	private PartitionKeyExtractorStrategy getPartitionKeyExtractorStrategy(
-			ProducerProperties producerProperties) {
+	@SuppressWarnings("deprecation")
+	private PartitionKeyExtractorStrategy getPartitionKeyExtractorStrategy(ProducerProperties producerProperties) {
+		PartitionKeyExtractorStrategy partitionKeyExtractor;
 		if (producerProperties.getPartitionKeyExtractorClass() != null) {
-			return getBean(producerProperties.getPartitionKeyExtractorClass().getName(),
-					PartitionKeyExtractorStrategy.class);
+			logger.warn("'partitionKeyExtractorClass' option is deprecated as of v2.0. Please configure partition "
+					+ "key extractor as a @Bean that implements 'PartitionKeyExtractorStrategy'. Additionally you can "
+					+ "specify 'spring.cloud.stream.bindings.output.producer.partitionKeyExtractorName' to specify which "
+					+ "bean to use in the event there are more then one.");
+			partitionKeyExtractor = instantiate(producerProperties.getPartitionKeyExtractorClass(), PartitionKeyExtractorStrategy.class);
 		}
-		return null;
+		else if (StringUtils.hasText(producerProperties.getPartitionKeyExtractorName())) {
+			partitionKeyExtractor = this.partitionKeyExtractors.get(producerProperties.getPartitionKeyExtractorName());
+			Assert.notNull(partitionKeyExtractor, "PartitionKeyExtractorStrategy bean with the name '" + producerProperties.getPartitionKeyExtractorName()
+				+ "' can not be found. Has it been configured (e.g., @Bean)?");
+		}
+		else {
+			Assert.isTrue(this.partitionKeyExtractors.size() <= 1,
+					"Multiple  beans of type 'PartitionKeyExtractorStrategy' found. " + this.partitionKeyExtractors + ". Please "
+							+ "use 'spring.cloud.stream.bindings.output.producer.partitionKeyExtractorName' property to specify "
+							+ "the name of the bean to be used.");
+			partitionKeyExtractor = CollectionUtils.isEmpty(this.partitionKeyExtractors) ?
+					null : this.partitionKeyExtractors.values().iterator().next();
+		}
+		return partitionKeyExtractor;
 	}
 
-	private PartitionSelectorStrategy getPartitionSelectorStrategy(
-			ProducerProperties producerProperties) {
+	@SuppressWarnings("deprecation")
+	private PartitionSelectorStrategy getPartitionSelectorStrategy(ProducerProperties producerProperties) {
+		PartitionSelectorStrategy partitionSelector;
 		if (producerProperties.getPartitionSelectorClass() != null) {
-			return getBean(producerProperties.getPartitionSelectorClass().getName(),
+			logger.warn("'partitionSelectorClass' option is deprecated as of v2.0. Please configure partition "
+					+ "selector as a @Bean that implements 'PartitionSelectorStrategy'. Additionally you can "
+					+ "specify 'spring.cloud.stream.bindings.output.producer.partitionSelectorName' to specify which "
+					+ "bean to use in the event there are more then one.");
+			partitionSelector = instantiate(producerProperties.getPartitionSelectorClass(),
 					PartitionSelectorStrategy.class);
 		}
-		return new DefaultPartitionSelector();
+		else if (StringUtils.hasText(producerProperties.getPartitionSelectorName())) {
+			partitionSelector = this.partitionSelectors.get(producerProperties.getPartitionSelectorName());
+			Assert.notNull(partitionSelector,
+					"PartitionSelectorStrategy bean with the name '" + producerProperties.getPartitionSelectorName()
+							+ "' can not be found. Has it been configured (e.g., @Bean)?");
+		}
+		else {
+			Assert.isTrue(this.partitionSelectors.size() <= 1,
+				"Multiple  beans of type 'PartitionSelectorStrategy' found. " + this.partitionSelectors + ". Please "
+					+ "use 'spring.cloud.stream.bindings.output.producer.partitionSelectorName' property to specify "
+					+ "the name of the bean to be used.");
+			partitionSelector = CollectionUtils.isEmpty(this.partitionSelectors)
+					? new DefaultPartitionSelector() : this.partitionSelectors.values().iterator().next();
+		}
+		return partitionSelector;
 	}
 
 	@SuppressWarnings("unchecked")
-	private <T> T getBean(String className, Class<T> type) {
-		if (this.beanFactory.containsBean(className)) {
-			return this.beanFactory.getBean(className, type);
+	private <T> T instantiate(Class<?> implClass, Class<T> type) {
+		try {
+			return (T) implClass.newInstance();
 		}
-		else {
-			synchronized (this) {
-				T bean;
-				Class<?> clazz;
-				try {
-					clazz = ClassUtils.forName(className, this.beanFactory.getBeanClassLoader());
-				}
-				catch (Exception e) {
-					throw new BinderException("Failed to load class: " + className, e);
-				}
-				try {
-					bean = (T) clazz.newInstance();
-					Assert.isInstanceOf(type, bean);
-					this.beanFactory.registerSingleton(className, bean);
-					this.beanFactory.initializeBean(bean, className);
-				}
-				catch (Exception e) {
-					throw new BinderException("Failed to instantiate class: " + className, e);
-				}
-				return bean;
-			}
+		catch (Exception e) {
+			throw new BinderException("Failed to instantiate class: " + implClass.getName(), e);
 		}
 	}
 
@@ -253,22 +283,22 @@ public class MessageConverterConfigurer
 			if (message instanceof ErrorMessage) {
 				return message;
 			}
-			
+
 			Message<?> postProcessedMessage = message;
 			MimeType contentType = this.mimeType;
 			if (message.getHeaders().containsKey(MessageHeaders.CONTENT_TYPE)) {
 				Object ct = message.getHeaders().get(MessageHeaders.CONTENT_TYPE);
 				contentType = ct instanceof String ? MimeType.valueOf((String)ct) : (MimeType)ct;
 			}
-			
-			boolean deserializationRequired = message.getPayload() instanceof byte[] && 
+
+			boolean deserializationRequired = message.getPayload() instanceof byte[] &&
 					("text".equalsIgnoreCase(contentType.getType()) ||
 					equalTypeAndSubType(MimeTypeUtils.APPLICATION_JSON, contentType) ||
 					equalTypeAndSubType(MessageConverterUtils.X_JAVA_SERIALIZED_OBJECT, contentType) ||
 					equalTypeAndSubType(MessageConverterUtils.X_JAVA_OBJECT, contentType));
-			
+
 			Object payload = deserializationRequired ? this.deserializePayload(message, contentType) : message.getPayload();
-			
+
 			if (payload != null) {
 				Object ct = message.getHeaders().get(BinderHeaders.BINDER_ORIGINAL_CONTENT_TYPE);
 				contentType = ct instanceof String ? MimeType.valueOf((String)ct) : (ct == null ? contentType : (MimeType)ct);
@@ -312,7 +342,7 @@ public class MessageConverterConfigurer
 								+ message.getHeaders().get(BinderHeaders.BINDER_ORIGINAL_CONTENT_TYPE), e);
 					}
 				}
-				
+
 				Assert.isTrue(!(equalTypeAndSubType(MessageConverterUtils.X_JAVA_OBJECT, contentTypeToUse) && targetClass == null),
 						"Cannot deserialize into message since 'contentType` is not "
 							+ "encoded with the actual target type."

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

@@ -28,6 +28,8 @@ import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean
 import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
 import org.springframework.boot.context.properties.EnableConfigurationProperties;
 import org.springframework.cloud.stream.binder.BinderFactory;
+import org.springframework.cloud.stream.binder.PartitionKeyExtractorStrategy;
+import org.springframework.cloud.stream.binder.PartitionSelectorStrategy;
 import org.springframework.cloud.stream.binding.AbstractBindingTargetFactory;
 import org.springframework.cloud.stream.binding.Bindable;
 import org.springframework.cloud.stream.binding.BinderAwareChannelResolver;
@@ -123,8 +125,10 @@ public class BindingServiceConfiguration {
 
 	@Bean
 	public MessageConverterConfigurer messageConverterConfigurer(BindingServiceProperties bindingServiceProperties,
-			CompositeMessageConverterFactory compositeMessageConverterFactory) {
-		return new MessageConverterConfigurer(bindingServiceProperties, compositeMessageConverterFactory);
+			CompositeMessageConverterFactory compositeMessageConverterFactory,
+			@Nullable Map<String, PartitionKeyExtractorStrategy> partitionKeyExtractors,
+			@Nullable Map<String, PartitionSelectorStrategy> partitionSelectors) {
+		return new MessageConverterConfigurer(bindingServiceProperties, compositeMessageConverterFactory, partitionKeyExtractors, partitionSelectors);
 	}
 
 	@Bean

spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binding/CustomPartitionedProducerTest.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2017 the original author or authors.
+ * Copyright 2017-2018 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.
@@ -20,11 +20,9 @@ import java.lang.reflect.Field;
 
 import org.junit.Assert;
 import org.junit.Test;
-import org.junit.runner.RunWith;
 
-import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.SpringApplication;
 import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
-import org.springframework.boot.test.context.SpringBootTest;
 import org.springframework.cloud.stream.annotation.EnableBinding;
 import org.springframework.cloud.stream.binder.PartitionHandler;
 import org.springframework.cloud.stream.binder.PartitionKeyExtractorStrategy;
@@ -33,6 +31,7 @@ import org.springframework.cloud.stream.messaging.Source;
 import org.springframework.cloud.stream.partitioning.CustomPartitionKeyExtractorClass;
 import org.springframework.cloud.stream.partitioning.CustomPartitionSelectorClass;
 import org.springframework.cloud.stream.utils.MockBinderRegistryConfiguration;
+import org.springframework.context.ApplicationContext;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Import;
 import org.springframework.context.annotation.PropertySource;
@@ -43,22 +42,23 @@ import org.springframework.integration.core.MessageSource;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.MessagingException;
 import org.springframework.messaging.support.ChannelInterceptor;
-import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
 import org.springframework.util.ReflectionUtils;
 
 /**
  * @author Ilayaperumal Gopinathan
+ * @author Oleg Zhurakousky
  */
-@RunWith(SpringJUnit4ClassRunner.class)
-@SpringBootTest(classes = CustomPartitionedProducerTest.TestSource.class)
 public class CustomPartitionedProducerTest {
 
-	@Autowired
-	private Source testSource;
-
 	@Test
 	public void testCustomPartitionedProducer() {
-		DirectChannel messageChannel = (DirectChannel) this.testSource.output();
+		ApplicationContext context = SpringApplication.run(CustomPartitionedProducerTest.TestSource.class, 
+				"--spring.jmx.enabled=false",
+				"--spring.main.web-application-type=none",
+				"--spring.cloud.stream.bindings.output.producer.partitionKeyExtractorClass=org.springframework.cloud.stream.partitioning.CustomPartitionKeyExtractorClass",
+				"--spring.cloud.stream.bindings.output.producer.partitionSelectorClass=org.springframework.cloud.stream.partitioning.CustomPartitionSelectorClass");
+		Source testSource = context.getBean(Source.class);
+		DirectChannel messageChannel = (DirectChannel) testSource.output();
 		for (ChannelInterceptor channelInterceptor : messageChannel.getChannelInterceptors()) {
 			if (channelInterceptor instanceof MessageConverterConfigurer.PartitioningInterceptor) {
 				Field partitionHandlerField = ReflectionUtils
@@ -80,6 +80,103 @@ public class CustomPartitionedProducerTest {
 			}
 		}
 	}
+	
+	@Test
+	public void testCustomPartitionedProducerByName() {
+		ApplicationContext context = SpringApplication.run(CustomPartitionedProducerTest.TestSource.class, 
+				"--spring.jmx.enabled=false",
+				"--spring.main.web-application-type=none",
+				"--spring.cloud.stream.bindings.output.producer.partitionKeyExtractorName=customPartitionKeyExtractor",
+				"--spring.cloud.stream.bindings.output.producer.partitionSelectorName=customPartitionSelector");
+		Source testSource = context.getBean(Source.class);
+		DirectChannel messageChannel = (DirectChannel) testSource.output();
+		for (ChannelInterceptor channelInterceptor : messageChannel.getChannelInterceptors()) {
+			if (channelInterceptor instanceof MessageConverterConfigurer.PartitioningInterceptor) {
+				Field partitionHandlerField = ReflectionUtils
+						.findField(MessageConverterConfigurer.PartitioningInterceptor.class, "partitionHandler");
+				ReflectionUtils.makeAccessible(partitionHandlerField);
+				PartitionHandler partitionHandler = (PartitionHandler) ReflectionUtils.getField(partitionHandlerField,
+						channelInterceptor);
+				Field partitonKeyExtractorField = ReflectionUtils.findField(PartitionHandler.class,
+						"partitionKeyExtractorStrategy");
+				ReflectionUtils.makeAccessible(partitonKeyExtractorField);
+				Field partitonSelectorField = ReflectionUtils.findField(PartitionHandler.class,
+						"partitionSelectorStrategy");
+				ReflectionUtils.makeAccessible(partitonSelectorField);
+				Assert.assertTrue(((PartitionKeyExtractorStrategy) ReflectionUtils.getField(partitonKeyExtractorField,
+						partitionHandler)).getClass().equals(CustomPartitionKeyExtractorClass.class));
+				Assert.assertTrue(
+						((PartitionSelectorStrategy) ReflectionUtils.getField(partitonSelectorField, partitionHandler))
+								.getClass().equals(CustomPartitionSelectorClass.class));
+			}
+		}
+	}
+	
+	@Test
+	public void testCustomPartitionedProducerAsSingletons() {
+		ApplicationContext context = SpringApplication.run(CustomPartitionedProducerTest.TestSource.class, 
+				"--spring.jmx.enabled=false", "--spring.main.web-application-type=none");
+		Source testSource = context.getBean(Source.class);
+		DirectChannel messageChannel = (DirectChannel) testSource.output();
+		for (ChannelInterceptor channelInterceptor : messageChannel.getChannelInterceptors()) {
+			if (channelInterceptor instanceof MessageConverterConfigurer.PartitioningInterceptor) {
+				Field partitionHandlerField = ReflectionUtils
+						.findField(MessageConverterConfigurer.PartitioningInterceptor.class, "partitionHandler");
+				ReflectionUtils.makeAccessible(partitionHandlerField);
+				PartitionHandler partitionHandler = (PartitionHandler) ReflectionUtils.getField(partitionHandlerField,
+						channelInterceptor);
+				Field partitonKeyExtractorField = ReflectionUtils.findField(PartitionHandler.class,
+						"partitionKeyExtractorStrategy");
+				ReflectionUtils.makeAccessible(partitonKeyExtractorField);
+				Field partitonSelectorField = ReflectionUtils.findField(PartitionHandler.class,
+						"partitionSelectorStrategy");
+				ReflectionUtils.makeAccessible(partitonSelectorField);
+				Assert.assertTrue(((PartitionKeyExtractorStrategy) ReflectionUtils.getField(partitonKeyExtractorField,
+						partitionHandler)).getClass().equals(CustomPartitionKeyExtractorClass.class));
+				Assert.assertTrue(
+						((PartitionSelectorStrategy) ReflectionUtils.getField(partitonSelectorField, partitionHandler))
+								.getClass().equals(CustomPartitionSelectorClass.class));
+			}
+		}
+	}
+	
+	public void testCustomPartitionedProducerMultipleInstances() {
+		ApplicationContext context = SpringApplication.run(CustomPartitionedProducerTest.TestSourceMultipleStrategies.class, 
+				"--spring.jmx.enabled=false", 
+				"--spring.main.web-application-type=none",
+				"--spring.cloud.stream.bindings.output.producer.partitionKeyExtractorName=customPartitionKeyExtractorOne",
+				"--spring.cloud.stream.bindings.output.producer.partitionSelectorName=customPartitionSelectorTwo");
+		Source testSource = context.getBean(Source.class);
+		DirectChannel messageChannel = (DirectChannel) testSource.output();
+		for (ChannelInterceptor channelInterceptor : messageChannel.getChannelInterceptors()) {
+			if (channelInterceptor instanceof MessageConverterConfigurer.PartitioningInterceptor) {
+				Field partitionHandlerField = ReflectionUtils
+						.findField(MessageConverterConfigurer.PartitioningInterceptor.class, "partitionHandler");
+				ReflectionUtils.makeAccessible(partitionHandlerField);
+				PartitionHandler partitionHandler = (PartitionHandler) ReflectionUtils.getField(partitionHandlerField,
+						channelInterceptor);
+				Field partitonKeyExtractorField = ReflectionUtils.findField(PartitionHandler.class,
+						"partitionKeyExtractorStrategy");
+				ReflectionUtils.makeAccessible(partitonKeyExtractorField);
+				Field partitonSelectorField = ReflectionUtils.findField(PartitionHandler.class,
+						"partitionSelectorStrategy");
+				ReflectionUtils.makeAccessible(partitonSelectorField);
+				Assert.assertTrue(((PartitionKeyExtractorStrategy) ReflectionUtils.getField(partitonKeyExtractorField,
+						partitionHandler)).getClass().equals(CustomPartitionKeyExtractorClass.class));
+				Assert.assertTrue(
+						((PartitionSelectorStrategy) ReflectionUtils.getField(partitonSelectorField, partitionHandler))
+								.getClass().equals(CustomPartitionSelectorClass.class));
+			}
+		}
+	}
+	
+	@Test(expected=Exception.class) 
+	// It actually throws UnsatisfiedDependencyException, but it is confusing when it comes to test
+	// But for the purposes of the test all we care about is that it fails
+	public void testCustomPartitionedProducerMultipleInstancesFailNoFilter() {
+		SpringApplication.run(CustomPartitionedProducerTest.TestSourceMultipleStrategies.class, 
+				"--spring.jmx.enabled=false", "--spring.main.web-application-type=none");
+	}
 
 	@EnableBinding(Source.class)
 	@EnableAutoConfiguration
@@ -87,6 +184,54 @@ public class CustomPartitionedProducerTest {
 	@PropertySource("classpath:/org/springframework/cloud/stream/binder/custom-partitioned-producer-test.properties")
 	public static class TestSource {
 
+		@Bean
+		public CustomPartitionSelectorClass customPartitionSelector() {
+			return new CustomPartitionSelectorClass();
+		}
+		
+		@Bean
+		public CustomPartitionKeyExtractorClass customPartitionKeyExtractor() {
+			return new CustomPartitionKeyExtractorClass();
+		}
+		
+		@Bean
+		@InboundChannelAdapter(value = Source.OUTPUT, poller = @Poller(fixedDelay = "5000", maxMessagesPerPoll = "1"))
+		public MessageSource<String> timerMessageSource() {
+			return new MessageSource<String>() {
+				@Override
+				public Message<String> receive() {
+					throw new MessagingException("test");
+				}
+			};
+		}
+	}
+	
+	@EnableBinding(Source.class)
+	@EnableAutoConfiguration
+	@Import(MockBinderRegistryConfiguration.class)
+	@PropertySource("classpath:/org/springframework/cloud/stream/binder/custom-partitioned-producer-test.properties")
+	public static class TestSourceMultipleStrategies {
+
+		@Bean
+		public CustomPartitionSelectorClass customPartitionSelectorOne() {
+			return new CustomPartitionSelectorClass();
+		}
+		
+		@Bean
+		public CustomPartitionSelectorClass customPartitionSelectorTwo() {
+			return new CustomPartitionSelectorClass();
+		}
+		
+		@Bean
+		public CustomPartitionKeyExtractorClass customPartitionKeyExtractorOne() {
+			return new CustomPartitionKeyExtractorClass();
+		}
+		
+		@Bean
+		public CustomPartitionKeyExtractorClass customPartitionKeyExtractorTwo() {
+			return new CustomPartitionKeyExtractorClass();
+		}
+		
 		@Bean
 		@InboundChannelAdapter(value = Source.OUTPUT, poller = @Poller(fixedDelay = "5000", maxMessagesPerPoll = "1"))
 		public MessageSource<String> timerMessageSource() {

spring-cloud-stream/src/test/java/org/springframework/cloud/stream/config/BinderPropertiesTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2017 the original author or authors.
+ * Copyright 2017-2018 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.
@@ -24,7 +24,7 @@ import org.junit.Test;
 
 import org.springframework.beans.factory.support.DefaultListableBeanFactory;
 import org.springframework.boot.actuate.context.properties.ConfigurationPropertiesReportEndpoint;
-import org.springframework.boot.actuate.context.properties.ConfigurationPropertiesReportEndpoint.ConfigurationPropertiesDescriptor;
+import org.springframework.boot.actuate.context.properties.ConfigurationPropertiesReportEndpoint.ContextConfigurationProperties;
 import org.springframework.context.support.StaticApplicationContext;
 
 import static org.junit.Assert.assertFalse;
@@ -61,7 +61,9 @@ public class BinderPropertiesTests {
 		ConfigurationPropertiesReportEndpoint endpoint = new ConfigurationPropertiesReportEndpoint();
 		endpoint.setApplicationContext(context);
 
-		ConfigurationPropertiesDescriptor configurationProperties = endpoint.configurationProperties();
+		
+		ContextConfigurationProperties configurationProperties = endpoint.configurationProperties().getContexts().values().iterator().next();
+		
 		Map<String, Object> properties = configurationProperties.getBeans().get("bindingServiceProperties").getProperties();
 		assertFalse(properties.containsKey("error"));
 		assertTrue(properties.containsKey("binders"));

spring-cloud-stream/src/test/java/org/springframework/cloud/stream/partitioning/PartitionedProducerTest.java

@@ -1,76 +0,0 @@
-/*
- * Copyright 2015-2017 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.
- * You may obtain a copy of the License at
- *
- *      http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.springframework.cloud.stream.partitioning;
-
-import org.junit.Assert;
-import org.junit.Test;
-import org.junit.runner.RunWith;
-import org.mockito.ArgumentCaptor;
-
-import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
-import org.springframework.boot.test.context.SpringBootTest;
-import org.springframework.cloud.stream.annotation.EnableBinding;
-import org.springframework.cloud.stream.binder.Binder;
-import org.springframework.cloud.stream.binder.BinderFactory;
-import org.springframework.cloud.stream.binder.ProducerProperties;
-import org.springframework.cloud.stream.messaging.Source;
-import org.springframework.cloud.stream.utils.MockBinderRegistryConfiguration;
-import org.springframework.context.annotation.Import;
-import org.springframework.context.annotation.PropertySource;
-import org.springframework.messaging.MessageChannel;
-import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
-
-import static org.hamcrest.Matchers.equalTo;
-import static org.mockito.Matchers.eq;
-import static org.mockito.Mockito.verify;
-import static org.mockito.Mockito.verifyNoMoreInteractions;
-
-/**
- * @author Marius Bogoevici
- * @author Ilayaperumal Gopinathan
- */
-@RunWith(SpringJUnit4ClassRunner.class)
-@SpringBootTest(classes = PartitionedProducerTest.TestSource.class)
-public class PartitionedProducerTest {
-
-	@Autowired
-	private BinderFactory binderFactory;
-
-	@Autowired
-	private Source testSource;
-
-	@Test
-	@SuppressWarnings("rawtypes")
-	public void testBindingPartitionedProducer() {
-		Binder binder = this.binderFactory.getBinder(null, MessageChannel.class);
-		ArgumentCaptor<ProducerProperties> argumentCaptor = ArgumentCaptor.forClass(ProducerProperties.class);
-		verify(binder).bindProducer(eq("partOut"), eq(this.testSource.output()), argumentCaptor.capture());
-		Assert.assertThat(argumentCaptor.getValue().getPartitionCount(), equalTo(3));
-		Assert.assertThat(argumentCaptor.getValue().getPartitionKeyExpression().getExpressionString(),
-				equalTo("payload"));
-		verifyNoMoreInteractions(binder);
-	}
-
-	@EnableBinding(Source.class)
-	@EnableAutoConfiguration
-	@Import(MockBinderRegistryConfiguration.class)
-	@PropertySource("classpath:/org/springframework/cloud/stream/binder/partitioned-producer-test.properties")
-	public static class TestSource {
-
-	}
-}

spring-cloud-stream/src/test/resources/org/springframework/cloud/stream/binder/custom-partitioned-producer-test.properties

@@ -1,4 +1,2 @@
 spring.cloud.stream.bindings.output.destination=partOut
 spring.cloud.stream.bindings.output.producer.partitionCount=3
-spring.cloud.stream.bindings.output.producer.partitionKeyExtractorClass=org.springframework.cloud.stream.partitioning.CustomPartitionKeyExtractorClass
-spring.cloud.stream.bindings.output.producer.partitionSelectorClass=org.springframework.cloud.stream.partitioning.CustomPartitionSelectorClass