From d85677f61e7e02b8d232140e2ae55488b38deff5 Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Fri, 23 Jan 2015 16:31:59 +0200 Subject: [PATCH] INTEXT-129: Namespace support for `KafkaMDCA` JIRA: https://jira.spring.io/browse/INTEXT-129 INTEXT-129: Polishing and Docs Doc Polishing --- ...afkaMessageDrivenChannelAdapterParser.java | 99 ++++++++++ .../config/xml/KafkaNamespaceHandler.java | 17 +- .../core/BrokerAddressListConfiguration.java | 5 +- .../KafkaMessageListenerContainer.java | 15 +- .../xml/spring-integration-kafka-1.0.xsd | 186 ++++++++++++++++-- ...rivenChannelAdapterParserTests-context.xml | 69 +++++++ ...essageDrivenChannelAdapterParserTests.java | 102 ++++++++++ ...apterWithXmlConfigurationTests-context.xml | 65 +++--- ...annelAdapterWithXmlConfigurationTests.java | 2 - .../DefaultConnectionFactoryTests.java | 14 +- .../listener/DefaultConnectionTests.java | 8 +- .../integration/kafka/rule/KafkaEmbedded.java | 43 ++-- .../integration/kafka/rule/KafkaRule.java | 2 +- .../integration/kafka/rule/KafkaRunning.java | 26 ++- 14 files changed, 545 insertions(+), 108 deletions(-) create mode 100644 spring-integration-kafka/src/main/java/org/springframework/integration/kafka/config/xml/KafkaMessageDrivenChannelAdapterParser.java create mode 100644 spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/KafkaMessageDrivenChannelAdapterParserTests-context.xml create mode 100644 spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/KafkaMessageDrivenChannelAdapterParserTests.java diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/config/xml/KafkaMessageDrivenChannelAdapterParser.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/config/xml/KafkaMessageDrivenChannelAdapterParser.java new file mode 100644 index 0000000000..8def6edf5c --- /dev/null +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/config/xml/KafkaMessageDrivenChannelAdapterParser.java @@ -0,0 +1,99 @@ +/* + * Copyright 2015 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.integration.kafka.config.xml; + +import org.w3c.dom.Element; + +import org.springframework.beans.factory.support.AbstractBeanDefinition; +import org.springframework.beans.factory.support.BeanDefinitionBuilder; +import org.springframework.beans.factory.xml.AbstractSingleBeanDefinitionParser; +import org.springframework.beans.factory.xml.ParserContext; +import org.springframework.integration.config.xml.AbstractChannelAdapterParser; +import org.springframework.integration.config.xml.IntegrationNamespaceUtils; +import org.springframework.integration.kafka.inbound.KafkaMessageDrivenChannelAdapter; +import org.springframework.integration.kafka.listener.KafkaMessageListenerContainer; +import org.springframework.util.StringUtils; + +/** + * @author Artem Bilan. + */ +public class KafkaMessageDrivenChannelAdapterParser extends AbstractChannelAdapterParser { + + @Override + protected AbstractBeanDefinition doParse(Element element, ParserContext parserContext, String channelName) { + BeanDefinitionBuilder builder = + BeanDefinitionBuilder.genericBeanDefinition(KafkaMessageDrivenChannelAdapter.class); + + String container = element.getAttribute("listener-container"); + String connectionFactory = element.getAttribute("connection-factory"); + String topics = element.getAttribute("topics"); + String offsetManager = element.getAttribute("offset-manager"); + String errorHandler = element.getAttribute("error-handler"); + String taskExecutor = element.getAttribute("task-executor"); + String concurrency = element.getAttribute("concurrency"); + String maxFetch = element.getAttribute("max-fetch"); + String queueSize = element.getAttribute("queue-size"); + + if (StringUtils.hasText(container) && + (StringUtils.hasText(connectionFactory) || StringUtils.hasText(topics) + || StringUtils.hasText(offsetManager) || StringUtils.hasText(errorHandler) + || StringUtils.hasText(taskExecutor) || StringUtils.hasText(concurrency) + || StringUtils.hasText(maxFetch) || StringUtils.hasText(queueSize))) { + parserContext.getReaderContext().error("The 'listener-container' is mutually exclusive with " + + "'connection-factory', 'topics', 'offset-manager', 'error-handler', 'task-executor', " + + "'concurrency', 'max-fetch' and 'queue-size'.", element); + } + + if (StringUtils.hasText(container)) { + builder.addConstructorArgReference(container); + } + else { + if (!StringUtils.hasText(connectionFactory)) { + parserContext.getReaderContext().error("The 'connection-factory' attribute is required when " + + "'listener-container' isn't provided.", element); + } + if (!StringUtils.hasText(topics)) { + parserContext.getReaderContext().error("The 'topics' attribute is required when " + + "'listener-container' isn't provided.", element); + } + + BeanDefinitionBuilder containerBuilder = + BeanDefinitionBuilder.genericBeanDefinition(KafkaMessageListenerContainer.class); + + containerBuilder.addConstructorArgReference(connectionFactory) + .addConstructorArgValue(topics); + IntegrationNamespaceUtils.setReferenceIfAttributeDefined(containerBuilder, element, "offset-manager"); + IntegrationNamespaceUtils.setReferenceIfAttributeDefined(containerBuilder, element, "error-handler"); + IntegrationNamespaceUtils.setReferenceIfAttributeDefined( + containerBuilder, element, "task-executor", "fetchTaskExecutor"); + IntegrationNamespaceUtils.setValueIfAttributeDefined(containerBuilder, element, "concurrency"); + IntegrationNamespaceUtils.setValueIfAttributeDefined(containerBuilder, element, "max-fetch"); + IntegrationNamespaceUtils.setValueIfAttributeDefined(containerBuilder, element, "queue-size"); + + builder.addConstructorArgValue(containerBuilder.getBeanDefinition()); + } + + builder.addPropertyReference("outputChannel", channelName); + IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "send-timeout"); + IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "error-channel"); + IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "key-decoder"); + IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "payload-decoder"); + + return builder.getBeanDefinition(); + } + +} diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/config/xml/KafkaNamespaceHandler.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/config/xml/KafkaNamespaceHandler.java index 3cc177b757..09e10d3b61 100644 --- a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/config/xml/KafkaNamespaceHandler.java +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/config/xml/KafkaNamespaceHandler.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2013 the original author or authors. + * Copyright 2002-2015 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. @@ -13,6 +13,7 @@ * See the License for the specific language governing permissions and * limitations under the License. */ + package org.springframework.integration.kafka.config.xml; import org.springframework.integration.config.xml.AbstractIntegrationNamespaceHandler; @@ -25,15 +26,15 @@ import org.springframework.integration.config.xml.AbstractIntegrationNamespaceHa * */ public class KafkaNamespaceHandler extends AbstractIntegrationNamespaceHandler { - /* (non-Javadoc) - * @see org.springframework.beans.factory.xml.NamespaceHandler#init() - */ + @Override public void init() { registerBeanDefinitionParser("zookeeper-connect", new ZookeeperConnectParser()); - registerBeanDefinitionParser("inbound-channel-adapter", new KafkaInboundChannelAdapterParser()); - registerBeanDefinitionParser("outbound-channel-adapter", new KafkaOutboundChannelAdapterParser()); - registerBeanDefinitionParser("producer-context", new KafkaProducerContextParser()); - registerBeanDefinitionParser("consumer-context", new KafkaConsumerContextParser()); + registerBeanDefinitionParser("inbound-channel-adapter", new KafkaInboundChannelAdapterParser()); + registerBeanDefinitionParser("outbound-channel-adapter", new KafkaOutboundChannelAdapterParser()); + registerBeanDefinitionParser("producer-context", new KafkaProducerContextParser()); + registerBeanDefinitionParser("consumer-context", new KafkaConsumerContextParser()); + registerBeanDefinitionParser("message-driven-channel-adapter", new KafkaMessageDrivenChannelAdapterParser()); } + } diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/core/BrokerAddressListConfiguration.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/core/BrokerAddressListConfiguration.java index e64e007d63..d013b09d75 100644 --- a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/core/BrokerAddressListConfiguration.java +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/core/BrokerAddressListConfiguration.java @@ -17,6 +17,7 @@ package org.springframework.integration.kafka.core; +import java.util.Arrays; import java.util.List; /** @@ -28,8 +29,8 @@ public class BrokerAddressListConfiguration extends AbstractConfiguration { private final List brokerAddresses; - public BrokerAddressListConfiguration(List brokerAddresses) { - this.brokerAddresses = brokerAddresses; + public BrokerAddressListConfiguration(BrokerAddress... brokerAddresses) { + this.brokerAddresses = Arrays.asList(brokerAddresses); } @Override diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/listener/KafkaMessageListenerContainer.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/listener/KafkaMessageListenerContainer.java index 9337f8d73d..3dfab05dcc 100644 --- a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/listener/KafkaMessageListenerContainer.java +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/listener/KafkaMessageListenerContainer.java @@ -87,9 +87,9 @@ public class KafkaMessageListenerContainer implements SmartLifecycle { private final KafkaTemplate kafkaTemplate; - private Partition[] partitions; + private final String[] topics; - private String[] topics; + private Partition[] partitions; public boolean autoStartup = true; @@ -123,6 +123,7 @@ public class KafkaMessageListenerContainer implements SmartLifecycle { Assert.noNullElements(partitions, "The list of partitions cannot contain null elements"); this.kafkaTemplate = new KafkaTemplate(connectionFactory); this.partitions = partitions; + this.topics = null; } public KafkaMessageListenerContainer(final ConnectionFactory connectionFactory, String... topics) { @@ -133,11 +134,6 @@ public class KafkaMessageListenerContainer implements SmartLifecycle { this.topics = topics; } - private static Partition[] getPartitionsForTopics(final ConnectionFactory connectionFactory, String[] topics) { - MutableList partitionList = flatCollect(topics, new GetPartitionsForTopic(connectionFactory)); - return partitionList.toArray(new Partition[partitionList.size()]); - } - public OffsetManager getOffsetManager() { return offsetManager; } @@ -297,6 +293,11 @@ public class KafkaMessageListenerContainer implements SmartLifecycle { return 0; } + private static Partition[] getPartitionsForTopics(final ConnectionFactory connectionFactory, String[] topics) { + MutableList partitionList = flatCollect(topics, new GetPartitionsForTopic(connectionFactory)); + return partitionList.toArray(new Partition[partitionList.size()]); + } + /** * Fetches data from Kafka for a group of partitions, located on the same broker. */ diff --git a/spring-integration-kafka/src/main/resources/org/springframework/integration/config/xml/spring-integration-kafka-1.0.xsd b/spring-integration-kafka/src/main/resources/org/springframework/integration/config/xml/spring-integration-kafka-1.0.xsd index 8d18422366..362cc66c97 100644 --- a/spring-integration-kafka/src/main/resources/org/springframework/integration/config/xml/spring-integration-kafka-1.0.xsd +++ b/spring-integration-kafka/src/main/resources/org/springframework/integration/config/xml/spring-integration-kafka-1.0.xsd @@ -382,18 +382,10 @@ - + + Maximum amount of time in milliseconds to wait when sending a message to the channel + if such channel may block. For example, a Queue Channel can block until space is available + if its maximum capacity has been reached. @@ -492,4 +484,174 @@ + + + + The definition for the Spring Integration Kafka + KafkaMessageDrivenChannelAdapter. + + + + + + + + Maximum amount of time in milliseconds to wait when sending a message to the channel + if such channel may block. For example, a Queue Channel can block until space is available + if its maximum capacity has been reached. + + + + + + + Message Channel to which error Messages should be sent. + + + + + + + + + + + + A 'kafka.serializer.Decoder' bean reference for decoding the 'key' + of the received the 'kafka.message.Message'. Defaults to 'kafka.serializer.DefaultDecoder' + without 'kafka.utils.VerifiableProperties'. + + + + + + + + + + + + A 'kafka.serializer.Decoder' bean reference for decoding the 'payload' + of the received the 'kafka.message.Message'. Defaults to 'kafka.serializer.DefaultDecoder' + without 'kafka.utils.VerifiableProperties'. + + + + + + + + + + + + A 'org.springframework.integration.kafka.listener.KafkaMessageListenerContainer' bean reference. + Mutually exclusive with 'connection-factory', 'topics', 'offset-manager', 'error-handler', + 'task-executor', 'concurrency', 'max-fetch', 'queue-size'. + + + + + + + + + + + + A 'org.springframework.integration.kafka.core.ConnectionFactory' bean reference. + Mutually exclusive with 'listener-container'. + + + + + + + + + + + + The comma-separated Kafka topics to listen to. + Mutually exclusive with 'listener-container'. + + + + + + + A 'org.springframework.integration.kafka.listener.OffsetManager' bean reference. + Defaults to 'org.springframework.integration.kafka.listener.MetadataStoreOffsetManager'. + Mutually exclusive with 'listener-container'. + + + + + + + + + + + + A 'org.springframework.integration.kafka.listener.ErrorHandler' bean reference. + Used from 'org.springframework.integration.kafka.listener.KafkaMessageListenerContainer' + to handle errors from 'kafka.message.Message' decoding and when the 'error-channel' + isn't provided to catch message producing errors. + Defaults to 'org.springframework.integration.kafka.listener.LoggingErrorHandler. + Mutually exclusive with 'listener-container'. + + + + + + + + + + + + A 'java.util.concurrent.Executor' bean reference to fetch messages from Kafka. + Defaults to the 'ThreadPoolExecutor' with the pool based on the number of 'partitionsByBroker'. + Mutually exclusive with 'listener-container'. + + + + + + + + + + + + The number of internal concurrent + 'org.springframework.integration.kafka.listener.QueueingMessageListenerInvoker'. + Defaults to '1'. + Mutually exclusive with 'listener-container'. + + + + + + + The maximum amount of data (in bytes) that pollers will fetch in one round. + Defaults to '300 * 1024'. + Mutually exclusive with 'listener-container'. + + + + + + + The maximum number of messages that are buffered by each concurrent 'MessageListener' runner. + Increasing the value may increase throughput, but also increases the memory consumption. + Defaults to '1024'. + Mutually exclusive with 'listener-container'. + + + + + + diff --git a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/KafkaMessageDrivenChannelAdapterParserTests-context.xml b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/KafkaMessageDrivenChannelAdapterParserTests-context.xml new file mode 100644 index 0000000000..da0bcb0dd2 --- /dev/null +++ b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/KafkaMessageDrivenChannelAdapterParserTests-context.xml @@ -0,0 +1,69 @@ + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + diff --git a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/KafkaMessageDrivenChannelAdapterParserTests.java b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/KafkaMessageDrivenChannelAdapterParserTests.java new file mode 100644 index 0000000000..3063631e2d --- /dev/null +++ b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/KafkaMessageDrivenChannelAdapterParserTests.java @@ -0,0 +1,102 @@ +/* + * Copyright 2015 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.integration.kafka.config.xml; + +import static org.junit.Assert.assertArrayEquals; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertSame; + +import java.util.concurrent.Executor; + +import org.junit.Test; +import org.junit.runner.RunWith; + +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.beans.factory.annotation.Qualifier; +import org.springframework.integration.channel.NullChannel; +import org.springframework.integration.channel.PublishSubscribeChannel; +import org.springframework.integration.kafka.core.ConnectionFactory; +import org.springframework.integration.kafka.inbound.KafkaMessageDrivenChannelAdapter; +import org.springframework.integration.kafka.listener.ErrorHandler; +import org.springframework.integration.kafka.listener.KafkaMessageListenerContainer; +import org.springframework.integration.kafka.listener.OffsetManager; +import org.springframework.integration.test.util.TestUtils; +import org.springframework.messaging.PollableChannel; +import org.springframework.test.context.ContextConfiguration; +import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; + +import kafka.serializer.Decoder; + +/** + * @author Artem Bilan. + */ +@RunWith(SpringJUnit4ClassRunner.class) +@ContextConfiguration +public class KafkaMessageDrivenChannelAdapterParserTests { + + @Autowired + private ConnectionFactory connectionFactory; + + @Autowired + private OffsetManager offsetManager; + + @Autowired + private ErrorHandler errorHandler; + + @Autowired + private Executor executor; + + @Autowired + private Decoder keyDecoder; + + @Autowired + private Decoder payloadDecoder; + + @Autowired + private NullChannel nullChannel; + + @Autowired + private PublishSubscribeChannel errorChannel; + + @Autowired + private KafkaMessageDrivenChannelAdapter kafkaListener; + + @Test + public void testKafkaMessageDrivenChannelAdapterParser() { + assertFalse(this.kafkaListener.isAutoStartup()); + assertFalse(this.kafkaListener.isRunning()); + assertEquals(100, this.kafkaListener.getPhase()); + assertSame(this.nullChannel, TestUtils.getPropertyValue(this.kafkaListener, "outputChannel")); + assertSame(this.errorChannel, TestUtils.getPropertyValue(this.kafkaListener, "errorChannel")); + assertSame(this.keyDecoder, TestUtils.getPropertyValue(this.kafkaListener, "keyDecoder")); + assertSame(this.payloadDecoder, TestUtils.getPropertyValue(this.kafkaListener, "payloadDecoder")); + KafkaMessageListenerContainer container = + TestUtils.getPropertyValue(this.kafkaListener, "messageListenerContainer", + KafkaMessageListenerContainer.class); + assertSame(this.connectionFactory, TestUtils.getPropertyValue(container, "kafkaTemplate.connectionFactory")); + assertSame(this.offsetManager, container.getOffsetManager()); + assertSame(this.errorHandler, container.getErrorHandler()); + assertSame(this.executor, container.getFetchTaskExecutor()); + assertEquals(10, container.getConcurrency()); + assertEquals(1000, container.getMaxFetch()); + assertEquals(1024, container.getQueueSize()); + assertEquals(1024, container.getQueueSize()); + assertArrayEquals(new String[] {"foo", "bar"}, TestUtils.getPropertyValue(container, "topics", String[].class)); + } + +} diff --git a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/listener/ChannelAdapterWithXmlConfigurationTests-context.xml b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/listener/ChannelAdapterWithXmlConfigurationTests-context.xml index d1ddd66a41..aaa9801d32 100644 --- a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/listener/ChannelAdapterWithXmlConfigurationTests-context.xml +++ b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/listener/ChannelAdapterWithXmlConfigurationTests-context.xml @@ -1,60 +1,39 @@ + http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context.xsd + http://www.springframework.org/schema/integration/kafka http://www.springframework.org/schema/integration/kafka/spring-integration-kafka.xsd"> - - - - - - - - - - - - - - - - + + + + - + + - - - - - - - - - - - - - - - - + + diff --git a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/listener/ChannelAdapterWithXmlConfigurationTests.java b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/listener/ChannelAdapterWithXmlConfigurationTests.java index 709436c5fb..a0cd4f9cba 100644 --- a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/listener/ChannelAdapterWithXmlConfigurationTests.java +++ b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/listener/ChannelAdapterWithXmlConfigurationTests.java @@ -48,8 +48,6 @@ public class ChannelAdapterWithXmlConfigurationTests extends AbstractMessageList @Test public void testConsumptionWithXmlConfiguration() throws Exception { - System.setProperty("kafka.test.port", - String.valueOf(kafkaEmbeddedBrokerRule.getBrokerAddresses().get(0).getPort())); System.setProperty("kafka.test.topic", TEST_TOPIC); createTopic(TEST_TOPIC, 1, 1, 1); diff --git a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/listener/DefaultConnectionFactoryTests.java b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/listener/DefaultConnectionFactoryTests.java index 5ec2b7a2be..71e94e7a41 100644 --- a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/listener/DefaultConnectionFactoryTests.java +++ b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/listener/DefaultConnectionFactoryTests.java @@ -55,15 +55,16 @@ public class DefaultConnectionFactoryTests extends AbstractBrokerTests { createTopic(TEST_TOPIC, 1, 1, 1); - List brokerAddresses = getKafkaRule().getBrokerAddresses(); + BrokerAddress[] brokerAddresses = getKafkaRule().getBrokerAddresses(); Partition partition = new Partition(TEST_TOPIC, 0); - DefaultConnectionFactory connectionFactory = new DefaultConnectionFactory(new BrokerAddressListConfiguration(brokerAddresses)); + DefaultConnectionFactory connectionFactory = + new DefaultConnectionFactory(new BrokerAddressListConfiguration(brokerAddresses)); connectionFactory.afterPropertiesSet(); - Connection connection = connectionFactory.connect(getKafkaRule().getBrokerAddresses().get(0)); + Connection connection = connectionFactory.connect(brokerAddresses[0]); Result leaders = connection.findLeaders(TEST_TOPIC); assertThat(leaders.getErrors().entrySet(), empty()); assertThat(leaders.getResults().entrySet(), hasSize(1)); - assertThat(leaders.getResults().get(partition), equalTo(getKafkaRule().getBrokerAddresses().get(0))); + assertThat(leaders.getResults().get(partition), equalTo(brokerAddresses[0])); } @Test @@ -78,10 +79,11 @@ public class DefaultConnectionFactoryTests extends AbstractBrokerTests { DefaultConnectionFactory connectionFactory = new DefaultConnectionFactory(new ZookeeperConfiguration(zookeeperConnect)); connectionFactory.afterPropertiesSet(); - Connection connection = connectionFactory.connect(getKafkaRule().getBrokerAddresses().get(0)); + Connection connection = connectionFactory.connect(getKafkaRule().getBrokerAddresses()[0]); Result leaders = connection.findLeaders(TEST_TOPIC); assertThat(leaders.getErrors().entrySet(), empty()); assertThat(leaders.getResults().entrySet(), hasSize(1)); - assertThat(leaders.getResults().get(partition), equalTo(getKafkaRule().getBrokerAddresses().get(0))); + assertThat(leaders.getResults().get(partition), equalTo(getKafkaRule().getBrokerAddresses()[0])); } + } diff --git a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/listener/DefaultConnectionTests.java b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/listener/DefaultConnectionTests.java index a818b47109..c054fabe47 100644 --- a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/listener/DefaultConnectionTests.java +++ b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/listener/DefaultConnectionTests.java @@ -51,7 +51,7 @@ public class DefaultConnectionTests extends AbstractBrokerTests { public void testFetchPartitionMetadata() throws Exception { createTopic(TEST_TOPIC, 1, 1, 1); Connection brokerConnection = - new DefaultConnection(getKafkaRule().getBrokerAddresses().get(0), "client", 64*1024, 3000, 1, 10000); + new DefaultConnection(getKafkaRule().getBrokerAddresses()[0], "client", 64*1024, 3000, 1, 10000); Partition partition = new Partition(TEST_TOPIC, 0); Result result = brokerConnection.fetchInitialOffset(-1, partition); Assert.assertEquals(0, result.getErrors().size()); @@ -65,7 +65,7 @@ public class DefaultConnectionTests extends AbstractBrokerTests { Producer producer = createStringProducer(0); producer.send( createMessages(10, TEST_TOPIC)); Connection brokerConnection = - new DefaultConnection(getKafkaRule().getBrokerAddresses().get(0), "client", 64*1024, 3000, 1, 10000); + new DefaultConnection(getKafkaRule().getBrokerAddresses()[0], "client", 64*1024, 3000, 1, 10000); Partition partition = new Partition(TEST_TOPIC, 0); FetchRequest fetchRequest = new FetchRequest(partition, 0L, 1000); Result result = brokerConnection.fetch(fetchRequest); @@ -88,7 +88,7 @@ public class DefaultConnectionTests extends AbstractBrokerTests { Producer producer = createStringProducer(1); producer.send( createMessages(10, TEST_TOPIC)); Connection brokerConnection = - new DefaultConnection(getKafkaRule().getBrokerAddresses().get(0), "client", 64*1024, 3000, 1, 10000); + new DefaultConnection(getKafkaRule().getBrokerAddresses()[0], "client", 64*1024, 3000, 1, 10000); Partition partition = new Partition(TEST_TOPIC, 0); FetchRequest fetchRequest = new FetchRequest(partition, 0L, 1000); Result result = brokerConnection.fetch(fetchRequest); @@ -118,7 +118,7 @@ public class DefaultConnectionTests extends AbstractBrokerTests { Producer producer = createStringProducer(2); producer.send( createMessages(10, TEST_TOPIC)); Connection brokerConnection = - new DefaultConnection(getKafkaRule().getBrokerAddresses().get(0), "client", 64*1024, 3000, 1, 10000); + new DefaultConnection(getKafkaRule().getBrokerAddresses()[0], "client", 64*1024, 3000, 1, 10000); Partition partition = new Partition(TEST_TOPIC, 0); FetchRequest fetchRequest = new FetchRequest(partition, 0L, 1000); Result result = brokerConnection.fetch(fetchRequest); diff --git a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/rule/KafkaEmbedded.java b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/rule/KafkaEmbedded.java index 1ade044eef..eecec93a08 100644 --- a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/rule/KafkaEmbedded.java +++ b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/rule/KafkaEmbedded.java @@ -20,12 +20,14 @@ package org.springframework.integration.kafka.rule; import static scala.collection.JavaConversions.asScalaBuffer; import java.util.ArrayList; +import java.util.Arrays; import java.util.List; import java.util.Properties; import com.gs.collections.api.block.function.Function; import com.gs.collections.impl.list.mutable.FastList; import com.gs.collections.impl.utility.ListIterate; + import kafka.Kafka; import kafka.server.KafkaConfig; import kafka.server.KafkaServer; @@ -35,15 +37,18 @@ import kafka.utils.TestZKUtils; import kafka.utils.Utils; import kafka.utils.ZKStringSerializer$; import kafka.zk.EmbeddedZookeeper; + import org.I0Itec.zkclient.ZkClient; import org.I0Itec.zkclient.exception.ZkInterruptedException; import org.junit.rules.ExternalResource; + import scala.collection.JavaConversions; import org.springframework.integration.kafka.core.BrokerAddress; /** * @author Marius Bogoevici + * @author Artem Bilan */ @SuppressWarnings("serial") public class KafkaEmbedded extends ExternalResource implements KafkaRule { @@ -76,7 +81,8 @@ public class KafkaEmbedded extends ExternalResource implements KafkaRule { zookeeper = new EmbeddedZookeeper(TestZKUtils.zookeeperConnect()); int zkConnectionTimeout = 6000; int zkSessionTimeout = 6000; - zookeeperClient = new ZkClient(TestZKUtils.zookeeperConnect(), zkSessionTimeout, zkConnectionTimeout, ZKStringSerializer$.MODULE$); + zookeeperClient = new ZkClient(TestZKUtils.zookeeperConnect(), zkSessionTimeout, zkConnectionTimeout, + ZKStringSerializer$.MODULE$); kafkaServers = new ArrayList(); for (int i = 0; i < count; i++) { Properties brokerConfigProperties = TestUtils.createBrokerConfig(i, kafkaPorts.get(i)); @@ -138,13 +144,17 @@ public class KafkaEmbedded extends ExternalResource implements KafkaRule { } @Override - public List getBrokerAddresses() { - return ListIterate.collect(kafkaServers, new Function() { - @Override - public BrokerAddress valueOf(KafkaServer kafkaServer) { - return new BrokerAddress(kafkaServer.config().hostName(), kafkaServer.config().port()); - } - }); + public BrokerAddress[] getBrokerAddresses() { + return ListIterate.collect(this.kafkaServers, + new Function() { + + @Override + public BrokerAddress valueOf(KafkaServer kafkaServer) { + return new BrokerAddress(kafkaServer.config().hostName(), kafkaServer.config().port()); + } + + }) + .toArray(new BrokerAddress[this.kafkaServers.size()]); } public void bounce(BrokerAddress brokerAddress) { @@ -162,16 +172,21 @@ public class KafkaEmbedded extends ExternalResource implements KafkaRule { @Override public String getBrokersAsString() { - return FastList.newList(getBrokerAddresses()).collect(new Function() { - @Override - public String valueOf(BrokerAddress object) { - return object.getHost() + ":" + object.getPort(); - } - }).makeString(","); + return FastList.newList(Arrays.asList(getBrokerAddresses())) + .collect(new Function() { + + @Override + public String valueOf(BrokerAddress object) { + return object.getHost() + ":" + object.getPort(); + } + + }) + .makeString(","); } @Override public boolean isEmbedded() { return true; } + } diff --git a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/rule/KafkaRule.java b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/rule/KafkaRule.java index 9b6bac37d2..8e28208895 100644 --- a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/rule/KafkaRule.java +++ b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/rule/KafkaRule.java @@ -34,7 +34,7 @@ public interface KafkaRule extends TestRule { ZkClient getZkClient(); - List getBrokerAddresses(); + BrokerAddress[] getBrokerAddresses(); String getBrokersAsString(); diff --git a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/rule/KafkaRunning.java b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/rule/KafkaRunning.java index e71dd0b663..e1cf4cfa8c 100644 --- a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/rule/KafkaRunning.java +++ b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/rule/KafkaRunning.java @@ -15,16 +15,19 @@ */ package org.springframework.integration.kafka.rule; +import java.util.Arrays; import java.util.List; import com.gs.collections.api.block.function.Function; import com.gs.collections.impl.block.factory.Functions; import com.gs.collections.impl.list.mutable.FastList; import com.gs.collections.impl.utility.ListIterate; + import kafka.cluster.Broker; import kafka.server.KafkaServer; import kafka.utils.ZKStringSerializer$; import kafka.utils.ZkUtils; + import org.I0Itec.zkclient.ZkClient; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; @@ -32,6 +35,7 @@ import org.junit.Assume; import org.junit.rules.TestWatcher; import org.junit.runner.Description; import org.junit.runners.model.Statement; + import scala.collection.JavaConversions; import scala.collection.Seq; @@ -72,19 +76,23 @@ public class KafkaRunning extends TestWatcher implements KafkaRule { @Override @SuppressWarnings("serial") - public List getBrokerAddresses() { + public BrokerAddress[] getBrokerAddresses() { Seq allBrokersInCluster = ZkUtils.getAllBrokersInCluster(zkClient); - return ListIterate.collect(JavaConversions.asJavaList(allBrokersInCluster), new Function() { - @Override - public BrokerAddress valueOf(Broker broker) { - return new BrokerAddress(broker.host(), broker.port()); - } - }); + return ListIterate.collect(JavaConversions.asJavaList(allBrokersInCluster), + new Function() { + + @Override + public BrokerAddress valueOf(Broker broker) { + return new BrokerAddress(broker.host(), broker.port()); + } + + }) + .toArray(new BrokerAddress[allBrokersInCluster.size()]); } @Override public String getBrokersAsString() { - return FastList.newList(getBrokerAddresses()).collect(Functions.getToString()).makeString(","); + return FastList.newList(Arrays.asList(getBrokerAddresses())).collect(Functions.getToString()).makeString(","); } @Override @@ -101,7 +109,7 @@ public class KafkaRunning extends TestWatcher implements KafkaRule { public Statement apply(Statement base, Description description) { try { this.zkClient = new ZkClient(ZOOKEEPER_CONNECT_STRING, 1000, 1000, ZKStringSerializer$.MODULE$); - if (getBrokerAddresses().size() == 0) { + if (getBrokerAddresses().length == 0) { throw new IllegalStateException("No running Kafka brokers"); } }