Allow instanceIndex and instanceCount to be set per binding

Fixes #654

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

@@ -947,6 +947,16 @@ backOffMultiplier::
   The backoff multiplier.
 +
 Default: `2.0`.
+instanceIndex::
+  When set to a value greater than equal to zero, allows customizing the instance index of this consumer (if different from `spring.cloud.stream.instanceIndex`).
+When set to a negative value, it will default to `spring.cloud.stream.instanceIndex`.
++
+Default: `-1`.
+instanceCount::
+  When set to a value greater than equal to zero, allows customizing the instance count of this consumer (if different from `spring.cloud.stream.instanceCount`).
+When set to a negative value, it will default to `spring.cloud.stream.instanceCount`.
++
+Default: `-1`.
 
 ==== Producer Properties
 

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

@@ -33,9 +33,9 @@ public class ConsumerProperties {
 
 	private boolean partitioned;
 
-	private int instanceCount = 1;
+	private int instanceCount = -1;
 
-	private int instanceIndex;
+	private int instanceIndex = -1;
 
 	private int maxAttempts = 3;
 

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

@@ -189,8 +189,13 @@ public class ChannelBindingServiceProperties implements ApplicationContextAware,
 		if (consumerProperties == null) {
 			consumerProperties = new ConsumerProperties();
 		}
-		consumerProperties.setInstanceCount(this.instanceCount);
-		consumerProperties.setInstanceIndex(this.instanceIndex);
+		// propagate instance count and instance index if not already set
+		if (consumerProperties.getInstanceCount() < 0) {
+			consumerProperties.setInstanceCount(this.instanceCount);
+		}
+		if (consumerProperties.getInstanceIndex() < 0) {
+			consumerProperties.setInstanceIndex(this.instanceIndex);
+		}
 		return consumerProperties;
 	}