Refactor Kafka binder configuration

- Move KafkaBinderConfigurationProperties out of the configuration class
 - Remove defaultProperties bean creation and use `locations` in `ConfigurationProperties` to load the default properties

Move properties to @PropertySource and add tests

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2015 the original author or authors.
+ * Copyright 2015-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.
@@ -17,110 +17,136 @@
 package org.springframework.cloud.stream.binder.kafka.config;
 
 import org.springframework.boot.context.properties.ConfigurationProperties;
+import org.springframework.cloud.stream.binder.kafka.KafkaMessageChannelBinder;
+import org.springframework.cloud.stream.binder.kafka.KafkaMessageChannelBinder.Mode;
+import org.springframework.util.StringUtils;
 
 /**
  * @author David Turanski
+ * @author Ilayaperumal Gopinathan
+ * @author Marius Bogoevici
  */
-@ConfigurationProperties(prefix = "spring.cloud.stream.binder.kafka.default")
- class KafkaBinderConfigurationProperties {
-	
-	private int batchSize;
-	
-	private long batchTimeout;
-	
-	private int requiredAcks;
-	
-	private int replicationFactor;
-	
-	private int concurrency;
-	
-	private String compressionCodec;
-	
-	private boolean autoCommitEnabled;
-	
-	private int fetchSize;
-	
-	private int minPartitionCount;
-	
-	private int queueSize;
+@ConfigurationProperties(prefix = "spring.cloud.stream.binder.kafka")
+class KafkaBinderConfigurationProperties {
 
-	public int getBatchSize() {
-		return batchSize;
+	private String[] zkNodes;
+
+	private String defaultZkPort;
+
+	private String[] brokers;
+
+	private String defaultBrokerPort;
+
+	private String[] headers;
+
+	private KafkaMessageChannelBinder.Mode mode;
+
+	private int offsetUpdateTimeWindow;
+
+	private int offsetUpdateCount;
+
+	private int offsetUpdateShutdownTimeout;
+
+	private boolean resetOffsets = false;
+
+	private KafkaMessageChannelBinder.StartOffset startOffset;
+
+	public String getZkConnectionString() {
+		return toConnectionString(this.zkNodes, this.defaultZkPort);
 	}
 
-	public void setBatchSize(int batchSize) {
-		this.batchSize = batchSize;
+	public String getKafkaConnectionString() {
+		return toConnectionString(this.brokers, this.defaultBrokerPort);
 	}
 
-	public long getBatchTimeout() {
-		return batchTimeout;
+	public String[] getHeaders() {
+		return headers;
 	}
 
-	public void setBatchTimeout(long batchTimeout) {
-		this.batchTimeout = batchTimeout;
+	public Mode getMode() {
+		return this.mode;
 	}
 
-	public int getRequiredAcks() {
-		return requiredAcks;
+	public int getOffsetUpdateTimeWindow() {
+		return this.offsetUpdateTimeWindow;
 	}
 
-	public void setRequiredAcks(int requiredAcks) {
-		this.requiredAcks = requiredAcks;
+	public int getOffsetUpdateCount() {
+		return this.offsetUpdateCount;
 	}
 
-	public int getReplicationFactor() {
-		return replicationFactor;
+	public int getOffsetUpdateShutdownTimeout() {
+		return this.offsetUpdateShutdownTimeout;
 	}
 
-	public void setReplicationFactor(int replicationFactor) {
-		this.replicationFactor = replicationFactor;
+	public void setZkNodes(String[] zkNodes) {
+		this.zkNodes = zkNodes;
 	}
 
-	public int getConcurrency() {
-		return concurrency;
+	public void setDefaultZkPort(String defaultZkPort) {
+		this.defaultZkPort = defaultZkPort;
 	}
 
-	public void setConcurrency(int concurrency) {
-		this.concurrency = concurrency;
+	public void setBrokers(String[] brokers) {
+		this.brokers = brokers;
 	}
 
-	public String getCompressionCodec() {
-		return compressionCodec;
+	public void setDefaultBrokerPort(String defaultBrokerPort) {
+		this.defaultBrokerPort = defaultBrokerPort;
 	}
 
-	public void setCompressionCodec(String compressionCodec) {
-		this.compressionCodec = compressionCodec;
+
+	public void setHeaders(String[] headers) {
+		this.headers = headers;
 	}
 
-	public boolean isAutoCommitEnabled() {
-		return autoCommitEnabled;
+	public void setMode(KafkaMessageChannelBinder.Mode mode) {
+		this.mode = mode;
 	}
 
-	public void setAutoCommitEnabled(boolean autoCommitEnabled) {
-		this.autoCommitEnabled = autoCommitEnabled;
+	public void setOffsetUpdateTimeWindow(int offsetUpdateTimeWindow) {
+		this.offsetUpdateTimeWindow = offsetUpdateTimeWindow;
 	}
 
-	public int getFetchSize() {
-		return fetchSize;
+	public void setOffsetUpdateCount(int offsetUpdateCount) {
+		this.offsetUpdateCount = offsetUpdateCount;
 	}
 
-	public void setFetchSize(int fetchSize) {
-		this.fetchSize = fetchSize;
+	public void setOffsetUpdateShutdownTimeout(int offsetUpdateShutdownTimeout) {
+		this.offsetUpdateShutdownTimeout = offsetUpdateShutdownTimeout;
 	}
 
-	public int getMinPartitionCount() {
-		return minPartitionCount;
+	public KafkaMessageChannelBinder.StartOffset getStartOffset() {
+		return startOffset;
 	}
 
-	public void setMinPartitionCount(int minPartitionCount) {
-		this.minPartitionCount = minPartitionCount;
+	public void setStartOffset(KafkaMessageChannelBinder.StartOffset startOffset) {
+		this.startOffset = startOffset;
 	}
 
-	public int getQueueSize() {
-		return queueSize;
+	public boolean isResetOffsets() {
+		return resetOffsets;
 	}
 
-	public void setQueueSize(int queueSize) {
-		this.queueSize = queueSize;
+	public void setResetOffsets(boolean resetOffsets) {
+		this.resetOffsets = resetOffsets;
+	}
+
+	/**
+	 * Converts an array of host values to a comma-separated String.
+	 *
+	 * It will append the default port value, if not already specified.
+	 */
+	private String toConnectionString(String[] hosts, String defaultPort) {
+		String[] fullyFormattedHosts = new String[hosts.length];
+		for (int i = 0; i < hosts.length; i++) {
+			if (hosts[i].contains(":") || StringUtils.isEmpty(defaultPort)) {
+				fullyFormattedHosts[i] = hosts[i];
+			}
+			else {
+				fullyFormattedHosts[i] = hosts[i] + ":" + defaultPort;
+			}
+		}
+		return StringUtils.arrayToCommaDelimitedString(fullyFormattedHosts);
 	}
 }

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

@@ -0,0 +1,127 @@
+/*
+ * 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.config;
+
+import org.springframework.boot.context.properties.ConfigurationProperties;
+
+/**
+ * @author David Turanski
+ * @author Ilayaperumal Gopinathan
+ * @author Marius Bogoevici
+ */
+@ConfigurationProperties(value = "spring.cloud.stream.binder.kafka.default")
+public class KafkaBinderDefaultProperties {
+
+	private int batchSize;
+
+	private long batchTimeout;
+
+	private int requiredAcks;
+
+	private int replicationFactor;
+
+	private int concurrency;
+
+	private String compressionCodec;
+
+	private boolean autoCommitEnabled;
+
+	private int fetchSize;
+
+	private int minPartitionCount;
+
+	private int queueSize;
+
+	public int getBatchSize() {
+		return batchSize;
+	}
+
+	public void setBatchSize(int batchSize) {
+		this.batchSize = batchSize;
+	}
+
+	public long getBatchTimeout() {
+		return batchTimeout;
+	}
+
+	public void setBatchTimeout(long batchTimeout) {
+		this.batchTimeout = batchTimeout;
+	}
+
+	public int getRequiredAcks() {
+		return requiredAcks;
+	}
+
+	public void setRequiredAcks(int requiredAcks) {
+		this.requiredAcks = requiredAcks;
+	}
+
+	public int getReplicationFactor() {
+		return replicationFactor;
+	}
+
+	public void setReplicationFactor(int replicationFactor) {
+		this.replicationFactor = replicationFactor;
+	}
+
+	public int getConcurrency() {
+		return concurrency;
+	}
+
+	public void setConcurrency(int concurrency) {
+		this.concurrency = concurrency;
+	}
+
+	public String getCompressionCodec() {
+		return compressionCodec;
+	}
+
+	public void setCompressionCodec(String compressionCodec) {
+		this.compressionCodec = compressionCodec;
+	}
+
+	public boolean isAutoCommitEnabled() {
+		return autoCommitEnabled;
+	}
+
+	public void setAutoCommitEnabled(boolean autoCommitEnabled) {
+		this.autoCommitEnabled = autoCommitEnabled;
+	}
+
+	public int getFetchSize() {
+		return fetchSize;
+	}
+
+	public void setFetchSize(int fetchSize) {
+		this.fetchSize = fetchSize;
+	}
+
+	public int getMinPartitionCount() {
+		return minPartitionCount;
+	}
+
+	public void setMinPartitionCount(int minPartitionCount) {
+		this.minPartitionCount = minPartitionCount;
+	}
+
+	public int getQueueSize() {
+		return queueSize;
+	}
+
+	public void setQueueSize(int queueSize) {
+		this.queueSize = queueSize;
+	}
+}

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

@@ -1,201 +0,0 @@
-/*
- * Copyright 2015-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.config;
-
-import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.boot.autoconfigure.PropertyPlaceholderAutoConfiguration;
-import org.springframework.boot.context.properties.ConfigurationProperties;
-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.integration.codec.Codec;
-import org.springframework.integration.kafka.support.ProducerListener;
-import org.springframework.integration.kafka.support.ZookeeperConnect;
-import org.springframework.util.ObjectUtils;
-import org.springframework.util.StringUtils;
-
-/**
- * @author David Turanski
- * @author Marius Bogoevici
- * @author Mark Fisher
- * @author Soby Chacko
- */
-@Configuration
-@Import({KryoCodecAutoConfiguration.class, PropertyPlaceholderAutoConfiguration.class})
-@ConfigurationProperties(prefix = "spring.cloud.stream.binder.kafka")
-public class KafkaMessageChannelBinderConfiguration {
-
-	private String[] zkNodes;
-
-	private String defaultZkPort;
-
-	private String[] brokers;
-
-	private String defaultBrokerPort;
-
-	private String[] headers;
-
-	private KafkaMessageChannelBinder.Mode mode;
-
-	private int offsetUpdateTimeWindow;
-
-	private int offsetUpdateCount;
-
-	private int offsetUpdateShutdownTimeout;
-
-	private boolean resetOffsets = false;
-
-	private KafkaMessageChannelBinder.StartOffset startOffset;
-
-	@Autowired
-	private Codec codec;
-
-	@Autowired
-	private KafkaBinderConfigurationProperties kafkaBinderConfigurationProperties;
-
-	@Autowired
-	private ProducerListener producerListener;
-
-	@Bean
-	ZookeeperConnect zookeeperConnect() {
-		ZookeeperConnect zookeeperConnect = new ZookeeperConnect();
-		zookeeperConnect.setZkConnect(getZkConnectionString());
-		return zookeeperConnect;
-	}
-
-	@Bean
-	KafkaMessageChannelBinder kafkaMessageChannelBinder() {
-		KafkaMessageChannelBinder kafkaMessageChannelBinder = ObjectUtils.isEmpty(headers) ?
-				new KafkaMessageChannelBinder(zookeeperConnect(), getKafkaConnectionString(), getZkConnectionString())
-				: new KafkaMessageChannelBinder(zookeeperConnect(), getKafkaConnectionString(), getZkConnectionString(),
-						headers);
-		kafkaMessageChannelBinder.setCodec(codec);
-		kafkaMessageChannelBinder.setMode(mode);
-		kafkaMessageChannelBinder.setOffsetUpdateTimeWindow(offsetUpdateTimeWindow);
-		kafkaMessageChannelBinder.setOffsetUpdateCount(offsetUpdateCount);
-		kafkaMessageChannelBinder.setOffsetUpdateShutdownTimeout(offsetUpdateShutdownTimeout);
-
-		kafkaMessageChannelBinder.setDefaultAutoCommitEnabled(kafkaBinderConfigurationProperties.isAutoCommitEnabled());
-		kafkaMessageChannelBinder.setDefaultBatchSize(kafkaBinderConfigurationProperties.getBatchSize());
-		kafkaMessageChannelBinder.setDefaultBatchTimeout(kafkaBinderConfigurationProperties.getBatchTimeout());
-		kafkaMessageChannelBinder.setDefaultCompressionCodec(kafkaBinderConfigurationProperties
-				.getCompressionCodec());
-		kafkaMessageChannelBinder.setDefaultConcurrency(kafkaBinderConfigurationProperties.getConcurrency());
-		kafkaMessageChannelBinder.setDefaultFetchSize(kafkaBinderConfigurationProperties.getFetchSize());
-		kafkaMessageChannelBinder.setDefaultMinPartitionCount(kafkaBinderConfigurationProperties
-				.getMinPartitionCount());
-		kafkaMessageChannelBinder.setDefaultQueueSize(kafkaBinderConfigurationProperties.getQueueSize());
-		kafkaMessageChannelBinder.setDefaultReplicationFactor(kafkaBinderConfigurationProperties
-				.getReplicationFactor());
-		kafkaMessageChannelBinder.setDefaultRequiredAcks(kafkaBinderConfigurationProperties.getRequiredAcks());
-
-		kafkaMessageChannelBinder.setResetOffsets(resetOffsets);
-		kafkaMessageChannelBinder.setStartOffset(startOffset);
-
-		kafkaMessageChannelBinder.setProducerListener(producerListener);
-
-		return kafkaMessageChannelBinder;
-	}
-
-	public void setZkNodes(String[] zkNodes) {
-		this.zkNodes = zkNodes;
-	}
-
-	public void setDefaultZkPort(String defaultZkPort) {
-		this.defaultZkPort = defaultZkPort;
-	}
-
-	public void setBrokers(String[] brokers) {
-		this.brokers = brokers;
-	}
-
-	public void setDefaultBrokerPort(String defaultBrokerPort) {
-		this.defaultBrokerPort = defaultBrokerPort;
-	}
-
-	public String[] getHeaders() {
-		return headers;
-	}
-
-	public void setHeaders(String[] headers) {
-		this.headers = headers;
-	}
-
-	public void setMode(KafkaMessageChannelBinder.Mode mode) {
-		this.mode = mode;
-	}
-
-	public void setOffsetUpdateTimeWindow(int offsetUpdateTimeWindow) {
-		this.offsetUpdateTimeWindow = offsetUpdateTimeWindow;
-	}
-
-	public void setOffsetUpdateCount(int offsetUpdateCount) {
-		this.offsetUpdateCount = offsetUpdateCount;
-	}
-
-	public void setOffsetUpdateShutdownTimeout(int offsetUpdateShutdownTimeout) {
-		this.offsetUpdateShutdownTimeout = offsetUpdateShutdownTimeout;
-	}
-
-	public void setCodec(Codec codec) {
-		this.codec = codec;
-	}
-
-	public String getZkConnectionString() {
-		return toConnectionString(this.zkNodes, this.defaultZkPort);
-	}
-
-	public String getKafkaConnectionString() {
-		return toConnectionString(this.brokers, this.defaultBrokerPort);
-	}
-
-	public KafkaMessageChannelBinder.StartOffset getStartOffset() {
-		return startOffset;
-	}
-
-	public void setStartOffset(KafkaMessageChannelBinder.StartOffset startOffset) {
-		this.startOffset = startOffset;
-	}
-
-	public boolean isResetOffsets() {
-		return resetOffsets;
-	}
-
-	public void setResetOffsets(boolean resetOffsets) {
-		this.resetOffsets = resetOffsets;
-	}
-
-	/**
-	 * Converts an array of host values to a comma-separated String.
-	 *
-	 * It will append the default port value, if not already specified.
-	 */
-	private String toConnectionString(String[] hosts, String defaultPort) {
-		String[] fullyFormattedHosts = new String[hosts.length];
-		for (int i = 0; i < hosts.length; i++) {
-			if (hosts[i].contains(":") || StringUtils.isEmpty(defaultPort)) {
-				fullyFormattedHosts[i] = hosts[i];
-			}
-			else {
-				fullyFormattedHosts[i] = hosts[i] + ":" + defaultPort;
-			}
-		}
-		return StringUtils.arrayToCommaDelimitedString(fullyFormattedHosts);
-	}
-}

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

@@ -5,7 +5,7 @@
  * 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
+ *      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,
@@ -16,32 +16,87 @@
 
 package org.springframework.cloud.stream.binder.kafka.config;
 
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.autoconfigure.PropertyPlaceholderAutoConfiguration;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
-import org.springframework.boot.autoconfigure.test.ImportAutoConfiguration;
 import org.springframework.boot.context.properties.EnableConfigurationProperties;
 import org.springframework.cloud.stream.binder.Binder;
+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;
+import org.springframework.integration.kafka.support.ZookeeperConnect;
+import org.springframework.util.ObjectUtils;
 
 /**
- * Bind to Kafka services.
- *
+ * @author David Turanski
  * @author Marius Bogoevici
  * @author Soby Chacko
+ * @author Mark Fisher
+ * @author Ilayaperumal Gopinathan
  */
 @Configuration
 @ConditionalOnMissingBean(Binder.class)
-@EnableConfigurationProperties({KafkaBinderConfigurationProperties.class, KafkaMessageChannelBinderConfiguration.class})
-@ImportAutoConfiguration(KafkaMessageChannelBinderConfiguration.class)
+@Import({KryoCodecAutoConfiguration.class, PropertyPlaceholderAutoConfiguration.class})
+@EnableConfigurationProperties({KafkaBinderConfigurationProperties.class, KafkaBinderDefaultProperties.class})
+@PropertySource("classpath:/META-INF/spring-cloud-stream/kafka-binder.properties")
 public class KafkaServiceAutoConfiguration {
 
-	@Configuration
-	@PropertySource("classpath:/META-INF/spring-cloud-stream/kafka-binder.properties")
-	public static class DefaultProperties {
+	@Autowired
+	private Codec codec;
 
+	@Autowired
+	private KafkaBinderDefaultProperties kafkaBinderDefaultProperties;
+
+	@Autowired
+	private KafkaBinderConfigurationProperties kafkaBinderConfigurationProperties;
+
+	@Autowired
+	private ProducerListener producerListener;
+
+	@Bean
+	ZookeeperConnect zookeeperConnect() {
+		ZookeeperConnect zookeeperConnect = new ZookeeperConnect();
+		zookeeperConnect.setZkConnect(kafkaBinderConfigurationProperties.getZkConnectionString());
+		return zookeeperConnect;
+	}
+
+	@Bean
+	KafkaMessageChannelBinder kafkaMessageChannelBinder() {
+		String[] headers = kafkaBinderConfigurationProperties.getHeaders();
+		String kafkaConnectionString = kafkaBinderConfigurationProperties.getKafkaConnectionString();
+		String zkConnectionString = kafkaBinderConfigurationProperties.getZkConnectionString();
+		KafkaMessageChannelBinder kafkaMessageChannelBinder = ObjectUtils.isEmpty(headers) ?
+				new KafkaMessageChannelBinder(zookeeperConnect(), kafkaConnectionString, zkConnectionString)
+				: new KafkaMessageChannelBinder(zookeeperConnect(), kafkaConnectionString, zkConnectionString,
+						headers);
+		kafkaMessageChannelBinder.setCodec(codec);
+		kafkaMessageChannelBinder.setMode(kafkaBinderConfigurationProperties.getMode());
+		kafkaMessageChannelBinder.setOffsetUpdateTimeWindow(kafkaBinderConfigurationProperties.getOffsetUpdateTimeWindow());
+		kafkaMessageChannelBinder.setOffsetUpdateCount(kafkaBinderConfigurationProperties.getOffsetUpdateCount());
+		kafkaMessageChannelBinder.setOffsetUpdateShutdownTimeout(kafkaBinderConfigurationProperties.getOffsetUpdateShutdownTimeout());
+
+		kafkaMessageChannelBinder.setResetOffsets(kafkaBinderConfigurationProperties.isResetOffsets());
+		kafkaMessageChannelBinder.setStartOffset(kafkaBinderConfigurationProperties.getStartOffset());
+
+		kafkaMessageChannelBinder.setDefaultAutoCommitEnabled(kafkaBinderDefaultProperties.isAutoCommitEnabled());
+		kafkaMessageChannelBinder.setDefaultBatchSize(kafkaBinderDefaultProperties.getBatchSize());
+		kafkaMessageChannelBinder.setDefaultBatchTimeout(kafkaBinderDefaultProperties.getBatchTimeout());
+		kafkaMessageChannelBinder.setDefaultCompressionCodec(kafkaBinderDefaultProperties.getCompressionCodec());
+		kafkaMessageChannelBinder.setDefaultConcurrency(kafkaBinderDefaultProperties.getConcurrency());
+		kafkaMessageChannelBinder.setDefaultFetchSize(kafkaBinderDefaultProperties.getFetchSize());
+		kafkaMessageChannelBinder.setDefaultMinPartitionCount(kafkaBinderDefaultProperties.getMinPartitionCount());
+		kafkaMessageChannelBinder.setDefaultQueueSize(kafkaBinderDefaultProperties.getQueueSize());
+		kafkaMessageChannelBinder.setDefaultReplicationFactor(kafkaBinderDefaultProperties.getReplicationFactor());
+		kafkaMessageChannelBinder.setDefaultRequiredAcks(kafkaBinderDefaultProperties.getRequiredAcks());
+
+		kafkaMessageChannelBinder.setProducerListener(producerListener);
+		return kafkaMessageChannelBinder;
 	}
 
 	@Bean

spring-cloud-stream-samples/multibinder-differentsystems/src/test/java/multibinder/TwoKafkaBindersApplicationTest.java

@@ -16,6 +16,10 @@
 
 package multibinder;
 
+import static org.hamcrest.Matchers.contains;
+import static org.hamcrest.Matchers.hasSize;
+
+import java.util.List;
 import java.util.UUID;
 
 import org.hamcrest.CoreMatchers;
@@ -26,13 +30,17 @@ import org.junit.ClassRule;
 import org.junit.Test;
 import org.junit.runner.RunWith;
 
+import org.springframework.beans.DirectFieldAccessor;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.boot.test.SpringApplicationConfiguration;
 import org.springframework.cloud.stream.binder.BinderFactory;
+import org.springframework.cloud.stream.binder.kafka.KafkaMessageChannelBinder;
 import org.springframework.cloud.stream.test.junit.kafka.KafkaTestSupport;
 import org.springframework.cloud.stream.test.junit.redis.RedisTestSupport;
 import org.springframework.integration.channel.DirectChannel;
 import org.springframework.integration.channel.QueueChannel;
+import org.springframework.integration.kafka.core.BrokerAddress;
+import org.springframework.integration.kafka.core.Configuration;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.MessageChannel;
 import org.springframework.messaging.support.MessageBuilder;
@@ -47,10 +55,10 @@ import org.springframework.test.context.web.WebAppConfiguration;
 public class TwoKafkaBindersApplicationTest {
 
 	@ClassRule
-	public static KafkaTestSupport kafkaTestSupport1 = new KafkaTestSupport();
+	public static KafkaTestSupport kafkaTestSupport1 = new KafkaTestSupport(true);
 
 	@ClassRule
-	public static KafkaTestSupport kafkaTestSupport2 = new KafkaTestSupport();
+	public static KafkaTestSupport kafkaTestSupport2 = new KafkaTestSupport(true);
 
 	@ClassRule
 	public static RedisTestSupport redisTestSupport = new RedisTestSupport();
@@ -68,6 +76,18 @@ public class TwoKafkaBindersApplicationTest {
 
 	@Test
 	public void contextLoads() {
+		KafkaMessageChannelBinder kafka1 = (KafkaMessageChannelBinder) binderFactory.getBinder("kafka1");
+		DirectFieldAccessor directFieldAccessor = new DirectFieldAccessor(kafka1.getConnectionFactory());
+		Configuration configuration = (Configuration) directFieldAccessor.getPropertyValue("configuration");
+		List<BrokerAddress> brokerAddresses = configuration.getBrokerAddresses();
+		Assert.assertThat(brokerAddresses, hasSize(1));
+		Assert.assertThat(brokerAddresses, contains(BrokerAddress.fromAddress(kafkaTestSupport1.getBrokerAddress())));
+		KafkaMessageChannelBinder kafka2 = (KafkaMessageChannelBinder) binderFactory.getBinder("kafka2");
+		DirectFieldAccessor directFieldAccessor2 = new DirectFieldAccessor(kafka2.getConnectionFactory());
+		Configuration configuration2 = (Configuration) directFieldAccessor2.getPropertyValue("configuration");
+		List<BrokerAddress> brokerAddresses2 = configuration2.getBrokerAddresses();
+		Assert.assertThat(brokerAddresses2, hasSize(1));
+		Assert.assertThat(brokerAddresses2, contains(BrokerAddress.fromAddress(kafkaTestSupport2.getBrokerAddress())));
 	}
 
 	@Test

spring-cloud-stream-test-support-internal/src/main/java/org/springframework/cloud/stream/test/junit/kafka/KafkaTestSupport.java

@@ -19,13 +19,6 @@ package org.springframework.cloud.stream.test.junit.kafka;
 
 import java.util.Properties;
 
-import kafka.server.KafkaConfig;
-import kafka.server.KafkaServer;
-import kafka.utils.SystemTime$;
-import kafka.utils.TestUtils;
-import kafka.utils.Utils;
-import kafka.utils.ZKStringSerializer$;
-import kafka.utils.ZkUtils;
 import org.I0Itec.zkclient.ZkClient;
 import org.I0Itec.zkclient.exception.ZkInterruptedException;
 import org.junit.Rule;
@@ -35,6 +28,14 @@ import org.slf4j.LoggerFactory;
 import org.springframework.cloud.stream.test.junit.AbstractExternalResourceTestSupport;
 import org.springframework.util.SocketUtils;
 
+import kafka.server.KafkaConfig;
+import kafka.server.KafkaServer;
+import kafka.utils.SystemTime$;
+import kafka.utils.TestUtils;
+import kafka.utils.Utils;
+import kafka.utils.ZKStringSerializer$;
+import kafka.utils.ZkUtils;
+
 
 /**
  * JUnit {@link Rule} that starts an embedded Kafka server (with an associated Zookeeper)
@@ -49,7 +50,7 @@ public class KafkaTestSupport extends AbstractExternalResourceTestSupport<String
 
 	private static final String SCS_KAFKA_TEST_EMBEDDED = "SCS_KAFKA_TEST_EMBEDDED";
 
-	public static final boolean embedded;
+	public static final boolean defaultEmbedded;
 
 	private static final String DEFAULT_ZOOKEEPER_CONNECT = "localhost:2181";
 
@@ -61,6 +62,8 @@ public class KafkaTestSupport extends AbstractExternalResourceTestSupport<String
 
 	private KafkaServer kafkaServer;
 
+	public final boolean embedded;
+
 	private final Properties brokerConfig = TestUtils.createBrokerConfig(0, TestUtils.choosePort(), false);
 
 	// caches previous failures to reach the external server - preventing repeated retries
@@ -68,12 +71,17 @@ public class KafkaTestSupport extends AbstractExternalResourceTestSupport<String
 
 	static {
 		// check if either the environment or Java property is set to use embedded tests
-		embedded = "true".equals(System.getenv(SCS_KAFKA_TEST_EMBEDDED)) || "true".equals(System.getProperty(SCS_KAFKA_TEST_EMBEDDED));
-		log.info(String.format("Testing with %s Kafka broker", embedded ? "embedded" : "external"));
+		defaultEmbedded = "true".equals(System.getenv(SCS_KAFKA_TEST_EMBEDDED)) || "true".equals(System.getProperty(SCS_KAFKA_TEST_EMBEDDED));
 	}
 
 	public KafkaTestSupport() {
+		this(defaultEmbedded);
+	}
+
+	public KafkaTestSupport(boolean embedded) {
 		super("KAFKA");
+		this.embedded = embedded;
+		log.info(String.format("Testing with %s Kafka broker", embedded ? "embedded" : "external"));
 	}
 
 	public KafkaServer getKafkaServer() {