Support consumption from multiple non-native part.

Currently, non-native partitioning (e.g. RabbitMQ) require
an instance per partition.
Add support to allow one instance to consume from multiple
partitions.

Changed instanceIndex from single valued to List

Added a new property to support index list to spawn multiple consumers

added test for the index list usage

added author to the changed files. Fixes-gh-1232

Using object utils to check if a list is empty

Only look at indexes when partitioned

Add docs for `instanceIndexList`

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

@@ -1887,18 +1887,23 @@ defaultRetryable::
 Whether exceptions thrown by the listener that are not listed in the `retryableExceptions` are retryable.
 +
 Default: `true`.
-instanceIndex::
-When set to a value greater than equal to zero, it allows customizing the instance index of this consumer (if different from `spring.cloud.stream.instanceIndex`).
-When set to a negative value, it defaults to `spring.cloud.stream.instanceIndex`.
-See `<<spring-cloud-stream-overview-instance-index-instance-count>>` for more information.
-+
-Default: `-1`.
 instanceCount::
 When set to a value greater than equal to zero, it allows customizing the instance count of this consumer (if different from `spring.cloud.stream.instanceCount`).
 When set to a negative value, it defaults to `spring.cloud.stream.instanceCount`.
 See `<<spring-cloud-stream-overview-instance-index-instance-count>>` for more information.
 +
 Default: `-1`.
+instanceIndex::
+When set to a value greater than equal to zero, it allows customizing the instance index of this consumer (if different from `spring.cloud.stream.instanceIndex`).
+When set to a negative value, it defaults to `spring.cloud.stream.instanceIndex`.
+Ignored if `instanceIndexList` is provided.
+See `<<spring-cloud-stream-overview-instance-index-instance-count>>` for more information.
++
+Default: `-1`.
+instanceIndexList::
+Used with binders that do not support native partitioning (such as RabbitMQ); allows an application instance to consume from more than one partition.
++
+Default: `-1`.
 retryableExceptions::
 A map of Throwable class names in the key and a boolean in the value.
 Specify those exceptions (and subclasses) that will or won't be retried.

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

@@ -35,6 +35,7 @@ import org.springframework.messaging.Message;
  * @author Soby Chacko
  * @author Oleg Zhurakousky
  * @author Nicolas Homble
+ * @author Michael Michailidis
  */
 @JsonInclude(JsonInclude.Include.NON_DEFAULT)
 public class ConsumerProperties {
@@ -74,6 +75,13 @@ public class ConsumerProperties {
 	 */
 	private int instanceIndex = -1;
 
+	/**
+	 * When set it will allow the customization of the consumer to spawn a consumer for
+	 * each item in the list. All negative indexes will be discarded. Default: null
+	 * NOTE: This setting will disable the instance-index
+	 */
+	private List<Integer> instanceIndexList;
+
 	/**
 	 * The number of attempts to process the message (including the first) in the event of
 	 * processing failures. This is a RetryTemplate configuration which is provided by the
@@ -211,6 +219,14 @@ public class ConsumerProperties {
 		this.instanceIndex = instanceIndex;
 	}
 
+	public List<Integer> getInstanceIndexList() {
+		return this.instanceIndexList;
+	}
+
+	public void setInstanceIndexList(List<Integer> instanceIndexList) {
+		this.instanceIndexList = instanceIndexList;
+	}
+
 	@Min(value = 1, message = "Max attempts should be greater than zero.")
 	public int getMaxAttempts() {
 		return this.maxAttempts;

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

@@ -53,10 +53,11 @@ import org.springframework.validation.beanvalidation.CustomValidatorBean;
  * @author Mark Fisher
  * @author Dave Syer
  * @author Marius Bogoevici
- * @author Ilayaperumal Gopinathan
+ * @author Ilayaperumal Gopinathan`
  * @author Gary Russell
  * @author Janne Valkealahti
  * @author Soby Chacko
+ * @author Michael Michailidis
  */
 public class BindingService {
 
@@ -118,13 +119,35 @@ public class BindingService {
 			String[] bindingTargets = StringUtils
 					.commaDelimitedListToStringArray(bindingTarget);
 			for (String target : bindingTargets) {
-				Binding<T> binding = input instanceof PollableSource
+				if (!consumerProperties.isPartitioned() || consumerProperties.getInstanceIndexList().isEmpty()) {
+					Binding<T> binding = input instanceof PollableSource
 						? doBindPollableConsumer(input, inputName, binder,
-								consumerProperties, target)
+						consumerProperties, target)
 						: doBindConsumer(input, inputName, binder, consumerProperties,
-								target);
+						target);
 
-				bindings.add(binding);
+					bindings.add(binding);
+				}
+				else {
+					for (Integer index : consumerProperties.getInstanceIndexList()) {
+						if (index < 0) {
+							continue;
+						}
+
+						ConsumerProperties consumerPropertiesTemp = new ExtendedConsumerProperties<>("");
+						BeanUtils.copyProperties(consumerProperties, consumerPropertiesTemp);
+
+						consumerPropertiesTemp.setInstanceIndex(index);
+
+						Binding<T> binding = input instanceof PollableSource
+							? doBindPollableConsumer(input, inputName, binder,
+							consumerPropertiesTemp, target)
+							: doBindConsumer(input, inputName, binder, consumerPropertiesTemp,
+							target);
+
+						bindings.add(binding);
+					}
+				}
 			}
 		}
 		bindings = Collections.unmodifiableCollection(bindings);

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

@@ -16,7 +16,9 @@
 
 package org.springframework.cloud.stream.config;
 
+import java.util.ArrayList;
 import java.util.HashMap;
+import java.util.List;
 import java.util.Map;
 import java.util.TreeMap;
 
@@ -49,11 +51,12 @@ import org.springframework.util.Assert;
  * @author Gary Russell
  * @author Ilayaperumal Gopinathan
  * @author Oleg Zhurakousky
+ * @author Michael Michailidis
  */
 @ConfigurationProperties("spring.cloud.stream")
 @JsonInclude(Include.NON_DEFAULT)
 public class BindingServiceProperties
-		implements ApplicationContextAware, InitializingBean {
+	implements ApplicationContextAware, InitializingBean {
 
 	private static final int DEFAULT_BINDING_RETRY_INTERVAL = 30;
 
@@ -66,6 +69,15 @@ public class BindingServiceProperties
 	@Value("${INSTANCE_INDEX:${CF_INSTANCE_INDEX:0}}")
 	private int instanceIndex;
 
+	/**
+	 * A list of instance id's from 0 to instanceCount-1. Used for partitioning and with
+	 * Kafka. NOTE: Could also be managed per individual binding
+	 * "spring.cloud.stream.bindings.foo.consumer.instance-index-list" where 'foo' is
+	 * the name of the binding. This setting will override the one set in
+	 * 'spring.cloud.stream.instance-index'
+	 */
+	private List<Integer> instanceIndexList = new ArrayList<>();
+
 	/**
 	 * The number of deployed instances of an application. Default: 1. NOTE: Could also be
 	 * managed per individual binding
@@ -82,7 +94,7 @@ public class BindingServiceProperties
 	 * application: 'spring.cloud.stream.bindings.input.contentType=text/plain'
 	 */
 	private Map<String, BindingProperties> bindings = new TreeMap<>(
-			String.CASE_INSENSITIVE_ORDER);
+		String.CASE_INSENSITIVE_ORDER);
 
 	/**
 	 * Additional per-binder properties (see {@link BinderProperties}) if more then one
@@ -146,6 +158,14 @@ public class BindingServiceProperties
 		this.instanceIndex = instanceIndex;
 	}
 
+	public List<Integer> getInstanceIndexList() {
+		return this.instanceIndexList;
+	}
+
+	public void setInstanceIndexList(List<Integer> instanceIndexList) {
+		this.instanceIndexList = instanceIndexList;
+	}
+
 	public int getInstanceCount() {
 		return this.instanceCount;
 	}
@@ -226,6 +246,9 @@ public class BindingServiceProperties
 		if (consumerProperties.getInstanceIndex() < 0) {
 			consumerProperties.setInstanceIndex(this.instanceIndex);
 		}
+		if (consumerProperties.getInstanceIndexList() == null) {
+			consumerProperties.setInstanceIndexList(this.instanceIndexList);
+		}
 		return consumerProperties;
 	}
 

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

@@ -17,6 +17,7 @@
 package org.springframework.cloud.stream.binding;
 
 import java.lang.reflect.Field;
+import java.util.Arrays;
 import java.util.Collection;
 import java.util.Collections;
 import java.util.HashMap;
@@ -103,6 +104,7 @@ import static org.mockito.Mockito.when;
  * @author Ilayaperumal Gopinathan
  * @author Janne Valkealahti
  * @author Soby Chacko
+ * @author Michael Michailidis
  */
 public class BindingServiceTests {
 
@@ -184,6 +186,67 @@ public class BindingServiceTests {
 		binderFactory.destroy();
 	}
 
+	@SuppressWarnings({ "unchecked", "rawtypes" })
+	@Test
+	public void testMultipleConsumerBindingsFromIndexList() throws Exception {
+		BindingServiceProperties properties = new BindingServiceProperties();
+		Map<String, BindingProperties> bindingProperties = new HashMap<>();
+		BindingProperties props = new BindingProperties();
+		props.setDestination("foo");
+
+		ConsumerProperties consumer = properties.getConsumerProperties("input");
+		consumer.setInstanceIndexList(Arrays.asList(0, 1));
+		consumer.setInstanceCount(2);
+		consumer.setPartitioned(true);
+		props.setConsumer(consumer);
+
+		final String inputChannelName = "input";
+		bindingProperties.put(inputChannelName, props);
+
+		properties.setBindings(bindingProperties);
+
+		DefaultBinderFactory binderFactory = createMockBinderFactory();
+
+		Binder binder = binderFactory.getBinder("mock", MessageChannel.class);
+		BindingService service = new BindingService(properties, binderFactory);
+		MessageChannel inputChannel = new DirectChannel();
+
+		Binding<MessageChannel> mockBinding1 = Mockito.mock(Binding.class, "FirstBinding");
+		Binding<MessageChannel> mockBinding2 = Mockito.mock(Binding.class, "SecondBinding");
+
+		ArgumentCaptor<ConsumerProperties> captor = ArgumentCaptor.forClass(ConsumerProperties.class);
+
+		when(binder.bindConsumer(eq("foo"), isNull(), same(inputChannel),
+			any(ConsumerProperties.class))).thenReturn(mockBinding1).thenReturn(mockBinding2);
+
+		Collection<Binding<MessageChannel>> bindings = service.bindConsumer(inputChannel,
+			"input");
+		assertThat(bindings).hasSize(2);
+
+		Iterator<Binding<MessageChannel>> iterator = bindings.iterator();
+		Binding<MessageChannel> binding1 = iterator.next();
+		Binding<MessageChannel> binding2 = iterator.next();
+
+		assertThat(binding1).isSameAs(mockBinding1);
+		assertThat(binding2).isSameAs(mockBinding2);
+
+		service.unbindConsumers("input");
+
+		verify(binder, times(2)).bindConsumer(eq("foo"), isNull(), same(inputChannel),
+			captor.capture());
+		verify(binding1).unbind();
+		verify(binding2).unbind();
+
+		List<ConsumerProperties> allValues = captor.getAllValues();
+
+		assertThat(allValues.size()).isEqualTo(2);
+
+		assertThat(allValues.get(0).getInstanceIndex()).isEqualTo(0);
+		assertThat(allValues.get(1).getInstanceIndex()).isEqualTo(1);
+
+		binderFactory.destroy();
+	}
+
 	@SuppressWarnings({ "unchecked", "rawtypes" })
 	@Test
 	public void testConsumerBindingWhenMultiplexingIsEnabled() throws Exception {