Adds support for extended binding properties

* Add parameterized ExtendedConsumerProperties and ExtendedProducerProperties;
* Added ExtendedPropertiesBinder with the ability of managing per-binding property extensions and interacting the core service;
* Add extensions for Kafka and Rabbit

Addressing PR comments

Changed binder-specific binding prefix to `spring.cloud.stream.binderType.bindings`

Removing kafka-binder.properties

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

@@ -0,0 +1,43 @@
+/*
+ * Copyright 2016 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.binder.kafka;
+
+/**
+ * @author Marius Bogoevici
+ */
+public class KafkaBindingProperties {
+
+	private KafkaConsumerProperties consumer = new KafkaConsumerProperties();
+
+	private KafkaProducerProperties producer = new KafkaProducerProperties();
+
+	public KafkaConsumerProperties getConsumer() {
+		return consumer;
+	}
+
+	public void setConsumer(KafkaConsumerProperties consumer) {
+		this.consumer = consumer;
+	}
+
+	public KafkaProducerProperties getProducer() {
+		return producer;
+	}
+
+	public void setProducer(KafkaProducerProperties producer) {
+		this.producer = producer;
+	}
+}

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

@@ -16,12 +16,10 @@
 
 package org.springframework.cloud.stream.binder.kafka;
 
-import org.springframework.cloud.stream.binder.ConsumerProperties;
-
 /**
  * @author Marius Bogoevici
  */
-public class KafkaConsumerProperties extends ConsumerProperties {
+public class KafkaConsumerProperties {
 
 	private int minPartitionCount = 1;
 

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

@@ -0,0 +1,60 @@
+/*
+ * Copyright 2016 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.binder.kafka;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import org.springframework.boot.context.properties.ConfigurationProperties;
+import org.springframework.cloud.stream.binder.ExtendedBindingProperties;
+
+/**
+ * @author Marius Bogoevici
+ */
+@ConfigurationProperties("spring.cloud.stream.kafka")
+public class KafkaExtendedBindingProperties implements ExtendedBindingProperties<KafkaConsumerProperties, KafkaProducerProperties> {
+
+	private Map<String, KafkaBindingProperties> bindings = new HashMap<>();
+
+	public Map<String, KafkaBindingProperties> getBindings() {
+		return bindings;
+	}
+
+	public void setBindings(Map<String, KafkaBindingProperties> bindings) {
+		this.bindings = bindings;
+	}
+
+	@Override
+	public KafkaConsumerProperties getExtendedConsumerProperties(String channelName) {
+		if (bindings.containsKey(channelName) && bindings.get(channelName).getConsumer() != null) {
+			return bindings.get(channelName).getConsumer();
+		}
+		else {
+			return new KafkaConsumerProperties();
+		}
+	}
+
+	@Override
+	public KafkaProducerProperties getExtendedProducerProperties(String channelName) {
+		if (bindings.containsKey(channelName) && bindings.get(channelName).getProducer() != null) {
+			return bindings.get(channelName).getProducer();
+		}
+		else {
+			return new KafkaProducerProperties();
+		}
+	}
+}

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

@@ -49,6 +49,9 @@ import org.springframework.cloud.stream.binder.BinderHeaders;
 import org.springframework.cloud.stream.binder.Binding;
 import org.springframework.cloud.stream.binder.DefaultBinding;
 import org.springframework.cloud.stream.binder.EmbeddedHeadersMessageConverter;
+import org.springframework.cloud.stream.binder.ExtendedConsumerProperties;
+import org.springframework.cloud.stream.binder.ExtendedProducerProperties;
+import org.springframework.cloud.stream.binder.ExtendedPropertiesBinder;
 import org.springframework.cloud.stream.binder.MessageValues;
 import org.springframework.cloud.stream.binder.PartitionHandler;
 import org.springframework.http.MediaType;
@@ -98,7 +101,7 @@ import org.springframework.util.StringUtils;
  * @author Mark Fisher
  * @author Soby Chacko
  */
-public class KafkaMessageChannelBinder extends AbstractBinder<MessageChannel, KafkaConsumerProperties, KafkaProducerProperties> {
+public class KafkaMessageChannelBinder extends AbstractBinder<MessageChannel, ExtendedConsumerProperties<KafkaConsumerProperties>, ExtendedProducerProperties<KafkaProducerProperties>> implements ExtendedPropertiesBinder<MessageChannel, KafkaConsumerProperties, KafkaProducerProperties> {
 
 	public static final ByteArraySerializer BYTE_ARRAY_SERIALIZER = new ByteArraySerializer();
 
@@ -147,6 +150,8 @@ public class KafkaMessageChannelBinder extends AbstractBinder<MessageChannel, Ka
 
 	private ProducerListener producerListener;
 
+	private KafkaExtendedBindingProperties extendedBindingProperties = new KafkaExtendedBindingProperties();
+
 	public KafkaMessageChannelBinder(ZookeeperConnect zookeeperConnect, String brokers, String zkAddress,
 									 String... headersToMap) {
 		this.zookeeperConnect = zookeeperConnect;
@@ -201,6 +206,10 @@ public class KafkaMessageChannelBinder extends AbstractBinder<MessageChannel, Ka
 		this.retryOperations = retryOperations;
 	}
 
+	public void setExtendedBindingProperties(KafkaExtendedBindingProperties extendedBindingProperties) {
+		this.extendedBindingProperties = extendedBindingProperties;
+	}
+
 	@Override
 	public void onInit() throws Exception {
 		ZookeeperConfiguration configuration = new ZookeeperConfiguration(this.zookeeperConnect);
@@ -284,12 +293,23 @@ public class KafkaMessageChannelBinder extends AbstractBinder<MessageChannel, Ka
 		this.zkConnectionTimeout = zkConnectionTimeout;
 	}
 
+	@Override
+	public KafkaConsumerProperties getExtendedConsumerProperties(String channelName) {
+		return extendedBindingProperties.getExtendedConsumerProperties(channelName);
+	}
+
+	@Override
+	public KafkaProducerProperties getExtendedProducerProperties(String channelName) {
+		return extendedBindingProperties.getExtendedProducerProperties(channelName);
+	}
+
 	Map<String, Collection<Partition>> getTopicsInUse() {
 		return this.topicsInUse;
 	}
 
 	@Override
-	protected Binding<MessageChannel> doBindConsumer(String name, String group, MessageChannel inputChannel, KafkaConsumerProperties properties) {
+	protected Binding<MessageChannel> doBindConsumer(String name, String group, MessageChannel inputChannel,
+			ExtendedConsumerProperties<KafkaConsumerProperties> properties) {
 		// If the caller provides a consumer group, use it; otherwise an anonymous consumer group
 		// is generated each time, such that each anonymous binding will receive all messages.
 		// Consumers reset offsets at the latest time by default, which allows them to receive only
@@ -300,14 +320,15 @@ public class KafkaMessageChannelBinder extends AbstractBinder<MessageChannel, Ka
 		// The reference point, if not set explicitly is the latest time for anonymous subscriptions and the
 		// earliest time for group subscriptions. This allows the latter to receive messages published before the group
 		// has been created.
-		long referencePoint = properties.getStartOffset() != null ?
-				properties.getStartOffset().getReferencePoint() : (anonymous ? OffsetRequest.LatestTime() : OffsetRequest.EarliestTime());
+		long referencePoint = properties.getExtension().getStartOffset() != null ?
+				properties.getExtension().getStartOffset().getReferencePoint() : (anonymous ? OffsetRequest.LatestTime() : OffsetRequest.EarliestTime());
 		return createKafkaConsumer(name, inputChannel, properties, consumerGroup, referencePoint);
 	}
 
 
 	@Override
-	public Binding<MessageChannel> doBindProducer(String name, MessageChannel moduleOutputChannel, KafkaProducerProperties properties) {
+	public Binding<MessageChannel> doBindProducer(String name, MessageChannel moduleOutputChannel,
+			ExtendedProducerProperties<KafkaProducerProperties> properties) {
 
 		Assert.isInstanceOf(SubscribableChannel.class, moduleOutputChannel);
 		if (logger.isInfoEnabled()) {
@@ -323,12 +344,12 @@ public class KafkaMessageChannelBinder extends AbstractBinder<MessageChannel, Ka
 
 		ProducerMetadata<byte[], byte[]> producerMetadata = new ProducerMetadata<>(name, byte[].class, byte[].class,
 				BYTE_ARRAY_SERIALIZER, BYTE_ARRAY_SERIALIZER);
-		producerMetadata.setSync(properties.isSync());
-		producerMetadata.setCompressionType(properties.getCompressionType());
-		producerMetadata.setBatchBytes(properties.getBufferSize());
+		producerMetadata.setSync(properties.getExtension().isSync());
+		producerMetadata.setCompressionType(properties.getExtension().getCompressionType());
+		producerMetadata.setBatchBytes(properties.getExtension().getBufferSize());
 		Properties additionalProps = new Properties();
 		additionalProps.put(ProducerConfig.ACKS_CONFIG, String.valueOf(requiredAcks));
-		additionalProps.put(ProducerConfig.LINGER_MS_CONFIG, String.valueOf(properties.getBatchTimeout()));
+		additionalProps.put(ProducerConfig.LINGER_MS_CONFIG, String.valueOf(properties.getExtension().getBatchTimeout()));
 		ProducerFactoryBean<byte[], byte[]> producerFB = new ProducerFactoryBean<>(producerMetadata, brokers, additionalProps);
 
 		try {
@@ -406,10 +427,10 @@ public class KafkaMessageChannelBinder extends AbstractBinder<MessageChannel, Ka
 	}
 
 	private Binding<MessageChannel> createKafkaConsumer(String name, final MessageChannel moduleInputChannel,
-														KafkaConsumerProperties properties, String group, long referencePoint) {
+			ExtendedConsumerProperties<KafkaConsumerProperties> properties, String group, long referencePoint) {
 
 		validateTopicName(name);
-		int minKafkaPartitions = properties.getMinPartitionCount();
+		int minKafkaPartitions = properties.getExtension().getMinPartitionCount();
 		int instance = properties.getInstanceCount();
 		if (instance == 0) {
 			throw new IllegalArgumentException("Instance count cannot be zero");
@@ -450,7 +471,7 @@ public class KafkaMessageChannelBinder extends AbstractBinder<MessageChannel, Ka
 		kafkaMessageDrivenChannelAdapter.setKeyDecoder(keyDecoder);
 		kafkaMessageDrivenChannelAdapter.setPayloadDecoder(valueDecoder);
 		kafkaMessageDrivenChannelAdapter.setOutputChannel(bridge);
-		kafkaMessageDrivenChannelAdapter.setAutoCommitOffset(properties.isAutoCommitOffset());
+		kafkaMessageDrivenChannelAdapter.setAutoCommitOffset(properties.getExtension().isAutoCommitOffset());
 		kafkaMessageDrivenChannelAdapter.afterPropertiesSet();
 		kafkaMessageDrivenChannelAdapter.start();
 
@@ -481,7 +502,8 @@ public class KafkaMessageChannelBinder extends AbstractBinder<MessageChannel, Ka
 		return consumerBinding;
 	}
 
-	KafkaMessageListenerContainer createMessageListenerContainer(KafkaConsumerProperties consumerProperties,
+	KafkaMessageListenerContainer createMessageListenerContainer(
+			ExtendedConsumerProperties<KafkaConsumerProperties> consumerProperties,
 																 String group, String topic, Collection<Partition> listenedPartitions,
 																 long referencePoint) {
 		Assert.isTrue(StringUtils.hasText(topic) ^ !CollectionUtils.isEmpty(listenedPartitions),
@@ -500,7 +522,7 @@ public class KafkaMessageChannelBinder extends AbstractBinder<MessageChannel, Ka
 		// if we have fewer target partitions than target concurrency, adjust accordingly
 		messageListenerContainer.setConcurrency(Math.min(consumerProperties.getConcurrency(), listenedPartitions.size()));
 		OffsetManager offsetManager = createOffsetManager(group, referencePoint);
-		if (consumerProperties.isResetOffsets()) {
+		if (consumerProperties.getExtension().isResetOffsets()) {
 			offsetManager.resetOffsets(listenedPartitions);
 		}
 		messageListenerContainer.setOffsetManager(offsetManager);
@@ -546,16 +568,16 @@ public class KafkaMessageChannelBinder extends AbstractBinder<MessageChannel, Ka
 
 	private class ReceivingHandler extends AbstractReplyProducingMessageHandler {
 
-		private KafkaConsumerProperties consumerProperties;
+		private ExtendedConsumerProperties<KafkaConsumerProperties> consumerProperties;
 
-		public ReceivingHandler(KafkaConsumerProperties consumerProperties) {
+		public ReceivingHandler(ExtendedConsumerProperties<KafkaConsumerProperties> consumerProperties) {
 			this.consumerProperties = consumerProperties;
 		}
 
 		@Override
 		@SuppressWarnings("unchecked")
 		protected Object handleRequestMessage(Message<?> requestMessage) {
-			if (Mode.embeddedHeaders.equals(consumerProperties.getMode())) {
+			if (Mode.embeddedHeaders.equals(consumerProperties.getExtension().getMode())) {
 				MessageValues messageValues;
 				try {
 					messageValues = embeddedHeadersMessageConverter.extractHeaders((Message<byte[]>) requestMessage,
@@ -596,7 +618,7 @@ public class KafkaMessageChannelBinder extends AbstractBinder<MessageChannel, Ka
 
 		private final String topicName;
 
-		private final KafkaProducerProperties producerProperties;
+		private final ExtendedProducerProperties<KafkaProducerProperties> producerProperties;
 
 		private final int numberOfKafkaPartitions;
 
@@ -604,7 +626,8 @@ public class KafkaMessageChannelBinder extends AbstractBinder<MessageChannel, Ka
 
 		private final PartitionHandler partitionHandler;
 
-		private SendingHandler(String topicName, KafkaProducerProperties properties, int numberOfPartitions,
+		private SendingHandler(String topicName, ExtendedProducerProperties<KafkaProducerProperties> properties,
+				int numberOfPartitions,
 							   ProducerConfiguration<byte[], byte[]> producerConfiguration) {
 			this.topicName = topicName;
 			producerProperties = properties;
@@ -626,13 +649,13 @@ public class KafkaMessageChannelBinder extends AbstractBinder<MessageChannel, Ka
 				targetPartition = roundRobin() % numberOfKafkaPartitions;
 			}
 
-			if (Mode.embeddedHeaders.equals(producerProperties.getMode())) {
+			if (Mode.embeddedHeaders.equals(producerProperties.getExtension().getMode())) {
 				MessageValues transformed = serializePayloadIfNecessary(message);
 				byte[] messageToSend = embeddedHeadersMessageConverter.embedHeaders(transformed,
 						KafkaMessageChannelBinder.this.headersToMap);
 				producerConfiguration.send(topicName, targetPartition, null, messageToSend);
 			}
-			else if (Mode.raw.equals(producerProperties.getMode())) {
+			else if (Mode.raw.equals(producerProperties.getExtension().getMode())) {
 				Object contentType = message.getHeaders().get(MessageHeaders.CONTENT_TYPE);
 				if (contentType != null
 						&& !contentType.equals(MediaType.APPLICATION_OCTET_STREAM_VALUE)) {

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

@@ -16,13 +16,12 @@
 
 package org.springframework.cloud.stream.binder.kafka;
 
-import org.springframework.cloud.stream.binder.ProducerProperties;
 import org.springframework.integration.kafka.support.ProducerMetadata;
 
 /**
  * @author Marius Bogoevici
  */
-public class KafkaProducerProperties extends ProducerProperties {
+public class KafkaProducerProperties {
 
 	private int bufferSize = 16384;
 

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

@@ -22,12 +22,12 @@ import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean
 import org.springframework.boot.context.properties.EnableConfigurationProperties;
 import org.springframework.cloud.stream.binder.Binder;
 import org.springframework.cloud.stream.binder.kafka.KafkaBinderHealthIndicator;
+import org.springframework.cloud.stream.binder.kafka.KafkaExtendedBindingProperties;
 import org.springframework.cloud.stream.binder.kafka.KafkaMessageChannelBinder;
 import org.springframework.cloud.stream.config.codec.kryo.KryoCodecAutoConfiguration;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 import org.springframework.context.annotation.Import;
-import org.springframework.context.annotation.PropertySource;
 import org.springframework.integration.codec.Codec;
 import org.springframework.integration.kafka.support.LoggingProducerListener;
 import org.springframework.integration.kafka.support.ProducerListener;
@@ -44,8 +44,7 @@ import org.springframework.util.ObjectUtils;
 @Configuration
 @ConditionalOnMissingBean(Binder.class)
 @Import({KryoCodecAutoConfiguration.class, PropertyPlaceholderAutoConfiguration.class})
-@EnableConfigurationProperties({KafkaBinderConfigurationProperties.class})
-@PropertySource("classpath:/META-INF/spring-cloud-stream/kafka-binder.properties")
+@EnableConfigurationProperties({KafkaBinderConfigurationProperties.class, KafkaExtendedBindingProperties.class})
 public class KafkaBinderConfiguration {
 
 	@Autowired
@@ -54,6 +53,9 @@ public class KafkaBinderConfiguration {
 	@Autowired
 	private KafkaBinderConfigurationProperties kafkaBinderConfigurationProperties;
 
+	@Autowired
+	private KafkaExtendedBindingProperties kafkaExtendedBindingProperties;
+
 	@Autowired
 	private ProducerListener producerListener;
 
@@ -89,6 +91,7 @@ public class KafkaBinderConfiguration {
 		kafkaMessageChannelBinder.setMaxWait(kafkaBinderConfigurationProperties.getMaxWait());
 
 		kafkaMessageChannelBinder.setProducerListener(producerListener);
+		kafkaMessageChannelBinder.setExtendedBindingProperties(kafkaExtendedBindingProperties);
 		return kafkaMessageChannelBinder;
 	}
 

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

@@ -26,37 +26,38 @@ import org.springframework.util.StringUtils;
  * @author Ilayaperumal Gopinathan
  * @author Marius Bogoevici
  */
-@ConfigurationProperties(prefix = "spring.cloud.stream.binder.kafka")
+@ConfigurationProperties(prefix = "spring.cloud.stream.kafka.binder")
 class KafkaBinderConfigurationProperties {
 
-	private String[] zkNodes;
+	private String[] zkNodes = new String[] {"localhost"};
 
-	private String defaultZkPort;
+	private String defaultZkPort = "2181";
 
-	private String[] brokers;
+	private String[] brokers = new String[] {"localhost"};
 
-	private String defaultBrokerPort;
+	private String defaultBrokerPort = "9092";
 
-	private String[] headers;
+	private String[] headers = new String[] {};
 
-	private KafkaMessageChannelBinder.Mode mode;
+	private KafkaMessageChannelBinder.Mode mode = Mode.embeddedHeaders;
 
-	private int offsetUpdateTimeWindow;
+	private int offsetUpdateTimeWindow = 10000;
 
-	private int offsetUpdateCount;
+	private int offsetUpdateCount = 0;
 
-	private int offsetUpdateShutdownTimeout;
+	private int offsetUpdateShutdownTimeout = 2000;
 
 	private int maxWait = 100;
+
 	/**
 	 * ZK session timeout in milliseconds.
 	 */
-	private int zkSessionTimeout;
+	private int zkSessionTimeout = 10000;
 
 	/**
 	 * ZK Connection timeout in milliseconds.
 	 */
-	private int zkConnectionTimeout;
+	private int zkConnectionTimeout = 10000;
 
 	private int requiredAcks = 1;
 
@@ -66,7 +67,7 @@ class KafkaBinderConfigurationProperties {
 
 	private int minPartitionCount = 1;
 
-	private int queueSize;
+	private int queueSize = 8192;
 
 	public String getZkConnectionString() {
 		return toConnectionString(this.zkNodes, this.defaultZkPort);

spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/main/resources/META-INF/spring-cloud-stream/kafka-binder.properties

@@ -1,18 +0,0 @@
-spring.cloud.stream.binder.kafka.brokers=${vcap.services.kafka.credentials.kafka.node_ips:localhost}
-spring.cloud.stream.binder.kafka.defaultBrokerPort=${vcap.services.kafka.credentials.kafka.port:9092}
-spring.cloud.stream.binder.kafka.zkNodes=${vcap.services.kafka.credentials.zookeeper.node_ips:localhost}
-spring.cloud.stream.binder.kafka.defaultZkPort=${vcap.services.kafka.credentials.zookeeper.port:2181}
-spring.cloud.stream.binder.kafka.mode=embeddedHeaders
-spring.cloud.stream.binder.kafka.offsetUpdateTimeWindow=10000
-spring.cloud.stream.binder.kafka.offsetUpdateCount=0
-spring.cloud.stream.binder.kafka.offsetUpdateShutdownTimeout=2000
-spring.cloud.stream.binder.kafka.zkSessionTimeout=10000
-spring.cloud.stream.binder.kafka.zkConnectionTimeout=10000
-spring.cloud.stream.binder.kafka.batchSize=16384
-spring.cloud.stream.binder.kafka.batchTimeout=0
-spring.cloud.stream.binder.kafka.requiredAcks=1
-spring.cloud.stream.binder.kafka.replicationFactor=1
-spring.cloud.stream.binder.kafka.compressionCodec=none
-spring.cloud.stream.binder.kafka.fetchSize=1048576
-spring.cloud.stream.binder.kafka.minPartitionCount=1
-spring.cloud.stream.binder.kafka.queueSize=8192

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

@@ -39,6 +39,8 @@ import org.junit.Test;
 
 import org.springframework.beans.DirectFieldAccessor;
 import org.springframework.cloud.stream.binder.Binding;
+import org.springframework.cloud.stream.binder.ExtendedConsumerProperties;
+import org.springframework.cloud.stream.binder.ExtendedProducerProperties;
 import org.springframework.cloud.stream.binder.PartitionCapableBinderTests;
 import org.springframework.cloud.stream.binder.Spy;
 import org.springframework.cloud.stream.test.junit.kafka.KafkaTestSupport;
@@ -65,7 +67,7 @@ import org.springframework.messaging.support.GenericMessage;
  * @author Mark Fisher
  * @author Ilayaperumal Gopinathan
  */
-public class KafkaBinderTests extends PartitionCapableBinderTests<KafkaTestBinder, KafkaConsumerProperties, KafkaProducerProperties> {
+public class KafkaBinderTests extends PartitionCapableBinderTests<KafkaTestBinder, ExtendedConsumerProperties<KafkaConsumerProperties>, ExtendedProducerProperties<KafkaProducerProperties>> {
 
 	private final String CLASS_UNDER_TEST_NAME = KafkaMessageChannelBinder.class.getSimpleName();
 
@@ -88,13 +90,13 @@ public class KafkaBinderTests extends PartitionCapableBinderTests<KafkaTestBinde
 	}
 
 	@Override
-	protected KafkaConsumerProperties createConsumerProperties() {
-		return new KafkaConsumerProperties();
+	protected ExtendedConsumerProperties<KafkaConsumerProperties> createConsumerProperties() {
+		return new ExtendedConsumerProperties<>(new KafkaConsumerProperties());
 	}
 
 	@Override
-	protected KafkaProducerProperties createProducerProperties() {
-		return new KafkaProducerProperties();
+	protected ExtendedProducerProperties<KafkaProducerProperties> createProducerProperties() {
+		return new ExtendedProducerProperties<>(new KafkaProducerProperties());
 	}
 
 	@Before
@@ -165,10 +167,10 @@ public class KafkaBinderTests extends PartitionCapableBinderTests<KafkaTestBinde
 		for (ProducerMetadata.CompressionType codec : codecs) {
 			DirectChannel moduleOutputChannel = new DirectChannel();
 			QueueChannel moduleInputChannel = new QueueChannel();
-			KafkaProducerProperties producerProperties = new KafkaProducerProperties();
-			producerProperties.setCompressionType(codec);
+			ExtendedProducerProperties<KafkaProducerProperties> producerProperties = createProducerProperties();
+			producerProperties.getExtension().setCompressionType(codec);
 			Binding<MessageChannel> producerBinding = binder.bindProducer("foo.0", moduleOutputChannel, producerProperties);
-			Binding<MessageChannel> consumerBinding = binder.bindConsumer("foo.0", "test", moduleInputChannel, new KafkaConsumerProperties());
+			Binding<MessageChannel> consumerBinding = binder.bindConsumer("foo.0", "test", moduleInputChannel, createConsumerProperties());
 			Message<?> message = org.springframework.integration.support.MessageBuilder.withPayload(ratherBigPayload).build();
 			// Let the consumer actually bind to the producer before sending a msg
 			binderBindUnbindLatency();
@@ -190,10 +192,10 @@ public class KafkaBinderTests extends PartitionCapableBinderTests<KafkaTestBinde
 
 		DirectChannel moduleOutputChannel = new DirectChannel();
 		QueueChannel moduleInputChannel = new QueueChannel();
-		KafkaProducerProperties producerProperties = new KafkaProducerProperties();
+		ExtendedProducerProperties<KafkaProducerProperties> producerProperties = createProducerProperties();
 		producerProperties.setPartitionCount(10);
-		KafkaConsumerProperties consumerProperties = new KafkaConsumerProperties();
-		consumerProperties.setMinPartitionCount(10);
+		ExtendedConsumerProperties<KafkaConsumerProperties> consumerProperties = createConsumerProperties();
+		consumerProperties.getExtension().setMinPartitionCount(10);
 		long uniqueBindingId = System.currentTimeMillis();
 		Binding<MessageChannel> producerBinding = binder.bindProducer("foo" + uniqueBindingId + ".0", moduleOutputChannel, producerProperties);
 		Binding<MessageChannel> consumerBinding = binder.bindConsumer("foo" + uniqueBindingId + ".0", null, moduleInputChannel, consumerProperties);
@@ -220,11 +222,11 @@ public class KafkaBinderTests extends PartitionCapableBinderTests<KafkaTestBinde
 
 		DirectChannel moduleOutputChannel = new DirectChannel();
 		QueueChannel moduleInputChannel = new QueueChannel();
-		KafkaProducerProperties producerProperties = new KafkaProducerProperties();
+		ExtendedProducerProperties<KafkaProducerProperties> producerProperties = createProducerProperties();
 		producerProperties.setPartitionCount(5);
 		producerProperties.setPartitionKeyExpression(spelExpressionParser.parseExpression("payload"));
-		KafkaConsumerProperties consumerProperties = new KafkaConsumerProperties();
-		consumerProperties.setMinPartitionCount(3);
+		ExtendedConsumerProperties<KafkaConsumerProperties> consumerProperties = createConsumerProperties();
+		consumerProperties.getExtension().setMinPartitionCount(3);
 		long uniqueBindingId = System.currentTimeMillis();
 		Binding<MessageChannel> producerBinding = binder.bindProducer("foo" + uniqueBindingId + ".0", moduleOutputChannel, producerProperties);
 		Binding<MessageChannel> consumerBinding = binder.bindConsumer("foo" + uniqueBindingId + ".0", null, moduleInputChannel, consumerProperties);
@@ -251,11 +253,11 @@ public class KafkaBinderTests extends PartitionCapableBinderTests<KafkaTestBinde
 
 		DirectChannel moduleOutputChannel = new DirectChannel();
 		QueueChannel moduleInputChannel = new QueueChannel();
-		KafkaProducerProperties producerProperties = new KafkaProducerProperties();
+		ExtendedProducerProperties<KafkaProducerProperties> producerProperties = createProducerProperties();
 		producerProperties.setPartitionCount(5);
 		producerProperties.setPartitionKeyExpression(spelExpressionParser.parseExpression("payload"));
-		KafkaConsumerProperties consumerProperties = new KafkaConsumerProperties();
-		consumerProperties.setMinPartitionCount(5);
+		ExtendedConsumerProperties<KafkaConsumerProperties> consumerProperties = createConsumerProperties();
+		consumerProperties.getExtension().setMinPartitionCount(5);
 		long uniqueBindingId = System.currentTimeMillis();
 		Binding<MessageChannel> producerBinding = binder.bindProducer("foo" + uniqueBindingId + ".0", moduleOutputChannel, producerProperties);
 		Binding<MessageChannel> consumerBinding = binder.bindConsumer("foo" + uniqueBindingId + ".0", null, moduleInputChannel, consumerProperties);
@@ -286,10 +288,10 @@ public class KafkaBinderTests extends PartitionCapableBinderTests<KafkaTestBinde
 		QueueChannel input1 = new QueueChannel();
 
 		String testTopicName = UUID.randomUUID().toString();
-		binder.bindProducer(testTopicName, output, new KafkaProducerProperties());
+		binder.bindProducer(testTopicName, output, createProducerProperties());
 		String testPayload1 = "foo-" + UUID.randomUUID().toString();
 		output.send(new GenericMessage<>(testPayload1.getBytes()));
-		binder.bindConsumer(testTopicName, "startOffsets", input1, new KafkaConsumerProperties());
+		binder.bindConsumer(testTopicName, "startOffsets", input1, createConsumerProperties());
 		Message<byte[]> receivedMessage1 = (Message<byte[]>) receive(input1);
 		assertThat(receivedMessage1, not(nullValue()));
 		assertThat(new String(receivedMessage1.getPayload()), equalTo(testPayload1));
@@ -308,11 +310,11 @@ public class KafkaBinderTests extends PartitionCapableBinderTests<KafkaTestBinde
 		QueueChannel input1 = new QueueChannel();
 
 		String testTopicName = UUID.randomUUID().toString();
-		binder.bindProducer(testTopicName, output, new KafkaProducerProperties());
+		binder.bindProducer(testTopicName, output, createProducerProperties());
 		String testPayload1 = "foo-" + UUID.randomUUID().toString();
 		output.send(new GenericMessage<>(testPayload1.getBytes()));
-		KafkaConsumerProperties properties = new KafkaConsumerProperties();
-		properties.setStartOffset(KafkaMessageChannelBinder.StartOffset.earliest);
+		ExtendedConsumerProperties<KafkaConsumerProperties> properties = createConsumerProperties();
+		properties.getExtension().setStartOffset(KafkaMessageChannelBinder.StartOffset.earliest);
 		binder.bindConsumer(testTopicName, "startOffsets", input1, properties);
 		Message<byte[]> receivedMessage1 = (Message<byte[]>) receive(input1);
 		assertThat(receivedMessage1, not(nullValue()));
@@ -332,12 +334,12 @@ public class KafkaBinderTests extends PartitionCapableBinderTests<KafkaTestBinde
 
 		String testTopicName = UUID.randomUUID().toString();
 
-		Binding<MessageChannel> producerBinding = binder.bindProducer(testTopicName, output, new KafkaProducerProperties());
+		Binding<MessageChannel> producerBinding = binder.bindProducer(testTopicName, output, createProducerProperties());
 		String testPayload1 = "foo-" + UUID.randomUUID().toString();
 		output.send(new GenericMessage<>(testPayload1.getBytes()));
-		KafkaConsumerProperties properties = new KafkaConsumerProperties();
-		properties.setResetOffsets(true);
-		properties.setStartOffset(KafkaMessageChannelBinder.StartOffset.earliest);
+		ExtendedConsumerProperties<KafkaConsumerProperties> properties = createConsumerProperties();
+		properties.getExtension().setResetOffsets(true);
+		properties.getExtension().setStartOffset(KafkaMessageChannelBinder.StartOffset.earliest);
 		Binding<MessageChannel> consumerBinding =
 				binder.bindConsumer(testTopicName, "startOffsets", input1, properties);
 		Message<byte[]> receivedMessage1 = (Message<byte[]>) receive(input1);
@@ -352,9 +354,9 @@ public class KafkaBinderTests extends PartitionCapableBinderTests<KafkaTestBinde
 		String testPayload3 = "foo-" + UUID.randomUUID().toString();
 		output.send(new GenericMessage<>(testPayload3.getBytes()));
 
-		KafkaConsumerProperties properties2 = new KafkaConsumerProperties();
-		properties2.setResetOffsets(true);
-		properties2.setStartOffset(KafkaMessageChannelBinder.StartOffset.earliest);
+		ExtendedConsumerProperties<KafkaConsumerProperties> properties2 = createConsumerProperties();
+		properties2.getExtension().setResetOffsets(true);
+		properties2.getExtension().setStartOffset(KafkaMessageChannelBinder.StartOffset.earliest);
 		consumerBinding =
 				binder.bindConsumer(testTopicName, "startOffsets", input1, properties2);
 		Message<byte[]> receivedMessage4 = (Message<byte[]>) receive(input1);
@@ -383,10 +385,10 @@ public class KafkaBinderTests extends PartitionCapableBinderTests<KafkaTestBinde
 		QueueChannel input1 = new QueueChannel();
 
 		String testTopicName = UUID.randomUUID().toString();
-		Binding<MessageChannel> producerBinding = binder.bindProducer(testTopicName, output, new KafkaProducerProperties());
+		Binding<MessageChannel> producerBinding = binder.bindProducer(testTopicName, output, createProducerProperties());
 		String testPayload1 = "foo-" + UUID.randomUUID().toString();
 		output.send(new GenericMessage<>(testPayload1.getBytes()));
-		KafkaConsumerProperties firstConsumerProperties = new KafkaConsumerProperties();
+		ExtendedConsumerProperties<KafkaConsumerProperties> firstConsumerProperties = createConsumerProperties();
 		Binding<MessageChannel> consumerBinding = binder.bindConsumer(testTopicName, "startOffsets", input1, firstConsumerProperties);
 		Message<byte[]> receivedMessage1 = (Message<byte[]>) receive(input1);
 		assertThat(receivedMessage1, not(nullValue()));
@@ -401,7 +403,7 @@ public class KafkaBinderTests extends PartitionCapableBinderTests<KafkaTestBinde
 		output.send(new GenericMessage<>(testPayload3.getBytes()));
 
 		consumerBinding =
-				binder.bindConsumer(testTopicName, "startOffsets", input1, new KafkaConsumerProperties());
+				binder.bindConsumer(testTopicName, "startOffsets", input1, createConsumerProperties());
 		Message<byte[]> receivedMessage3 = (Message<byte[]>) receive(input1);
 		assertThat(receivedMessage3, not(nullValue()));
 		assertThat(new String(receivedMessage3.getPayload()), equalTo(testPayload3));
@@ -421,8 +423,8 @@ public class KafkaBinderTests extends PartitionCapableBinderTests<KafkaTestBinde
 		DirectChannel output = new DirectChannel();
 
 		String testTopicName = UUID.randomUUID().toString();
-		KafkaProducerProperties properties = new KafkaProducerProperties();
-		properties.setSync(true);
+		ExtendedProducerProperties<KafkaProducerProperties> properties = createProducerProperties();
+		properties.getExtension().setSync(true);
 		Binding<MessageChannel> producerBinding = binder.bindProducer(testTopicName, output, properties);
 		DirectFieldAccessor accessor = new DirectFieldAccessor(extractEndpoint(producerBinding));
 		MessageHandler handler = (MessageHandler) accessor.getPropertyValue("handler");

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

@@ -18,7 +18,12 @@ package org.springframework.cloud.stream.binder.kafka;
 
 import java.util.List;
 
+import com.esotericsoftware.kryo.Kryo;
+import com.esotericsoftware.kryo.Registration;
+
 import org.springframework.cloud.stream.binder.AbstractTestBinder;
+import org.springframework.cloud.stream.binder.ExtendedConsumerProperties;
+import org.springframework.cloud.stream.binder.ExtendedProducerProperties;
 import org.springframework.cloud.stream.test.junit.kafka.KafkaTestSupport;
 import org.springframework.cloud.stream.test.junit.kafka.TestKafkaCluster;
 import org.springframework.context.support.GenericApplicationContext;
@@ -30,9 +35,6 @@ import org.springframework.integration.kafka.support.ProducerListener;
 import org.springframework.integration.kafka.support.ZookeeperConnect;
 import org.springframework.integration.tuple.TupleKryoRegistrar;
 
-import com.esotericsoftware.kryo.Kryo;
-import com.esotericsoftware.kryo.Registration;
-
 
 /**
  * Test support class for {@link KafkaMessageChannelBinder}.
@@ -43,7 +45,7 @@ import com.esotericsoftware.kryo.Registration;
  * @author Gary Russell
  * @author Soby Chacko
  */
-public class KafkaTestBinder extends AbstractTestBinder<KafkaMessageChannelBinder, KafkaConsumerProperties, KafkaProducerProperties> {
+public class KafkaTestBinder extends AbstractTestBinder<KafkaMessageChannelBinder, ExtendedConsumerProperties<KafkaConsumerProperties>, ExtendedProducerProperties<KafkaProducerProperties>> {
 
 	public KafkaTestBinder(KafkaTestSupport kafkaTestSupport) {
 

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

@@ -31,6 +31,8 @@ import org.junit.Test;
 
 import org.springframework.cloud.stream.binder.BinderHeaders;
 import org.springframework.cloud.stream.binder.Binding;
+import org.springframework.cloud.stream.binder.ExtendedConsumerProperties;
+import org.springframework.cloud.stream.binder.ExtendedProducerProperties;
 import org.springframework.integration.IntegrationMessageHeaderAccessor;
 import org.springframework.integration.channel.DirectChannel;
 import org.springframework.integration.channel.QueueChannel;
@@ -52,7 +54,7 @@ public class RawModeKafkaBinderTests extends KafkaBinderTests {
 	@Override
 	public void testPartitionedModuleJava() throws Exception {
 		KafkaTestBinder binder = getBinder();
-		KafkaProducerProperties properties = new KafkaProducerProperties();
+		ExtendedProducerProperties<KafkaProducerProperties> properties = createProducerProperties();
 		properties.setPartitionKeyExtractorClass(RawKafkaPartitionTestSupport.class);
 		properties.setPartitionSelectorClass(RawKafkaPartitionTestSupport.class);
 		properties.setPartitionCount(3);
@@ -61,7 +63,7 @@ public class RawModeKafkaBinderTests extends KafkaBinderTests {
 		output.setBeanName("test.output");
 		Binding<MessageChannel> outputBinding = binder.bindProducer("partJ.0", output, properties);
 
-		KafkaConsumerProperties consumerProperties = new KafkaConsumerProperties();
+		ExtendedConsumerProperties<KafkaConsumerProperties> consumerProperties = createConsumerProperties();
 		consumerProperties.setConcurrency(2);
 		consumerProperties.setInstanceCount(3);
 		consumerProperties.setInstanceIndex(0);
@@ -105,7 +107,7 @@ public class RawModeKafkaBinderTests extends KafkaBinderTests {
 	@Override
 	public void testPartitionedModuleSpEL() throws Exception {
 		KafkaTestBinder binder = getBinder();
-		KafkaProducerProperties properties = new KafkaProducerProperties();
+		ExtendedProducerProperties<KafkaProducerProperties> properties = createProducerProperties();
 		properties.setPartitionKeyExpression(spelExpressionParser.parseExpression("payload[0]"));
 		properties.setPartitionSelectorExpression(spelExpressionParser.parseExpression("hashCode()"));
 		properties.setPartitionCount(3);
@@ -121,7 +123,7 @@ public class RawModeKafkaBinderTests extends KafkaBinderTests {
 
 		}
 
-		KafkaConsumerProperties consumerProperties = new KafkaConsumerProperties();
+		ExtendedConsumerProperties<KafkaConsumerProperties> consumerProperties = createConsumerProperties();
 		consumerProperties.setConcurrency(2);
 		consumerProperties.setInstanceIndex(0);
 		consumerProperties.setInstanceCount(3);
@@ -173,8 +175,8 @@ public class RawModeKafkaBinderTests extends KafkaBinderTests {
 		KafkaTestBinder binder = getBinder();
 		DirectChannel moduleOutputChannel = new DirectChannel();
 		QueueChannel moduleInputChannel = new QueueChannel();
-		Binding<MessageChannel> producerBinding = binder.bindProducer("foo.0", moduleOutputChannel, new KafkaProducerProperties());
-		Binding<MessageChannel> consumerBinding = binder.bindConsumer("foo.0", "test", moduleInputChannel, new KafkaConsumerProperties());
+		Binding<MessageChannel> producerBinding = binder.bindProducer("foo.0", moduleOutputChannel, createProducerProperties());
+		Binding<MessageChannel> consumerBinding = binder.bindConsumer("foo.0", "test", moduleInputChannel, createConsumerProperties());
 		Message<?> message = MessageBuilder.withPayload("foo".getBytes()).build();
 		// Let the consumer actually bind to the producer before sending a msg
 		binderBindUnbindLatency();
@@ -202,14 +204,14 @@ public class RawModeKafkaBinderTests extends KafkaBinderTests {
 		QueueChannel module1InputChannel = new QueueChannel();
 		QueueChannel module2InputChannel = new QueueChannel();
 		QueueChannel module3InputChannel = new QueueChannel();
-		Binding<MessageChannel> producerBinding = binder.bindProducer("baz.0", moduleOutputChannel, new KafkaProducerProperties());
-		Binding<MessageChannel> input1Binding = binder.bindConsumer("baz.0", "test", module1InputChannel, new KafkaConsumerProperties());
+		Binding<MessageChannel> producerBinding = binder.bindProducer("baz.0", moduleOutputChannel, createProducerProperties());
+		Binding<MessageChannel> input1Binding = binder.bindConsumer("baz.0", "test", module1InputChannel, createConsumerProperties());
 		// A new module is using the tap as an input channel
 		String fooTapName = "baz.0";
-		Binding<MessageChannel> input2Binding = binder.bindConsumer(fooTapName, "tap1", module2InputChannel, new KafkaConsumerProperties());
+		Binding<MessageChannel> input2Binding = binder.bindConsumer(fooTapName, "tap1", module2InputChannel, createConsumerProperties());
 		// Another new module is using tap as an input channel
 		String barTapName = "baz.0";
-		Binding<MessageChannel> input3Binding = binder.bindConsumer(barTapName, "tap2", module3InputChannel, new KafkaConsumerProperties());
+		Binding<MessageChannel> input3Binding = binder.bindConsumer(barTapName, "tap2", module3InputChannel, createConsumerProperties());
 
 		Message<?> message = MessageBuilder.withPayload("foo".getBytes()).build();
 		boolean success = false;
@@ -245,7 +247,7 @@ public class RawModeKafkaBinderTests extends KafkaBinderTests {
 		assertNull(receive(module3InputChannel));
 
 		// re-subscribed tap does receive the message
-		input3Binding = binder.bindConsumer(barTapName, "tap2", module3InputChannel, new KafkaConsumerProperties());
+		input3Binding = binder.bindConsumer(barTapName, "tap2", module3InputChannel, createConsumerProperties());
 		assertNotNull(receive(module3InputChannel));
 
 		// clean up

spring-cloud-stream-binders/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitBindingProperties.java

@@ -0,0 +1,43 @@
+/*
+ * Copyright 2016 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.binder.rabbit;
+
+/**
+ * @author Marius Bogoevici
+ */
+public class RabbitBindingProperties {
+
+	private RabbitConsumerProperties consumer = new RabbitConsumerProperties();
+
+	private RabbitProducerProperties producer = new RabbitProducerProperties();
+
+	public RabbitConsumerProperties getConsumer() {
+		return consumer;
+	}
+
+	public void setConsumer(RabbitConsumerProperties consumer) {
+		this.consumer = consumer;
+	}
+
+	public RabbitProducerProperties getProducer() {
+		return producer;
+	}
+
+	public void setProducer(RabbitProducerProperties producer) {
+		this.producer = producer;
+	}
+}

spring-cloud-stream-binders/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitConsumerProperties.java

@@ -17,13 +17,12 @@
 package org.springframework.cloud.stream.binder.rabbit;
 
 import org.springframework.amqp.core.AcknowledgeMode;
-import org.springframework.cloud.stream.binder.ConsumerProperties;
 import org.springframework.util.Assert;
 
 /**
  * @author Marius Bogoevici
  */
-public class RabbitConsumerProperties extends ConsumerProperties {
+public class RabbitConsumerProperties {
 
 	private String prefix = "";
 

spring-cloud-stream-binders/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitExtendedBindingProperties.java

@@ -0,0 +1,60 @@
+/*
+ * Copyright 2016 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.binder.rabbit;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import org.springframework.boot.context.properties.ConfigurationProperties;
+import org.springframework.cloud.stream.binder.ExtendedBindingProperties;
+
+/**
+ * @author Marius Bogoevici
+ */
+@ConfigurationProperties("spring.cloud.stream.rabbit")
+public class RabbitExtendedBindingProperties implements ExtendedBindingProperties<RabbitConsumerProperties, RabbitProducerProperties> {
+
+	private Map<String, RabbitBindingProperties> bindings = new HashMap<>();
+
+	public Map<String, RabbitBindingProperties> getBindings() {
+		return bindings;
+	}
+
+	public void setBindings(Map<String, RabbitBindingProperties> bindings) {
+		this.bindings = bindings;
+	}
+
+	@Override
+	public RabbitConsumerProperties getExtendedConsumerProperties(String channelName) {
+		if (bindings.containsKey(channelName) && bindings.get(channelName).getConsumer() != null) {
+			return bindings.get(channelName).getConsumer();
+		}
+		else {
+			return new RabbitConsumerProperties();
+		}
+	}
+
+	@Override
+	public RabbitProducerProperties getExtendedProducerProperties(String channelName) {
+		if (bindings.containsKey(channelName) && bindings.get(channelName).getProducer() != null) {
+			return bindings.get(channelName).getProducer();
+		}
+		else {
+			return new RabbitProducerProperties();
+		}
+	}
+}

spring-cloud-stream-binders/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitMessageChannelBinder.java

@@ -62,6 +62,9 @@ import org.springframework.beans.factory.support.DefaultListableBeanFactory;
 import org.springframework.cloud.stream.binder.AbstractBinder;
 import org.springframework.cloud.stream.binder.Binding;
 import org.springframework.cloud.stream.binder.DefaultBinding;
+import org.springframework.cloud.stream.binder.ExtendedConsumerProperties;
+import org.springframework.cloud.stream.binder.ExtendedProducerProperties;
+import org.springframework.cloud.stream.binder.ExtendedPropertiesBinder;
 import org.springframework.cloud.stream.binder.MessageValues;
 import org.springframework.cloud.stream.binder.PartitionHandler;
 import org.springframework.context.Lifecycle;
@@ -99,8 +102,8 @@ import org.springframework.util.StringUtils;
  * @author David Turanski
  * @author Marius Bogoevici
  */
-public class RabbitMessageChannelBinder extends AbstractBinder<MessageChannel, RabbitConsumerProperties,
-			RabbitProducerProperties> {
+public class RabbitMessageChannelBinder extends AbstractBinder<MessageChannel, ExtendedConsumerProperties<RabbitConsumerProperties>,
+		ExtendedProducerProperties<RabbitProducerProperties>> implements ExtendedPropertiesBinder<MessageChannel, RabbitConsumerProperties, RabbitProducerProperties> {
 
 	public static final AnonymousQueue.Base64UrlNamingStrategy ANONYMOUS_GROUP_NAME_GENERATOR
 			= new AnonymousQueue.Base64UrlNamingStrategy("anonymous.");
@@ -151,6 +154,8 @@ public class RabbitMessageChannelBinder extends AbstractBinder<MessageChannel, R
 
 	private volatile boolean clustered;
 
+	private RabbitExtendedBindingProperties extendedBindingProperties = new RabbitExtendedBindingProperties();
+
 	public RabbitMessageChannelBinder(ConnectionFactory connectionFactory) {
 		Assert.notNull(connectionFactory, "connectionFactory must not be null");
 		this.connectionFactory = connectionFactory;
@@ -201,6 +206,10 @@ public class RabbitMessageChannelBinder extends AbstractBinder<MessageChannel, R
 		this.password = password;
 	}
 
+	public void setExtendedBindingProperties(RabbitExtendedBindingProperties extendedBindingProperties) {
+		this.extendedBindingProperties = extendedBindingProperties;
+	}
+
 	public void setVhost(String vhost) {
 		this.vhost = vhost;
 	}
@@ -225,23 +234,34 @@ public class RabbitMessageChannelBinder extends AbstractBinder<MessageChannel, R
 		}
 	}
 
+	@Override
+	public RabbitConsumerProperties getExtendedConsumerProperties(String channelName) {
+		return extendedBindingProperties.getExtendedConsumerProperties(channelName);
+	}
+
+
+	@Override
+	public RabbitProducerProperties getExtendedProducerProperties(String channelName) {
+		return extendedBindingProperties.getExtendedProducerProperties(channelName);
+	}
+
 	@Override
 	public Binding<MessageChannel> doBindConsumer(String name, String group, MessageChannel inputChannel,
-			RabbitConsumerProperties properties) {
+			ExtendedConsumerProperties<RabbitConsumerProperties> properties) {
 		boolean anonymousConsumer = !StringUtils.hasText(group);
 		String baseQueueName = anonymousConsumer ? groupedName(name, ANONYMOUS_GROUP_NAME_GENERATOR.generateName())
 				: groupedName(name, group);
 		if (this.logger.isInfoEnabled()) {
 			this.logger.info("declaring queue for inbound: " + baseQueueName + ", bound to: " + name);
 		}
-		String prefix = properties.getPrefix();
+		String prefix = properties.getExtension().getPrefix();
 		String exchangeName = applyPrefix(prefix, name);
 		TopicExchange exchange = new TopicExchange(exchangeName);
 		declareExchange(exchangeName, exchange);
 
 		String queueName = applyPrefix(prefix, baseQueueName);
 		boolean partitioned = !anonymousConsumer && properties.isPartitioned();
-		boolean durable = !anonymousConsumer && properties.isDurableSubscription();
+		boolean durable = !anonymousConsumer && properties.getExtension().isDurableSubscription();
 		Queue queue;
 
 		if (anonymousConsumer) {
@@ -254,7 +274,7 @@ public class RabbitMessageChannelBinder extends AbstractBinder<MessageChannel, R
 			}
 			if (durable) {
 				queue = new Queue(queueName, true, false, false,
-						queueArgs(queueName, properties.getPrefix(), properties.isAutoBindDlq()));
+						queueArgs(queueName, properties.getExtension().getPrefix(), properties.getExtension().isAutoBindDlq()));
 			}
 			else {
 				queue = new Queue(queueName, false, false, true);
@@ -272,7 +292,7 @@ public class RabbitMessageChannelBinder extends AbstractBinder<MessageChannel, R
 		}
 		Binding<MessageChannel> binding = doRegisterConsumer(baseQueueName, group, inputChannel, queue, properties);
 		if (durable) {
-			autoBindDLQ(applyPrefix(prefix, baseQueueName), queueName, properties.getPrefix(), properties.isAutoBindDlq());
+			autoBindDLQ(applyPrefix(prefix, baseQueueName), queueName, properties.getExtension().getPrefix(), properties.getExtension().isAutoBindDlq());
 		}
 		return binding;
 	}
@@ -287,34 +307,34 @@ public class RabbitMessageChannelBinder extends AbstractBinder<MessageChannel, R
 	}
 
 	private Binding<MessageChannel> doRegisterConsumer(final String name, String group, MessageChannel moduleInputChannel, Queue queue,
-													   final RabbitConsumerProperties properties) {
+			final ExtendedConsumerProperties<RabbitConsumerProperties> properties) {
 		DefaultBinding<MessageChannel> consumerBinding;
 		SimpleMessageListenerContainer listenerContainer = new SimpleMessageListenerContainer(
 				this.connectionFactory);
-		listenerContainer.setAcknowledgeMode(properties.getAcknowledgeMode());
-		listenerContainer.setChannelTransacted(properties.isTransacted());
-		listenerContainer.setDefaultRequeueRejected(properties.isRequeueRejected());
+		listenerContainer.setAcknowledgeMode(properties.getExtension().getAcknowledgeMode());
+		listenerContainer.setChannelTransacted(properties.getExtension().isTransacted());
+		listenerContainer.setDefaultRequeueRejected(properties.getExtension().isRequeueRejected());
 
 		int concurrency = properties.getConcurrency();
 		concurrency = concurrency > 0 ? concurrency : 1;
 		listenerContainer.setConcurrentConsumers(concurrency);
-		int maxConcurrency = properties.getMaxConcurrency();
+		int maxConcurrency = properties.getExtension().getMaxConcurrency();
 		if (maxConcurrency > concurrency) {
 			listenerContainer.setMaxConcurrentConsumers(maxConcurrency);
 		}
 
-		listenerContainer.setPrefetchCount(properties.getPrefetch());
-		listenerContainer.setTxSize(properties.getTxSize());
+		listenerContainer.setPrefetchCount(properties.getExtension().getPrefetch());
+		listenerContainer.setTxSize(properties.getExtension().getTxSize());
 		listenerContainer.setTaskExecutor(new SimpleAsyncTaskExecutor(queue.getName() + "-"));
 		listenerContainer.setQueues(queue);
 		int maxAttempts = properties.getMaxAttempts();
-		if (maxAttempts > 1 || properties.isRepublishToDlq()) {
+		if (maxAttempts > 1 || properties.getExtension().isRepublishToDlq()) {
 			RetryOperationsInterceptor retryInterceptor = RetryInterceptorBuilder.stateless()
 					.maxAttempts(maxAttempts)
 					.backOffOptions(properties.getBackOffInitialInterval(),
 							properties.getBackOffMultiplier(),
 							properties.getBackOffMaxInterval())
-					.recoverer(determineRecoverer(name, properties.getPrefix(), properties.isRepublishToDlq()))
+					.recoverer(determineRecoverer(name, properties.getExtension().getPrefix(), properties.getExtension().isRepublishToDlq()))
 					.build();
 			listenerContainer.setAdviceChain(new Advice[] { retryInterceptor });
 		}
@@ -329,14 +349,14 @@ public class RabbitMessageChannelBinder extends AbstractBinder<MessageChannel, R
 		adapter.setOutputChannel(bridgeToModuleChannel);
 		adapter.setBeanName("inbound." + name);
 		DefaultAmqpHeaderMapper mapper = new DefaultAmqpHeaderMapper();
-		mapper.setRequestHeaderNames(properties.getRequestHeaderPatterns());
-		mapper.setReplyHeaderNames(properties.getReplyHeaderPatterns());
+		mapper.setRequestHeaderNames(properties.getExtension().getRequestHeaderPatterns());
+		mapper.setReplyHeaderNames(properties.getExtension().getReplyHeaderPatterns());
 		adapter.setHeaderMapper(mapper);
 		adapter.afterPropertiesSet();
 		consumerBinding = new DefaultBinding<MessageChannel>(name, group, moduleInputChannel, adapter) {
 			@Override
 			protected void afterUnbind() {
-				cleanAutoDeclareContext(properties.getPrefix(), name);
+				cleanAutoDeclareContext(properties.getExtension().getPrefix(), name);
 			}
 		};
 		ReceivingHandler convertingBridge = new ReceivingHandler();
@@ -361,9 +381,10 @@ public class RabbitMessageChannelBinder extends AbstractBinder<MessageChannel, R
 		}
 	}
 
-	private AmqpOutboundEndpoint buildOutboundEndpoint(final String name, RabbitProducerProperties properties,
+	private AmqpOutboundEndpoint buildOutboundEndpoint(final String name,
+			ExtendedProducerProperties<RabbitProducerProperties> properties,
 													   RabbitTemplate rabbitTemplate) {
-		String prefix = properties.getPrefix();
+		String prefix = properties.getExtension().getPrefix();
 		String exchangeName = applyPrefix(prefix, name);
 		TopicExchange exchange = new TopicExchange(exchangeName);
 		declareExchange(exchangeName, exchange);
@@ -378,9 +399,9 @@ public class RabbitMessageChannelBinder extends AbstractBinder<MessageChannel, R
 		for (String requiredGroupName : properties.getRequiredGroups()) {
 			String baseQueueName = exchangeName + "." + requiredGroupName;
 			if (!properties.isPartitioned()) {
-				Queue queue = new Queue(baseQueueName, true, false, false, queueArgs(baseQueueName, prefix, properties.isAutoBindDlq()));
+				Queue queue = new Queue(baseQueueName, true, false, false, queueArgs(baseQueueName, prefix, properties.getExtension().isAutoBindDlq()));
 				declareQueue(baseQueueName, queue);
-				autoBindDLQ(baseQueueName, baseQueueName, properties.getPrefix(), properties.isAutoBindDlq());
+				autoBindDLQ(baseQueueName, baseQueueName, properties.getExtension().getPrefix(), properties.getExtension().isAutoBindDlq());
 				org.springframework.amqp.core.Binding binding = BindingBuilder.bind(queue).to(exchange).with(name);
 				declareBinding(baseQueueName, binding);
 			}
@@ -390,9 +411,9 @@ public class RabbitMessageChannelBinder extends AbstractBinder<MessageChannel, R
 					String partitionSuffix = "-" + i;
 					String partitionQueueName = baseQueueName + partitionSuffix;
 					Queue queue = new Queue(partitionQueueName, true, false, false,
-							queueArgs(partitionQueueName, properties.getPrefix(), properties.isAutoBindDlq()));
+							queueArgs(partitionQueueName, properties.getExtension().getPrefix(), properties.getExtension().isAutoBindDlq()));
 					declareQueue(queue.getName(), queue);
-					autoBindDLQ(baseQueueName, baseQueueName + partitionSuffix, properties.getPrefix(), properties.isAutoBindDlq());
+					autoBindDLQ(baseQueueName, baseQueueName + partitionSuffix, properties.getExtension().getPrefix(), properties.getExtension().isAutoBindDlq());
 					declareBinding(queue.getName(), BindingBuilder.bind(queue).to(exchange).with(name + partitionSuffix));
 				}
 			}
@@ -401,23 +422,25 @@ public class RabbitMessageChannelBinder extends AbstractBinder<MessageChannel, R
 		return endpoint;
 	}
 
-	private void configureOutboundHandler(AmqpOutboundEndpoint handler, RabbitProducerProperties producerProperties) {
+	private void configureOutboundHandler(AmqpOutboundEndpoint handler,
+			ExtendedProducerProperties<RabbitProducerProperties> producerProperties) {
 		DefaultAmqpHeaderMapper mapper = new DefaultAmqpHeaderMapper();
-		mapper.setRequestHeaderNames(producerProperties.getRequestHeaderPatterns());
-		mapper.setReplyHeaderNames(producerProperties.getReplyHeaderPatterns());
+		mapper.setRequestHeaderNames(producerProperties.getExtension().getRequestHeaderPatterns());
+		mapper.setReplyHeaderNames(producerProperties.getExtension().getReplyHeaderPatterns());
 		handler.setHeaderMapper(mapper);
-		handler.setDefaultDeliveryMode(producerProperties.getDeliveryMode());
+		handler.setDefaultDeliveryMode(producerProperties.getExtension().getDeliveryMode());
 		handler.setBeanFactory(this.getBeanFactory());
 		handler.afterPropertiesSet();
 	}
 
 	@Override
-	public Binding<MessageChannel> doBindProducer(String name, MessageChannel outputChannel, RabbitProducerProperties producerProperties) {
-		String exchangeName = applyPrefix(producerProperties.getPrefix(), name);
+	public Binding<MessageChannel> doBindProducer(String name, MessageChannel outputChannel,
+			ExtendedProducerProperties<RabbitProducerProperties> producerProperties) {
+		String exchangeName = applyPrefix(producerProperties.getExtension().getPrefix(), name);
 		TopicExchange exchange = new TopicExchange(exchangeName);
 		declareExchange(exchangeName, exchange);
 		AmqpOutboundEndpoint endpoint = this.buildOutboundEndpoint(name, producerProperties,
-				buildRabbitTemplate(producerProperties));
+				buildRabbitTemplate(producerProperties.getExtension()));
 		return doRegisterProducer(name, outputChannel, endpoint, producerProperties);
 	}
 
@@ -445,12 +468,13 @@ public class RabbitMessageChannelBinder extends AbstractBinder<MessageChannel, R
 	}
 
 	private Binding<MessageChannel> doRegisterProducer(final String name, MessageChannel moduleOutputChannel,
-													   AmqpOutboundEndpoint delegate, RabbitProducerProperties properties) {
+			AmqpOutboundEndpoint delegate, ExtendedProducerProperties<RabbitProducerProperties> properties) {
 		return this.doRegisterProducer(name, moduleOutputChannel, delegate, null, properties);
 	}
 
 	private Binding<MessageChannel> doRegisterProducer(final String name, MessageChannel moduleOutputChannel,
-													   AmqpOutboundEndpoint delegate, String replyTo, RabbitProducerProperties properties) {
+			AmqpOutboundEndpoint delegate, String replyTo,
+			ExtendedProducerProperties<RabbitProducerProperties> properties) {
 		Assert.isInstanceOf(SubscribableChannel.class, moduleOutputChannel);
 		MessageHandler handler = new SendingHandler(delegate, replyTo, properties);
 		EventDrivenConsumer consumer = new EventDrivenConsumer((SubscribableChannel) moduleOutputChannel, handler);
@@ -589,11 +613,12 @@ public class RabbitMessageChannelBinder extends AbstractBinder<MessageChannel, R
 
 		private final String replyTo;
 
-		private final RabbitProducerProperties producerProperties;
+		private final ExtendedProducerProperties<RabbitProducerProperties> producerProperties;
 
 		private final PartitionHandler partitionHandler;
 
-		private SendingHandler(MessageHandler delegate, String replyTo, RabbitProducerProperties properties) {
+		private SendingHandler(MessageHandler delegate, String replyTo,
+				ExtendedProducerProperties<RabbitProducerProperties> properties) {
 			this.delegate = delegate;
 			this.replyTo = replyTo;
 			producerProperties = properties;

spring-cloud-stream-binders/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitProducerProperties.java

@@ -17,13 +17,12 @@
 package org.springframework.cloud.stream.binder.rabbit;
 
 import org.springframework.amqp.core.MessageDeliveryMode;
-import org.springframework.cloud.stream.binder.ProducerProperties;
 
 /**
  * @author Marius Bogoevici
  * @author Gary Russell
  */
-public class RabbitProducerProperties extends ProducerProperties {
+public class RabbitProducerProperties {
 
 	private String prefix = "";
 

spring-cloud-stream-binders/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/config/RabbitBinderConfigurationProperties.java

@@ -22,7 +22,7 @@ import org.springframework.core.io.Resource;
 /**
  * @author David Turanski
  */
-@ConfigurationProperties(prefix = "spring.cloud.stream.binder.rabbit")
+@ConfigurationProperties(prefix = "spring.cloud.stream.rabbit.binder")
 class RabbitBinderConfigurationProperties {
 
 	private String[] addresses = new String[0];

spring-cloud-stream-binders/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/config/RabbitMessageChannelBinderConfiguration.java

@@ -28,6 +28,7 @@ import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.boot.autoconfigure.PropertyPlaceholderAutoConfiguration;
 import org.springframework.boot.context.properties.EnableConfigurationProperties;
 import org.springframework.cloud.stream.binder.rabbit.ConnectionFactorySettings;
+import org.springframework.cloud.stream.binder.rabbit.RabbitExtendedBindingProperties;
 import org.springframework.cloud.stream.binder.rabbit.RabbitMessageChannelBinder;
 import org.springframework.cloud.stream.config.codec.kryo.KryoCodecAutoConfiguration;
 import org.springframework.context.annotation.Bean;
@@ -37,7 +38,7 @@ import org.springframework.integration.codec.Codec;
 
 @Configuration
 @Import({PropertyPlaceholderAutoConfiguration.class, KryoCodecAutoConfiguration.class})
-@EnableConfigurationProperties({RabbitBinderConfigurationProperties.class})
+@EnableConfigurationProperties({RabbitBinderConfigurationProperties.class, RabbitExtendedBindingProperties.class})
 public class RabbitMessageChannelBinderConfiguration {
 
 	@Autowired
@@ -49,6 +50,9 @@ public class RabbitMessageChannelBinderConfiguration {
 	@Autowired
 	private RabbitBinderConfigurationProperties rabbitBinderConfigurationProperties;
 
+	@Autowired
+	private RabbitExtendedBindingProperties rabbitExtendedBindingProperties;
+
 	@Bean
 	RabbitMessageChannelBinder rabbitMessageChannelBinder() {
 		RabbitMessageChannelBinder binder = new RabbitMessageChannelBinder(rabbitConnectionFactory);
@@ -63,6 +67,7 @@ public class RabbitMessageChannelBinderConfiguration {
 		binder.setUsername(rabbitBinderConfigurationProperties.getUsername());
 		binder.setUseSSL(rabbitBinderConfigurationProperties.isUseSSL());
 		binder.setVhost(rabbitBinderConfigurationProperties.getVhost());
+		binder.setExtendedBindingProperties(rabbitExtendedBindingProperties);
 		return binder;
 	}
 

spring-cloud-stream-binders/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitBinderTests.java

@@ -55,6 +55,8 @@ import org.springframework.amqp.support.postprocessor.DelegatingDecompressingPos
 import org.springframework.amqp.utils.test.TestUtils;
 import org.springframework.beans.DirectFieldAccessor;
 import org.springframework.cloud.stream.binder.Binding;
+import org.springframework.cloud.stream.binder.ExtendedConsumerProperties;
+import org.springframework.cloud.stream.binder.ExtendedProducerProperties;
 import org.springframework.cloud.stream.binder.PartitionCapableBinderTests;
 import org.springframework.cloud.stream.binder.PartitionKeyExtractorStrategy;
 import org.springframework.cloud.stream.binder.PartitionSelectorStrategy;
@@ -79,7 +81,7 @@ import org.springframework.messaging.support.GenericMessage;
  * @author Gary Russell
  * @author David Turanski
  */
-public class RabbitBinderTests extends PartitionCapableBinderTests<RabbitTestBinder, RabbitConsumerProperties, RabbitProducerProperties> {
+public class RabbitBinderTests extends PartitionCapableBinderTests<RabbitTestBinder, ExtendedConsumerProperties<RabbitConsumerProperties>, ExtendedProducerProperties<RabbitProducerProperties>> {
 
 	private final String CLASS_UNDER_TEST_NAME = RabbitMessageChannelBinder.class.getSimpleName();
 
@@ -97,13 +99,13 @@ public class RabbitBinderTests extends PartitionCapableBinderTests<RabbitTestBin
 	}
 
 	@Override
-	protected RabbitConsumerProperties createConsumerProperties() {
-		return new RabbitConsumerProperties();
+	protected ExtendedConsumerProperties<RabbitConsumerProperties> createConsumerProperties() {
+		return new ExtendedConsumerProperties<>(new RabbitConsumerProperties());
 	}
 
 	@Override
-	protected RabbitProducerProperties createProducerProperties() {
-		return new RabbitProducerProperties();
+	protected ExtendedProducerProperties<RabbitProducerProperties> createProducerProperties() {
+		return new ExtendedProducerProperties<>(new RabbitProducerProperties());
 	}
 
 	@Override
@@ -116,8 +118,8 @@ public class RabbitBinderTests extends PartitionCapableBinderTests<RabbitTestBin
 		RabbitTestBinder binder = getBinder();
 		DirectChannel moduleOutputChannel = new DirectChannel();
 		DirectChannel moduleInputChannel = new DirectChannel();
-		Binding<MessageChannel> producerBinding = binder.bindProducer("bad.0", moduleOutputChannel, new RabbitProducerProperties());
-		Binding<MessageChannel> consumerBinding = binder.bindConsumer("bad.0", "test", moduleInputChannel, new RabbitConsumerProperties());
+		Binding<MessageChannel> producerBinding = binder.bindProducer("bad.0", moduleOutputChannel, createProducerProperties());
+		Binding<MessageChannel> consumerBinding = binder.bindConsumer("bad.0", "test", moduleInputChannel, createConsumerProperties());
 		Message<?> message = MessageBuilder.withPayload("bad").setHeader(MessageHeaders.CONTENT_TYPE,
 				"foo/bar").build();
 		final CountDownLatch latch = new CountDownLatch(3);
@@ -138,15 +140,15 @@ public class RabbitBinderTests extends PartitionCapableBinderTests<RabbitTestBin
 	@Test
 	public void testConsumerProperties() throws Exception {
 		RabbitTestBinder binder = getBinder();
-		RabbitConsumerProperties properties = new RabbitConsumerProperties();
-		properties.setTransacted(true);
+		ExtendedConsumerProperties<RabbitConsumerProperties> properties = createConsumerProperties();
+		properties.getExtension().setTransacted(true);
 		Binding<MessageChannel> consumerBinding = binder.bindConsumer("props.0", null, new DirectChannel(), properties);
 		AbstractEndpoint endpoint = extractEndpoint(consumerBinding);
 		SimpleMessageListenerContainer container = TestUtils.getPropertyValue(endpoint, "messageListenerContainer",
 				SimpleMessageListenerContainer.class);
 		assertEquals(AcknowledgeMode.AUTO, container.getAcknowledgeMode());
 		assertThat(container.getQueueNames()[0],
-				startsWith(properties.getPrefix()));
+				startsWith(properties.getExtension().getPrefix()));
 		assertTrue(TestUtils.getPropertyValue(container, "transactional", Boolean.class));
 		assertEquals(1, TestUtils.getPropertyValue(container, "concurrentConsumers"));
 		assertNull(TestUtils.getPropertyValue(container, "maxConcurrentConsumers"));
@@ -161,19 +163,19 @@ public class RabbitBinderTests extends PartitionCapableBinderTests<RabbitTestBin
 		consumerBinding.unbind();
 		assertFalse(endpoint.isRunning());
 
-		properties = new RabbitConsumerProperties();
-		properties.setAcknowledgeMode(AcknowledgeMode.NONE);
+		properties = createConsumerProperties();
+		properties.getExtension().setAcknowledgeMode(AcknowledgeMode.NONE);
 		properties.setBackOffInitialInterval(2000);
 		properties.setBackOffMaxInterval(20000);
 		properties.setBackOffMultiplier(5.0);
 		properties.setConcurrency(2);
 		properties.setMaxAttempts(23);
-		properties.setMaxConcurrency(3);
-		properties.setPrefix("foo.");
-		properties.setPrefetch(20);
-		properties.setRequestHeaderPatterns(new String[] {"foo"});
-		properties.setRequeueRejected(false);
-		properties.setTxSize(10);
+		properties.getExtension().setMaxConcurrency(3);
+		properties.getExtension().setPrefix("foo.");
+		properties.getExtension().setPrefetch(20);
+		properties.getExtension().setRequestHeaderPatterns(new String[] {"foo"});
+		properties.getExtension().setRequeueRejected(false);
+		properties.getExtension().setTxSize(10);
 		properties.setInstanceIndex(0);
 		consumerBinding = binder.bindConsumer("props.0", "test", new DirectChannel(), properties);
 
@@ -189,7 +191,7 @@ public class RabbitBinderTests extends PartitionCapableBinderTests<RabbitTestBin
 	@Test
 	public void testProducerProperties() throws Exception {
 		RabbitTestBinder binder = getBinder();
-		Binding<MessageChannel> producerBinding = binder.bindProducer("props.0", new DirectChannel(), new RabbitProducerProperties());
+		Binding<MessageChannel> producerBinding = binder.bindProducer("props.0", new DirectChannel(), createProducerProperties());
 		@SuppressWarnings("unchecked")
 		AbstractEndpoint endpoint = extractEndpoint(producerBinding);
 		MessageDeliveryMode mode = TestUtils.getPropertyValue(endpoint, "handler.delegate.defaultDeliveryMode",
@@ -202,16 +204,16 @@ public class RabbitBinderTests extends PartitionCapableBinderTests<RabbitTestBin
 		assertFalse(endpoint.isRunning());
 		assertFalse(TestUtils.getPropertyValue(endpoint, "handler.delegate.amqpTemplate.transactional", Boolean.class));
 
-		RabbitProducerProperties properties = new RabbitProducerProperties();
-		properties.setPrefix("foo.");
-		properties.setDeliveryMode(MessageDeliveryMode.NON_PERSISTENT);
-		properties.setRequestHeaderPatterns(new String[] {"foo"});
+		ExtendedProducerProperties<RabbitProducerProperties> properties = createProducerProperties();
+		properties.getExtension().setPrefix("foo.");
+		properties.getExtension().setDeliveryMode(MessageDeliveryMode.NON_PERSISTENT);
+		properties.getExtension().setRequestHeaderPatterns(new String[] {"foo"});
 		properties.setPartitionKeyExpression(spelExpressionParser.parseExpression("'foo'"));
 		properties.setPartitionKeyExtractorClass(TestPartitionKeyExtractorClass.class);
 		properties.setPartitionSelectorExpression(spelExpressionParser.parseExpression("0"));
 		properties.setPartitionSelectorClass(TestPartitionSelectorClass.class);
 		properties.setPartitionCount(1);
-		properties.setTransacted(true);
+		properties.getExtension().setTransacted(true);
 
 		producerBinding = binder.bindProducer("props.0", new DirectChannel(), properties);
 		endpoint = extractEndpoint(producerBinding);
@@ -235,12 +237,12 @@ public class RabbitBinderTests extends PartitionCapableBinderTests<RabbitTestBin
 
 		RabbitTestBinder binder = getBinder();
 
-		RabbitConsumerProperties properties = new RabbitConsumerProperties();
-		properties.setPrefix(TEST_PREFIX);
-		properties.setAutoBindDlq(true);
-		properties.setDurableSubscription(true);
+		ExtendedConsumerProperties<RabbitConsumerProperties> properties = createConsumerProperties();
+		properties.getExtension().setPrefix(TEST_PREFIX);
+		properties.getExtension().setAutoBindDlq(true);
+		properties.getExtension().setDurableSubscription(true);
 		properties.setMaxAttempts(1); // disable retry
-		properties.setRequeueRejected(false);
+		properties.getExtension().setRequeueRejected(false);
 		DirectChannel moduleInputChannel = new DirectChannel();
 		moduleInputChannel.setBeanName("durableTest");
 		moduleInputChannel.subscribe(new MessageHandler() {
@@ -276,12 +278,12 @@ public class RabbitBinderTests extends PartitionCapableBinderTests<RabbitTestBin
 		RabbitAdmin admin = new RabbitAdmin(this.rabbitAvailableRule.getResource());
 
 		RabbitTestBinder binder = getBinder();
-		RabbitConsumerProperties properties = new RabbitConsumerProperties();
-		properties.setPrefix(TEST_PREFIX);
-		properties.setAutoBindDlq(true);
-		properties.setDurableSubscription(false);
+		ExtendedConsumerProperties<RabbitConsumerProperties> properties = createConsumerProperties();
+		properties.getExtension().setPrefix(TEST_PREFIX);
+		properties.getExtension().setAutoBindDlq(true);
+		properties.getExtension().setDurableSubscription(false);
 		properties.setMaxAttempts(1); // disable retry
-		properties.setRequeueRejected(false);
+		properties.getExtension().setRequeueRejected(false);
 		DirectChannel moduleInputChannel = new DirectChannel();
 		moduleInputChannel.setBeanName("nondurabletest");
 		moduleInputChannel.subscribe(new MessageHandler() {
@@ -301,12 +303,12 @@ public class RabbitBinderTests extends PartitionCapableBinderTests<RabbitTestBin
 	@Test
 	public void testAutoBindDLQ() throws Exception {
 		RabbitTestBinder binder = getBinder();
-		RabbitConsumerProperties properties = new RabbitConsumerProperties();
-		properties.setPrefix(TEST_PREFIX);
-		properties.setAutoBindDlq(true);
+		ExtendedConsumerProperties<RabbitConsumerProperties> properties = createConsumerProperties();
+		properties.getExtension().setPrefix(TEST_PREFIX);
+		properties.getExtension().setAutoBindDlq(true);
 		properties.setMaxAttempts(1); // disable retry
-		properties.setRequeueRejected(false);
-		properties.setDurableSubscription(true);
+		properties.getExtension().setRequeueRejected(false);
+		properties.getExtension().setDurableSubscription(true);
 		DirectChannel moduleInputChannel = new DirectChannel();
 		moduleInputChannel.setBeanName("dlqTest");
 		moduleInputChannel.subscribe(new MessageHandler() {
@@ -346,11 +348,11 @@ public class RabbitBinderTests extends PartitionCapableBinderTests<RabbitTestBin
 	@Test
 	public void testAutoBindDLQPartionedConsumerFirst() throws Exception {
 		RabbitTestBinder binder = getBinder();
-		RabbitConsumerProperties properties = new RabbitConsumerProperties();
-		properties.setPrefix("bindertest.");
-		properties.setAutoBindDlq(true);
+		ExtendedConsumerProperties<RabbitConsumerProperties> properties = createConsumerProperties();
+		properties.getExtension().setPrefix("bindertest.");
+		properties.getExtension().setAutoBindDlq(true);
 		properties.setMaxAttempts(1); // disable retry
-		properties.setRequeueRejected(false);
+		properties.getExtension().setRequeueRejected(false);
 		properties.setPartitioned(true);
 		properties.setInstanceIndex(0);
 		DirectChannel input0 = new DirectChannel();
@@ -364,9 +366,9 @@ public class RabbitBinderTests extends PartitionCapableBinderTests<RabbitTestBin
 		Binding<MessageChannel> input1Binding = binder.bindConsumer("partDLQ.0", "dlqPartGrp", input1, properties);
 		Binding<MessageChannel> defaultConsumerBinding2 = binder.bindConsumer("partDLQ.0", "default", new QueueChannel(), properties);
 
-		RabbitProducerProperties producerProperties = new RabbitProducerProperties();
-		producerProperties.setPrefix("bindertest.");
-		producerProperties.setAutoBindDlq(true);
+		ExtendedProducerProperties<RabbitProducerProperties> producerProperties = createProducerProperties();
+		producerProperties.getExtension().setPrefix("bindertest.");
+		producerProperties.getExtension().setAutoBindDlq(true);
 		producerProperties.setPartitionKeyExtractorClass(PartitionTestSupport.class);
 		producerProperties.setPartitionSelectorClass(PartitionTestSupport.class);
 		producerProperties.setPartitionCount(2);
@@ -432,10 +434,10 @@ public class RabbitBinderTests extends PartitionCapableBinderTests<RabbitTestBin
 	@Test
 	public void testAutoBindDLQPartitionedProducerFirst() throws Exception {
 		RabbitTestBinder binder = getBinder();
-		RabbitProducerProperties properties = new RabbitProducerProperties();
+		ExtendedProducerProperties<RabbitProducerProperties> properties = createProducerProperties();
 
-		properties.setPrefix("bindertest.");
-		properties.setAutoBindDlq(true);
+		properties.getExtension().setPrefix("bindertest.");
+		properties.getExtension().setAutoBindDlq(true);
 		properties.setRequiredGroups("dlqPartGrp");
 		properties.setPartitionKeyExtractorClass(PartitionTestSupport.class);
 		properties.setPartitionSelectorClass(PartitionTestSupport.class);
@@ -444,11 +446,11 @@ public class RabbitBinderTests extends PartitionCapableBinderTests<RabbitTestBin
 		output.setBeanName("test.output");
 		Binding<MessageChannel> outputBinding = binder.bindProducer("partDLQ.1", output, properties);
 
-		RabbitConsumerProperties consumerProperties = new RabbitConsumerProperties();
-		consumerProperties.setPrefix("bindertest.");
-		consumerProperties.setAutoBindDlq(true);
+		ExtendedConsumerProperties<RabbitConsumerProperties> consumerProperties = createConsumerProperties();
+		consumerProperties.getExtension().setPrefix("bindertest.");
+		consumerProperties.getExtension().setAutoBindDlq(true);
 		consumerProperties.setMaxAttempts(1); // disable retry
-		consumerProperties.setRequeueRejected(false);
+		consumerProperties.getExtension().setRequeueRejected(false);
 		consumerProperties.setPartitioned(true);
 		consumerProperties.setInstanceIndex(0);
 		DirectChannel input0 = new DirectChannel();
@@ -527,13 +529,13 @@ public class RabbitBinderTests extends PartitionCapableBinderTests<RabbitTestBin
 		admin.declareQueue(queue);
 
 		RabbitTestBinder binder = getBinder();
-		RabbitConsumerProperties properties = new RabbitConsumerProperties();
-		properties.setPrefix(TEST_PREFIX);
-		properties.setAutoBindDlq(true);
-		properties.setRepublishToDlq(true);
+		ExtendedConsumerProperties<RabbitConsumerProperties> properties = createConsumerProperties();
+		properties.getExtension().setPrefix(TEST_PREFIX);
+		properties.getExtension().setAutoBindDlq(true);
+		properties.getExtension().setRepublishToDlq(true);
 		properties.setMaxAttempts(1); // disable retry
-		properties.setRequeueRejected(false);
-		properties.setDurableSubscription(true);
+		properties.getExtension().setRequeueRejected(false);
+		properties.getExtension().setDurableSubscription(true);
 		DirectChannel moduleInputChannel = new DirectChannel();
 		moduleInputChannel.setBeanName("dlqPubTest");
 		moduleInputChannel.subscribe(new MessageHandler() {
@@ -569,20 +571,20 @@ public class RabbitBinderTests extends PartitionCapableBinderTests<RabbitTestBin
 	public void testBatchingAndCompression() throws Exception {
 		RabbitTemplate template = new RabbitTemplate(this.rabbitAvailableRule.getResource());
 		RabbitTestBinder binder = getBinder();
-		RabbitProducerProperties properties = new RabbitProducerProperties();
-		properties.setDeliveryMode(MessageDeliveryMode.NON_PERSISTENT);
-		properties.setBatchingEnabled(true);
-		properties.setBatchSize(2);
-		properties.setBatchBufferLimit(100000);
-		properties.setBatchTimeout(30000);
-		properties.setCompress(true);
+		ExtendedProducerProperties<RabbitProducerProperties> properties = createProducerProperties();
+		properties.getExtension().setDeliveryMode(MessageDeliveryMode.NON_PERSISTENT);
+		properties.getExtension().setBatchingEnabled(true);
+		properties.getExtension().setBatchSize(2);
+		properties.getExtension().setBatchBufferLimit(100000);
+		properties.getExtension().setBatchTimeout(30000);
+		properties.getExtension().setCompress(true);
 		properties.setRequiredGroups("default");
 
 		DirectChannel output = new DirectChannel();
 		output.setBeanName("batchingProducer");
 		Binding<MessageChannel> producerBinding = binder.bindProducer("batching.0", output, properties);
 
-		while (template.receive(properties.getPrefix() + "batching.0.default") != null) {
+		while (template.receive(properties.getExtension().getPrefix() + "batching.0.default") != null) {
 		}
 
 		Log logger = spy(TestUtils.getPropertyValue(binder, "binder.compressingPostProcessor.logger", Log.class));
@@ -605,7 +607,7 @@ public class RabbitBinderTests extends PartitionCapableBinderTests<RabbitTestBin
 
 		QueueChannel input = new QueueChannel();
 		input.setBeanName("batchingConsumer");
-		Binding<MessageChannel> consumerBinding = binder.bindConsumer("batching.0", "test", input, new RabbitConsumerProperties());
+		Binding<MessageChannel> consumerBinding = binder.bindConsumer("batching.0", "test", input, createConsumerProperties());
 
 		output.send(new GenericMessage<>("foo".getBytes()));
 		output.send(new GenericMessage<>("bar".getBytes()));
@@ -633,16 +635,16 @@ public class RabbitBinderTests extends PartitionCapableBinderTests<RabbitTestBin
 		RabbitMessageChannelBinder rabbitBinder = new RabbitMessageChannelBinder(cf);
 		RabbitTestBinder binder = new RabbitTestBinder(cf, rabbitBinder);
 
-		RabbitProducerProperties properties = new RabbitProducerProperties();
-		properties.setPrefix("latebinder.");
-		properties.setAutoBindDlq(true);
+		ExtendedProducerProperties<RabbitProducerProperties> properties = createProducerProperties();
+		properties.getExtension().setPrefix("latebinder.");
+		properties.getExtension().setAutoBindDlq(true);
 
 		MessageChannel moduleOutputChannel = new DirectChannel();
 		Binding<MessageChannel> late0ProducerBinding = binder.bindProducer("late.0", moduleOutputChannel, properties);
 
 		QueueChannel moduleInputChannel = new QueueChannel();
-		RabbitConsumerProperties rabbitConsumerProperties = new RabbitConsumerProperties();
-		rabbitConsumerProperties.setPrefix("latebinder.");
+		ExtendedConsumerProperties<RabbitConsumerProperties> rabbitConsumerProperties = createConsumerProperties();
+		rabbitConsumerProperties.getExtension().setPrefix("latebinder.");
 		Binding<MessageChannel> late0ConsumerBinding = binder.bindConsumer("late.0", "test", moduleInputChannel, rabbitConsumerProperties);
 
 		properties.setPartitionKeyExpression(spelExpressionParser.parseExpression("payload.equals('0') ? 0 : 1"));
@@ -655,31 +657,31 @@ public class RabbitBinderTests extends PartitionCapableBinderTests<RabbitTestBin
 		QueueChannel partInputChannel0 = new QueueChannel();
 		QueueChannel partInputChannel1 = new QueueChannel();
 
-		RabbitConsumerProperties partLateConsumerProperties = new RabbitConsumerProperties();
-		partLateConsumerProperties.setPrefix("latebinder.");
+		ExtendedConsumerProperties<RabbitConsumerProperties> partLateConsumerProperties = createConsumerProperties();
+		partLateConsumerProperties.getExtension().setPrefix("latebinder.");
 		partLateConsumerProperties.setPartitioned(true);
 		partLateConsumerProperties.setInstanceIndex(0);
 		Binding<MessageChannel> partlate0Consumer0Binding = binder.bindConsumer("partlate.0", "test", partInputChannel0, partLateConsumerProperties);
 		partLateConsumerProperties.setInstanceIndex(1);
 		Binding<MessageChannel> partlate0Consumer1Binding = binder.bindConsumer("partlate.0", "test", partInputChannel1, partLateConsumerProperties);
 
-		RabbitProducerProperties noDlqProducerProperties = new RabbitProducerProperties();
-		noDlqProducerProperties.setPrefix("latebinder.");
+		ExtendedProducerProperties<RabbitProducerProperties> noDlqProducerProperties = createProducerProperties();
+		noDlqProducerProperties.getExtension().setPrefix("latebinder.");
 		MessageChannel noDLQOutputChannel = new DirectChannel();
 		Binding<MessageChannel> noDlqProducerBinding = binder.bindProducer("lateNoDLQ.0", noDLQOutputChannel, noDlqProducerProperties);
 
 		QueueChannel noDLQInputChannel = new QueueChannel();
-		RabbitConsumerProperties noDlqConsumerProperties = new RabbitConsumerProperties();
-		noDlqConsumerProperties.setPrefix("latebinder.");
+		ExtendedConsumerProperties<RabbitConsumerProperties> noDlqConsumerProperties = createConsumerProperties();
+		noDlqConsumerProperties.getExtension().setPrefix("latebinder.");
 		Binding<MessageChannel> noDlqConsumerBinding = binder.bindConsumer("lateNoDLQ.0", "test", noDLQInputChannel, noDlqConsumerProperties);
 
 		MessageChannel outputChannel = new DirectChannel();
 		Binding<MessageChannel> pubSubProducerBinding = binder.bindProducer("latePubSub", outputChannel, noDlqProducerProperties);
 		QueueChannel pubSubInputChannel = new QueueChannel();
-		noDlqConsumerProperties.setDurableSubscription(false);
+		noDlqConsumerProperties.getExtension().setDurableSubscription(false);
 		Binding<MessageChannel> nonDurableConsumerBinding = binder.bindConsumer("latePubSub", "lategroup", pubSubInputChannel, noDlqConsumerProperties);
 		QueueChannel durablePubSubInputChannel = new QueueChannel();
-		noDlqConsumerProperties.setDurableSubscription(true);
+		noDlqConsumerProperties.getExtension().setDurableSubscription(true);
 		Binding<MessageChannel> durableConsumerBinding = binder.bindConsumer("latePubSub", "lateDurableGroup", durablePubSubInputChannel, noDlqConsumerProperties);
 
 		proxy.start();

spring-cloud-stream-binders/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitTestBinder.java

@@ -23,6 +23,8 @@ import org.springframework.amqp.rabbit.connection.ConnectionFactory;
 import org.springframework.amqp.rabbit.core.RabbitAdmin;
 import org.springframework.cloud.stream.binder.AbstractTestBinder;
 import org.springframework.cloud.stream.binder.Binding;
+import org.springframework.cloud.stream.binder.ExtendedConsumerProperties;
+import org.springframework.cloud.stream.binder.ExtendedProducerProperties;
 import org.springframework.context.support.GenericApplicationContext;
 import org.springframework.integration.codec.kryo.PojoCodec;
 import org.springframework.integration.context.IntegrationContextUtils;
@@ -37,7 +39,7 @@ import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;
  * @author David Turanski
  * @author Mark Fisher
  */
-public class RabbitTestBinder extends AbstractTestBinder<RabbitMessageChannelBinder, RabbitConsumerProperties, RabbitProducerProperties> {
+public class RabbitTestBinder extends AbstractTestBinder<RabbitMessageChannelBinder, ExtendedConsumerProperties<RabbitConsumerProperties>, ExtendedProducerProperties<RabbitProducerProperties>> {
 
 	private final RabbitAdmin rabbitAdmin;
 
@@ -65,18 +67,20 @@ public class RabbitTestBinder extends AbstractTestBinder<RabbitMessageChannelBin
 	}
 
 	@Override
-	public Binding<MessageChannel> bindConsumer(String name, String group, MessageChannel moduleInputChannel, RabbitConsumerProperties properties) {
+	public Binding<MessageChannel> bindConsumer(String name, String group, MessageChannel moduleInputChannel,
+			ExtendedConsumerProperties<RabbitConsumerProperties> properties) {
 		if (group != null) {
-			this.queues.add(properties.getPrefix() + name + ("." + group));
+			this.queues.add(properties.getExtension().getPrefix() + name + ("." + group));
 		}
-		this.exchanges.add(properties.getPrefix() + name);
+		this.exchanges.add(properties.getExtension().getPrefix() + name);
 		return super.bindConsumer(name, group, moduleInputChannel, properties);
 	}
 
 	@Override
-	public Binding<MessageChannel> bindProducer(String name, MessageChannel moduleOutputChannel, RabbitProducerProperties properties) {
-		this.queues.add(properties.getPrefix() + name + ".default");
-		this.exchanges.add(properties.getPrefix() + name);
+	public Binding<MessageChannel> bindProducer(String name, MessageChannel moduleOutputChannel,
+			ExtendedProducerProperties<RabbitProducerProperties> properties) {
+		this.queues.add(properties.getExtension().getPrefix() + name + ".default");
+		this.exchanges.add(properties.getExtension().getPrefix() + name);
 		return super.bindProducer(name, moduleOutputChannel, properties);
 	}
 

spring-cloud-stream-binders/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/integration/RabbitBinderModuleTests.java

@@ -23,6 +23,7 @@ import static org.hamcrest.CoreMatchers.not;
 import static org.hamcrest.collection.IsMapContaining.hasKey;
 import static org.junit.Assert.assertNotNull;
 import static org.junit.Assert.assertThat;
+import static org.junit.Assert.assertTrue;
 
 import java.util.ArrayList;
 import java.util.List;
@@ -36,6 +37,8 @@ import org.mockito.Mockito;
 import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
 import org.springframework.amqp.rabbit.connection.ConnectionFactory;
 import org.springframework.amqp.rabbit.core.RabbitAdmin;
+import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
+import org.springframework.amqp.utils.test.TestUtils;
 import org.springframework.beans.DirectFieldAccessor;
 import org.springframework.boot.SpringApplication;
 import org.springframework.boot.actuate.health.CompositeHealthIndicator;
@@ -45,11 +48,14 @@ import org.springframework.boot.autoconfigure.SpringBootApplication;
 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.Binding;
 import org.springframework.cloud.stream.binder.rabbit.RabbitMessageChannelBinder;
+import org.springframework.cloud.stream.binding.ChannelBindingService;
 import org.springframework.cloud.stream.messaging.Processor;
 import org.springframework.cloud.stream.test.junit.rabbit.RabbitTestSupport;
 import org.springframework.context.ConfigurableApplicationContext;
 import org.springframework.context.annotation.Bean;
+import org.springframework.messaging.MessageChannel;
 
 /**
  * @author Marius Bogoevici
@@ -101,6 +107,45 @@ public class RabbitBinderModuleTests {
 		assertThat(healthIndicators.get("rabbit").health().getStatus(), equalTo(Status.UP));
 	}
 
+	@Test
+	public void testParentConnectionFactoryInheritedByDefaultAndRabbitSettingsPropagated() {
+		context = SpringApplication.run(SimpleProcessor.class,
+				"--server.port=0",
+				"--spring.cloud.stream.rabbit.bindings.input.consumer.transacted=true",
+				"--spring.cloud.stream.rabbit.bindings.output.producer.transacted=true");
+		BinderFactory<?> binderFactory = context.getBean(BinderFactory.class);
+		Binder binder = binderFactory.getBinder(null);
+		assertThat(binder, instanceOf(RabbitMessageChannelBinder.class));
+		ChannelBindingService channelBindingService = context.getBean(ChannelBindingService.class);
+		DirectFieldAccessor channelBindingServiceAccessor = new DirectFieldAccessor(channelBindingService);
+		Map<String, List<Binding<MessageChannel>>> consumerBindings = (Map<String, List<Binding<MessageChannel>>>)
+				channelBindingServiceAccessor.getPropertyValue("consumerBindings");
+		Binding<MessageChannel> inputBinding = consumerBindings.get("input").get(0);
+		SimpleMessageListenerContainer container = TestUtils.getPropertyValue(inputBinding,
+				"endpoint.messageListenerContainer",
+				SimpleMessageListenerContainer.class);
+		assertTrue(TestUtils.getPropertyValue(container, "transactional", Boolean.class));
+		Map<String, Binding<MessageChannel>> producerBindings =
+				(Map<String, Binding<MessageChannel>>) TestUtils.getPropertyValue(channelBindingService, "producerBindings");
+		Binding<MessageChannel> outputBinding = producerBindings.get("output");
+		assertTrue(TestUtils.getPropertyValue(outputBinding, "endpoint.handler.delegate.amqpTemplate.transactional", Boolean.class));
+		DirectFieldAccessor binderFieldAccessor = new DirectFieldAccessor(binder);
+		ConnectionFactory binderConnectionFactory =
+				(ConnectionFactory) binderFieldAccessor.getPropertyValue("connectionFactory");
+		assertThat(binderConnectionFactory, instanceOf(CachingConnectionFactory.class));
+		ConnectionFactory connectionFactory = context.getBean(ConnectionFactory.class);
+		assertThat(binderConnectionFactory, is(connectionFactory));
+		CompositeHealthIndicator bindersHealthIndicator =
+				context.getBean("bindersHealthIndicator", CompositeHealthIndicator.class);
+		DirectFieldAccessor directFieldAccessor = new DirectFieldAccessor(bindersHealthIndicator);
+		assertNotNull(bindersHealthIndicator);
+		@SuppressWarnings("unchecked")
+		Map<String, HealthIndicator> healthIndicators =
+				(Map<String, HealthIndicator>) directFieldAccessor.getPropertyValue("indicators");
+		assertThat(healthIndicators, hasKey("rabbit"));
+		assertThat(healthIndicators.get("rabbit").health().getStatus(), equalTo(Status.UP));
+	}
+
 	@Test
 	public void testParentConnectionFactoryInheritedIfOverridden() {
 		context = new SpringApplication(SimpleProcessor.class, ConnectionFactoryConfiguration.class).run("--server.port=0");

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

@@ -0,0 +1,30 @@
+/*
+ * Copyright 2016 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.binder;
+
+/**
+ * Properties that extend the common binding properties for a particular binder implementation.
+ *
+ * @author Marius Bogoevici
+ * @author Mark Fisher
+ */
+public interface ExtendedBindingProperties<C, P> {
+
+	C getExtendedConsumerProperties(String channelName);
+
+	P getExtendedProducerProperties(String channelName);
+}

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

@@ -0,0 +1,35 @@
+/*
+ * Copyright 2016 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.binder;
+
+/**
+ * Extension of {@link ConsumerProperties} to be used with an {@link ExtendedPropertiesBinder}.
+ *
+ * @author Marius Bogoevici
+ */
+public class ExtendedConsumerProperties<T> extends ConsumerProperties {
+
+	private T extension;
+
+	public ExtendedConsumerProperties(T extension) {
+		this.extension = extension;
+	}
+
+	public T getExtension() {
+		return extension;
+	}
+}

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

@@ -0,0 +1,34 @@
+/*
+ * Copyright 2016 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.binder;
+
+/**
+ * @author Marius Bogoevici
+ */
+public class ExtendedProducerProperties<T> extends ProducerProperties {
+
+	private T extension;
+
+	public ExtendedProducerProperties(T extension) {
+		this.extension = extension;
+	}
+
+	public T getExtension() {
+		return extension;
+	}
+
+}

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

@@ -0,0 +1,29 @@
+/*
+ * Copyright 2016 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.binder;
+
+/**
+ * Extension of {@link Binder} that takes {@link ExtendedConsumerProperties} and {@link ExtendedProducerProperties}
+ * as arguments. In addition to supporting binding operations, it allows the binder to provide values for the
+ * additional properties it expects on the bindings.
+ *
+ * @author Marius Bogoevici
+ */
+public interface ExtendedPropertiesBinder<T, C, P>
+		extends Binder<T, ExtendedConsumerProperties<C>, ExtendedProducerProperties<P>>, ExtendedBindingProperties<C, P> {
+
+}

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

@@ -109,10 +109,7 @@ public class BinderAwareChannelResolver extends BeanFactoryMessageChannelDestina
 					@SuppressWarnings("unchecked")
 					Binder<MessageChannel, ?, ProducerProperties> binder =
 							(Binder<MessageChannel, ?, ProducerProperties>) binderFactory.getBinder(binderName);
-					Class<? extends ProducerProperties> producerPropertiesClass =
-							ChannelBindingService.resolveProducerPropertiesType(binder);
-					ProducerProperties producerProperties =
-							this.channelBindingServiceProperties.getProducerProperties(channelName, producerPropertiesClass);
+					ProducerProperties producerProperties = this.channelBindingServiceProperties.getProducerProperties(channelName);
 					String destinationName = this.channelBindingServiceProperties.getBindingDestination(channelName);
 					this.dynamicDestinationsBindable.addOutputBinding(beanName,
 							binder.bindProducer(destinationName, channel, producerProperties));

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

@@ -25,13 +25,16 @@ import java.util.Map;
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 
+import org.springframework.beans.BeanUtils;
 import org.springframework.cloud.stream.binder.Binder;
 import org.springframework.cloud.stream.binder.BinderFactory;
 import org.springframework.cloud.stream.binder.Binding;
 import org.springframework.cloud.stream.binder.ConsumerProperties;
+import org.springframework.cloud.stream.binder.ExtendedConsumerProperties;
+import org.springframework.cloud.stream.binder.ExtendedProducerProperties;
+import org.springframework.cloud.stream.binder.ExtendedPropertiesBinder;
 import org.springframework.cloud.stream.binder.ProducerProperties;
 import org.springframework.cloud.stream.config.ChannelBindingServiceProperties;
-import org.springframework.core.ResolvableType;
 import org.springframework.messaging.MessageChannel;
 import org.springframework.util.CollectionUtils;
 import org.springframework.util.StringUtils;
@@ -71,12 +74,17 @@ public class ChannelBindingService {
 		List<Binding<MessageChannel>> bindings = new ArrayList<>();
 		Binder<MessageChannel, ConsumerProperties, ?> binder =
 				(Binder<MessageChannel, ConsumerProperties, ?>) getBinderForChannel(inputChannelName);
-		Class<? extends ConsumerProperties> propertiesClass = resolveConsumerPropertiesType(binder);
 		ConsumerProperties consumerProperties =
-				this.channelBindingServiceProperties.getConsumerProperties(inputChannelName, propertiesClass);
+				this.channelBindingServiceProperties.getConsumerProperties(inputChannelName);
+		if (binder instanceof ExtendedPropertiesBinder) {
+			ExtendedPropertiesBinder extendedPropertiesBinder = (ExtendedPropertiesBinder) binder;
+			Object extension = extendedPropertiesBinder.getExtendedConsumerProperties(inputChannelName);
+			ExtendedConsumerProperties extendedConsumerProperties = new ExtendedConsumerProperties(extension);
+			BeanUtils.copyProperties(consumerProperties, extendedConsumerProperties);
+			consumerProperties = extendedConsumerProperties;
+		}
 		for (String target : channelBindingTargets) {
-			Binding<MessageChannel> binding = binder.bindConsumer(target, channelBindingServiceProperties.getGroup(inputChannelName),
-					inputChannel, consumerProperties);
+			Binding<MessageChannel> binding = binder.bindConsumer(target, channelBindingServiceProperties.getGroup(inputChannelName), inputChannel, consumerProperties);
 			bindings.add(binding);
 		}
 		this.consumerBindings.put(inputChannelName, bindings);
@@ -88,9 +96,14 @@ public class ChannelBindingService {
 		String channelBindingTarget = this.channelBindingServiceProperties.getBindingDestination(outputChannelName);
 		Binder<MessageChannel, ?, ProducerProperties> binder =
 				(Binder<MessageChannel, ?, ProducerProperties>) getBinderForChannel(outputChannelName);
-		Class<? extends ProducerProperties> propertiesClass = resolveProducerPropertiesType(binder);
-		ProducerProperties producerProperties =
-				this.channelBindingServiceProperties.getProducerProperties(outputChannelName, propertiesClass);
+		ProducerProperties producerProperties = this.channelBindingServiceProperties.getProducerProperties(outputChannelName);
+		if (binder instanceof ExtendedPropertiesBinder) {
+			ExtendedPropertiesBinder extendedPropertiesBinder = (ExtendedPropertiesBinder) binder;
+			Object extension = extendedPropertiesBinder.getExtendedProducerProperties(outputChannelName);
+			ExtendedProducerProperties extendedProducerProperties = new ExtendedProducerProperties<>(extension);
+			BeanUtils.copyProperties(producerProperties, extendedProducerProperties);
+			producerProperties = extendedProducerProperties;
+		}
 		Binding<MessageChannel> binding = binder.bindProducer(channelBindingTarget, outputChannel, producerProperties);
 		this.producerBindings.put(outputChannelName, binding);
 		return binding;
@@ -122,45 +135,4 @@ public class ChannelBindingService {
 		String transport = this.channelBindingServiceProperties.getBinder(channelName);
 		return binderFactory.getBinder(transport);
 	}
-
-
-	static Class<? extends ConsumerProperties> resolveConsumerPropertiesType(Binder<?, ?, ?> binder) {
-		return resolveTypeForBinder(binder, ConsumerProperties.class);
-	}
-
-	static Class<? extends ProducerProperties> resolveProducerPropertiesType(Binder<?, ?, ?> binder) {
-		return resolveTypeForBinder(binder, ProducerProperties.class);
-	}
-
-	@SuppressWarnings("unchecked")
-	static <T> Class<? extends T> resolveTypeForBinder(Binder<?, ?, ?> binder, Class<? extends T> upperBound) {
-		Class<? extends T> propertiesClass = null;
-		ResolvableType currentType = ResolvableType.forType(binder.getClass());
-		while (!Object.class.equals(currentType.getRawClass()) && propertiesClass == null) {
-			ResolvableType[] interfaces = currentType.getInterfaces();
-			ResolvableType binderResolvableType = null;
-			for (ResolvableType interfaceType : interfaces) {
-				if (Binder.class.equals(interfaceType.getRawClass())) {
-					binderResolvableType = interfaceType;
-					break;
-				}
-			}
-			if (binderResolvableType == null) {
-				currentType = currentType.getSuperType();
-			}
-			else {
-				ResolvableType[] generics = binderResolvableType.getGenerics();
-				for (ResolvableType generic : generics) {
-					Class<?> resolvedParameter = generic.resolve();
-					if (resolvedParameter != null && upperBound.isAssignableFrom(resolvedParameter)) {
-						propertiesClass = (Class<? extends T>) resolvedParameter;
-					}
-				}
-			}
-		}
-		if (propertiesClass == null) {
-			propertiesClass = upperBound;
-		}
-		return propertiesClass;
-	}
 }

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

@@ -16,9 +16,15 @@
 
 package org.springframework.cloud.stream.config;
 
+import javax.validation.constraints.AssertTrue;
+
 import com.fasterxml.jackson.annotation.JsonInclude;
 import com.fasterxml.jackson.annotation.JsonInclude.Include;
 
+import org.springframework.cloud.stream.binder.ConsumerProperties;
+import org.springframework.cloud.stream.binder.ProducerProperties;
+import org.springframework.validation.annotation.Validated;
+
 /**
  * Contains the properties of a binding.
  * @author Marius Bogoevici
@@ -26,6 +32,7 @@ import com.fasterxml.jackson.annotation.JsonInclude.Include;
  * @author Gary Russell
  */
 @JsonInclude(value = Include.NON_DEFAULT)
+@Validated
 public class BindingProperties {
 
 	private static final String COMMA = ",";
@@ -49,6 +56,11 @@ public class BindingProperties {
 
 	private String binder;
 
+
+	private ConsumerProperties consumer = null;
+
+	private ProducerProperties producer = null;
+
 	public String getDestination() {
 		return this.destination;
 	}
@@ -81,6 +93,27 @@ public class BindingProperties {
 		this.binder = binder;
 	}
 
+	public ConsumerProperties getConsumer() {
+		return consumer;
+	}
+
+	public void setConsumer(ConsumerProperties consumer) {
+		this.consumer = consumer;
+	}
+
+	public ProducerProperties getProducer() {
+		return producer;
+	}
+
+	public void setProducer(ProducerProperties producer) {
+		this.producer = producer;
+	}
+
+	@AssertTrue(message = "A binding must not set both producer and consumer properties.")
+	public boolean onlyOneOfProducerOrConsumerSet() {
+		return consumer == null || producer == null;
+	}
+
 	public String toString() {
 		StringBuilder sb = new StringBuilder();
 		sb.append("destination=" + this.destination);

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

@@ -16,10 +16,7 @@
 
 package org.springframework.cloud.stream.config;
 
-import java.beans.PropertyDescriptor;
-import java.util.ArrayList;
 import java.util.HashMap;
-import java.util.List;
 import java.util.Map;
 import java.util.Properties;
 import java.util.TreeMap;
@@ -27,14 +24,9 @@ import java.util.TreeMap;
 import com.fasterxml.jackson.annotation.JsonInclude;
 import com.fasterxml.jackson.annotation.JsonInclude.Include;
 
-import org.springframework.beans.BeanUtils;
 import org.springframework.beans.BeansException;
-import org.springframework.beans.MutablePropertyValues;
-import org.springframework.beans.factory.BeanInitializationException;
 import org.springframework.beans.factory.InitializingBean;
 import org.springframework.beans.factory.annotation.Value;
-import org.springframework.boot.bind.PropertySourcesPropertyValues;
-import org.springframework.boot.bind.RelaxedDataBinder;
 import org.springframework.boot.context.properties.ConfigurationProperties;
 import org.springframework.cloud.stream.binder.ConsumerProperties;
 import org.springframework.cloud.stream.binder.ProducerProperties;
@@ -56,17 +48,6 @@ import org.springframework.util.StringUtils;
 @JsonInclude(Include.NON_DEFAULT)
 public class ChannelBindingServiceProperties implements ApplicationContextAware, InitializingBean {
 
-	private final static String[] bindingPropertyFields;
-
-	static {
-		PropertyDescriptor[] propertyDescriptors = BeanUtils.getPropertyDescriptors(BindingProperties.class);
-		List<String> propertyNames = new ArrayList<>();
-		for (PropertyDescriptor propertyDescriptor : propertyDescriptors) {
-			propertyNames.add(propertyDescriptor.getName());
-		}
-		bindingPropertyFields = propertyNames.toArray(new String[propertyNames.size()]);
-	}
-
 	private ConversionService conversionService;
 
 	@Value("${INSTANCE_INDEX:${CF_INSTANCE_INDEX:0}}")
@@ -197,48 +178,27 @@ public class ChannelBindingServiceProperties implements ApplicationContextAware,
 		return properties;
 	}
 
-	public <T extends ConsumerProperties> T getConsumerProperties(String inputChannelName, Class<T> beanClass) {
+	public ConsumerProperties getConsumerProperties(String inputChannelName) {
 		Assert.notNull(inputChannelName, "The input channel name cannot be null");
-		Assert.notNull(beanClass, "The bean class cannot be null");
-		T consumerProperties = populateProperties(inputChannelName, beanClass, consumerDefaults);
+		ConsumerProperties consumerProperties = getBindingProperties(inputChannelName).getConsumer();
+		if (consumerProperties == null) {
+			consumerProperties = new ConsumerProperties();
+		}
 		consumerProperties.setInstanceCount(this.instanceCount);
 		consumerProperties.setInstanceIndex(this.instanceIndex);
 		return consumerProperties;
 	}
 
 
-	public <T extends ProducerProperties> T getProducerProperties(String outputChannelName, Class<T> beanClass) {
+	public ProducerProperties getProducerProperties(String outputChannelName) {
 		Assert.notNull(outputChannelName, "The output channel name cannot be null");
-		Assert.notNull(beanClass, "The bean class cannot be null");
-		T producerProperties = populateProperties(outputChannelName, beanClass, producerDefaults);
+		ProducerProperties producerProperties = getBindingProperties(outputChannelName).getProducer();
+		if (producerProperties == null) {
+			producerProperties = new ProducerProperties();
+		}
 		return producerProperties;
 	}
 
-
-	private <C> C populateProperties(String channelName, Class<C> propertiesClass, Properties defaults) {
-		C beanInstance;
-		try {
-			beanInstance = propertiesClass.newInstance();
-		}
-		catch (InstantiationException | IllegalAccessException e) {
-			throw new BeanInitializationException(e.getMessage());
-		}
-		// bind defaults first
-		RelaxedDataBinder dataBinder = new RelaxedDataBinder(beanInstance);
-		dataBinder.setIgnoreUnknownFields(ignoreUnknownProperties);
-		dataBinder.setDisallowedFields(bindingPropertyFields);
-		dataBinder.bind(new MutablePropertyValues(defaults));
-		// bind configured properties next, if available
-		if (applicationContext != null && applicationContext.getEnvironment() != null) {
-			dataBinder = new RelaxedDataBinder(beanInstance, "spring.cloud.stream.bindings." + channelName);
-			dataBinder.setConversionService(conversionService);
-			dataBinder.setIgnoreUnknownFields(ignoreUnknownProperties);
-			dataBinder.setDisallowedFields(bindingPropertyFields);
-			dataBinder.bind(new PropertySourcesPropertyValues(applicationContext.getEnvironment().getPropertySources()));
-		}
-		return beanInstance;
-	}
-
 	public BindingProperties getBindingProperties(String channelName) {
 		BindingProperties bindingProperties = bindings.containsKey(channelName) ?
 				bindings.get(channelName) : new BindingProperties();

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

@@ -1,130 +0,0 @@
-/*
- * Copyright 2016 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.binder;
-
-import static org.hamcrest.MatcherAssert.assertThat;
-import static org.hamcrest.collection.IsArrayContainingInOrder.arrayContaining;
-import static org.mockito.Matchers.eq;
-import static org.mockito.Matchers.same;
-import static org.mockito.Mockito.verify;
-import static org.mockito.Mockito.verifyNoMoreInteractions;
-
-import org.hamcrest.CoreMatchers;
-import org.junit.Test;
-import org.mockito.ArgumentCaptor;
-
-import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
-import org.springframework.boot.builder.SpringApplicationBuilder;
-import org.springframework.cloud.stream.annotation.EnableBinding;
-import org.springframework.cloud.stream.utils.MockBinderRegistryConfiguration;
-import org.springframework.context.ConfigurableApplicationContext;
-import org.springframework.context.annotation.Import;
-
-/**
- * @author Marius Bogoevici
- */
-public class DefaultSettingsTests {
-
-	@SuppressWarnings("unchecked")
-	@Test
-	public void testDefaultSettings() {
-		ConfigurableApplicationContext applicationContext = createBuilder().run(
-				"--spring.cloud.stream.bindings.foo.destination=fooDest",
-				"--spring.cloud.stream.bindings.bar.destination=barDest",
-				"--spring.cloud.stream.bindings.baz.destination=bazDest",
-				"--spring.cloud.stream.bindings.qux.destination=quxDest",
-				"--spring.cloud.stream.bindings.foo.group=fooBarGroup",
-				"--spring.cloud.stream.bindings.bar.group=fooBarGroup",
-				"--spring.cloud.stream.consumerDefaults.concurrency=2",
-				"--spring.cloud.stream.producerDefaults.requiredGroups=quxbazReq1,quxbazReq2");
-
-		Binder binder = applicationContext.getBean(Binder.class);
-		FooChannels fooChannels = applicationContext.getBean(FooChannels.class);
-
-		ArgumentCaptor<ConsumerProperties> fooConsumerProperties = ArgumentCaptor.forClass(ConsumerProperties.class);
-		verify(binder).bindConsumer(eq("fooDest"), eq("fooBarGroup"), same(fooChannels.foo()),
-				fooConsumerProperties.capture());
-		assertThat(fooConsumerProperties.getValue().getConcurrency(), CoreMatchers.equalTo(2));
-		ArgumentCaptor<ConsumerProperties> barConsumerProperties = ArgumentCaptor.forClass(ConsumerProperties.class);
-		verify(binder).bindConsumer(eq("barDest"), eq("fooBarGroup"), same(fooChannels.bar()),
-				barConsumerProperties.capture());
-		assertThat(barConsumerProperties.getValue().getConcurrency(), CoreMatchers.equalTo(2));
-
-		ArgumentCaptor<ProducerProperties> bazProducerProperties = ArgumentCaptor.forClass(ProducerProperties.class);
-		verify(binder).bindProducer(eq("bazDest"), same(fooChannels.baz()), bazProducerProperties.capture());
-		assertThat(bazProducerProperties.getValue().getRequiredGroups(), arrayContaining("quxbazReq1", "quxbazReq2"));
-
-		ArgumentCaptor<ProducerProperties> quxProducerProperties = ArgumentCaptor.forClass(ProducerProperties.class);
-		verify(binder).bindProducer(eq("quxDest"), same(fooChannels.qux()), quxProducerProperties.capture());
-		assertThat(quxProducerProperties.getValue().getRequiredGroups(), arrayContaining("quxbazReq1", "quxbazReq2"));
-
-		verifyNoMoreInteractions(binder);
-		applicationContext.close();
-	}
-
-	@SuppressWarnings("unchecked")
-	@Test
-	public void testDefaultSettingsOverridden() {
-		ConfigurableApplicationContext applicationContext = createBuilder().run(
-				"--spring.cloud.stream.bindings.foo.destination=fooDest",
-				"--spring.cloud.stream.bindings.bar.destination=barDest",
-				"--spring.cloud.stream.bindings.baz.destination=bazDest",
-				"--spring.cloud.stream.bindings.qux.destination=quxDest",
-				"--spring.cloud.stream.bindings.foo.group=fooBarGroup",
-				"--spring.cloud.stream.bindings.bar.group=fooBarGroup",
-				"--spring.cloud.stream.consumerDefaults.concurrency=2",
-				"--spring.cloud.stream.bindings.bar.concurrency=4",
-				"--spring.cloud.stream.producerDefaults.requiredGroups=quxbazReq1,quxbazReq2",
-				"--spring.cloud.stream.bindings.qux.requiredGroups=quxbazReq3,quxbazReq4");
-
-		Binder binder = applicationContext.getBean(Binder.class);
-		FooChannels fooChannels = applicationContext.getBean(FooChannels.class);
-
-		ArgumentCaptor<ConsumerProperties> fooConsumerProperties = ArgumentCaptor.forClass(ConsumerProperties.class);
-		verify(binder).bindConsumer(eq("fooDest"), eq("fooBarGroup"), same(fooChannels.foo()),
-				fooConsumerProperties.capture());
-		assertThat(fooConsumerProperties.getValue().getConcurrency(), CoreMatchers.equalTo(2));
-		ArgumentCaptor<ConsumerProperties> barConsumerProperties = ArgumentCaptor.forClass(ConsumerProperties.class);
-		verify(binder).bindConsumer(eq("barDest"), eq("fooBarGroup"), same(fooChannels.bar()),
-				barConsumerProperties.capture());
-		assertThat(barConsumerProperties.getValue().getConcurrency(), CoreMatchers.equalTo(4));
-
-		ArgumentCaptor<ProducerProperties> bazProducerProperties = ArgumentCaptor.forClass(ProducerProperties.class);
-		verify(binder).bindProducer(eq("bazDest"), same(fooChannels.baz()), bazProducerProperties.capture());
-		assertThat(bazProducerProperties.getValue().getRequiredGroups(), arrayContaining("quxbazReq1", "quxbazReq2"));
-
-		ArgumentCaptor<ProducerProperties> quxProducerProperties = ArgumentCaptor.forClass(ProducerProperties.class);
-		verify(binder).bindProducer(eq("quxDest"), same(fooChannels.qux()), quxProducerProperties.capture());
-		assertThat(quxProducerProperties.getValue().getRequiredGroups(), arrayContaining("quxbazReq3", "quxbazReq4"));
-
-		verifyNoMoreInteractions(binder);
-		applicationContext.close();
-	}
-
-	private SpringApplicationBuilder createBuilder() {
-		return new SpringApplicationBuilder(TestFooChannels.class)
-				.web(false);
-	}
-
-
-	@EnableBinding(FooChannels.class)
-	@EnableAutoConfiguration
-	@Import(MockBinderRegistryConfiguration.class)
-	public static class TestFooChannels {
-
-	}
-}

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

@@ -1,161 +0,0 @@
-/*
- * Copyright 2016 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.binding;
-
-import org.junit.Assert;
-import org.junit.Test;
-
-import org.springframework.cloud.stream.binder.AbstractBinder;
-import org.springframework.cloud.stream.binder.Binder;
-import org.springframework.cloud.stream.binder.Binding;
-import org.springframework.cloud.stream.binder.ConsumerProperties;
-import org.springframework.cloud.stream.binder.ProducerProperties;
-
-
-/**
- * @author Marius Bogoevici
- */
-public class PropertiesClassResolutionTests {
-
-	@Test
-	public void testDefaultResolution() {
-		SimpleBinderImplementation testBinder = new SimpleBinderImplementation();
-		Assert.assertEquals(ConsumerProperties.class, ChannelBindingService.resolveConsumerPropertiesType(testBinder));
-		Assert.assertEquals(ProducerProperties.class, ChannelBindingService.resolveProducerPropertiesType(testBinder));
-	}
-
-	@Test
-	public void testBinderImplementorWithCustomTypes() {
-		BinderImplementationWithCustomTypes testBinder = new BinderImplementationWithCustomTypes();
-		Assert.assertEquals(SubclassConsumerProperties.class, ChannelBindingService.resolveConsumerPropertiesType(testBinder));
-		Assert.assertEquals(SubclassProducerProperties.class, ChannelBindingService.resolveProducerPropertiesType(testBinder));
-	}
-
-	@Test
-	public void testExtendsAbstractBinderWithDefaultTypes() {
-		ExtendsAbstractBinderWithDefaultTypes testBinder = new ExtendsAbstractBinderWithDefaultTypes();
-		Assert.assertEquals(ConsumerProperties.class, ChannelBindingService.resolveConsumerPropertiesType(testBinder));
-		Assert.assertEquals(ProducerProperties.class, ChannelBindingService.resolveProducerPropertiesType(testBinder));
-	}
-
-	@Test
-	public void testExtendsAbstractBinderWithCustomTypes() {
-		ExtendsAbstractBinderWithCustomTypes testBinder = new ExtendsAbstractBinderWithCustomTypes();
-		Assert.assertEquals(SubclassConsumerProperties.class, ChannelBindingService.resolveConsumerPropertiesType(testBinder));
-		Assert.assertEquals(SubclassProducerProperties.class, ChannelBindingService.resolveProducerPropertiesType(testBinder));
-	}
-
-	@Test
-	public void testGenericBinderWithDefaultTypes() {
-		GenericBinderWithDefaultTypes testBinder = new GenericBinderWithDefaultTypes();
-		Assert.assertEquals(ConsumerProperties.class, ChannelBindingService.resolveConsumerPropertiesType(testBinder));
-		Assert.assertEquals(ProducerProperties.class, ChannelBindingService.resolveProducerPropertiesType(testBinder));
-	}
-
-	@Test
-	public void testGenericBinderWithCustomTypes() {
-		GenericBinderWithCustomTypes testBinder = new GenericBinderWithCustomTypes();
-		Assert.assertEquals(SubclassConsumerProperties.class, ChannelBindingService.resolveConsumerPropertiesType(testBinder));
-		Assert.assertEquals(SubclassProducerProperties.class, ChannelBindingService.resolveProducerPropertiesType(testBinder));
-	}
-
-	private class SimpleBinderImplementation implements Binder<Object, ConsumerProperties, ProducerProperties> {
-
-		@Override
-		public Binding<Object> bindConsumer(String name, String group, Object inboundBindTarget, ConsumerProperties consumerProperties) {
-			return null;
-		}
-
-		@Override
-		public Binding<Object> bindProducer(String name, Object outboundBindTarget, ProducerProperties producerProperties) {
-			return null;
-		}
-	}
-
-	private class BinderImplementationWithCustomTypes implements Binder<Object, SubclassConsumerProperties, SubclassProducerProperties> {
-
-		@Override
-		public Binding<Object> bindConsumer(String name, String group, Object inboundBindTarget, SubclassConsumerProperties consumerProperties) {
-			return null;
-		}
-
-		@Override
-		public Binding<Object> bindProducer(String name, Object outboundBindTarget, SubclassProducerProperties producerProperties) {
-			return null;
-		}
-	}
-
-	private class ExtendsAbstractBinderWithDefaultTypes extends AbstractBinder<Object, ConsumerProperties, ProducerProperties> {
-
-		@Override
-		protected Binding<Object> doBindConsumer(String name, String group, Object inputTarget, ConsumerProperties properties) {
-			return null;
-		}
-
-		@Override
-		protected Binding<Object> doBindProducer(String name, Object outboundBindTarget, ProducerProperties properties) {
-			return null;
-		}
-	}
-
-	private class ExtendsAbstractBinderWithCustomTypes extends AbstractBinder<Object, SubclassConsumerProperties, SubclassProducerProperties> {
-
-		@Override
-		protected Binding<Object> doBindConsumer(String name, String group, Object inputTarget, SubclassConsumerProperties properties) {
-			return null;
-		}
-
-		@Override
-		protected Binding<Object> doBindProducer(String name, Object outboundBindTarget, SubclassProducerProperties properties) {
-			return null;
-		}
-	}
-
-	private class GenericBinderWithDefaultTypes<C extends ConsumerProperties, P extends ProducerProperties> extends AbstractBinder<Object, C, P> {
-
-		@Override
-		protected Binding<Object> doBindConsumer(String name, String group, Object inputTarget, C properties) {
-			return null;
-		}
-
-		@Override
-		protected Binding<Object> doBindProducer(String name, Object outboundBindTarget, P properties) {
-			return null;
-		}
-	}
-
-	private class GenericBinderWithCustomTypes<C extends SubclassConsumerProperties, P extends SubclassProducerProperties> extends AbstractBinder<Object, C, P> {
-
-		@Override
-		protected Binding<Object> doBindConsumer(String name, String group, Object inputTarget, C properties) {
-			return null;
-		}
-
-		@Override
-		protected Binding<Object> doBindProducer(String name, Object outboundBindTarget, P properties) {
-			return null;
-		}
-	}
-
-	private class SubclassConsumerProperties extends ConsumerProperties {
-
-	}
-
-	private class SubclassProducerProperties extends ProducerProperties {
-
-	}
-}

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

@@ -1,5 +1,5 @@
 spring.cloud.stream.bindings.input.destination=partIn
-spring.cloud.stream.bindings.input.partitioned=true
+spring.cloud.stream.bindings.input.consumer.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

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