Spring/spring-integration
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

INTEXT-129: Namespace support for KafkaMDCA

Browse Source 
JIRA: https://jira.spring.io/browse/INTEXT-129

INTEXT-129: Polishing and Docs

Doc Polishing
This commit is contained in:
Artem Bilan
2015-01-23 16:31:59 +02:00
committed by Artem Bilan
parent 2cae7420a3
commit d85677f61e
14 changed files with 545 additions and 108 deletions
Download Patch File Download Diff File Expand all files Collapse all files

99
spring-integration-kafka/src/main/java/org/springframework/integration/kafka/config/xml/KafkaMessageDrivenChannelAdapterParser.java Normal file
View File

@@ -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();
}
}

17
spring-integration-kafka/src/main/java/org/springframework/integration/kafka/config/xml/KafkaNamespaceHandler.java
View File

@@ -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());
}
}

5
spring-integration-kafka/src/main/java/org/springframework/integration/kafka/core/BrokerAddressListConfiguration.java
View File

@@ -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<BrokerAddress> brokerAddresses;
public BrokerAddressListConfiguration(List<BrokerAddress> brokerAddresses) {
this.brokerAddresses = brokerAddresses;
public BrokerAddressListConfiguration(BrokerAddress... brokerAddresses) {
this.brokerAddresses = Arrays.asList(brokerAddresses);
}
@Override

15
spring-integration-kafka/src/main/java/org/springframework/integration/kafka/listener/KafkaMessageListenerContainer.java
View File

@@ -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<Partition> 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<Partition> 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.
*/

186
spring-integration-kafka/src/main/resources/org/springframework/integration/config/xml/spring-integration-kafka-1.0.xsd
View File

@@ -382,18 +382,10 @@
<xsd:attributeGroup ref="integration:channelAdapterAttributes"/>
<xsd:attribute name="send-timeout" type="xsd:string">
<xsd:annotation>
<xsd:documentation><![CDATA[
Allows you to specify how long this inbound-channel-adapter
will wait for the message (containing the retrieved entities)
to be sent successfully to the message channel, before throwing
an exception.
Keep in mind that when sending to a DirectChannel, the
invocation will occur in the sender's thread so the failing
of the send operation may be caused by other components
further downstream. By default the Inbound Channel Adapter
will wait indefinitely. The value is specified in milliseconds.
]]>
<xsd:documentation>
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.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
@@ -492,4 +484,174 @@
</xsd:complexType>
</xsd:element>
<xsd:element name="message-driven-channel-adapter">
<xsd:annotation>
<xsd:documentation>
The definition for the Spring Integration Kafka
KafkaMessageDrivenChannelAdapter.
</xsd:documentation>
</xsd:annotation>
<xsd:complexType>
<xsd:attributeGroup ref="integration:channelAdapterAttributes"/>
<xsd:attribute name="send-timeout" type="xsd:string">
<xsd:annotation>
<xsd:documentation>
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.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="error-channel" type="xsd:string">
<xsd:annotation>
<xsd:documentation>
Message Channel to which error Messages should be sent.
</xsd:documentation>
<xsd:appinfo>
<tool:annotation kind="ref">
<tool:expected-type type="org.springframework.messaging.MessageChannel"/>
</tool:annotation>
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="key-decoder" type="xsd:string">
<xsd:annotation>
<xsd:documentation>
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'.
</xsd:documentation>
<xsd:appinfo>
<tool:annotation kind="ref">
<tool:expected-type type="kafka.serializer.Decoder"/>
</tool:annotation>
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="payload-decoder" type="xsd:string">
<xsd:annotation>
<xsd:documentation>
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'.
</xsd:documentation>
<xsd:appinfo>
<tool:annotation kind="ref">
<tool:expected-type type="kafka.serializer.Decoder"/>
</tool:annotation>
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="listener-container" type="xsd:string">
<xsd:annotation>
<xsd:documentation>
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'.
</xsd:documentation>
<xsd:appinfo>
<tool:annotation kind="ref">
<tool:expected-type type="org.springframework.integration.kafka.core.ConnectionFactory"/>
</tool:annotation>
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="connection-factory" type="xsd:string">
<xsd:annotation>
<xsd:documentation>
A 'org.springframework.integration.kafka.core.ConnectionFactory' bean reference.
Mutually exclusive with 'listener-container'.
</xsd:documentation>
<xsd:appinfo>
<tool:annotation kind="ref">
<tool:expected-type type="org.springframework.integration.kafka.core.ConnectionFactory"/>
</tool:annotation>
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="topics" type="xsd:string">
<xsd:annotation>
<xsd:documentation>
The comma-separated Kafka topics to listen to.
Mutually exclusive with 'listener-container'.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="offset-manager" type="xsd:string">
<xsd:annotation>
<xsd:documentation>
A 'org.springframework.integration.kafka.listener.OffsetManager' bean reference.
Defaults to 'org.springframework.integration.kafka.listener.MetadataStoreOffsetManager'.
Mutually exclusive with 'listener-container'.
</xsd:documentation>
<xsd:appinfo>
<tool:annotation kind="ref">
<tool:expected-type type="org.springframework.integration.kafka.listener.OffsetManager"/>
</tool:annotation>
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="error-handler" type="xsd:string">
<xsd:annotation>
<xsd:documentation>
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'.
</xsd:documentation>
<xsd:appinfo>
<tool:annotation kind="ref">
<tool:expected-type type="org.springframework.integration.kafka.listener.ErrorHandler"/>
</tool:annotation>
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="task-executor" type="xsd:string">
<xsd:annotation>
<xsd:documentation>
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'.
</xsd:documentation>
<xsd:appinfo>
<tool:annotation kind="ref">
<tool:expected-type type="java.util.concurrent.Executor"/>
</tool:annotation>
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="concurrency" type="xsd:string">
<xsd:annotation>
<xsd:documentation>
The number of internal concurrent
'org.springframework.integration.kafka.listener.QueueingMessageListenerInvoker'.
Defaults to '1'.
Mutually exclusive with 'listener-container'.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="max-fetch" type="xsd:string">
<xsd:annotation>
<xsd:documentation>
The maximum amount of data (in bytes) that pollers will fetch in one round.
Defaults to '300 * 1024'.
Mutually exclusive with 'listener-container'.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="queue-size" type="xsd:string">
<xsd:annotation>
<xsd:documentation>
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'.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
</xsd:complexType>
</xsd:element>
</xsd:schema>

69
spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/KafkaMessageDrivenChannelAdapterParserTests-context.xml Normal file
View File

@@ -0,0 +1,69 @@
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:int-kafka="http://www.springframework.org/schema/integration/kafka"
xmlns:context="http://www.springframework.org/schema/context"
xmlns:task="http://www.springframework.org/schema/task"
xsi:schemaLocation="http://www.springframework.org/schema/integration/kafka http://www.springframework.org/schema/integration/kafka/spring-integration-kafka.xsd
http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context.xsd
http://www.springframework.org/schema/task http://www.springframework.org/schema/task/spring-task.xsd">
<context:property-placeholder/>
<bean id="connectionFactory" class="org.mockito.Mockito" factory-method="mock">
<constructor-arg value="org.springframework.integration.kafka.core.ConnectionFactory"/>
</bean>
<bean id="offsetManager" class="org.mockito.Mockito" factory-method="mock">
<constructor-arg value="org.springframework.integration.kafka.listener.OffsetManager"/>
</bean>
<bean id="errorHandler" class="org.mockito.Mockito" factory-method="mock">
<constructor-arg value="org.springframework.integration.kafka.listener.ErrorHandler"/>
</bean>
<bean id="keyDecoder" class="org.mockito.Mockito" factory-method="mock">
<constructor-arg value="kafka.serializer.Decoder"/>
</bean>
<bean id="payloadDecoder" class="org.mockito.Mockito" factory-method="mock">
<constructor-arg value="kafka.serializer.Decoder"/>
</bean>
<task:executor id="executor"/>
<int-kafka:message-driven-channel-adapter
id="kafkaListener"
auto-startup="false"
phase="100"
send-timeout="5000"
channel="nullChannel"
error-channel="errorChannel"
error-handler="errorHandler"
connection-factory="connectionFactory"
key-decoder="keyDecoder"
payload-decoder="payloadDecoder"
offset-manager="offsetManager"
task-executor="executor"
queue-size="${queue.size:1024}"
concurrency="${concurrency:10}"
max-fetch="${max.fetch:1000}"
topics="${topics:foo,bar}"/>
<!-- Invalid config
<bean id="container" class="org.mockito.Mockito" factory-method="mock">
<constructor-arg value="org.springframework.integration.kafka.listener.KafkaMessageListenerContainer"/>
</bean>
<int-kafka:message-driven-channel-adapter
channel="nullChannel"
topics="foo"
listener-container="container"/>
-->
</beans>

102
spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/KafkaMessageDrivenChannelAdapterParserTests.java Normal file
View File

@@ -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));
}
}

65
spring-integration-kafka/src/test/java/org/springframework/integration/kafka/listener/ChannelAdapterWithXmlConfigurationTests-context.xml
View File

@@ -1,60 +1,39 @@
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:int="http://www.springframework.org/schema/integration"
xmlns:util="http://www.springframework.org/schema/util"
xmlns:context="http://www.springframework.org/schema/context"
xmlns="http://www.springframework.org/schema/beans"
xsi:schemaLocation="
http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
xmlns:int="http://www.springframework.org/schema/integration"
xmlns:context="http://www.springframework.org/schema/context"
xmlns="http://www.springframework.org/schema/beans"
xmlns:int-kafka="http://www.springframework.org/schema/integration/kafka"
xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration.xsd
http://www.springframework.org/schema/util http://www.springframework.org/schema/util/spring-util.xsd
http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context.xsd">
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">
<context:property-placeholder/>
<util:list id="kafkaBrokerAddresses">
<bean class="org.springframework.integration.kafka.core.BrokerAddress">
<constructor-arg index="0" value="localhost"/>
<constructor-arg index="1" value="${kafka.test.port}"/>
</bean>
</util:list>
<util:list id="partitions">
<bean class="org.springframework.integration.kafka.core.Partition">
<constructor-arg index="0" value="${kafka.test.topic}"/>
<constructor-arg index="1" value="0"/>
</bean>
</util:list>
<bean id="kafkaConfiguration" class="org.springframework.integration.kafka.core.BrokerAddressListConfiguration">
<constructor-arg index="0" ref="kafkaBrokerAddresses"/>
<int-kafka:zookeeper-connect id="zkConnect" zk-connect="#{T (kafka.utils.TestZKUtils).zookeeperConnect()}"/>
<bean id="kafkaConfiguration" class="org.springframework.integration.kafka.core.ZookeeperConfiguration">
<constructor-arg ref="zkConnect"/>
</bean>
<bean id="connectionFactory" class="org.springframework.integration.kafka.core.DefaultConnectionFactory">
<constructor-arg index="0" ref="kafkaConfiguration"/>
<constructor-arg ref="kafkaConfiguration"/>
</bean>
<bean id="decoder" class="org.springframework.integration.kafka.serializer.common.StringDecoder"/>
<bean id="kafkaMessageListenerContainer"
class="org.springframework.integration.kafka.listener.KafkaMessageListenerContainer">
<constructor-arg index="0" ref="connectionFactory"/>
<constructor-arg index="1" ref="partitions"/>
<property name="maxFetch" value="100"/>
</bean>
<bean id="kafkaInboundChannelAdapter"
class="org.springframework.integration.kafka.inbound.KafkaMessageDrivenChannelAdapter">
<constructor-arg index="0" ref="kafkaMessageListenerContainer"/>
<property name="outputChannel" ref="output"/>
<property name="keyDecoder">
<bean class="org.springframework.integration.kafka.serializer.common.StringDecoder"/>
</property>
<property name="payloadDecoder">
<bean class="org.springframework.integration.kafka.serializer.common.StringDecoder"/>
</property>
</bean>
<int-kafka:message-driven-channel-adapter
id="adapter"
channel="output"
connection-factory="connectionFactory"
key-decoder="decoder"
payload-decoder="decoder"
max-fetch="100"
topics="${kafka.test.topic}"/>
<int:channel id="output">
<int:queue capacity="100"/>
</int:channel>
</beans>

2
spring-integration-kafka/src/test/java/org/springframework/integration/kafka/listener/ChannelAdapterWithXmlConfigurationTests.java
View File

@@ -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);

14
spring-integration-kafka/src/test/java/org/springframework/integration/kafka/listener/DefaultConnectionFactoryTests.java
View File

@@ -55,15 +55,16 @@ public class DefaultConnectionFactoryTests extends AbstractBrokerTests {
createTopic(TEST_TOPIC, 1, 1, 1);
List<BrokerAddress> 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<BrokerAddress> 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<BrokerAddress> 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]));
}
}

8
spring-integration-kafka/src/test/java/org/springframework/integration/kafka/listener/DefaultConnectionTests.java
View File

@@ -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<Long> result = brokerConnection.fetchInitialOffset(-1, partition);
Assert.assertEquals(0, result.getErrors().size());
@@ -65,7 +65,7 @@ public class DefaultConnectionTests extends AbstractBrokerTests {
Producer<String, String> 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<KafkaMessageBatch> result = brokerConnection.fetch(fetchRequest);
@@ -88,7 +88,7 @@ public class DefaultConnectionTests extends AbstractBrokerTests {
Producer<String, String> 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<KafkaMessageBatch> result = brokerConnection.fetch(fetchRequest);
@@ -118,7 +118,7 @@ public class DefaultConnectionTests extends AbstractBrokerTests {
Producer<String, String> 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<KafkaMessageBatch> result = brokerConnection.fetch(fetchRequest);

43
spring-integration-kafka/src/test/java/org/springframework/integration/kafka/rule/KafkaEmbedded.java
View File

@@ -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<KafkaServer>();
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<BrokerAddress> getBrokerAddresses() {
return ListIterate.collect(kafkaServers, new Function<KafkaServer, BrokerAddress>() {
@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<KafkaServer, BrokerAddress>() {
@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<BrokerAddress, String>() {
@Override
public String valueOf(BrokerAddress object) {
return object.getHost() + ":" + object.getPort();
}
}).makeString(",");
return FastList.newList(Arrays.asList(getBrokerAddresses()))
.collect(new Function<BrokerAddress, String>() {
@Override
public String valueOf(BrokerAddress object) {
return object.getHost() + ":" + object.getPort();
}
})
.makeString(",");
}
@Override
public boolean isEmbedded() {
return true;
}
}

2
spring-integration-kafka/src/test/java/org/springframework/integration/kafka/rule/KafkaRule.java
View File

@@ -34,7 +34,7 @@ public interface KafkaRule extends TestRule {
ZkClient getZkClient();
List<BrokerAddress> getBrokerAddresses();
BrokerAddress[] getBrokerAddresses();
String getBrokersAsString();

26
spring-integration-kafka/src/test/java/org/springframework/integration/kafka/rule/KafkaRunning.java
View File

@@ -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<BrokerAddress> getBrokerAddresses() {
public BrokerAddress[] getBrokerAddresses() {
Seq<Broker> allBrokersInCluster = ZkUtils.getAllBrokersInCluster(zkClient);
return ListIterate.collect(JavaConversions.asJavaList(allBrokersInCluster), new Function<Broker, BrokerAddress>() {
@Override
public BrokerAddress valueOf(Broker broker) {
return new BrokerAddress(broker.host(), broker.port());
}
});
return ListIterate.collect(JavaConversions.asJavaList(allBrokersInCluster),
new Function<Broker, BrokerAddress>() {
@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");
}
}