diff --git a/samples/kafka/build.gradle b/samples/kafka/build.gradle index 0c3e6ee..5b02cbd 100644 --- a/samples/kafka/build.gradle +++ b/samples/kafka/build.gradle @@ -1,21 +1,19 @@ description = 'Spring Integration Kafka Sample' buildscript { - - repositories { - mavenCentral() - mavenLocal() - } - dependencies { - classpath 'org.apache.maven:maven-artifact:2.2.1' // 3.x won't work - classpath 'org.apache.avro:avro-compiler:1.7.3' // use Avro 1.7.4 to compile the Avro files - //classpath 'org.clojars.miguno:avro-gradle-plugin:1.7.2' - classpath "org.apache.avro:avro-gradle-plugin:1.7.2" - } - + repositories { + mavenCentral() + //mavenLocal() + maven { url 'https://repo.spring.io/simple/ext-release-local' } + } + dependencies { + classpath 'org.apache.maven:maven-artifact:2.2.1' // 3.x won't work + classpath 'org.apache.avro:avro-compiler:1.7.3' // use Avro 1.7.4 to compile the Avro files + //classpath 'org.clojars.miguno:avro-gradle-plugin:1.7.2' + classpath "org.apache.avro:avro-gradle-plugin:1.7.2" + } } - apply plugin: 'base' apply plugin: 'java' apply plugin: 'eclipse' @@ -25,46 +23,42 @@ apply plugin: 'maven' apply plugin: 'avro-gradle-plugin' ext { - avroTaskGroup = "Avro" - avroSource = "schemas" - avroDest = "target/generated-avro-sources/main/java" + avroTaskGroup = "Avro" + avroSource = "schemas" + avroDest = "target/generated-avro-sources/main/java" } repositories { - -mavenLocal() -maven { - url 'https://repository.apache.org/content/groups/public' - } - maven { url 'https://repo.springsource.org/libs-milestone' } - + maven { + url 'https://repository.apache.org/content/groups/public' + } + maven { url 'https://repo.springsource.org/libs-milestone' } + mavenLocal() } dependencies { - compile([ - "org.apache.avro:avro:1.7.3", - "org.apache.avro:avro-compiler:1.7.3" - ]) - compile "org.springframework:spring-beans:3.1.3.RELEASE" - compile "org.springframework:spring-context:3.1.3.RELEASE" - compile "org.springframework:spring-expression:3.1.3.RELEASE" - compile "org.springframework.integration:spring-integration-stream:2.2.0.RELEASE" + compile([ + "org.apache.avro:avro:1.7.3", + "org.apache.avro:avro-compiler:1.7.3" + ]) + compile "org.springframework:spring-beans:3.1.3.RELEASE" + compile "org.springframework:spring-context:3.1.3.RELEASE" + compile "org.springframework:spring-expression:3.1.3.RELEASE" + compile "org.springframework.integration:spring-integration-stream:2.2.0.RELEASE" compile("org.springframework.integration:spring-integration-kafka:0.5.0.BUILD-SNAPSHOT") { - exclude module: 'log4j' - exclude module: 'jms' - exclude module: 'jmxtools' - exclude module: 'jmxri' + exclude module: 'log4j' + exclude module: 'jms' + exclude module: 'jmxtools' + exclude module: 'jmxri' } -compile("log4j:log4j:1.2.15") { + compile("log4j:log4j:1.2.15") { exclude module: 'mail' - exclude module: 'jms' - exclude module: 'jmx' - exclude module: 'jmxtools' - exclude module: 'jmxri' - - } - - compile "commons-logging:commons-logging:1.1.1" + exclude module: 'jms' + exclude module: 'jmx' + exclude module: 'jmxtools' + exclude module: 'jmxri' + } + compile "commons-logging:commons-logging:1.1.1" } compileAvro.group = avroTaskGroup @@ -73,27 +67,27 @@ compileAvro.source = avroSource compileAvro.destinationDir = file(avroDest) task cleanAvro(type: Delete) { - group = avroTaskGroup - description = "deletes generated avro code" - delete avroDest + group = avroTaskGroup + description = "deletes generated avro code" + delete avroDest } compileJava.dependsOn compileAvro sourceSets { - main { - java { - srcDir avroDest - } - resources { - srcDir avroSource - } - } + main { + java { + srcDir avroDest + } + resources { + srcDir avroSource + } + } } task wrapper(type: Wrapper) { description = 'Generates gradlew[.bat] scripts' - gradleVersion = '1.3' + gradleVersion = '1.8' } eclipse { diff --git a/samples/kafka/gradle/wrapper/gradle-wrapper.jar b/samples/kafka/gradle/wrapper/gradle-wrapper.jar index e3b3376..7f1e239 100644 Binary files a/samples/kafka/gradle/wrapper/gradle-wrapper.jar and b/samples/kafka/gradle/wrapper/gradle-wrapper.jar differ diff --git a/samples/kafka/gradle/wrapper/gradle-wrapper.properties b/samples/kafka/gradle/wrapper/gradle-wrapper.properties index 2b88cdf..238e92a 100644 --- a/samples/kafka/gradle/wrapper/gradle-wrapper.properties +++ b/samples/kafka/gradle/wrapper/gradle-wrapper.properties @@ -1,6 +1,6 @@ -#Fri Sep 28 16:40:22 CST 2012 +#Mon Oct 28 16:21:39 EDT 2013 distributionBase=GRADLE_USER_HOME distributionPath=wrapper/dists zipStoreBase=GRADLE_USER_HOME zipStorePath=wrapper/dists -distributionUrl=http\://services.gradle.org/distributions/gradle-1.1-bin.zip +distributionUrl=http\://services.gradle.org/distributions/gradle-1.8-bin.zip diff --git a/samples/kafka/gradlew b/samples/kafka/gradlew index cf12650..e61422d 100755 --- a/samples/kafka/gradlew +++ b/samples/kafka/gradlew @@ -101,7 +101,7 @@ if [ "$cygwin" = "false" -a "$darwin" = "false" ] ; then warn "Could not set maximum file descriptor limit: $MAX_FD" fi else - warn "Could not query businessSystem maximum file descriptor limit: $MAX_FD_LIMIT" + warn "Could not query maximum file descriptor limit: $MAX_FD_LIMIT" fi fi diff --git a/samples/kafka/gradlew.bat b/samples/kafka/gradlew.bat index 8a0b282..aec9973 100644 --- a/samples/kafka/gradlew.bat +++ b/samples/kafka/gradlew.bat @@ -1,90 +1,90 @@ -@if "%DEBUG%" == "" @echo off -@rem ########################################################################## -@rem -@rem Gradle startup script for Windows -@rem -@rem ########################################################################## - -@rem Set local scope for the variables with windows NT shell -if "%OS%"=="Windows_NT" setlocal - -@rem Add default JVM options here. You can also use JAVA_OPTS and GRADLE_OPTS to pass JVM options to this script. -set DEFAULT_JVM_OPTS= - -set DIRNAME=%~dp0 -if "%DIRNAME%" == "" set DIRNAME=. -set APP_BASE_NAME=%~n0 -set APP_HOME=%DIRNAME% - -@rem Find java.exe -if defined JAVA_HOME goto findJavaFromJavaHome - -set JAVA_EXE=java.exe -%JAVA_EXE% -version >NUL 2>&1 -if "%ERRORLEVEL%" == "0" goto init - -echo. -echo ERROR: JAVA_HOME is not set and no 'java' command could be found in your PATH. -echo. -echo Please set the JAVA_HOME variable in your environment to match the -echo location of your Java installation. - -goto fail - -:findJavaFromJavaHome -set JAVA_HOME=%JAVA_HOME:"=% -set JAVA_EXE=%JAVA_HOME%/bin/java.exe - -if exist "%JAVA_EXE%" goto init - -echo. -echo ERROR: JAVA_HOME is set to an invalid directory: %JAVA_HOME% -echo. -echo Please set the JAVA_HOME variable in your environment to match the -echo location of your Java installation. - -goto fail - -:init -@rem Get command-line arguments, handling Windowz variants - -if not "%OS%" == "Windows_NT" goto win9xME_args -if "%@eval[2+2]" == "4" goto 4NT_args - -:win9xME_args -@rem Slurp the command line arguments. -set CMD_LINE_ARGS= -set _SKIP=2 - -:win9xME_args_slurp -if "x%~1" == "x" goto execute - -set CMD_LINE_ARGS=%* -goto execute - -:4NT_args -@rem Get arguments from the 4NT Shell from JP Software -set CMD_LINE_ARGS=%$ - -:execute -@rem Setup the command line - -set CLASSPATH=%APP_HOME%\gradle\wrapper\gradle-wrapper.jar - -@rem Execute Gradle -"%JAVA_EXE%" %DEFAULT_JVM_OPTS% %JAVA_OPTS% %GRADLE_OPTS% "-Dorg.gradle.appname=%APP_BASE_NAME%" -classpath "%CLASSPATH%" org.gradle.wrapper.GradleWrapperMain %CMD_LINE_ARGS% - -:end -@rem End local scope for the variables with windows NT shell -if "%ERRORLEVEL%"=="0" goto mainEnd - -:fail -rem Set variable GRADLE_EXIT_CONSOLE if you need the _script_ return code instead of -rem the _cmd.exe /c_ return code! -if not "" == "%GRADLE_EXIT_CONSOLE%" exit 1 -exit /b 1 - -:mainEnd -if "%OS%"=="Windows_NT" endlocal - -:omega +@if "%DEBUG%" == "" @echo off +@rem ########################################################################## +@rem +@rem Gradle startup script for Windows +@rem +@rem ########################################################################## + +@rem Set local scope for the variables with windows NT shell +if "%OS%"=="Windows_NT" setlocal + +@rem Add default JVM options here. You can also use JAVA_OPTS and GRADLE_OPTS to pass JVM options to this script. +set DEFAULT_JVM_OPTS= + +set DIRNAME=%~dp0 +if "%DIRNAME%" == "" set DIRNAME=. +set APP_BASE_NAME=%~n0 +set APP_HOME=%DIRNAME% + +@rem Find java.exe +if defined JAVA_HOME goto findJavaFromJavaHome + +set JAVA_EXE=java.exe +%JAVA_EXE% -version >NUL 2>&1 +if "%ERRORLEVEL%" == "0" goto init + +echo. +echo ERROR: JAVA_HOME is not set and no 'java' command could be found in your PATH. +echo. +echo Please set the JAVA_HOME variable in your environment to match the +echo location of your Java installation. + +goto fail + +:findJavaFromJavaHome +set JAVA_HOME=%JAVA_HOME:"=% +set JAVA_EXE=%JAVA_HOME%/bin/java.exe + +if exist "%JAVA_EXE%" goto init + +echo. +echo ERROR: JAVA_HOME is set to an invalid directory: %JAVA_HOME% +echo. +echo Please set the JAVA_HOME variable in your environment to match the +echo location of your Java installation. + +goto fail + +:init +@rem Get command-line arguments, handling Windowz variants + +if not "%OS%" == "Windows_NT" goto win9xME_args +if "%@eval[2+2]" == "4" goto 4NT_args + +:win9xME_args +@rem Slurp the command line arguments. +set CMD_LINE_ARGS= +set _SKIP=2 + +:win9xME_args_slurp +if "x%~1" == "x" goto execute + +set CMD_LINE_ARGS=%* +goto execute + +:4NT_args +@rem Get arguments from the 4NT Shell from JP Software +set CMD_LINE_ARGS=%$ + +:execute +@rem Setup the command line + +set CLASSPATH=%APP_HOME%\gradle\wrapper\gradle-wrapper.jar + +@rem Execute Gradle +"%JAVA_EXE%" %DEFAULT_JVM_OPTS% %JAVA_OPTS% %GRADLE_OPTS% "-Dorg.gradle.appname=%APP_BASE_NAME%" -classpath "%CLASSPATH%" org.gradle.wrapper.GradleWrapperMain %CMD_LINE_ARGS% + +:end +@rem End local scope for the variables with windows NT shell +if "%ERRORLEVEL%"=="0" goto mainEnd + +:fail +rem Set variable GRADLE_EXIT_CONSOLE if you need the _script_ return code instead of +rem the _cmd.exe /c_ return code! +if not "" == "%GRADLE_EXIT_CONSOLE%" exit 1 +exit /b 1 + +:mainEnd +if "%OS%"=="Windows_NT" endlocal + +:omega diff --git a/samples/kafka/src/main/java/org/springframework/integration/samples/kafka/inbound/InboundRunner.java b/samples/kafka/src/main/java/org/springframework/integration/samples/kafka/inbound/InboundRunner.java index 0fdf9a8..bd24f48 100644 --- a/samples/kafka/src/main/java/org/springframework/integration/samples/kafka/inbound/InboundRunner.java +++ b/samples/kafka/src/main/java/org/springframework/integration/samples/kafka/inbound/InboundRunner.java @@ -18,10 +18,10 @@ package org.springframework.integration.samples.kafka.inbound; import org.springframework.context.support.ClassPathXmlApplicationContext; public class InboundRunner { - private static final String CONFIG = "kafkaInboundAdapterParserTests-context.xml"; + private static final String CONFIG = "kafkaInboundAdapterParserTests-context.xml"; - public static void main(final String args[]) { - final ClassPathXmlApplicationContext ctx = new ClassPathXmlApplicationContext(CONFIG, InboundRunner.class); - ctx.start(); - } + public static void main(final String args[]) { + final ClassPathXmlApplicationContext ctx = new ClassPathXmlApplicationContext(CONFIG, InboundRunner.class); + ctx.start(); + } } diff --git a/samples/kafka/src/main/java/org/springframework/integration/samples/kafka/outbound/CustomPartitioner.java b/samples/kafka/src/main/java/org/springframework/integration/samples/kafka/outbound/CustomPartitioner.java index eff5c22..a6f3318 100644 --- a/samples/kafka/src/main/java/org/springframework/integration/samples/kafka/outbound/CustomPartitioner.java +++ b/samples/kafka/src/main/java/org/springframework/integration/samples/kafka/outbound/CustomPartitioner.java @@ -23,15 +23,15 @@ import kafka.producer.Partitioner; * This class is for internal use only and therefore is at default access level */ class CustomPartitioner implements Partitioner { - /** - * Uses the key to calculate a partition bucket id for routing - * the data to the appropriate broker partition - * @return an integer between 0 and numPartitions-1 - */ - @Override - public int partition(final T key, final int numPartitions) { - final String s = (String) key; - final Integer i = Integer.parseInt(s); - return i % numPartitions; - } + /** + * Uses the key to calculate a partition bucket id for routing + * the data to the appropriate broker partition + * @return an integer between 0 and numPartitions-1 + */ + @Override + public int partition(final T key, final int numPartitions) { + final String s = (String) key; + final Integer i = Integer.parseInt(s); + return i % numPartitions; + } } diff --git a/samples/kafka/src/main/java/org/springframework/integration/samples/kafka/outbound/OutboundRunner.java b/samples/kafka/src/main/java/org/springframework/integration/samples/kafka/outbound/OutboundRunner.java index c27ebb7..f285c8d 100644 --- a/samples/kafka/src/main/java/org/springframework/integration/samples/kafka/outbound/OutboundRunner.java +++ b/samples/kafka/src/main/java/org/springframework/integration/samples/kafka/outbound/OutboundRunner.java @@ -23,37 +23,55 @@ import org.springframework.integration.samples.kafka.user.User; import org.springframework.integration.support.MessageBuilder; public class OutboundRunner { - private static final String CONFIG = "kafkaOutboundAdapterParserTests-context.xml"; - private static final Log LOG = LogFactory.getLog(OutboundRunner.class); + private static final String CONFIG = "kafkaOutboundAdapterParserTests-context.xml"; + private static final Log LOG = LogFactory.getLog(OutboundRunner.class); - public static void main(final String args[]) { - final ClassPathXmlApplicationContext ctx = new ClassPathXmlApplicationContext(CONFIG, OutboundRunner.class); - ctx.start(); + public static void main(final String args[]) { + final ClassPathXmlApplicationContext ctx = new ClassPathXmlApplicationContext(CONFIG, OutboundRunner.class); + ctx.start(); - final MessageChannel channel = ctx.getBean("inputToKafka", MessageChannel.class); - LOG.info(channel.getClass()); + final MessageChannel channel = ctx.getBean("inputToKafka", MessageChannel.class); + LOG.info(channel.getClass()); - //sending 100,000 messages to Kafka server for topic test1 - for (int i = 0; i < 500; i++) { + //sending 100,000 messages to Kafka server for topic test1 + for (int i = 0; i < 500; i++) { final User user = new User(); user.setFirstName("fname" + i); user.setLastName("lname" + i); - channel.send( - MessageBuilder.withPayload(user) - .setHeader("messageKey", String.valueOf(i)) - .setHeader("topic", "test1").build()); + channel.send( + MessageBuilder.withPayload(user) + .setHeader("messageKey", String.valueOf(i)) + .setHeader("topic", "test1").build()); - LOG.info("message sent " + i); - } + LOG.info("message sent " + i); + } - //sending 5,000 messages to kafka server for topic test2 - for (int i = 0; i < 50; i++) { - channel.send( - MessageBuilder.withPayload("hello Fom ob adapter test2 - " + i) - .setHeader("messageKey", String.valueOf(i)) - .setHeader("topic", "test2").build()); + //sending 5,000 messages to kafka server for topic test2 + for (int i = 0; i < 50; i++) { + channel.send( + MessageBuilder.withPayload("hello Fom ob adapter test2 - " + i) + .setHeader("messageKey", String.valueOf(i)) + .setHeader("topic", "test2").build()); - LOG.info("message sent " + i); - } - } + LOG.info("message sent " + i); + } + + //Send some messages to multiple topics matching regex. + for (int i = 0; i < 10; i++) { + channel.send( + MessageBuilder.withPayload("hello Fom ob adapter regextopic1 - " + i) + .setHeader("messageKey", String.valueOf(i)) + .setHeader("topic", "regextopic1").build()); + + LOG.info("message sent " + i); + } + for (int i = 0; i < 10; i++) { + channel.send( + MessageBuilder.withPayload("hello Fom ob adapter regextopic2 - " + i) + .setHeader("messageKey", String.valueOf(i)) + .setHeader("topic", "regextopic2").build()); + + LOG.info("message sent " + i); + } + } } diff --git a/samples/kafka/src/main/java/org/springframework/integration/samples/kafka/outbound/PartitionlessTransformer.java b/samples/kafka/src/main/java/org/springframework/integration/samples/kafka/outbound/PartitionlessTransformer.java index 49470ad..5846ae6 100644 --- a/samples/kafka/src/main/java/org/springframework/integration/samples/kafka/outbound/PartitionlessTransformer.java +++ b/samples/kafka/src/main/java/org/springframework/integration/samples/kafka/outbound/PartitionlessTransformer.java @@ -29,28 +29,28 @@ import java.util.Map; * @author Soby Chacko */ public class PartitionlessTransformer implements Transformer { - @Override - @SuppressWarnings("unchecked") - public Message transform(final Message message) { + @Override + @SuppressWarnings("unchecked") + public Message transform(final Message message) { - final Map>> origData = - (Map>>) message.getPayload(); + final Map>> origData = + (Map>>) message.getPayload(); - final Map> nonPartitionedData = new HashMap>(); + final Map> nonPartitionedData = new HashMap>(); - for(final String topic : origData.keySet()) { - final Map> partitionedData = origData.get(topic); - final Collection> nonPartitionedDataFromTopic = partitionedData.values(); + for(final String topic : origData.keySet()) { + final Map> partitionedData = origData.get(topic); + final Collection> nonPartitionedDataFromTopic = partitionedData.values(); - final List mergedList = new ArrayList(); + final List mergedList = new ArrayList(); - for (final List l : nonPartitionedDataFromTopic){ - mergedList.addAll(l); - } + for (final List l : nonPartitionedDataFromTopic){ + mergedList.addAll(l); + } - nonPartitionedData.put(topic, mergedList); - } + nonPartitionedData.put(topic, mergedList); + } - return MessageBuilder.withPayload(nonPartitionedData).build(); - } + return MessageBuilder.withPayload(nonPartitionedData).build(); + } } diff --git a/samples/kafka/src/main/resources/org/springframework/integration/samples/kafka/inbound/kafkaInboundAdapterParserTests-context.xml b/samples/kafka/src/main/resources/org/springframework/integration/samples/kafka/inbound/kafkaInboundAdapterParserTests-context.xml index a7b8d51..5dce6af 100644 --- a/samples/kafka/src/main/resources/org/springframework/integration/samples/kafka/inbound/kafkaInboundAdapterParserTests-context.xml +++ b/samples/kafka/src/main/resources/org/springframework/integration/samples/kafka/inbound/kafkaInboundAdapterParserTests-context.xml @@ -1,60 +1,63 @@ - + + - + - + - + + + - - - + + + - - - + + + - - - + + + + + + + + + + + + + - - - - - - - - - - + - - - + diff --git a/samples/kafka/src/main/resources/org/springframework/integration/samples/kafka/outbound/kafkaOutboundAdapterParserTests-context.xml b/samples/kafka/src/main/resources/org/springframework/integration/samples/kafka/outbound/kafkaOutboundAdapterParserTests-context.xml index 80d3eef..1fa2155 100644 --- a/samples/kafka/src/main/resources/org/springframework/integration/samples/kafka/outbound/kafkaOutboundAdapterParserTests-context.xml +++ b/samples/kafka/src/main/resources/org/springframework/integration/samples/kafka/outbound/kafkaOutboundAdapterParserTests-context.xml @@ -1,49 +1,52 @@ + xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" + xmlns:int="http://www.springframework.org/schema/integration" + xmlns:int-kafka="http://www.springframework.org/schema/integration/kafka" + 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/integration http://www.springframework.org/schema/integration/spring-integration.xsd + http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd + http://www.springframework.org/schema/task http://www.springframework.org/schema/task/spring-task.xsd"> - - - + + + - - - - - + + + - - - + - - - + + + - + + + - - - - - - + + + + + + + + + diff --git a/spring-integration-kafka/README.md b/spring-integration-kafka/README.md index db0a57f..16f8681 100644 --- a/spring-integration-kafka/README.md +++ b/spring-integration-kafka/README.md @@ -86,6 +86,9 @@ Producer context is at the heart of the kafka outbound adapter. Here is an examp topic="test2" compression-codec="default" async="true"/> + ``` @@ -97,7 +100,7 @@ If you go by the above example, there are two producers generated from this conf test1 and another for test2. Each producer can take the following: broker-list List of comma separated brokers that this producer connects to - topic Topic name + topic Topic name or Java regex pattern of topic name compression-codec Compression method to be used. Default is no compression. Supported compression codec are gzip and snappy. Anything else would result in no compression value-encoder Serializer to be used for encoding messages. @@ -180,7 +183,7 @@ currently supports only the High Level Consumer. Here are the details of configu auto-startup="false" channel="inputFromKafka"> - + ``` Since this inbound channel adapter uses a Polling Channel under the hood, it must be configured with a Poller. A notable difference @@ -204,10 +207,18 @@ Inbound Kafka Adapter must specify a kafka-consumer-context-ref element and here + + + - + ``` +`consumer-configuration` supports consuming from specific topic using a `topic` child element or from multiple topics matching a topic regex using `topic-filter` child element. `topic-filter` supports both whitelist and blacklist filter based on `exclude` attribute. + Consumer context requires a reference to a zookeeper-connect which dictates all the zookeeper specific configuration details. Here is how a zookeeper-connect is configured. @@ -315,8 +326,3 @@ Map. If your use case does not require ordering of messages during consumption, then you can easily pass this payload to a standard SI transformer and just get a full dump of the actual payload sent by Kafka. - - - - - diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/config/xml/KafkaConsumerContextParser.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/config/xml/KafkaConsumerContextParser.java index e116eec..c21f7f1 100644 --- a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/config/xml/KafkaConsumerContextParser.java +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/config/xml/KafkaConsumerContextParser.java @@ -15,7 +15,10 @@ */ package org.springframework.integration.kafka.config.xml; -import kafka.serializer.DefaultDecoder; +import java.util.HashMap; +import java.util.List; +import java.util.Map; + import org.springframework.beans.factory.config.BeanDefinition; import org.springframework.beans.factory.config.BeanDefinitionHolder; import org.springframework.beans.factory.support.AbstractBeanDefinition; @@ -29,17 +32,14 @@ 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.xml.DomUtils; import org.w3c.dom.Element; -import java.util.ArrayList; -import java.util.HashMap; -import java.util.List; -import java.util.Map; - /** * @author Soby Chacko + * @author Rajasekar Elango * @since 0.5 */ public class KafkaConsumerContextParser extends AbstractSingleBeanDefinitionParser { @@ -57,7 +57,6 @@ public class KafkaConsumerContextParser extends AbstractSingleBeanDefinitionPars parseConsumerConfigurations(consumerConfigurations, parserContext, builder, element); } - @SuppressWarnings("unchecked") private void parseConsumerConfigurations(final Element consumerConfigurations, final ParserContext parserContext, final BeanDefinitionBuilder builder, final Element parentElem) { for (final Element consumerConfiguration : DomUtils.getChildElementsByTagName(consumerConfigurations, "consumer-configuration")) { @@ -75,14 +74,24 @@ public class KafkaConsumerContextParser extends AbstractSingleBeanDefinitionPars final Map topicStreamsMap = new HashMap(); - for (final Element topicConfiguration : DomUtils.getChildElementsByTagName(consumerConfiguration, "topic")) { - final String topic = topicConfiguration.getAttribute("id"); - final String streams = topicConfiguration.getAttribute("streams"); - final Integer streamsInt = Integer.valueOf(streams); - topicStreamsMap.put(topic, streamsInt); + final List topicConfigurations = DomUtils.getChildElementsByTagName(consumerConfiguration, "topic"); + + if (topicConfigurations != null){ + for (final Element topicConfiguration : topicConfigurations) { + final String topic = topicConfiguration.getAttribute("id"); + final String streams = topicConfiguration.getAttribute("streams"); + final Integer streamsInt = Integer.valueOf(streams); + topicStreamsMap.put(topic, streamsInt); + } + consumerMetadataBuilder.addPropertyValue("topicStreamMap", topicStreamsMap); } - consumerMetadataBuilder.addPropertyValue("topicStreamMap", topicStreamsMap); + final Element topicFilter = DomUtils.getChildElementByTagName(consumerConfiguration, "topic-filter"); + + if (topicFilter != null){ + final TopicFilterConfiguration topicFilterConfiguration = new TopicFilterConfiguration(topicFilter.getAttribute("pattern"),Integer.valueOf(topicFilter.getAttribute("streams")), Boolean.valueOf(topicFilter.getAttribute("exclude"))); + consumerMetadataBuilder.addPropertyValue("topicFilterConfiguration", topicFilterConfiguration); + } final BeanDefinition consumerMetadataBeanDef = consumerMetadataBuilder.getBeanDefinition(); registerBeanDefinition(new BeanDefinitionHolder(consumerMetadataBeanDef, "consumerMetadata_" + consumerConfiguration.getAttribute("group-id")), diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/config/xml/KafkaProducerContextParser.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/config/xml/KafkaProducerContextParser.java index dadfe47..ee2a8bb 100644 --- a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/config/xml/KafkaProducerContextParser.java +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/config/xml/KafkaProducerContextParser.java @@ -30,9 +30,6 @@ import org.springframework.util.StringUtils; import org.springframework.util.xml.DomUtils; import org.w3c.dom.Element; -import java.util.HashMap; -import java.util.Map; - /** * @author Soby Chacko * @since 0.5 @@ -52,7 +49,6 @@ public class KafkaProducerContextParser extends AbstractSimpleBeanDefinitionPars parseProducerConfigurations(topics, parserContext); } - @SuppressWarnings("unchecked") private void parseProducerConfigurations(final Element topics, final ParserContext parserContext) { for (final Element producerConfiguration : DomUtils.getChildElementsByTagName(topics, "producer-configuration")){ final BeanDefinitionBuilder producerConfigurationBuilder = BeanDefinitionBuilder.genericBeanDefinition(ProducerConfiguration.class); diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/serializer/avro/AvroDatumSupport.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/serializer/avro/AvroDatumSupport.java index 6165b1d..808c5a0 100644 --- a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/serializer/avro/AvroDatumSupport.java +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/serializer/avro/AvroDatumSupport.java @@ -21,7 +21,6 @@ public abstract class AvroDatumSupport { this.avroSerializer = new AvroSerializer(); } - @SuppressWarnings("unchecked") public byte[] toBytes(final T source, final DatumWriter writer) { try { return avroSerializer.serialize(source, writer); @@ -31,7 +30,6 @@ public abstract class AvroDatumSupport { return null; } - @SuppressWarnings("unchecked") public T fromBytes(final byte[] bytes, final DatumReader reader) { try { return avroSerializer.deserialize(bytes, reader); diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/serializer/avro/AvroReflectDatumBackedKafkaDecoder.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/serializer/avro/AvroReflectDatumBackedKafkaDecoder.java index 7e9b7c9..68a6663 100644 --- a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/serializer/avro/AvroReflectDatumBackedKafkaDecoder.java +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/serializer/avro/AvroReflectDatumBackedKafkaDecoder.java @@ -16,17 +16,15 @@ package org.springframework.integration.kafka.serializer.avro; import kafka.serializer.Decoder; + import org.apache.avro.io.DatumReader; import org.apache.avro.reflect.ReflectDatumReader; -import org.apache.commons.logging.Log; -import org.apache.commons.logging.LogFactory; /** * @author Soby Chacko * @since 0.5 */ public class AvroReflectDatumBackedKafkaDecoder extends AvroDatumSupport implements Decoder { - private static final Log LOG = LogFactory.getLog(AvroReflectDatumBackedKafkaDecoder.class); private final DatumReader reader; @@ -35,9 +33,7 @@ public class AvroReflectDatumBackedKafkaDecoder extends AvroDatumSupport i } @Override - @SuppressWarnings("unchecked") public T fromBytes(final byte[] bytes) { return fromBytes(bytes, reader); } } - diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/serializer/avro/AvroReflectDatumBackedKafkaEncoder.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/serializer/avro/AvroReflectDatumBackedKafkaEncoder.java index 6f1c0cc..4d51d09 100644 --- a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/serializer/avro/AvroReflectDatumBackedKafkaEncoder.java +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/serializer/avro/AvroReflectDatumBackedKafkaEncoder.java @@ -16,17 +16,15 @@ package org.springframework.integration.kafka.serializer.avro; import kafka.serializer.Encoder; + import org.apache.avro.io.DatumWriter; import org.apache.avro.reflect.ReflectDatumWriter; -import org.apache.commons.logging.Log; -import org.apache.commons.logging.LogFactory; /** * @author Soby Chacko * @since 0.5 */ public class AvroReflectDatumBackedKafkaEncoder extends AvroDatumSupport implements Encoder { - private static final Log LOG = LogFactory.getLog(AvroReflectDatumBackedKafkaEncoder.class); private final DatumWriter writer; @@ -35,7 +33,6 @@ public class AvroReflectDatumBackedKafkaEncoder extends AvroDatumSupport i } @Override - @SuppressWarnings("unchecked") public byte[] toBytes(final T source) { return toBytes(source, writer); } diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/serializer/avro/AvroSpecificDatumBackedKafkaDecoder.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/serializer/avro/AvroSpecificDatumBackedKafkaDecoder.java index 622a9e3..9ad7ee6 100644 --- a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/serializer/avro/AvroSpecificDatumBackedKafkaDecoder.java +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/serializer/avro/AvroSpecificDatumBackedKafkaDecoder.java @@ -1,10 +1,9 @@ package org.springframework.integration.kafka.serializer.avro; import kafka.serializer.Decoder; + import org.apache.avro.io.DatumReader; import org.apache.avro.specific.SpecificDatumReader; -import org.apache.commons.logging.Log; -import org.apache.commons.logging.LogFactory; /** * @author Soby Chacko @@ -12,8 +11,6 @@ import org.apache.commons.logging.LogFactory; */ public class AvroSpecificDatumBackedKafkaDecoder extends AvroDatumSupport implements Decoder { - private static final Log LOG = LogFactory.getLog(AvroSpecificDatumBackedKafkaDecoder.class); - private final DatumReader reader; public AvroSpecificDatumBackedKafkaDecoder(final Class specificRecordBase) { @@ -21,7 +18,6 @@ public class AvroSpecificDatumBackedKafkaDecoder extends AvroDatumSupport } @Override - @SuppressWarnings("unchecked") public T fromBytes(final byte[] bytes) { return fromBytes(bytes, reader); } diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/serializer/avro/AvroSpecificDatumBackedKafkaEncoder.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/serializer/avro/AvroSpecificDatumBackedKafkaEncoder.java index 410dfd6..92fb4ac 100644 --- a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/serializer/avro/AvroSpecificDatumBackedKafkaEncoder.java +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/serializer/avro/AvroSpecificDatumBackedKafkaEncoder.java @@ -1,10 +1,9 @@ package org.springframework.integration.kafka.serializer.avro; import kafka.serializer.Encoder; + import org.apache.avro.io.DatumWriter; import org.apache.avro.specific.SpecificDatumWriter; -import org.apache.commons.logging.Log; -import org.apache.commons.logging.LogFactory; /** * @author Soby Chacko @@ -12,8 +11,6 @@ import org.apache.commons.logging.LogFactory; */ public class AvroSpecificDatumBackedKafkaEncoder extends AvroDatumSupport implements Encoder { - private static final Log LOG = LogFactory.getLog(AvroSpecificDatumBackedKafkaEncoder.class); - private final DatumWriter writer; public AvroSpecificDatumBackedKafkaEncoder(final Class specificRecordClazz) { @@ -21,7 +18,6 @@ public class AvroSpecificDatumBackedKafkaEncoder extends AvroDatumSupport } @Override - @SuppressWarnings("unchecked") public byte[] toBytes(final T source) { return toBytes(source, writer); } diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/support/ConsumerConfiguration.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/support/ConsumerConfiguration.java index 25fe67f..8376eff 100644 --- a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/support/ConsumerConfiguration.java +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/support/ConsumerConfiguration.java @@ -1,63 +1,52 @@ /* - * Copyright 2002-2013 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. + * Copyright 2002-2013 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.support; +import java.util.*; +import java.util.concurrent.*; + import kafka.consumer.ConsumerTimeoutException; import kafka.consumer.KafkaStream; import kafka.javaapi.consumer.ConsumerConnector; import kafka.message.MessageAndMetadata; + import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; import org.springframework.integration.MessagingException; -import java.util.ArrayList; -import java.util.LinkedList; -import java.util.List; -import java.util.Map; -import java.util.concurrent.Callable; -import java.util.concurrent.ConcurrentHashMap; -import java.util.concurrent.ExecutorService; -import java.util.concurrent.Executors; -import java.util.concurrent.Future; - /** * @author Soby Chacko + * @author Rajasekar Elango * @since 0.5 */ -public class ConsumerConfiguration { +public class ConsumerConfiguration { private static final Log LOGGER = LogFactory.getLog(ConsumerConfiguration.class); - private final ConsumerMetadata consumerMetadata; + private final ConsumerMetadata consumerMetadata; private final ConsumerConnectionProvider consumerConnectionProvider; - private final MessageLeftOverTracker messageLeftOverTracker; + private final MessageLeftOverTracker messageLeftOverTracker; private ConsumerConnector consumerConnector; private volatile int count = 0; private int maxMessages = 1; + private Collection>> consumerMessageStreams; - private ExecutorService executorService = Executors.newCachedThreadPool(); + private final ExecutorService executorService = Executors.newCachedThreadPool(); - public ConsumerConfiguration(final ConsumerMetadata consumerMetadata, - final ConsumerConnectionProvider consumerConnectionProvider, - final MessageLeftOverTracker messageLeftOverTracker) { + public ConsumerConfiguration(final ConsumerMetadata consumerMetadata, + final ConsumerConnectionProvider consumerConnectionProvider, + final MessageLeftOverTracker messageLeftOverTracker) { this.consumerMetadata = consumerMetadata; this.consumerConnectionProvider = consumerConnectionProvider; this.messageLeftOverTracker = messageLeftOverTracker; } - public ConsumerMetadata getConsumerMetadata() { + public ConsumerMetadata getConsumerMetadata() { return consumerMetadata; } @@ -65,23 +54,23 @@ public class ConsumerConfiguration { count = messageLeftOverTracker.getCurrentCount(); final Object lock = new Object(); - final List>>> tasks = new LinkedList>>>(); + final List>>> tasks = new LinkedList>>>(); - final Map>> consumerMap = getConsumerMapWithMessageStreams(); - for (final List> streams : consumerMap.values()) { - for (final KafkaStream stream : streams) { - tasks.add(new Callable>>() { + for (final List> streams : createConsumerMessageStreams()) { + for (final KafkaStream stream : streams) { + tasks.add(new Callable>>() { @Override - public List> call() throws Exception { - final List> rawMessages = new ArrayList>(); + public List> call() throws Exception { + final List> rawMessages = new ArrayList>(); try { while (count < maxMessages) { - final MessageAndMetadata messageAndMetadata = stream.iterator().next(); + final MessageAndMetadata messageAndMetadata = stream.iterator().next(); synchronized (lock) { if (count < maxMessages) { rawMessages.add(messageAndMetadata); count++; - } else { + } + else { messageLeftOverTracker.addMessageAndMetadata(messageAndMetadata); } } @@ -97,18 +86,20 @@ public class ConsumerConfiguration { return executeTasks(tasks); } - private Map>> executeTasks(final List>>> tasks) { + private Map>> executeTasks( + final List>>> tasks) { final Map>> messages = new ConcurrentHashMap>>(); messages.putAll(getLeftOverMessageMap()); try { - for (final Future>> result : executorService.invokeAll(tasks)) { + for (final Future>> result : executorService.invokeAll(tasks)) { if (!result.get().isEmpty()) { final String topic = result.get().get(0).topic(); if (!messages.containsKey(topic)) { messages.put(topic, getPayload(result.get())); - } else { + } + else { final Map> existingPayloadMap = messages.get(topic); getPayload(result.get(), existingPayloadMap); @@ -126,21 +117,21 @@ public class ConsumerConfiguration { return messages; } - @SuppressWarnings("unchecked") private Map>> getLeftOverMessageMap() { final Map>> messages = new ConcurrentHashMap>>(); - for (final MessageAndMetadata mamd : messageLeftOverTracker.getMessageLeftOverFromPreviousPoll()) { + for (final MessageAndMetadata mamd : messageLeftOverTracker.getMessageLeftOverFromPreviousPoll()) { final String topic = mamd.topic(); if (!messages.containsKey(topic)) { - final List> l = new ArrayList>(); + final List> l = new ArrayList>(); l.add(mamd); messages.put(topic, getPayload(l)); - } else { + } + else { final Map> existingPayloadMap = messages.get(topic); - final List> l = new ArrayList>(); + final List> l = new ArrayList>(); l.add(mamd); getPayload(l, existingPayloadMap); } @@ -149,15 +140,16 @@ public class ConsumerConfiguration { return messages; } - private Map> getPayload(final List> messageAndMetadatas) { + private Map> getPayload(final List> messageAndMetadatas) { final Map> payloadMap = new ConcurrentHashMap>(); - for (final MessageAndMetadata messageAndMetadata : messageAndMetadatas) { + for (final MessageAndMetadata messageAndMetadata : messageAndMetadatas) { if (!payloadMap.containsKey(messageAndMetadata.partition())) { final List payload = new ArrayList(); payload.add(messageAndMetadata.message()); payloadMap.put(messageAndMetadata.partition(), payload); - } else { + } + else { final List payload = payloadMap.get(messageAndMetadata.partition()); payload.add(messageAndMetadata.message()); } @@ -167,25 +159,52 @@ public class ConsumerConfiguration { return payloadMap; } - private void getPayload(final List> messageAndMetadatas, final Map> existingPayloadMap) { - for (final MessageAndMetadata messageAndMetadata : messageAndMetadatas) { + private void getPayload(final List> messageAndMetadatas, + final Map> existingPayloadMap) { + for (final MessageAndMetadata messageAndMetadata : messageAndMetadatas) { if (!existingPayloadMap.containsKey(messageAndMetadata.partition())) { final List payload = new ArrayList(); payload.add(messageAndMetadata.message()); existingPayloadMap.put(messageAndMetadata.partition(), payload); - } else { + } + else { final List payload = existingPayloadMap.get(messageAndMetadata.partition()); payload.add(messageAndMetadata.message()); } } } - @SuppressWarnings("unchecked") - public Map>> getConsumerMapWithMessageStreams() { - return getConsumerConnector().createMessageStreams( - consumerMetadata.getTopicStreamMap(), - consumerMetadata.getKeyDecoder(), - consumerMetadata.getValueDecoder()); + private Collection>> createConsumerMessageStreams() { + if (consumerMessageStreams == null) { + if (!(consumerMetadata.getTopicStreamMap() == null || consumerMetadata.getTopicStreamMap().isEmpty())) { + consumerMessageStreams = createMessageStreamsForTopic().values(); + } + else { + consumerMessageStreams = new ArrayList>>(); + consumerMessageStreams.add(createMessageStreamsForTopicFilter()); + } + } + return consumerMessageStreams; + } + + public Map>> createMessageStreamsForTopic() { + return getConsumerConnector().createMessageStreams(consumerMetadata.getTopicStreamMap(), + consumerMetadata.getKeyDecoder(), consumerMetadata.getValueDecoder()); + } + + public List> createMessageStreamsForTopicFilter() { + List> messageStream = new ArrayList>(); + TopicFilterConfiguration topicFilterConfiguration = consumerMetadata.getTopicFilterConfiguration(); + if (topicFilterConfiguration != null) { + messageStream = getConsumerConnector().createMessageStreamsByFilter( + topicFilterConfiguration.getTopicFilter(), topicFilterConfiguration.getNumberOfStreams(), + consumerMetadata.getKeyDecoder(), consumerMetadata.getValueDecoder()); + } + else { + LOGGER.warn("No Topic Filter Configuration defined"); + } + + return messageStream; } public int getMaxMessages() { diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/support/ConsumerMetadata.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/support/ConsumerMetadata.java index 476e66a..55bb6d5 100644 --- a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/support/ConsumerMetadata.java +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/support/ConsumerMetadata.java @@ -15,15 +15,16 @@ */ package org.springframework.integration.kafka.support; +import java.util.Map; + import kafka.serializer.Decoder; import kafka.serializer.DefaultDecoder; import org.springframework.beans.factory.InitializingBean; import org.springframework.integration.kafka.core.KafkaConsumerDefaults; -import java.util.Map; - /** * @author Soby Chacko + * @author Rajasekar Elango * @since 0.5 */ public class ConsumerMetadata implements InitializingBean { @@ -46,6 +47,7 @@ public class ConsumerMetadata implements InitializingBean { private Decoder valueDecoder; private Decoder keyDecoder; private Map topicStreamMap; + private TopicFilterConfiguration topicFilterConfiguration; public String getGroupId() { return groupId; @@ -186,4 +188,14 @@ public class ConsumerMetadata implements InitializingBean { setKeyDecoder((Decoder) getValueDecoder()); } } + + public TopicFilterConfiguration getTopicFilterConfiguration() { + return topicFilterConfiguration; + } + + public void setTopicFilterConfiguration( + TopicFilterConfiguration topicFilterConfiguration) { + this.topicFilterConfiguration = topicFilterConfiguration; + } + } diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/support/KafkaProducerContext.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/support/KafkaProducerContext.java index bdb7e11..0fdb3c2 100644 --- a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/support/KafkaProducerContext.java +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/support/KafkaProducerContext.java @@ -15,23 +15,24 @@ */ package org.springframework.integration.kafka.support; -import org.springframework.beans.BeansException; -import org.springframework.beans.factory.BeanFactory; -import org.springframework.beans.factory.BeanFactoryAware; -import org.springframework.beans.factory.ListableBeanFactory; -import org.springframework.integration.Message; - import java.util.Collection; import java.util.Map; +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; +import org.springframework.beans.BeansException; +import org.springframework.beans.factory.*; +import org.springframework.integration.Message; + /** * @author Soby Chacko + * @author Rajasekar Elango * @since 0.5 */ public class KafkaProducerContext implements BeanFactoryAware { + private static final Log LOGGER = LogFactory.getLog(KafkaProducerContext.class); private Map> topicsConfiguration; - @SuppressWarnings("unchecked") public void send(final Message message) throws Exception { final ProducerConfiguration producerConfiguration = getTopicConfiguration(message.getHeaders().get("topic", String.class)); @@ -41,15 +42,15 @@ public class KafkaProducerContext implements BeanFactoryAware { } } - private ProducerConfiguration getTopicConfiguration(final String topic){ + public ProducerConfiguration getTopicConfiguration(final String topic) { final Collection> topics = topicsConfiguration.values(); for (final ProducerConfiguration producerConfiguration : topics){ - if (producerConfiguration.getProducerMetadata().getTopic().equals(topic)){ + if (topic.matches(producerConfiguration.getProducerMetadata().getTopic())){ return producerConfiguration; } } - + LOGGER.error("No is producer-configuration defined for topic " + topic + ". cannot send message"); return null; } diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/support/ProducerConfiguration.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/support/ProducerConfiguration.java index 71026ae..e7e325e 100644 --- a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/support/ProducerConfiguration.java +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/support/ProducerConfiguration.java @@ -1,40 +1,33 @@ /* - * Copyright 2002-2013 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. + * Copyright 2002-2013 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.support; +import java.io.*; + import kafka.javaapi.producer.Producer; import kafka.producer.KeyedMessage; import kafka.serializer.DefaultEncoder; + import org.apache.commons.lang.builder.EqualsBuilder; import org.apache.commons.lang.builder.HashCodeBuilder; import org.springframework.integration.Message; -import java.io.ByteArrayOutputStream; -import java.io.IOException; -import java.io.ObjectOutputStream; - /** * @author Soby Chacko + * @author Rajasekar Elango * @since 0.5 */ -public class ProducerConfiguration { - private final Producer producer; - private final ProducerMetadata producerMetadata; +public class ProducerConfiguration { + private final Producer producer; + private final ProducerMetadata producerMetadata; - public ProducerConfiguration(final ProducerMetadata producerMetadata, final Producer producer){ + public ProducerConfiguration(final ProducerMetadata producerMetadata, final Producer producer) { this.producerMetadata = producerMetadata; this.producer = producer; } @@ -46,10 +39,12 @@ public class ProducerConfiguration { public void send(final Message message) throws Exception { final V v = getPayload(message); + String topic = message.getHeaders().get("topic", String.class); if (message.getHeaders().containsKey("messageKey")) { - producer.send(new KeyedMessage(producerMetadata.getTopic(), getKey(message), v)); - } else { - producer.send(new KeyedMessage(producerMetadata.getTopic(), v)); + producer.send(new KeyedMessage(topic, getKey(message), v)); + } + else { + producer.send(new KeyedMessage(topic, v)); } } @@ -57,7 +52,8 @@ public class ProducerConfiguration { private V getPayload(final Message message) throws Exception { if (producerMetadata.getValueEncoder().getClass().isAssignableFrom(DefaultEncoder.class)) { return (V) getByteStream(message.getPayload()); - } else if (message.getPayload().getClass().isAssignableFrom(producerMetadata.getValueClassType())) { + } + else if (message.getPayload().getClass().isAssignableFrom(producerMetadata.getValueClassType())) { return producerMetadata.getValueClassType().cast(message.getPayload()); } @@ -75,13 +71,13 @@ public class ProducerConfiguration { return message.getHeaders().get("messageKey", producerMetadata.getKeyClassType()); } - private static boolean isRawByteArray(final Object obj){ + private static boolean isRawByteArray(final Object obj) { return obj instanceof byte[]; } private static byte[] getByteStream(final Object obj) throws IOException { - if (isRawByteArray(obj)){ - return (byte[])obj; + if (isRawByteArray(obj)) { + return (byte[]) obj; } final ByteArrayOutputStream out = new ByteArrayOutputStream(); @@ -92,7 +88,7 @@ public class ProducerConfiguration { } @Override - public boolean equals(final Object obj){ + public boolean equals(final Object obj) { return EqualsBuilder.reflectionEquals(this, obj); } @@ -100,4 +96,11 @@ public class ProducerConfiguration { public int hashCode() { return HashCodeBuilder.reflectionHashCode(this); } + + @Override + public String toString() { + StringBuilder builder = new StringBuilder(); + builder.append("ProducerConfiguration [producerMetadata=").append(producerMetadata).append("]"); + return builder.toString(); + } } diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/support/ProducerMetadata.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/support/ProducerMetadata.java index 69b432e..e6dca25 100644 --- a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/support/ProducerMetadata.java +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/support/ProducerMetadata.java @@ -18,12 +18,14 @@ package org.springframework.integration.kafka.support; import kafka.producer.Partitioner; import kafka.serializer.DefaultEncoder; import kafka.serializer.Encoder; + import org.apache.commons.lang.builder.EqualsBuilder; import org.apache.commons.lang.builder.HashCodeBuilder; import org.springframework.beans.factory.InitializingBean; /** * @author Soby Chacko + * @author Rajasekar Elango * @since 0.5 */ public class ProducerMetadata implements InitializingBean { @@ -137,4 +139,14 @@ public class ProducerMetadata implements InitializingBean { public int hashCode() { return HashCodeBuilder.reflectionHashCode(this); } + + @Override + public String toString() { + StringBuilder builder = new StringBuilder(); + builder.append("ProducerMetadata [keyEncoder=").append(keyEncoder).append(", valueEncoder=") + .append(valueEncoder).append(", topic=").append(topic).append(", compressionCodec=") + .append(compressionCodec).append(", partitioner=").append(partitioner).append(", async=").append(async) + .append(", batchNumMessages=").append(batchNumMessages).append("]"); + return builder.toString(); + } } diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/support/TopicFilterConfiguration.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/support/TopicFilterConfiguration.java new file mode 100644 index 0000000..7170c6b --- /dev/null +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/support/TopicFilterConfiguration.java @@ -0,0 +1,46 @@ +/* + * Copyright 2002-2013 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.support; + +import kafka.consumer.Blacklist; +import kafka.consumer.TopicFilter; +import kafka.consumer.Whitelist; + +/** + * @author Rajasekar Elango + * @since 0.5 + */ +public class TopicFilterConfiguration { + + private final int numberOfStreams; + private TopicFilter topicFilter; + + public TopicFilterConfiguration(final String pattern, final int numberOfStreams, final boolean exclude) { + this.numberOfStreams = numberOfStreams; + if (exclude) { + topicFilter = new Blacklist(pattern); + } + else { + topicFilter = new Whitelist(pattern); + } + } + + public TopicFilter getTopicFilter() { + return topicFilter; + } + + public int getNumberOfStreams() { + return numberOfStreams; + } + + @Override + public String toString() { + return new StringBuilder(topicFilter.toString()).append(" : ").append(numberOfStreams).toString(); + } +} 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 1c716fe..b3009d8 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 @@ -232,14 +232,56 @@ ]]> - + - + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + avroBackedKafkaEncoder = new AvroReflectDatumBackedKafkaEncoder(TestObject.class); @@ -45,7 +45,6 @@ public class AvroReflectDatumBackedKafkaSerializerTest { } @Test - @SuppressWarnings("unchecked") public void anotherTest() { final AvroReflectDatumBackedKafkaEncoder avroBackedKafkaEncoder = new AvroReflectDatumBackedKafkaEncoder(java.lang.String.class); final String testString = "Testing Avro"; diff --git a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/serializer/AvroSpecificDatumBackedKafkaSerializerTest.java b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/serializer/AvroSpecificDatumBackedKafkaSerializerTest.java index 8d18bef..3850a5a 100644 --- a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/serializer/AvroSpecificDatumBackedKafkaSerializerTest.java +++ b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/serializer/AvroSpecificDatumBackedKafkaSerializerTest.java @@ -13,7 +13,6 @@ import org.springframework.integration.kafka.test.utils.User; public class AvroSpecificDatumBackedKafkaSerializerTest { @Test - @SuppressWarnings("unchecked") public void testEncodeDecodeFromSpecificDatumSchema() { final AvroSpecificDatumBackedKafkaEncoder avroBackedKafkaEncoder = new AvroSpecificDatumBackedKafkaEncoder(User.class); diff --git a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/support/ConsumerConfigurationTests.java b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/support/ConsumerConfigurationTests.java index 19c267a..9e81076 100644 --- a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/support/ConsumerConfigurationTests.java +++ b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/support/ConsumerConfigurationTests.java @@ -16,18 +16,9 @@ package org.springframework.integration.kafka.support; import static org.junit.Assert.assertNull; -import static org.mockito.Mockito.atLeast; -import static org.mockito.Mockito.atMost; -import static org.mockito.Mockito.mock; -import static org.mockito.Mockito.times; -import static org.mockito.Mockito.verify; -import static org.mockito.Mockito.when; +import static org.mockito.Mockito.*; -import java.util.ArrayList; -import java.util.Collection; -import java.util.HashMap; -import java.util.List; -import java.util.Map; +import java.util.*; import kafka.consumer.ConsumerIterator; import kafka.consumer.KafkaStream; @@ -42,7 +33,7 @@ import org.mockito.stubbing.Answer; /** * @author Soby Chacko - * @author Gunnar Hillert + * @author Rajasekar Elango * @since 0.5 */ public class ConsumerConfigurationTests { @@ -55,6 +46,10 @@ public class ConsumerConfigurationTests { final MessageLeftOverTracker messageLeftOverTracker = mock(MessageLeftOverTracker.class); final ConsumerConnector consumerConnector = mock(ConsumerConnector.class); + Map topicStreamMap = new HashMap(); + topicStreamMap.put("topic1", 1); + when(consumerMetadata.getTopicStreamMap()).thenReturn(topicStreamMap); + when(consumerConnectionProvider.getConsumerConnector()).thenReturn(consumerConnector); final ConsumerConfiguration consumerConfiguration = new ConsumerConfiguration(consumerMetadata, @@ -67,7 +62,7 @@ public class ConsumerConfigurationTests { final Map>> messageStreams = new HashMap>>(); messageStreams.put("topic", streams); - when(consumerConfiguration.getConsumerMapWithMessageStreams()).thenReturn(messageStreams); + when(consumerConfiguration.createMessageStreamsForTopic()).thenReturn(messageStreams); final ConsumerIterator iterator = mock(ConsumerIterator.class); when(stream.iterator()).thenReturn(iterator); final MessageAndMetadata messageAndMetadata = mock(MessageAndMetadata.class); @@ -95,6 +90,10 @@ public class ConsumerConfigurationTests { mock(ConsumerConnectionProvider.class); final MessageLeftOverTracker messageLeftOverTracker = mock(MessageLeftOverTracker.class); + Map topicStreamMap = new HashMap(); + topicStreamMap.put("topic1", 1); + when(consumerMetadata.getTopicStreamMap()).thenReturn(topicStreamMap); + final ConsumerConnector consumerConnector = mock(ConsumerConnector.class); when(consumerConnectionProvider.getConsumerConnector()).thenReturn(consumerConnector); @@ -113,7 +112,7 @@ public class ConsumerConfigurationTests { final Map>> messageStreams = new HashMap>>(); messageStreams.put("topic", streams); - when(consumerConfiguration.getConsumerMapWithMessageStreams()).thenReturn(messageStreams); + when(consumerConfiguration.createMessageStreamsForTopic()).thenReturn(messageStreams); final ConsumerIterator iterator1 = mock(ConsumerIterator.class); final ConsumerIterator iterator2 = mock(ConsumerIterator.class); final ConsumerIterator iterator3 = mock(ConsumerIterator.class); @@ -160,7 +159,12 @@ public class ConsumerConfigurationTests { final ConsumerMetadata consumerMetadata = mock(ConsumerMetadata.class); final ConsumerConnectionProvider consumerConnectionProvider = mock(ConsumerConnectionProvider.class); - final MessageLeftOverTracker messageLeftOverTracker = mock(MessageLeftOverTracker.class); + final MessageLeftOverTracker messageLeftOverTracker = mock(MessageLeftOverTracker.class); + + Map topicStreamMap = new HashMap(); + topicStreamMap.put("topic1", 1); + when(consumerMetadata.getTopicStreamMap()).thenReturn(topicStreamMap); + final ConsumerConnector consumerConnector = mock(ConsumerConnector.class); @@ -182,7 +186,7 @@ public class ConsumerConfigurationTests { messageStreams.put("topic2", streams); messageStreams.put("topic3", streams); - when(consumerConfiguration.getConsumerMapWithMessageStreams()).thenReturn(messageStreams); + when(consumerConfiguration.createMessageStreamsForTopic()).thenReturn(messageStreams); final ConsumerIterator iterator1 = mock(ConsumerIterator.class); final ConsumerIterator iterator2 = mock(ConsumerIterator.class); final ConsumerIterator iterator3 = mock(ConsumerIterator.class); @@ -211,6 +215,7 @@ public class ConsumerConfigurationTests { when(messageAndMetadata1.partition()).thenAnswer(getAnswer()); final Map>> messages = consumerConfiguration.receive(); + int sum = 0; final Collection>> values = messages.values(); @@ -224,6 +229,8 @@ public class ConsumerConfigurationTests { Assert.assertEquals(sum, 9); } + + private Answer getAnswer() { return new Answer() { private int count = 0; @@ -250,6 +257,9 @@ public class ConsumerConfigurationTests { final MessageLeftOverTracker messageLeftOverTracker = mock(MessageLeftOverTracker.class); final ConsumerConnector consumerConnector = mock(ConsumerConnector.class); + Map topicStreamMap = new HashMap(); + topicStreamMap.put("topic1", 1); + when(consumerMetadata.getTopicStreamMap()).thenReturn(topicStreamMap); when(messageLeftOverTracker.getCurrentCount()).thenReturn(3); final MessageAndMetadata m1 = new MessageAndMetadata("key1", "value1", "topic1", 1, 1L); final MessageAndMetadata m2 = new MessageAndMetadata("key2", "value2", "topic2", 1, 1L); @@ -273,11 +283,10 @@ public class ConsumerConfigurationTests { streams.add(stream); final Map>> messageStreams = new HashMap>>(); messageStreams.put("topic1", streams); - - when(consumerConfiguration.getConsumerMapWithMessageStreams()).thenReturn(messageStreams); - final ConsumerIterator iterator = mock(ConsumerIterator.class); - when(stream.iterator()).thenReturn((ConsumerIterator) iterator); - final MessageAndMetadata messageAndMetadata = mock(MessageAndMetadata.class); + when(consumerConfiguration.createMessageStreamsForTopic()).thenReturn(messageStreams); + final ConsumerIterator iterator = mock(ConsumerIterator.class); + when(stream.iterator()).thenReturn(iterator); + final MessageAndMetadata messageAndMetadata = mock(MessageAndMetadata.class); when(iterator.next()).thenReturn(messageAndMetadata); when(messageAndMetadata.message()).thenReturn("got message"); when(messageAndMetadata.topic()).thenReturn("topic1"); @@ -329,7 +338,7 @@ public class ConsumerConfigurationTests { final ConsumerConfiguration consumerConfiguration = new ConsumerConfiguration(mockedConsumerMetadata, mockedConsumerConnectionProvider, mockedMessageLeftOverTracker); - consumerConfiguration.getConsumerMapWithMessageStreams(); + consumerConfiguration.createMessageStreamsForTopic(); verify(mockedConsumerMetadata, atLeast(1)).getTopicStreamMap(); verify(mockedConsumerConnector, atLeast(1)).createMessageStreams(topicsStreamMap, null, null); @@ -368,13 +377,122 @@ public class ConsumerConfigurationTests { final ConsumerConfiguration consumerConfiguration = new ConsumerConfiguration(mockedConsumerMetadata, mockedConsumerConnectionProvider, mockedMessageLeftOverTracker); - consumerConfiguration.getConsumerMapWithMessageStreams(); + consumerConfiguration.createMessageStreamsForTopic(); verify(mockedConsumerMetadata, atLeast(1)).getTopicStreamMap(); verify(mockedConsumerConnector, atMost(0)).createMessageStreams(topicsStreamMap); verify(mockedConsumerConnector, atLeast(1)).createMessageStreams(topicsStreamMap, mockedKeyDecoder, mockedValueDecoder); } + + @Test + @SuppressWarnings("unchecked") + public void testReceiveMessageForTopicFilterFromSingleStream() { + final ConsumerMetadata consumerMetadata = mock(ConsumerMetadata.class); + final ConsumerConnectionProvider consumerConnectionProvider = + mock(ConsumerConnectionProvider.class); + final MessageLeftOverTracker messageLeftOverTracker = mock(MessageLeftOverTracker.class); + final ConsumerConnector consumerConnector = mock(ConsumerConnector.class); + + when(consumerMetadata.getTopicFilterConfiguration()).thenReturn(new TopicFilterConfiguration(".*", 1, false)); + + when(consumerConnectionProvider.getConsumerConnector()).thenReturn(consumerConnector); + + final ConsumerConfiguration consumerConfiguration = new ConsumerConfiguration(consumerMetadata, + consumerConnectionProvider, messageLeftOverTracker); + consumerConfiguration.setMaxMessages(1); + + final KafkaStream stream = mock(KafkaStream.class); + final List> streams = new ArrayList>(); + streams.add(stream); + + when(consumerConfiguration.createMessageStreamsForTopicFilter()).thenReturn(streams); + final ConsumerIterator iterator = mock(ConsumerIterator.class); + when(stream.iterator()).thenReturn(iterator); + final MessageAndMetadata messageAndMetadata = mock(MessageAndMetadata.class); + when(iterator.next()).thenReturn(messageAndMetadata); + when(messageAndMetadata.message()).thenReturn("got message"); + when(messageAndMetadata.topic()).thenReturn("topic"); + when(messageAndMetadata.partition()).thenReturn(1); + + final Map>> messages = consumerConfiguration.receive(); + Assert.assertEquals(messages.size(), 1); + Assert.assertEquals(messages.get("topic").size(), 1); + Assert.assertEquals(messages.get("topic").get(1).get(0), "got message"); + + verify(stream, times(1)).iterator(); + verify(iterator, times(1)).next(); + verify(messageAndMetadata, times(1)).message(); + verify(messageAndMetadata, times(1)).topic(); + } + + @Test + @SuppressWarnings("unchecked") + public void testReceiveMessageForTopicFilterFromMultipleStreams() { + final ConsumerMetadata consumerMetadata = mock(ConsumerMetadata.class); + final ConsumerConnectionProvider consumerConnectionProvider = + mock(ConsumerConnectionProvider.class); + final MessageLeftOverTracker messageLeftOverTracker = mock(MessageLeftOverTracker.class); + + when(consumerMetadata.getTopicFilterConfiguration()).thenReturn(new TopicFilterConfiguration(".*", 1, false)); + + final ConsumerConnector consumerConnector = mock(ConsumerConnector.class); + + when(consumerConnectionProvider.getConsumerConnector()).thenReturn(consumerConnector); + + final ConsumerConfiguration consumerConfiguration = new ConsumerConfiguration(consumerMetadata, + consumerConnectionProvider, messageLeftOverTracker); + consumerConfiguration.setMaxMessages(3); + + final KafkaStream stream1 = mock(KafkaStream.class); + final KafkaStream stream2 = mock(KafkaStream.class); + final KafkaStream stream3 = mock(KafkaStream.class); + final List> streams = new ArrayList>(); + streams.add(stream1); + streams.add(stream2); + streams.add(stream3); + + when(consumerConfiguration.createMessageStreamsForTopicFilter()).thenReturn(streams); + final ConsumerIterator iterator1 = mock(ConsumerIterator.class); + final ConsumerIterator iterator2 = mock(ConsumerIterator.class); + final ConsumerIterator iterator3 = mock(ConsumerIterator.class); + + when(stream1.iterator()).thenReturn(iterator1); + when(stream2.iterator()).thenReturn(iterator2); + when(stream3.iterator()).thenReturn(iterator3); + final MessageAndMetadata messageAndMetadata1 = mock(MessageAndMetadata.class); + final MessageAndMetadata messageAndMetadata2 = mock(MessageAndMetadata.class); + final MessageAndMetadata messageAndMetadata3 = mock(MessageAndMetadata.class); + + when(iterator1.next()).thenReturn(messageAndMetadata1); + when(iterator2.next()).thenReturn(messageAndMetadata2); + when(iterator3.next()).thenReturn(messageAndMetadata3); + + when(messageAndMetadata1.message()).thenReturn("got message"); + when(messageAndMetadata1.topic()).thenReturn("topic"); + when(messageAndMetadata1.partition()).thenReturn(1); + + when(messageAndMetadata2.message()).thenReturn("got message"); + when(messageAndMetadata2.topic()).thenReturn("topic"); + when(messageAndMetadata2.partition()).thenReturn(2); + + when(messageAndMetadata3.message()).thenReturn("got message"); + when(messageAndMetadata3.topic()).thenReturn("topic"); + when(messageAndMetadata3.partition()).thenReturn(3); + + final Map>> messages = consumerConfiguration.receive(); + Assert.assertEquals(messages.size(), 1); + int sum = 0; + + final Map> values = messages.get("topic"); + + for (final List l : values.values()) { + sum += l.size(); + } + + Assert.assertEquals(sum, 3); + } + private boolean valueFound(final List l, final String value){ for (final Object o : l){ if (value.equals(o)){ diff --git a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/support/KafkaProducerContextTests.java b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/support/KafkaProducerContextTests.java new file mode 100644 index 0000000..c1bc1ed --- /dev/null +++ b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/support/KafkaProducerContextTests.java @@ -0,0 +1,65 @@ +/* + * Copyright 2002-2013 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.support; + +import java.util.HashMap; +import java.util.Map; + +import org.junit.Assert; + +import kafka.javaapi.producer.Producer; + +import org.junit.Test; +import org.mockito.Mockito; +import org.springframework.beans.factory.ListableBeanFactory; + +/** + * @author Rajasekar Elango + * @since 0.5 + */ +public class KafkaProducerContextTests { + + @SuppressWarnings({ "rawtypes", "unchecked" }) + @Test + public void testTopicRegexForProducerConfiguration(){ + + final KafkaProducerContext kafkaProducerContext = new KafkaProducerContext(); + final ListableBeanFactory beanFactory = Mockito.mock(ListableBeanFactory.class); + + final ProducerMetadata producerMetadata = Mockito.mock(ProducerMetadata.class); + + String testRegex = "test.*"; + + Mockito.when(producerMetadata.getTopic()).thenReturn(testRegex); + final Producer producer = Mockito.mock(Producer.class); + + final ProducerConfiguration producerConfiguration = new ProducerConfiguration(producerMetadata, producer); + + + final Map topicConfigurations = new HashMap(); + topicConfigurations.put(testRegex, producerConfiguration); + + Mockito.when(beanFactory.getBeansOfType(ProducerConfiguration.class)).thenReturn(topicConfigurations); + kafkaProducerContext.setBeanFactory(beanFactory); + + Assert.assertNotNull(kafkaProducerContext.getTopicConfiguration("test1")); + Assert.assertNotNull(kafkaProducerContext.getTopicConfiguration("test2")); + Assert.assertNotNull(kafkaProducerContext.getTopicConfiguration("testabc")); + Assert.assertNull(kafkaProducerContext.getTopicConfiguration("dontmatch_testRegex")); + + } + +} diff --git a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/test/utils/User.java b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/test/utils/User.java index c6ad7dd..8ceada8 100644 --- a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/test/utils/User.java +++ b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/test/utils/User.java @@ -46,7 +46,6 @@ public class User extends org.apache.avro.specific.SpecificRecordBase implements } // Used by DatumReader. Applications should not call. - @SuppressWarnings(value = "unchecked") public void put(int field$, java.lang.Object value$) { switch (field$) { case 0: @@ -92,5 +91,3 @@ public class User extends org.apache.avro.specific.SpecificRecordBase implements this.lastName = value; } } - -