diff --git a/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/config/KafkaBinderConfigurationProperties.java b/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/config/KafkaBinderConfigurationProperties.java index 9b6c7ee29..99f63f00e 100644 --- a/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/config/KafkaBinderConfigurationProperties.java +++ b/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); } } diff --git a/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/config/KafkaBinderDefaultProperties.java b/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/config/KafkaBinderDefaultProperties.java new file mode 100644 index 000000000..c70c5344f --- /dev/null +++ b/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; + } +} diff --git a/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/config/KafkaMessageChannelBinderConfiguration.java b/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/config/KafkaMessageChannelBinderConfiguration.java deleted file mode 100644 index 5ce84b513..000000000 --- a/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/config/KafkaMessageChannelBinderConfiguration.java +++ /dev/null @@ -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); - } -} diff --git a/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/config/KafkaServiceAutoConfiguration.java b/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/config/KafkaServiceAutoConfiguration.java index 3318b106b..fb94c1858 100644 --- a/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/config/KafkaServiceAutoConfiguration.java +++ b/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 diff --git a/spring-cloud-stream-samples/multibinder-differentsystems/src/test/java/multibinder/TwoKafkaBindersApplicationTest.java b/spring-cloud-stream-samples/multibinder-differentsystems/src/test/java/multibinder/TwoKafkaBindersApplicationTest.java index 918e97f87..6b16ce0e9 100644 --- a/spring-cloud-stream-samples/multibinder-differentsystems/src/test/java/multibinder/TwoKafkaBindersApplicationTest.java +++ b/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 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 brokerAddresses2 = configuration2.getBrokerAddresses(); + Assert.assertThat(brokerAddresses2, hasSize(1)); + Assert.assertThat(brokerAddresses2, contains(BrokerAddress.fromAddress(kafkaTestSupport2.getBrokerAddress()))); } @Test diff --git a/spring-cloud-stream-test-support-internal/src/main/java/org/springframework/cloud/stream/test/junit/kafka/KafkaTestSupport.java b/spring-cloud-stream-test-support-internal/src/main/java/org/springframework/cloud/stream/test/junit/kafka/KafkaTestSupport.java index e86dc1454..fb15767d5 100644 --- a/spring-cloud-stream-test-support-internal/src/main/java/org/springframework/cloud/stream/test/junit/kafka/KafkaTestSupport.java +++ b/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