Add per-binding configuration for partitioning

- consumer and producer partitioning properties are defined per-binding
- unit tests
- Kafka binding fixes for partitioning

spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java

@@ -391,6 +391,7 @@ public class KafkaMessageChannelBinder extends MessageChannelBinderSupport {
 			retryTemplate.setBackOffPolicy(backOffPolicy);
 			retryOperations = retryTemplate;
 		}
+		super.afterPropertiesSet();
 	}
 
 	/**

spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaTestBinder.java

@@ -56,10 +56,10 @@ public class KafkaTestBinder extends AbstractTestBinder<KafkaMessageChannelBinde
 			binder.setCodec(getCodec());
 			binder.setDefaultBatchingEnabled(false);
 			binder.setMode(mode);
-			binder.afterPropertiesSet();
 			GenericApplicationContext context = new GenericApplicationContext();
 			context.refresh();
 			binder.setApplicationContext(context);
+			binder.afterPropertiesSet();
 			this.setBinder(binder);
 		}
 		catch (Exception e) {

spring-cloud-stream-binders/spring-cloud-stream-binder-spi/src/main/java/org/springframework/cloud/stream/binder/MessageChannelBinderSupport.java

@@ -146,6 +146,7 @@ public abstract class MessageChannelBinderSupport
 					BinderProperties.PARTITION_KEY_EXTRACTOR_CLASS,
 					BinderProperties.PARTITION_SELECTOR_CLASS,
 					BinderProperties.PARTITION_SELECTOR_EXPRESSION,
+					BinderProperties.MIN_PARTITION_COUNT
 			}));
 
 	protected static final Set<Object> PRODUCER_BATCHING_BASIC_PROPERTIES = new HashSet<Object>(

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

@@ -23,7 +23,8 @@ import org.springframework.util.Assert;
 import org.springframework.util.StringUtils;
 
 /**
- * Handles the binding of input/output channels by delegating to an underlying {@link Binder}.
+ * Handles the binding of input/output channels by delegating to an underlying
+ * {@link Binder}.
  *
  * @author Mark Fisher
  * @author Dave Syer
@@ -35,59 +36,50 @@ public class ChannelBindingService {
 
 	private ChannelBindingProperties channelBindingProperties;
 
-	public ChannelBindingService(ChannelBindingProperties channelBindingProperties,
-															 Binder<MessageChannel> binder) {
+	public ChannelBindingService(ChannelBindingProperties channelBindingProperties, Binder<MessageChannel> binder) {
 		this.channelBindingProperties = channelBindingProperties;
 		this.binder = binder;
 	}
 
-
 	public void bindConsumer(MessageChannel inputChannel, String inputChannelName) {
-		String channelBindingTarget = this.channelBindingProperties
-				.getBindingPath(inputChannelName);
+		String channelBindingTarget = this.channelBindingProperties.getBindingDestination(inputChannelName);
 		if (isChannelPubSub(channelBindingTarget)) {
-			this.binder.bindPubSubConsumer(removePrefix(channelBindingTarget), inputChannel,
-					this.channelBindingProperties.getConsumerProperties());
+			this.binder.bindPubSubConsumer(removePrefix(channelBindingTarget),
+					inputChannel, this.channelBindingProperties.getConsumerProperties(inputChannelName));
 		}
 		else {
 			this.binder.bindConsumer(channelBindingTarget, inputChannel,
-					this.channelBindingProperties.getConsumerProperties());
+					this.channelBindingProperties.getConsumerProperties(inputChannelName));
 		}
 	}
 
 	public void bindProducer(MessageChannel outputChannel, String outputChannelName) {
-		String channelBindingTarget = this.channelBindingProperties
-				.getBindingPath(outputChannelName);
+		String channelBindingTarget = this.channelBindingProperties.getBindingDestination(outputChannelName);
 		if (isChannelPubSub(channelBindingTarget)) {
-			this.binder.bindPubSubProducer(removePrefix(channelBindingTarget), outputChannel,
-					this.channelBindingProperties.getProducerProperties());
+			this.binder.bindPubSubProducer(removePrefix(channelBindingTarget),
+					outputChannel, this.channelBindingProperties.getProducerProperties(outputChannelName));
 		}
 		else {
 			this.binder.bindProducer(channelBindingTarget, outputChannel,
-					this.channelBindingProperties.getProducerProperties());
+					this.channelBindingProperties.getProducerProperties(outputChannelName));
 		}
 	}
 
 	private boolean isChannelPubSub(String bindingTarget) {
-		Assert.isTrue(StringUtils.hasText(bindingTarget),
-				"Binding target should not be empty/null.");
+		Assert.isTrue(StringUtils.hasText(bindingTarget), "Binding target should not be empty/null.");
 		return bindingTarget.startsWith("topic:");
 	}
 
 	private String removePrefix(String bindingTarget) {
-		Assert.isTrue(StringUtils.hasText(bindingTarget),
-				"Binding target should not be empty/null.");
+		Assert.isTrue(StringUtils.hasText(bindingTarget), "Binding target should not be empty/null.");
 		return bindingTarget.substring(bindingTarget.indexOf(":") + 1);
 	}
 
 	public void unbindConsumers(String inputChannelName) {
-		this.binder.unbindConsumers
-				(inputChannelName);
+		this.binder.unbindConsumers(inputChannelName);
 	}
 
 	public void unbindProducers(String outputChannelName) {
 		this.binder.unbindProducers(outputChannelName);
 	}
-
-
 }

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

@@ -20,7 +20,9 @@ import java.util.Map;
 import java.util.Properties;
 import java.util.TreeMap;
 
+import org.springframework.beans.factory.annotation.Value;
 import org.springframework.boot.context.properties.ConfigurationProperties;
+import org.springframework.cloud.stream.binder.BinderProperties;
 
 import com.fasterxml.jackson.annotation.JsonInclude;
 import com.fasterxml.jackson.annotation.JsonInclude.Include;
@@ -33,7 +35,16 @@ import com.fasterxml.jackson.annotation.JsonInclude.Include;
 @JsonInclude(Include.NON_DEFAULT)
 public class ChannelBindingProperties {
 
-	public static final String PATH = "path";
+	public static final String DESTINATION = "destination";
+
+	public static final String PARTITION_COUNT = "partitionCount";
+
+	public static final String PARTITIONED = "partitioned";
+
+	@Value("${INSTANCE_INDEX:${CF_INSTANCE_INDEX:0}}")
+	private int instanceIndex;
+
+	private int instanceCount = 1;
 
 	private Properties consumerProperties = new Properties();
 
@@ -41,7 +52,7 @@ public class ChannelBindingProperties {
 
 	private Map<String,Object> bindings = new TreeMap<>(String.CASE_INSENSITIVE_ORDER);
 
-	public Properties getConsumerProperties() {
+	private Properties getConsumerProperties() {
 		return this.consumerProperties;
 	}
 
@@ -65,7 +76,24 @@ public class ChannelBindingProperties {
 		this.bindings = bindings;
 	}
 
-	public String getBindingPath(String channelName) {
+	public int getInstanceIndex() {
+		return instanceIndex;
+	}
+
+	public void setInstanceIndex(int instanceIndex) {
+		this.instanceIndex = instanceIndex;
+	}
+
+	public int getInstanceCount() {
+		return instanceCount;
+	}
+
+	public void setInstanceCount(int instanceCount) {
+		this.instanceCount = instanceCount;
+	}
+
+
+	public String getBindingDestination(String channelName) {
 		Object binding = bindings.get(channelName);
 		// we may shortcut directly to the path
 		if (binding != null) {
@@ -74,7 +102,7 @@ public class ChannelBindingProperties {
 			}
 			else if (binding instanceof Map) {
 				Map<?, ?> bindingProperties = (Map<?, ?>) binding;
-				Object bindingPath = bindingProperties.get(PATH);
+				Object bindingPath = bindingProperties.get(DESTINATION);
 				if (bindingPath != null) {
 					return bindingPath.toString();
 				}
@@ -84,8 +112,114 @@ public class ChannelBindingProperties {
 		return channelName;
 	}
 
-	public String getTapChannelName(String channelName) {
-		return "tap:" + getBindingPath(channelName);
+	public int getPartitionCount(String channelName) {
+		Object binding = bindings.get(channelName);
+		// we may shortcut directly to the path
+		if (binding instanceof Map) {
+			try {
+				Map<?, ?> bindingProperties = (Map<?, ?>) binding;
+				Object bindingPath = bindingProperties.get(PARTITION_COUNT);
+				if (bindingPath != null) {
+					return Integer.parseInt(bindingPath.toString());
+				}
+			} catch (NumberFormatException e) {
+				// ignore and just return 1
+			}
+		}
+		return 1;
 	}
-	
+
+	public boolean isPartitionedConsumer(String channelName) {
+		Object binding = bindings.get(channelName);
+		// if the setting is just a target shortcut
+		if (binding == null || binding instanceof String) {
+			return false;
+		}
+		else if (binding instanceof Map) {
+			Map<?, ?> bindingProperties = (Map<?, ?>) binding;
+			Object bindingPath = bindingProperties.get(PARTITIONED);
+			if (bindingPath != null) {
+				return Boolean.valueOf(bindingPath.toString());
+			}
+		}
+		// just return the channel name if not found
+		return false;
+	}
+
+	public boolean isPartitionedProducer(String channelName) {
+		Object binding = bindings.get(channelName);
+		// if the setting is just a target shortcut
+		if (binding == null || binding instanceof String) {
+			return false;
+		}
+		else if (binding instanceof Map) {
+			Map<?, ?> bindingProperties = (Map<?, ?>) binding;
+			return bindingProperties.get(BinderProperties.PARTITION_KEY_EXPRESSION) != null
+					|| bindingProperties.get(BinderProperties.PARTITION_KEY_EXTRACTOR_CLASS) != null;
+		}
+		return false;
+	}
+
+	/**
+	 * Merge general properties provided by 'spring.cloud.stream.consumerProperties.*' with individual binding
+	 * properties supplied via binders.
+	 *
+	 * @param inputChannelName the input channel name
+	 * @return merged consumer properties
+	 */
+	public Properties getConsumerProperties(String inputChannelName) {
+		if (isPartitionedConsumer(inputChannelName)) {
+			Properties channelConsumerProperties = new Properties();
+			channelConsumerProperties.putAll(consumerProperties);
+			channelConsumerProperties.setProperty(BinderProperties.COUNT,
+					Integer.toString(getInstanceCount()));
+			channelConsumerProperties.setProperty(BinderProperties.PARTITION_INDEX,
+					Integer.toString(getInstanceIndex()));
+			return channelConsumerProperties;
+		}
+		else {
+			return getConsumerProperties();
+		}
+	}
+
+	/**
+	 * Merge general properties provided by 'spring.cloud.stream.producerProperties.*' with individual binding
+	 * properties supplied via binders.
+	 *
+	 * @param outputChannelName the output channel name
+	 * @return merged producer properties
+	 */
+	public Properties getProducerProperties(String outputChannelName) {
+		if (isPartitionedProducer(outputChannelName)) {
+			Properties channelProducerProperties = new Properties();
+			channelProducerProperties.putAll(this.producerProperties);
+			channelProducerProperties.setProperty(BinderProperties.NEXT_MODULE_COUNT,
+					Integer.toString(getPartitionCount(outputChannelName)));
+			copyChannelBindingProperty(outputChannelName, channelProducerProperties,
+					BinderProperties.PARTITION_KEY_EXPRESSION);
+			copyChannelBindingProperty(outputChannelName, channelProducerProperties,
+					BinderProperties.PARTITION_KEY_EXTRACTOR_CLASS);
+			copyChannelBindingProperty(outputChannelName, channelProducerProperties,
+					BinderProperties.PARTITION_SELECTOR_CLASS);
+			copyChannelBindingProperty(outputChannelName, channelProducerProperties,
+					BinderProperties.PARTITION_SELECTOR_EXPRESSION);
+			return channelProducerProperties;
+		}
+		else {
+			return this.producerProperties;
+		}
+	}
+
+	@SuppressWarnings("unchecked")
+	private void copyChannelBindingProperty(String outputChannelName, Properties targetProperties, String propertyName) {
+		Map<String, Object> channelBindingProperties = (Map<String, Object>) bindings.get(outputChannelName);
+		if (null != channelBindingProperties && channelBindingProperties.containsKey(propertyName)) {
+			targetProperties.setProperty(propertyName, (String) channelBindingProperties.get(propertyName));
+		}
+	}
+
+	public String getTapChannelName(String channelName) {
+		return "tap:" + getBindingDestination(channelName);
+	}
+
 }

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

@@ -0,0 +1,92 @@
+/*
+ * Copyright 2015 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 static org.hamcrest.core.IsEqual.equalTo;
+import static org.hamcrest.text.IsEqualIgnoringCase.equalToIgnoringCase;
+import static org.mockito.Matchers.eq;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.verifyNoMoreInteractions;
+
+import java.util.Properties;
+
+import org.junit.Assert;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.mockito.ArgumentCaptor;
+import org.mockito.ArgumentMatcher;
+
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
+import org.springframework.boot.test.SpringApplicationConfiguration;
+import org.springframework.cloud.stream.annotation.Bindings;
+import org.springframework.cloud.stream.annotation.EnableBinding;
+import org.springframework.cloud.stream.binder.Binder;
+import org.springframework.cloud.stream.binder.BinderProperties;
+import org.springframework.cloud.stream.messaging.Sink;
+import org.springframework.cloud.stream.messaging.Source;
+import org.springframework.cloud.stream.utils.MockBinderConfiguration;
+import org.springframework.context.annotation.Import;
+import org.springframework.context.annotation.PropertySource;
+import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
+
+/**
+ * @author Marius Bogoevici
+ */
+@RunWith(SpringJUnit4ClassRunner.class)
+@SpringApplicationConfiguration(PartitionedConsumerTest.TestSink.class)
+
+public class PartitionedConsumerTest {
+
+	@SuppressWarnings("rawtypes")
+	@Autowired
+	private Binder binder;
+
+	@Autowired @Bindings(TestSink.class)
+	private Sink testSource;
+
+	@Test
+	@SuppressWarnings("unchecked")
+	public void testBindingPartitionedConsumer() {
+		ArgumentCaptor<Properties> argumentCaptor = ArgumentCaptor.forClass(Properties.class);
+		verify(binder).bindConsumer(eq("partIn"), eq(testSource.input()), argumentCaptor.capture());
+		Assert.assertThat(argumentCaptor.getValue().getProperty(BinderProperties.PARTITION_INDEX), equalTo("0"));
+		Assert.assertThat(argumentCaptor.getValue().getProperty(BinderProperties.COUNT),
+				equalTo("2"));
+		verifyNoMoreInteractions(binder);
+	}
+
+
+	@EnableBinding(Sink.class)
+	@EnableAutoConfiguration
+	@Import(MockBinderConfiguration.class)
+	@PropertySource("classpath:/org/springframework/cloud/stream/binder/partitioned-consumer-test.properties")
+	public static class TestSink {
+
+	}
+
+	class PropertiesArgumentMatcher extends ArgumentMatcher<Properties> {
+		@Override
+		public boolean matches(Object argument) {
+			if (!(argument instanceof Properties)) {
+				return false;
+			}
+			return true;
+		}
+	}
+
+}

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

@@ -0,0 +1,90 @@
+/*
+ * Copyright 2015 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 static org.hamcrest.core.IsEqual.equalTo;
+import static org.mockito.Matchers.eq;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.verifyNoMoreInteractions;
+
+import java.util.Properties;
+
+import org.junit.Assert;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.mockito.ArgumentCaptor;
+import org.mockito.ArgumentMatcher;
+
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
+import org.springframework.boot.test.SpringApplicationConfiguration;
+import org.springframework.cloud.stream.annotation.Bindings;
+import org.springframework.cloud.stream.annotation.EnableBinding;
+import org.springframework.cloud.stream.binder.Binder;
+import org.springframework.cloud.stream.binder.BinderProperties;
+import org.springframework.cloud.stream.messaging.Source;
+import org.springframework.cloud.stream.utils.MockBinderConfiguration;
+import org.springframework.context.annotation.Import;
+import org.springframework.context.annotation.PropertySource;
+import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
+
+/**
+ * @author Marius Bogoevici
+ */
+@RunWith(SpringJUnit4ClassRunner.class)
+@SpringApplicationConfiguration(PartitionedProducerTest.TestSource.class)
+
+public class PartitionedProducerTest {
+
+	@SuppressWarnings("rawtypes")
+	@Autowired
+	private Binder binder;
+
+	@Autowired @Bindings(TestSource.class)
+	private Source testSource;
+
+	@Test
+	@SuppressWarnings("unchecked")
+	public void testBindingPartitionedProducer() {
+		ArgumentCaptor<Properties> argumentCaptor = ArgumentCaptor.forClass(Properties.class);
+		verify(binder).bindProducer(eq("partOut"), eq(testSource.output()), argumentCaptor.capture());
+		Assert.assertThat(argumentCaptor.getValue().getProperty(BinderProperties.NEXT_MODULE_COUNT), equalTo("3"));
+		Assert.assertThat(argumentCaptor.getValue().getProperty(BinderProperties.PARTITION_KEY_EXPRESSION),
+				equalTo("payload"));
+		verifyNoMoreInteractions(binder);
+	}
+
+
+	@EnableBinding(Source.class)
+	@EnableAutoConfiguration
+	@Import(MockBinderConfiguration.class)
+	@PropertySource("classpath:/org/springframework/cloud/stream/binder/partitioned-producer-test.properties")
+	public static class TestSource {
+
+	}
+
+	class PropertiesArgumentMatcher extends ArgumentMatcher<Properties> {
+		@Override
+		public boolean matches(Object argument) {
+			if (!(argument instanceof Properties)) {
+				return false;
+			}
+			return true;
+		}
+	}
+
+}

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

@@ -0,0 +1,5 @@
+spring.cloud.stream.bindings.input.destination=partIn
+spring.cloud.stream.bindings.input.partitioned=true
+spring.cloud.stream.instanceCount=2
+spring.cloud.stream.instanceIndex=0
+

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

@@ -0,0 +1,4 @@
+spring.cloud.stream.bindings.output.destination=partOut
+spring.cloud.stream.bindings.output.partitionKeyExpression=payload
+spring.cloud.stream.bindings.output.partitionCount=3
+