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

Kafka-GH-92: Allow PP for the <topic-filter>

Browse Source 
Fixes https://github.com/spring-projects/spring-integration-extensions/issues/92
This commit is contained in:
Artem Bilan
2014-08-12 22:57:50 +03:00
committed by Artem Bilan
parent 999878f048
commit bacf8d2ceb
6 changed files with 473 additions and 419 deletions
Download Patch File Download Diff File Expand all files Collapse all files

34
spring-integration-kafka/src/main/java/org/springframework/integration/kafka/config/xml/KafkaConsumerContextParser.java
View File

@@ -1,5 +1,5 @@
/* /*
* Copyright 2002-2013 the original author or authors. * Copyright 2002-2014 the original author or authors.
* *
* Licensed under the Apache License, Version 2.0 (the "License"); * Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License. * you may not use this file except in compliance with the License.
@@ -13,9 +13,14 @@
* See the License for the specific language governing permissions and * See the License for the specific language governing permissions and
* limitations under the License. * limitations under the License.
*/ */
package org.springframework.integration.kafka.config.xml; package org.springframework.integration.kafka.config.xml;
import java.util.*; import java.util.HashMap;
import java.util.List;
import java.util.Map;
import org.w3c.dom.Element;
import org.springframework.beans.factory.config.BeanDefinition; import org.springframework.beans.factory.config.BeanDefinition;
import org.springframework.beans.factory.config.BeanDefinitionHolder; import org.springframework.beans.factory.config.BeanDefinitionHolder;
@@ -24,14 +29,20 @@ import org.springframework.beans.factory.support.BeanDefinitionBuilder;
import org.springframework.beans.factory.xml.AbstractSingleBeanDefinitionParser; import org.springframework.beans.factory.xml.AbstractSingleBeanDefinitionParser;
import org.springframework.beans.factory.xml.ParserContext; import org.springframework.beans.factory.xml.ParserContext;
import org.springframework.integration.config.xml.IntegrationNamespaceUtils; import org.springframework.integration.config.xml.IntegrationNamespaceUtils;
import org.springframework.integration.kafka.support.*; import org.springframework.integration.kafka.support.ConsumerConfigFactoryBean;
import org.springframework.integration.kafka.support.ConsumerConfiguration;
import org.springframework.integration.kafka.support.ConsumerConnectionProvider;
import org.springframework.integration.kafka.support.ConsumerMetadata;
import org.springframework.integration.kafka.support.KafkaConsumerContext;
import org.springframework.integration.kafka.support.MessageLeftOverTracker;
import org.springframework.integration.kafka.support.TopicFilterConfiguration;
import org.springframework.util.StringUtils; import org.springframework.util.StringUtils;
import org.springframework.util.xml.DomUtils; import org.springframework.util.xml.DomUtils;
import org.w3c.dom.Element;
/** /**
* @author Soby Chacko * @author Soby Chacko
* @author Rajasekar Elango * @author Rajasekar Elango
* @author Artem Bilan
* @since 0.5 * @since 0.5
*/ */
public class KafkaConsumerContextParser extends AbstractSingleBeanDefinitionParser { public class KafkaConsumerContextParser extends AbstractSingleBeanDefinitionParser {
@@ -50,7 +61,7 @@ public class KafkaConsumerContextParser extends AbstractSingleBeanDefinitionPars
} }
private void parseConsumerConfigurations(final Element consumerConfigurations, final ParserContext parserContext, private void parseConsumerConfigurations(final Element consumerConfigurations, final ParserContext parserContext,
final BeanDefinitionBuilder builder, final Element parentElem) { final BeanDefinitionBuilder builder, final Element parentElem) {
for (final Element consumerConfiguration : DomUtils.getChildElementsByTagName(consumerConfigurations, "consumer-configuration")) { for (final Element consumerConfiguration : DomUtils.getChildElementsByTagName(consumerConfigurations, "consumer-configuration")) {
final BeanDefinitionBuilder consumerConfigurationBuilder = BeanDefinitionBuilder.genericBeanDefinition(ConsumerConfiguration.class); final BeanDefinitionBuilder consumerConfigurationBuilder = BeanDefinitionBuilder.genericBeanDefinition(ConsumerConfiguration.class);
final BeanDefinitionBuilder consumerMetadataBuilder = BeanDefinitionBuilder.genericBeanDefinition(ConsumerMetadata.class); final BeanDefinitionBuilder consumerMetadataBuilder = BeanDefinitionBuilder.genericBeanDefinition(ConsumerMetadata.class);
@@ -68,7 +79,7 @@ public class KafkaConsumerContextParser extends AbstractSingleBeanDefinitionPars
final List<Element> topicConfigurations = DomUtils.getChildElementsByTagName(consumerConfiguration, "topic"); final List<Element> topicConfigurations = DomUtils.getChildElementsByTagName(consumerConfiguration, "topic");
if (topicConfigurations != null){ if (topicConfigurations != null) {
for (final Element topicConfiguration : topicConfigurations) { for (final Element topicConfiguration : topicConfigurations) {
final String topic = topicConfiguration.getAttribute("id"); final String topic = topicConfiguration.getAttribute("id");
final String streams = topicConfiguration.getAttribute("streams"); final String streams = topicConfiguration.getAttribute("streams");
@@ -80,9 +91,14 @@ public class KafkaConsumerContextParser extends AbstractSingleBeanDefinitionPars
final Element topicFilter = DomUtils.getChildElementByTagName(consumerConfiguration, "topic-filter"); final Element topicFilter = DomUtils.getChildElementByTagName(consumerConfiguration, "topic-filter");
if (topicFilter != null){ if (topicFilter != null) {
final TopicFilterConfiguration topicFilterConfiguration = new TopicFilterConfiguration(topicFilter.getAttribute("pattern"),Integer.valueOf(topicFilter.getAttribute("streams")), Boolean.valueOf(topicFilter.getAttribute("exclude"))); BeanDefinition topicFilterConfigurationBeanDefinition =
consumerMetadataBuilder.addPropertyValue("topicFilterConfiguration", topicFilterConfiguration); BeanDefinitionBuilder.genericBeanDefinition(TopicFilterConfiguration.class)
.addConstructorArgValue(topicFilter.getAttribute("pattern"))
.addConstructorArgValue(topicFilter.getAttribute("streams"))
.addConstructorArgValue(topicFilter.getAttribute("exclude"))
.getBeanDefinition();
consumerMetadataBuilder.addPropertyValue("topicFilterConfiguration", topicFilterConfigurationBeanDefinition);
} }
final BeanDefinition consumerMetadataBeanDef = consumerMetadataBuilder.getBeanDefinition(); final BeanDefinition consumerMetadataBeanDef = consumerMetadataBuilder.getBeanDefinition();

35
spring-integration-kafka/src/main/java/org/springframework/integration/kafka/support/TopicFilterConfiguration.java
View File

@@ -1,11 +1,19 @@
/* /*
* Copyright 2002-2013 the original author or authors. Licensed under the Apache License, Version 2.0 (the "License"); * Copyright 2002-2014 the original author or authors.
* 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 * Licensed under the Apache License, Version 2.0 (the "License");
* distributed under the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, * you may not use this file except in compliance with the License.
* either express or implied. See the License for the specific language governing permissions and limitations under the * You may obtain a copy of the License at
* License. *
* 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.support; package org.springframework.integration.kafka.support;
import kafka.consumer.Blacklist; import kafka.consumer.Blacklist;
@@ -14,33 +22,36 @@ import kafka.consumer.Whitelist;
/** /**
* @author Rajasekar Elango * @author Rajasekar Elango
* @author Artem Bilan
* @since 0.5 * @since 0.5
*/ */
public class TopicFilterConfiguration { public class TopicFilterConfiguration {
private final int numberOfStreams; private final int numberOfStreams;
private TopicFilter topicFilter;
private final TopicFilter topicFilter;
public TopicFilterConfiguration(final String pattern, final int numberOfStreams, final boolean exclude) { public TopicFilterConfiguration(final String pattern, final int numberOfStreams, final boolean exclude) {
this.numberOfStreams = numberOfStreams; this.numberOfStreams = numberOfStreams;
if (exclude) { if (exclude) {
topicFilter = new Blacklist(pattern); this.topicFilter = new Blacklist(pattern);
} }
else { else {
topicFilter = new Whitelist(pattern); this.topicFilter = new Whitelist(pattern);
} }
} }
public TopicFilter getTopicFilter() { public TopicFilter getTopicFilter() {
return topicFilter; return this.topicFilter;
} }
public int getNumberOfStreams() { public int getNumberOfStreams() {
return numberOfStreams; return this.numberOfStreams;
} }
@Override @Override
public String toString() { public String toString() {
return new StringBuilder(topicFilter.toString()).append(" : ").append(numberOfStreams).toString(); return this.topicFilter.toString() + " : " + this.numberOfStreams;
} }
} }

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

@@ -1,389 +1,392 @@
<?xml version="1.0" encoding="UTF-8"?> <?xml version="1.0" encoding="UTF-8"?>
<xsd:schema xmlns="http://www.springframework.org/schema/integration/kafka" <xsd:schema xmlns="http://www.springframework.org/schema/integration/kafka"
xmlns:xsd="http://www.w3.org/2001/XMLSchema" xmlns:beans="http://www.springframework.org/schema/beans" xmlns:xsd="http://www.w3.org/2001/XMLSchema" xmlns:beans="http://www.springframework.org/schema/beans"
xmlns:tool="http://www.springframework.org/schema/tool" xmlns:tool="http://www.springframework.org/schema/tool"
xmlns:integration="http://www.springframework.org/schema/integration" xmlns:integration="http://www.springframework.org/schema/integration"
targetNamespace="http://www.springframework.org/schema/integration/kafka" targetNamespace="http://www.springframework.org/schema/integration/kafka"
elementFormDefault="qualified" attributeFormDefault="unqualified"> elementFormDefault="qualified" attributeFormDefault="unqualified">
<xsd:import namespace="http://www.springframework.org/schema/beans"/> <xsd:import namespace="http://www.springframework.org/schema/beans"/>
<xsd:import namespace="http://www.springframework.org/schema/tool"/> <xsd:import namespace="http://www.springframework.org/schema/tool"/>
<xsd:import namespace="http://www.springframework.org/schema/integration" <xsd:import namespace="http://www.springframework.org/schema/integration"
schemaLocation="http://www.springframework.org/schema/integration/spring-integration-4.0.xsd"/> schemaLocation="http://www.springframework.org/schema/integration/spring-integration-4.0.xsd"/>
<xsd:annotation> <xsd:annotation>
<xsd:documentation><![CDATA[ <xsd:documentation><![CDATA[
Defines the configuration elements for the Spring Integration Defines the configuration elements for the Spring Integration
Kafka Adapter. Kafka Adapter.
]]></xsd:documentation> ]]></xsd:documentation>
</xsd:annotation> </xsd:annotation>
<xsd:element name="zookeeper-connect"> <xsd:element name="zookeeper-connect">
<xsd:annotation> <xsd:annotation>
<xsd:documentation><![CDATA[ <xsd:documentation><![CDATA[
Defines a Kafka server information. Defines a Kafka server information.
]]></xsd:documentation> ]]></xsd:documentation>
</xsd:annotation> </xsd:annotation>
<xsd:complexType> <xsd:complexType>
<xsd:attribute name="id" type="xsd:string" use="required"/> <xsd:attribute name="id" type="xsd:string" use="required"/>
<xsd:attribute name="zk-connect" use="optional"> <xsd:attribute name="zk-connect" use="optional">
<xsd:annotation> <xsd:annotation>
<xsd:documentation><![CDATA[ <xsd:documentation><![CDATA[
Indicates the Kafka server URL Indicates the Kafka server URL
]]></xsd:documentation> ]]></xsd:documentation>
<xsd:appinfo> <xsd:appinfo>
<tool:annotation kind="direct"> <tool:annotation kind="direct">
<tool:expected-type type="java.lang.String"/> <tool:expected-type type="java.lang.String"/>
</tool:annotation> </tool:annotation>
</xsd:appinfo> </xsd:appinfo>
</xsd:annotation> </xsd:annotation>
</xsd:attribute> </xsd:attribute>
<xsd:attribute name="zk-connection-timeout" use="optional"> <xsd:attribute name="zk-connection-timeout" use="optional">
<xsd:annotation> <xsd:annotation>
<xsd:documentation><![CDATA[ <xsd:documentation><![CDATA[
Indicates the Kafka consumer zkConnectionTimeout value Indicates the Kafka consumer zkConnectionTimeout value
]]></xsd:documentation> ]]></xsd:documentation>
<xsd:appinfo> <xsd:appinfo>
<tool:annotation kind="direct"> <tool:annotation kind="direct">
<tool:expected-type type="java.lang.Integer"/> <tool:expected-type type="java.lang.Integer"/>
</tool:annotation> </tool:annotation>
</xsd:appinfo> </xsd:appinfo>
</xsd:annotation> </xsd:annotation>
</xsd:attribute> </xsd:attribute>
<xsd:attribute name="zk-session-timeout" use="optional"> <xsd:attribute name="zk-session-timeout" use="optional">
<xsd:annotation> <xsd:annotation>
<xsd:documentation><![CDATA[ <xsd:documentation><![CDATA[
Indicates the Kafka consumer group id Indicates the Kafka consumer group id
]]></xsd:documentation> ]]></xsd:documentation>
<xsd:appinfo> <xsd:appinfo>
<tool:annotation kind="direct"> <tool:annotation kind="direct">
<tool:expected-type type="java.lang.String"/> <tool:expected-type type="java.lang.String"/>
</tool:annotation> </tool:annotation>
</xsd:appinfo> </xsd:appinfo>
</xsd:annotation> </xsd:annotation>
</xsd:attribute> </xsd:attribute>
<xsd:attribute name="zk-sync-time" use="optional"> <xsd:attribute name="zk-sync-time" use="optional">
<xsd:annotation> <xsd:annotation>
<xsd:documentation><![CDATA[ <xsd:documentation><![CDATA[
Indicates the Kafka consumer group id Indicates the Kafka consumer group id
]]></xsd:documentation> ]]></xsd:documentation>
<xsd:appinfo> <xsd:appinfo>
<tool:annotation kind="direct"> <tool:annotation kind="direct">
<tool:expected-type type="java.lang.String"/> <tool:expected-type type="java.lang.String"/>
</tool:annotation> </tool:annotation>
</xsd:appinfo> </xsd:appinfo>
</xsd:annotation> </xsd:annotation>
</xsd:attribute> </xsd:attribute>
<xsd:attribute name="auto-commit-interval" use="optional"> <xsd:attribute name="auto-commit-interval" use="optional">
<xsd:annotation> <xsd:annotation>
<xsd:documentation><![CDATA[ <xsd:documentation><![CDATA[
Indicates the Kafka consumer group id Indicates the Kafka consumer group id
]]></xsd:documentation> ]]></xsd:documentation>
<xsd:appinfo> <xsd:appinfo>
<tool:annotation kind="direct"> <tool:annotation kind="direct">
<tool:expected-type type="java.lang.String"/> <tool:expected-type type="java.lang.String"/>
</tool:annotation> </tool:annotation>
</xsd:appinfo> </xsd:appinfo>
</xsd:annotation> </xsd:annotation>
</xsd:attribute> </xsd:attribute>
</xsd:complexType> </xsd:complexType>
</xsd:element> </xsd:element>
<xsd:element name="producer-context"> <xsd:element name="producer-context">
<xsd:annotation> <xsd:annotation>
<xsd:documentation><![CDATA[ <xsd:documentation><![CDATA[
Defines a producer context. Defines a producer context.
]]></xsd:documentation> ]]></xsd:documentation>
</xsd:annotation> </xsd:annotation>
<xsd:complexType> <xsd:complexType>
<xsd:sequence> <xsd:sequence>
<xsd:element name="producer-configurations" minOccurs="0" maxOccurs="1"> <xsd:element name="producer-configurations" minOccurs="0" maxOccurs="1">
<xsd:annotation> <xsd:annotation>
<xsd:documentation><![CDATA[ <xsd:documentation><![CDATA[
Groups kafka topic configurations. Groups kafka topic configurations.
]]></xsd:documentation> ]]></xsd:documentation>
</xsd:annotation> </xsd:annotation>
<xsd:complexType> <xsd:complexType>
<xsd:choice> <xsd:choice>
<xsd:element name="producer-configuration" maxOccurs="unbounded"> <xsd:element name="producer-configuration" maxOccurs="unbounded">
<xsd:annotation> <xsd:annotation>
<xsd:documentation><![CDATA[ <xsd:documentation><![CDATA[
Declares a kafka topic configuration which drives how and where a topic is sent to broker/s. Declares a kafka topic configuration which drives how and where a topic is sent to broker/s.
]]></xsd:documentation> ]]></xsd:documentation>
</xsd:annotation> </xsd:annotation>
<xsd:complexType> <xsd:complexType>
<xsd:attribute name="topic" type="xsd:string" use="required"> <xsd:attribute name="topic" type="xsd:string" use="required">
<xsd:annotation> <xsd:annotation>
<xsd:documentation> <xsd:documentation>
The topic configured by this configuration. The topic configured by this configuration.
</xsd:documentation> </xsd:documentation>
</xsd:annotation> </xsd:annotation>
</xsd:attribute> </xsd:attribute>
<xsd:attribute name="broker-list" use="required"> <xsd:attribute name="broker-list" use="required">
<xsd:annotation> <xsd:annotation>
<xsd:documentation><![CDATA[ <xsd:documentation><![CDATA[
list of comma separated kafka brokers. list of comma separated kafka brokers.
]]></xsd:documentation> ]]></xsd:documentation>
<xsd:appinfo> <xsd:appinfo>
<tool:annotation kind="direct"> <tool:annotation kind="direct">
<tool:expected-type type="java.lang.String"/> <tool:expected-type type="java.lang.String"/>
</tool:annotation> </tool:annotation>
</xsd:appinfo> </xsd:appinfo>
</xsd:annotation> </xsd:annotation>
</xsd:attribute> </xsd:attribute>
<xsd:attribute name="value-encoder" use="optional" type="xsd:string"> <xsd:attribute name="value-encoder" use="optional" type="xsd:string">
<xsd:annotation> <xsd:annotation>
<xsd:documentation> <xsd:documentation>
Custom implemenation of a Kafka Encoder for encoding message values. Custom implemenation of a Kafka Encoder for encoding message values.
</xsd:documentation> </xsd:documentation>
</xsd:annotation> </xsd:annotation>
</xsd:attribute> </xsd:attribute>
<xsd:attribute name="key-encoder" use="optional" <xsd:attribute name="key-encoder" use="optional"
type="xsd:string"> type="xsd:string">
<xsd:annotation> <xsd:annotation>
<xsd:documentation> <xsd:documentation>
Custom implemenation of a Kafka Encoder for encoding message keys. Custom implemenation of a Kafka Encoder for encoding message keys.
</xsd:documentation> </xsd:documentation>
</xsd:annotation> </xsd:annotation>
</xsd:attribute> </xsd:attribute>
<xsd:attribute name="key-class-type" use="optional" <xsd:attribute name="key-class-type" use="optional"
type="xsd:string"> type="xsd:string">
<xsd:annotation> <xsd:annotation>
<xsd:documentation> <xsd:documentation>
Class type used for the key Class type used for the key
</xsd:documentation> </xsd:documentation>
</xsd:annotation> </xsd:annotation>
</xsd:attribute> </xsd:attribute>
<xsd:attribute name="value-class-type" use="optional" <xsd:attribute name="value-class-type" use="optional"
type="xsd:string"> type="xsd:string">
<xsd:annotation> <xsd:annotation>
<xsd:documentation> <xsd:documentation>
Class type used for the value Class type used for the value
</xsd:documentation> </xsd:documentation>
</xsd:annotation> </xsd:annotation>
</xsd:attribute> </xsd:attribute>
<xsd:attribute name="compression-codec" use="optional"> <xsd:attribute name="compression-codec" use="optional">
<xsd:annotation> <xsd:annotation>
<xsd:documentation><![CDATA[ <xsd:documentation><![CDATA[
Indicates the type of compression codec used for message compression. Indicates the type of compression codec used for message compression.
]]></xsd:documentation> ]]></xsd:documentation>
<xsd:appinfo> <xsd:appinfo>
<tool:annotation kind="direct"> <tool:annotation kind="direct">
<tool:expected-type type="java.lang.String"/> <tool:expected-type type="java.lang.String"/>
</tool:annotation> </tool:annotation>
</xsd:appinfo> </xsd:appinfo>
</xsd:annotation> </xsd:annotation>
</xsd:attribute> </xsd:attribute>
<xsd:attribute name="partitioner" use="optional" <xsd:attribute name="partitioner" use="optional"
type="xsd:string"> type="xsd:string">
<xsd:annotation> <xsd:annotation>
<xsd:documentation> <xsd:documentation>
Custom Kafka key partitioner. Custom Kafka key partitioner.
</xsd:documentation> </xsd:documentation>
</xsd:annotation> </xsd:annotation>
</xsd:attribute> </xsd:attribute>
<xsd:attribute name="async" use="optional" <xsd:attribute name="async" use="optional"
type="xsd:boolean"> type="xsd:boolean">
<xsd:annotation> <xsd:annotation>
<xsd:documentation> <xsd:documentation>
Indicates if this producer is async or not. Indicates if this producer is async or not.
</xsd:documentation> </xsd:documentation>
</xsd:annotation> </xsd:annotation>
</xsd:attribute> </xsd:attribute>
<xsd:attribute name="batch-num-messages" use="optional" <xsd:attribute name="batch-num-messages" use="optional"
type="xsd:string"> type="xsd:string">
<xsd:annotation> <xsd:annotation>
<xsd:documentation> <xsd:documentation>
number of messages to batch at this producer. number of messages to batch at this producer.
</xsd:documentation> </xsd:documentation>
</xsd:annotation> </xsd:annotation>
</xsd:attribute> </xsd:attribute>
</xsd:complexType> </xsd:complexType>
</xsd:element> </xsd:element>
</xsd:choice> </xsd:choice>
</xsd:complexType> </xsd:complexType>
</xsd:element> </xsd:element>
</xsd:sequence> </xsd:sequence>
<xsd:attribute name="id" type="xsd:string" use="required"/> <xsd:attribute name="id" type="xsd:string" use="required"/>
<xsd:attribute name="producer-properties" use="optional" <xsd:attribute name="producer-properties" use="optional"
type="xsd:string"> type="xsd:string">
<xsd:annotation> <xsd:annotation>
<xsd:documentation> <xsd:documentation>
Kafka producer properties to use for all producers Kafka producer properties to use for all producers
</xsd:documentation> </xsd:documentation>
</xsd:annotation> </xsd:annotation>
</xsd:attribute> </xsd:attribute>
</xsd:complexType> </xsd:complexType>
</xsd:element> </xsd:element>
<xsd:element name="consumer-context"> <xsd:element name="consumer-context">
<xsd:annotation> <xsd:annotation>
<xsd:documentation><![CDATA[ <xsd:documentation><![CDATA[
Defines a producer context. Defines a producer context.
]]></xsd:documentation> ]]></xsd:documentation>
</xsd:annotation> </xsd:annotation>
<xsd:complexType> <xsd:complexType>
<xsd:sequence> <xsd:sequence>
<xsd:element name="consumer-configurations" minOccurs="0" maxOccurs="1"> <xsd:element name="consumer-configurations" minOccurs="0" maxOccurs="1">
<xsd:annotation> <xsd:annotation>
<xsd:documentation><![CDATA[ <xsd:documentation><![CDATA[
Groups kafka topic configurations. Groups kafka topic configurations.
]]></xsd:documentation> ]]></xsd:documentation>
</xsd:annotation> </xsd:annotation>
<xsd:complexType> <xsd:complexType>
<xsd:choice> <xsd:choice>
<xsd:element name="consumer-configuration" maxOccurs="unbounded"> <xsd:element name="consumer-configuration" maxOccurs="unbounded">
<xsd:annotation> <xsd:annotation>
<xsd:documentation><![CDATA[ <xsd:documentation><![CDATA[
Declares a kafka topic configuration which drives how and where a topic is sent to broker/s. Declares a kafka topic configuration which drives how and where a topic is sent to broker/s.
]]></xsd:documentation> ]]></xsd:documentation>
</xsd:annotation> </xsd:annotation>
<xsd:complexType> <xsd:complexType>
<xsd:choice> <xsd:choice>
<xsd:element name="topic" maxOccurs="unbounded"> <xsd:element name="topic" maxOccurs="unbounded">
<xsd:complexType> <xsd:complexType>
<xsd:attribute name="id" type="xsd:string" use="required"/> <xsd:attribute name="id" type="xsd:string" use="required"/>
<xsd:attribute name="streams" type="xsd:string" use="required"/> <xsd:attribute name="streams" type="xsd:string" use="required"/>
</xsd:complexType> </xsd:complexType>
</xsd:element> </xsd:element>
<xsd:element name="topic-filter" maxOccurs="1"> <xsd:element name="topic-filter" maxOccurs="1">
<xsd:complexType> <xsd:complexType>
<xsd:attribute name="pattern" type="xsd:string" use="required"> <xsd:attribute name="pattern" type="xsd:string" use="required">
<xsd:annotation> <xsd:annotation>
<xsd:documentation><![CDATA[ <xsd:documentation><![CDATA[
Regex pattern to match topic Regex pattern to match topic
]]></xsd:documentation> ]]></xsd:documentation>
<xsd:appinfo> <xsd:appinfo>
<tool:annotation kind="direct"> <tool:annotation kind="direct">
<tool:expected-type type="java.lang.String"/> <tool:expected-type type="java.lang.String"/>
</tool:annotation> </tool:annotation>
</xsd:appinfo> </xsd:appinfo>
</xsd:annotation> </xsd:annotation>
</xsd:attribute> </xsd:attribute>
<xsd:attribute name="streams" type="xsd:string" use="required"> <xsd:attribute name="streams" type="xsd:string" use="required">
<xsd:annotation> <xsd:annotation>
<xsd:documentation><![CDATA[ <xsd:documentation><![CDATA[
Number of streams (threads) to use to consume messages Number of streams (threads) to use to consume messages
]]></xsd:documentation> ]]></xsd:documentation>
<xsd:appinfo> <xsd:appinfo>
<tool:annotation kind="direct"> <tool:annotation kind="direct">
<tool:expected-type type="java.lang.String"/> <tool:expected-type type="java.lang.String"/>
</tool:annotation> </tool:annotation>
</xsd:appinfo> </xsd:appinfo>
</xsd:annotation> </xsd:annotation>
</xsd:attribute> </xsd:attribute>
<xsd:attribute name="exclude" type="xsd:boolean" use="optional" default="false"> <xsd:attribute name="exclude" use="optional" default="false">
<xsd:annotation> <xsd:annotation>
<xsd:documentation><![CDATA[ <xsd:documentation><![CDATA[
If exclude is false, it uses whitelist to include topics matching given pattern. If exclude is false, it uses whitelist to include topics matching given pattern.
If exclude is true, it uses blacklist to exclude topics matching given patttern. If exclude is true, it uses blacklist to exclude topics matching given pattern.
Default value is false. Default value is false.
]]></xsd:documentation> ]]></xsd:documentation>
<xsd:appinfo> <xsd:appinfo>
<tool:annotation kind="direct"> <tool:annotation kind="direct">
<tool:expected-type type="java.lang.String"/> <tool:expected-type type="java.lang.String"/>
</tool:annotation> </tool:annotation>
</xsd:appinfo> </xsd:appinfo>
</xsd:annotation> </xsd:annotation>
</xsd:attribute> <xsd:simpleType>
</xsd:complexType> <xsd:union memberTypes="xsd:boolean xsd:string"/>
</xsd:element> </xsd:simpleType>
</xsd:choice> </xsd:attribute>
<xsd:attribute name="max-messages" use="optional"> </xsd:complexType>
<xsd:annotation> </xsd:element>
<xsd:documentation><![CDATA[ </xsd:choice>
<xsd:attribute name="max-messages" use="optional">
<xsd:annotation>
<xsd:documentation><![CDATA[
Indicates max messages to aggregate in a single call to receive Indicates max messages to aggregate in a single call to receive
]]></xsd:documentation> ]]></xsd:documentation>
<xsd:appinfo> <xsd:appinfo>
<tool:annotation kind="direct"> <tool:annotation kind="direct">
<tool:expected-type type="java.lang.String"/> <tool:expected-type type="java.lang.String"/>
</tool:annotation> </tool:annotation>
</xsd:appinfo> </xsd:appinfo>
</xsd:annotation> </xsd:annotation>
</xsd:attribute> </xsd:attribute>
<xsd:attribute name="group-id" use="required"> <xsd:attribute name="group-id" use="required">
<xsd:annotation> <xsd:annotation>
<xsd:documentation><![CDATA[ <xsd:documentation><![CDATA[
Indicates the Kafka consumer group id Indicates the Kafka consumer group id
]]></xsd:documentation> ]]></xsd:documentation>
<xsd:appinfo> <xsd:appinfo>
<tool:annotation kind="direct"> <tool:annotation kind="direct">
<tool:expected-type type="java.lang.String"/> <tool:expected-type type="java.lang.String"/>
</tool:annotation> </tool:annotation>
</xsd:appinfo> </xsd:appinfo>
</xsd:annotation> </xsd:annotation>
</xsd:attribute> </xsd:attribute>
<xsd:attribute name="value-decoder" use="optional" <xsd:attribute name="value-decoder" use="optional"
type="xsd:string"> type="xsd:string">
<xsd:annotation> <xsd:annotation>
<xsd:documentation> <xsd:documentation>
Kafka Server Bean Name Kafka Server Bean Name
</xsd:documentation> </xsd:documentation>
</xsd:annotation> </xsd:annotation>
</xsd:attribute> </xsd:attribute>
<xsd:attribute name="key-decoder" use="optional" <xsd:attribute name="key-decoder" use="optional"
type="xsd:string"> type="xsd:string">
<xsd:annotation> <xsd:annotation>
<xsd:documentation> <xsd:documentation>
Kafka Server Bean Name Kafka Server Bean Name
</xsd:documentation> </xsd:documentation>
</xsd:annotation> </xsd:annotation>
</xsd:attribute> </xsd:attribute>
</xsd:complexType> </xsd:complexType>
</xsd:element> </xsd:element>
</xsd:choice> </xsd:choice>
</xsd:complexType> </xsd:complexType>
</xsd:element> </xsd:element>
</xsd:sequence> </xsd:sequence>
<xsd:attribute name="id" type="xsd:string" use="required"/> <xsd:attribute name="id" type="xsd:string" use="required"/>
<xsd:attribute name="consumer-timeout" use="optional"> <xsd:attribute name="consumer-timeout" use="optional">
<xsd:annotation> <xsd:annotation>
<xsd:documentation><![CDATA[ <xsd:documentation><![CDATA[
Indicates the Kafka consumer timeout ms Indicates the Kafka consumer timeout ms
]]></xsd:documentation> ]]></xsd:documentation>
<xsd:appinfo> <xsd:appinfo>
<tool:annotation kind="direct"> <tool:annotation kind="direct">
<tool:expected-type type="java.lang.String"/> <tool:expected-type type="java.lang.String"/>
</tool:annotation> </tool:annotation>
</xsd:appinfo> </xsd:appinfo>
</xsd:annotation> </xsd:annotation>
</xsd:attribute> </xsd:attribute>
<xsd:attribute name="zookeeper-connect" use="required" <xsd:attribute name="zookeeper-connect" use="required"
type="xsd:string"> type="xsd:string">
<xsd:annotation> <xsd:annotation>
<xsd:documentation> <xsd:documentation>
Kafka Server Bean Name Kafka Server Bean Name
</xsd:documentation> </xsd:documentation>
</xsd:annotation> </xsd:annotation>
</xsd:attribute> </xsd:attribute>
<xsd:attribute name="consumer-properties" use="optional" <xsd:attribute name="consumer-properties" use="optional"
type="xsd:string"> type="xsd:string">
<xsd:annotation> <xsd:annotation>
<xsd:documentation> <xsd:documentation>
Kafka consumer properties to use for all consumers Kafka consumer properties to use for all consumers
</xsd:documentation> </xsd:documentation>
</xsd:annotation> </xsd:annotation>
</xsd:attribute> </xsd:attribute>
</xsd:complexType> </xsd:complexType>
</xsd:element> </xsd:element>
<xsd:element name="inbound-channel-adapter"> <xsd:element name="inbound-channel-adapter">
<xsd:annotation> <xsd:annotation>
<xsd:documentation> <xsd:documentation>
The definition for the Spring Integration Kafka The definition for the Spring Integration Kafka
Inbound Channel Adapter. Inbound Channel Adapter.
</xsd:documentation> </xsd:documentation>
</xsd:annotation> </xsd:annotation>
<xsd:complexType> <xsd:complexType>
<xsd:sequence> <xsd:sequence>
<xsd:element ref="integration:poller" minOccurs="0" maxOccurs="1"/> <xsd:element ref="integration:poller" minOccurs="0" maxOccurs="1"/>
</xsd:sequence> </xsd:sequence>
<xsd:attributeGroup ref="integration:channelAdapterAttributes"/> <xsd:attributeGroup ref="integration:channelAdapterAttributes"/>
<xsd:attribute name="send-timeout" type="xsd:string"> <xsd:attribute name="send-timeout" type="xsd:string">
<xsd:annotation> <xsd:annotation>
<xsd:documentation><![CDATA[ <xsd:documentation><![CDATA[
Allows you to specify how long this inbound-channel-adapter Allows you to specify how long this inbound-channel-adapter
will wait for the message (containing the retrieved entities) will wait for the message (containing the retrieved entities)
to be sent successfully to the message channel, before throwing to be sent successfully to the message channel, before throwing
@@ -395,59 +398,59 @@
further downstream. By default the Inbound Channel Adapter further downstream. By default the Inbound Channel Adapter
will wait indefinitely. The value is specified in milliseconds. will wait indefinitely. The value is specified in milliseconds.
]]> ]]>
</xsd:documentation> </xsd:documentation>
</xsd:annotation> </xsd:annotation>
</xsd:attribute> </xsd:attribute>
<xsd:attribute name="group-id" use="optional"> <xsd:attribute name="group-id" use="optional">
<xsd:annotation> <xsd:annotation>
<xsd:documentation><![CDATA[ <xsd:documentation><![CDATA[
Indicates the Kafka consumer group id Indicates the Kafka consumer group id
]]></xsd:documentation> ]]></xsd:documentation>
<xsd:appinfo> <xsd:appinfo>
<tool:annotation kind="direct"> <tool:annotation kind="direct">
<tool:expected-type type="java.lang.String"/> <tool:expected-type type="java.lang.String"/>
</tool:annotation> </tool:annotation>
</xsd:appinfo> </xsd:appinfo>
</xsd:annotation> </xsd:annotation>
</xsd:attribute> </xsd:attribute>
<xsd:attribute name="kafka-consumer-context-ref" use="required" type="xsd:string"> <xsd:attribute name="kafka-consumer-context-ref" use="required" type="xsd:string">
<xsd:annotation> <xsd:annotation>
<xsd:documentation> <xsd:documentation>
Kafka Server Bean Name Kafka Server Bean Name
</xsd:documentation> </xsd:documentation>
</xsd:annotation> </xsd:annotation>
</xsd:attribute> </xsd:attribute>
</xsd:complexType> </xsd:complexType>
</xsd:element> </xsd:element>
<xsd:element name="outbound-channel-adapter"> <xsd:element name="outbound-channel-adapter">
<xsd:annotation> <xsd:annotation>
<xsd:documentation> <xsd:documentation>
Defines kafka outbound channel adapter that writes the contents of the Defines kafka outbound channel adapter that writes the contents of the
Message to kafka broker. Message to kafka broker.
</xsd:documentation> </xsd:documentation>
</xsd:annotation> </xsd:annotation>
<xsd:complexType> <xsd:complexType>
<xsd:sequence> <xsd:sequence>
<xsd:element ref="integration:poller" minOccurs="0" maxOccurs="1"/> <xsd:element ref="integration:poller" minOccurs="0" maxOccurs="1"/>
</xsd:sequence> </xsd:sequence>
<xsd:attributeGroup ref="integration:channelAdapterAttributes"/> <xsd:attributeGroup ref="integration:channelAdapterAttributes"/>
<xsd:attribute name="kafka-producer-context-ref" use="required" type="xsd:string"> <xsd:attribute name="kafka-producer-context-ref" use="required" type="xsd:string">
<xsd:annotation> <xsd:annotation>
<xsd:documentation> <xsd:documentation>
Kafka producer context reference. Kafka producer context reference.
</xsd:documentation> </xsd:documentation>
</xsd:annotation> </xsd:annotation>
</xsd:attribute> </xsd:attribute>
<xsd:attribute name="order"> <xsd:attribute name="order">
<xsd:annotation> <xsd:annotation>
<xsd:documentation> <xsd:documentation>
Specifies the order for invocation when this endpoint is connected as a Specifies the order for invocation when this endpoint is connected as a
subscriber to a SubscribableChannel. subscriber to a SubscribableChannel.
</xsd:documentation> </xsd:documentation>
</xsd:annotation> </xsd:annotation>
</xsd:attribute> </xsd:attribute>
</xsd:complexType> </xsd:complexType>
</xsd:element> </xsd:element>
</xsd:schema> </xsd:schema>

24
spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/KafkaConsumerContextParserTests-context.xml
View File

@@ -1,9 +1,11 @@
<?xml version="1.0" encoding="UTF-8"?> <?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans" <beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:int-kafka="http://www.springframework.org/schema/integration/kafka" xmlns:int-kafka="http://www.springframework.org/schema/integration/kafka"
xsi:schemaLocation="http://www.springframework.org/schema/integration/kafka http://www.springframework.org/schema/integration/kafka/spring-integration-kafka.xsd xmlns:context="http://www.springframework.org/schema/context"
http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd"> xmlns:util="http://www.springframework.org/schema/util"
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/util http://www.springframework.org/schema/util/spring-util.xsd">
<int-kafka:zookeeper-connect id="zookeeperConnect" zk-connect="localhost:2181" zk-connection-timeout="6000" <int-kafka:zookeeper-connect id="zookeeperConnect" zk-connect="localhost:2181" zk-connection-timeout="6000"
zk-session-timeout="6000" zk-session-timeout="6000"
@@ -25,6 +27,14 @@
</property> </property>
</bean> </bean>
<util:properties id="placeholderProperties">
<prop key="topic.filter.pattern">foo</prop>
<prop key="topic.filter.streams">10</prop>
<prop key="topic.filter.exclude">true</prop>
</util:properties>
<context:property-placeholder properties-ref="placeholderProperties"/>
<int-kafka:consumer-context id="consumerContext" <int-kafka:consumer-context id="consumerContext"
consumer-timeout="4000" consumer-timeout="4000"
zookeeper-connect="zookeeperConnect" zookeeper-connect="zookeeperConnect"
@@ -34,9 +44,11 @@
value-decoder="valueDecoder" value-decoder="valueDecoder"
key-decoder="valueDecoder" key-decoder="valueDecoder"
max-messages="5000"> max-messages="5000">
<int-kafka:topic id="test1" streams="4"/> <int-kafka:topic-filter pattern="${topic.filter.pattern}"
<int-kafka:topic id="test2" streams="4"/> streams="${topic.filter.streams}"
exclude="${topic.filter.exclude}"/>
</int-kafka:consumer-configuration> </int-kafka:consumer-configuration>
</int-kafka:consumer-configurations> </int-kafka:consumer-configurations>
</int-kafka:consumer-context> </int-kafka:consumer-context>
</beans> </beans>

25
spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/KafkaConsumerContextParserTests.java
View File

@@ -1,5 +1,5 @@
/* /*
* Copyright 2002-2013 the original author or authors. * Copyright 2002-2014 the original author or authors.
* *
* Licensed under the Apache License, Version 2.0 (the "License"); * Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License. * you may not use this file except in compliance with the License.
@@ -15,23 +15,29 @@
*/ */
package org.springframework.integration.kafka.config.xml; package org.springframework.integration.kafka.config.xml;
import org.junit.Assert; import static org.junit.Assert.*;
import kafka.consumer.Blacklist;
import org.hamcrest.Matchers;
import org.junit.Test; import org.junit.Test;
import org.junit.runner.RunWith; import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ApplicationContext; import org.springframework.context.ApplicationContext;
import org.springframework.integration.kafka.support.ConsumerMetadata; import org.springframework.integration.kafka.support.ConsumerMetadata;
import org.springframework.integration.kafka.support.KafkaConsumerContext; import org.springframework.integration.kafka.support.KafkaConsumerContext;
import org.springframework.integration.kafka.support.TopicFilterConfiguration;
import org.springframework.test.context.ContextConfiguration; import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
/** /**
* @author Soby Chacko * @author Soby Chacko
* @author Artem Bilan
* @since 0.5 * @since 0.5
*/ */
@RunWith(SpringJUnit4ClassRunner.class) @RunWith(SpringJUnit4ClassRunner.class)
@ContextConfiguration @ContextConfiguration
public class KafkaConsumerContextParserTests<K,V> { public class KafkaConsumerContextParserTests<K, V> {
@Autowired @Autowired
private ApplicationContext appContext; private ApplicationContext appContext;
@@ -39,10 +45,15 @@ public class KafkaConsumerContextParserTests<K,V> {
@Test @Test
@SuppressWarnings("unchecked") @SuppressWarnings("unchecked")
public void testConsumerContextConfiguration() { public void testConsumerContextConfiguration() {
final KafkaConsumerContext<K,V> consumerContext = appContext.getBean("consumerContext", KafkaConsumerContext.class); final KafkaConsumerContext<K, V> consumerContext =
Assert.assertNotNull(consumerContext); appContext.getBean("consumerContext", KafkaConsumerContext.class);
assertNotNull(consumerContext);
final ConsumerMetadata<K,V> cm = appContext.getBean("consumerMetadata_default1", ConsumerMetadata.class); ConsumerMetadata<K, V> cm = appContext.getBean("consumerMetadata_default1", ConsumerMetadata.class);
Assert.assertNotNull(cm); assertNotNull(cm);
TopicFilterConfiguration topicFilterConfiguration = cm.getTopicFilterConfiguration();
assertEquals("foo : 10", topicFilterConfiguration.toString());
assertThat(topicFilterConfiguration.getTopicFilter(), Matchers.instanceOf(Blacklist.class));
} }
} }

7
spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/kafkaInboundAdapterCommon-context.xml
View File

@@ -9,9 +9,10 @@
http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.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/context http://www.springframework.org/schema/context/spring-context.xsd">
<context:component-scan base-package="org.springframework.integration.kafka.inbound"></context:component-scan> <context:component-scan base-package="org.springframework.integration.kafka.inbound"/>
<int:channel id="inputFromKafka">
</int:channel> <int:channel id="inputFromKafka"/>
<stream:stdout-channel-adapter id="stdout" channel="inputFromKafka" append-newline="true"/> <stream:stdout-channel-adapter id="stdout" channel="inputFromKafka" append-newline="true"/>
</beans> </beans>