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

Merge pull request #58 from relango/merge-master-kafka-topic-filter

Browse Source 
* relango-merge-master-kafka-topic-filter:
  INTEXT-77 Add Kafka topic filter support for producer + consumer
This commit is contained in:
Gunnar Hillert
2013-10-28 17:08:47 -04:00
parent b588aa4667 3e7b7d1331
commit ab3b62e0aa
31 changed files with 795 additions and 470 deletions
Download Patch File Download Diff File Expand all files Collapse all files

106
samples/kafka/build.gradle
View File

@@ -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 {

BIN
samples/kafka/gradle/wrapper/gradle-wrapper.jar vendored
View File

Binary file not shown.

4
samples/kafka/gradle/wrapper/gradle-wrapper.properties vendored
View File

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

2
samples/kafka/gradlew vendored
View File

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

180
samples/kafka/gradlew.bat vendored
View File

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

10
samples/kafka/src/main/java/org/springframework/integration/samples/kafka/inbound/InboundRunner.java
View File

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

22
samples/kafka/src/main/java/org/springframework/integration/samples/kafka/outbound/CustomPartitioner.java
View File

@@ -23,15 +23,15 @@ import kafka.producer.Partitioner;
* This class is for internal use only and therefore is at default access level
*/
class CustomPartitioner<T> implements Partitioner<T> {
/**
* 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;
}
}

66
samples/kafka/src/main/java/org/springframework/integration/samples/kafka/outbound/OutboundRunner.java
View File

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

34
samples/kafka/src/main/java/org/springframework/integration/samples/kafka/outbound/PartitionlessTransformer.java
View File

@@ -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<String, Map<Integer, List<Object>>> origData =
(Map<String, Map<Integer, List<Object>>>) message.getPayload();
final Map<String, Map<Integer, List<Object>>> origData =
(Map<String, Map<Integer, List<Object>>>) message.getPayload();
final Map<String, List<Object>> nonPartitionedData = new HashMap<String, List<Object>>();
final Map<String, List<Object>> nonPartitionedData = new HashMap<String, List<Object>>();
for(final String topic : origData.keySet()) {
final Map<Integer, List<Object>> partitionedData = origData.get(topic);
final Collection<List<Object>> nonPartitionedDataFromTopic = partitionedData.values();
for(final String topic : origData.keySet()) {
final Map<Integer, List<Object>> partitionedData = origData.get(topic);
final Collection<List<Object>> nonPartitionedDataFromTopic = partitionedData.values();
final List<Object> mergedList = new ArrayList<Object>();
final List<Object> mergedList = new ArrayList<Object>();
for (final List<Object> l : nonPartitionedDataFromTopic){
mergedList.addAll(l);
}
for (final List<Object> l : nonPartitionedDataFromTopic){
mergedList.addAll(l);
}
nonPartitionedData.put(topic, mergedList);
}
nonPartitionedData.put(topic, mergedList);
}
return MessageBuilder.withPayload(nonPartitionedData).build();
}
return MessageBuilder.withPayload(nonPartitionedData).build();
}
}

95
samples/kafka/src/main/resources/org/springframework/integration/samples/kafka/inbound/kafkaInboundAdapterParserTests-context.xml
View File

@@ -1,60 +1,63 @@
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:int="http://www.springframework.org/schema/integration"
xmlns:stream="http://www.springframework.org/schema/integration/stream"
xmlns:int-kafka="http://www.springframework.org/schema/integration/kafka"
xsi:schemaLocation="http://www.springframework.org/schema/integration/stream http://www.springframework.org/schema/integration/stream/spring-integration-stream.xsd
http://www.springframework.org/schema/integration/kafka http://www.springframework.org/schema/integration/kafka/spring-integration-kafka.xsd
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:int="http://www.springframework.org/schema/integration"
xmlns:stream="http://www.springframework.org/schema/integration/stream"
xmlns:int-kafka="http://www.springframework.org/schema/integration/kafka"
xsi:schemaLocation="http://www.springframework.org/schema/integration/stream http://www.springframework.org/schema/integration/stream/spring-integration-stream.xsd
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">
<int:channel id="inputFromKafka"/>
<int:channel id="inputFromKafka"/>
<int:channel id="inputFromTransformer"/>
<int:channel id="inputFromTransformer"/>
<stream:stdout-channel-adapter id="stdout" channel="inputFromKafka" append-newline="true"/>
<stream:stdout-channel-adapter id="stdout" channel="inputFromKafka" append-newline="true"/>
<int-kafka:zookeeper-connect id="zookeeperConnect" zk-connect="localhost:2181"
zk-connection-timeout="6000"
zk-session-timeout="6000"
zk-sync-time="2000"/>
<int-kafka:zookeeper-connect id="zookeeperConnect" zk-connect="localhost:2181"
zk-connection-timeout="6000"
zk-session-timeout="6000"
zk-sync-time="2000"/>
<int-kafka:inbound-channel-adapter id="kafkaInboundChannelAdapter"
kafka-consumer-context-ref="consumerContext"
auto-startup="false"
channel="inputFromKafka">
<int:poller fixed-delay="1" time-unit="MILLISECONDS"/>
</int-kafka:inbound-channel-adapter>
<int-kafka:inbound-channel-adapter id="kafkaInboundChannelAdapter"
kafka-consumer-context-ref="consumerContext"
auto-startup="false"
channel="inputFromKafka">
<int:poller fixed-delay="1" time-unit="MILLISECONDS"/>
</int-kafka:inbound-channel-adapter>
<bean id="kafkaReflectionDecoder" class="org.springframework.integration.kafka.serializer.avro.AvroReflectDatumBackedKafkaDecoder">
<constructor-arg type="java.lang.Class" value="java.lang.String"/>
</bean>
<bean id="kafkaReflectionDecoder" class="org.springframework.integration.kafka.serializer.avro.AvroReflectDatumBackedKafkaDecoder">
<constructor-arg type="java.lang.Class" value="java.lang.String"/>
</bean>
<bean id="kafkaSpecificDecoder" class="org.springframework.integration.kafka.serializer.avro.AvroSpecificDatumBackedKafkaDecoder">
<constructor-arg value="org.springframework.integration.samples.kafka.user.User" />
</bean>
<bean id="kafkaSpecificDecoder" class="org.springframework.integration.kafka.serializer.avro.AvroSpecificDatumBackedKafkaDecoder">
<constructor-arg value="org.springframework.integration.samples.kafka.user.User" />
</bean>
<int-kafka:consumer-context id="consumerContext"
consumer-timeout="1000"
zookeeper-connect="zookeeperConnect">
<int-kafka:consumer-configurations>
<int-kafka:consumer-configuration group-id="default1"
value-decoder="kafkaSpecificDecoder"
key-decoder="kafkaReflectionDecoder"
max-messages="5000">
<int-kafka:topic id="test1" streams="4"/>
</int-kafka:consumer-configuration>
<int-kafka:consumer-configuration group-id="default2"
max-messages="50">
<int-kafka:topic id="test2" streams="4"/>
</int-kafka:consumer-configuration>
<int-kafka:consumer-configuration group-id="default3"
max-messages="10">
<int-kafka:topic-filter pattern="regextopic.*" streams="4" exclude="false"/>
</int-kafka:consumer-configuration>
</int-kafka:consumer-configurations>
</int-kafka:consumer-context>
<int-kafka:consumer-context id="consumerContext"
consumer-timeout="1000"
zookeeper-connect="zookeeperConnect">
<int-kafka:consumer-configurations>
<int-kafka:consumer-configuration group-id="default"
value-decoder="kafkaSpecificDecoder"
key-decoder="kafkaReflectionDecoder"
max-messages="5000">
<int-kafka:topic id="test1" streams="4"/>
</int-kafka:consumer-configuration>
<int-kafka:consumer-configuration group-id="default1"
max-messages="50">
<int-kafka:topic id="test2" streams="4"/>
</int-kafka:consumer-configuration>
</int-kafka:consumer-configurations>
</int-kafka:consumer-context>
<bean id="partitionlessTransformer" class="org.springframework.integration.samples.kafka.outbound.PartitionlessTransformer"/>
<bean id="partitionlessTransformer" class="org.springframework.integration.samples.kafka.outbound.PartitionlessTransformer"/>
<int:transformer ref="partitionlessTransformer" method="transform"
input-channel="inputFromKafka"
output-channel="inputFromTransformer"/>
<int:transformer ref="partitionlessTransformer" method="transform"
input-channel="inputFromKafka"
output-channel="inputFromTransformer"/>
</beans>

83
samples/kafka/src/main/resources/org/springframework/integration/samples/kafka/outbound/kafkaOutboundAdapterParserTests-context.xml
View File

@@ -1,49 +1,52 @@
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:int="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">
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">
<int:channel id="inputToKafka">
<int:queue/>
</int:channel>
<int:channel id="inputToKafka">
<int:queue/>
</int:channel>
<int-kafka:outbound-channel-adapter kafka-producer-context-ref="kafkaProducerContext"
auto-startup="false"
channel="inputToKafka">
<int:poller fixed-delay="1000" time-unit="MILLISECONDS" receive-timeout="0" task-executor="taskExecutor"/>
</int-kafka:outbound-channel-adapter>
<task:executor id="taskExecutor" pool-size="5" keep-alive="120" queue-capacity="500"/>
<int-kafka:outbound-channel-adapter kafka-producer-context-ref="kafkaProducerContext"
auto-startup="false"
channel="inputToKafka">
<int:poller fixed-delay="1000" time-unit="MILLISECONDS" receive-timeout="0" task-executor="taskExecutor"/>
</int-kafka:outbound-channel-adapter>
<bean id="kafkaReflectionEncoder" class="org.springframework.integration.kafka.serializer.avro.AvroReflectDatumBackedKafkaEncoder">
<constructor-arg value="java.lang.String" />
</bean>
<task:executor id="taskExecutor" pool-size="5" keep-alive="120" queue-capacity="500"/>
<bean id="kafkaSpecificEncoder" class="org.springframework.integration.kafka.serializer.avro.AvroSpecificDatumBackedKafkaEncoder">
<constructor-arg value="org.springframework.integration.samples.kafka.user.User" />
</bean>
<bean id="kafkaReflectionEncoder" class="org.springframework.integration.kafka.serializer.avro.AvroReflectDatumBackedKafkaEncoder">
<constructor-arg value="java.lang.String" />
</bean>
<bean id="customPartitioner" class="org.springframework.integration.samples.kafka.outbound.CustomPartitioner"/>
<bean id="kafkaSpecificEncoder" class="org.springframework.integration.kafka.serializer.avro.AvroSpecificDatumBackedKafkaEncoder">
<constructor-arg value="org.springframework.integration.samples.kafka.user.User" />
</bean>
<int-kafka:producer-context id="kafkaProducerContext">
<int-kafka:producer-configurations>
<int-kafka:producer-configuration broker-list="localhost:9092"
key-class-type="java.lang.String"
value-class-type="org.springframework.integration.samples.kafka.user.User"
topic="test1"
value-encoder="kafkaSpecificEncoder"
key-encoder="kafkaReflectionEncoder"
compression-codec="default"
partitioner="customPartitioner"/>
<int-kafka:producer-configuration broker-list="localhost:9092"
topic="test2"
compression-codec="default"/>
</int-kafka:producer-configurations>
</int-kafka:producer-context>
<bean id="customPartitioner" class="org.springframework.integration.samples.kafka.outbound.CustomPartitioner"/>
<int-kafka:producer-context id="kafkaProducerContext">
<int-kafka:producer-configurations>
<int-kafka:producer-configuration broker-list="localhost:9092"
key-class-type="java.lang.String"
value-class-type="org.springframework.integration.samples.kafka.user.User"
topic="test1"
value-encoder="kafkaSpecificEncoder"
key-encoder="kafkaReflectionEncoder"
compression-codec="default"
partitioner="customPartitioner"/>
<int-kafka:producer-configuration broker-list="localhost:9092"
topic="test2"
compression-codec="default"/>
<int-kafka:producer-configuration broker-list="localhost:9092"
topic="regextopic.*"
compression-codec="default"/>
</int-kafka:producer-configurations>
</int-kafka:producer-context>
</beans>

22
spring-integration-kafka/README.md
View File

@@ -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"/>
<int-kafka:producer-configuration broker-list="localhost:9092"
topic="regextopic.*"
compression-codec="default"/>
</int-kafka:producer-configurations>
</int-kafka:producer-context>
```
@@ -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">
<int:poller fixed-delay="10" time-unit="MILLISECONDS" max-messages-per-poll="5"/>
</int-kafka:inbound-channel-adapter>
</int-kafka:inbound-channel-adapter>
```
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
<int-kafka:topic id="test1" streams="4"/>
<int-kafka:topic id="test2" streams="4"/>
</int-kafka:consumer-configuration>
<int-kafka:consumer-configuration group-id="default3"
value-decoder="kafkaSpecificDecoder"
key-decoder="kafkaReflectionDecoder"
max-messages="10">
<int-kafka:topic-filter pattern="regextopic.*" streams="4" exclude="false"/>
</int-kafka:consumer-configuration>
</int-kafka:consumer-configurations>
</int-kafka:consumer-context>
</int-kafka:consumer-context>
```
`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.

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

@@ -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<String, Integer> topicStreamsMap = new HashMap<String, Integer>();
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<Element> 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")),

4
spring-integration-kafka/src/main/java/org/springframework/integration/kafka/config/xml/KafkaProducerContextParser.java
View File

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

2
spring-integration-kafka/src/main/java/org/springframework/integration/kafka/serializer/avro/AvroDatumSupport.java
View File

@@ -21,7 +21,6 @@ public abstract class AvroDatumSupport<T> {
this.avroSerializer = new AvroSerializer<T>();
}
@SuppressWarnings("unchecked")
public byte[] toBytes(final T source, final DatumWriter<T> writer) {
try {
return avroSerializer.serialize(source, writer);
@@ -31,7 +30,6 @@ public abstract class AvroDatumSupport<T> {
return null;
}
@SuppressWarnings("unchecked")
public T fromBytes(final byte[] bytes, final DatumReader<T> reader) {
try {
return avroSerializer.deserialize(bytes, reader);

6
spring-integration-kafka/src/main/java/org/springframework/integration/kafka/serializer/avro/AvroReflectDatumBackedKafkaDecoder.java
View File

@@ -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<T> extends AvroDatumSupport<T> implements Decoder<T> {
private static final Log LOG = LogFactory.getLog(AvroReflectDatumBackedKafkaDecoder.class);
private final DatumReader<T> reader;
@@ -35,9 +33,7 @@ public class AvroReflectDatumBackedKafkaDecoder<T> extends AvroDatumSupport<T> i
}
@Override
@SuppressWarnings("unchecked")
public T fromBytes(final byte[] bytes) {
return fromBytes(bytes, reader);
}
}

5
spring-integration-kafka/src/main/java/org/springframework/integration/kafka/serializer/avro/AvroReflectDatumBackedKafkaEncoder.java
View File

@@ -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<T> extends AvroDatumSupport<T> implements Encoder<T> {
private static final Log LOG = LogFactory.getLog(AvroReflectDatumBackedKafkaEncoder.class);
private final DatumWriter<T> writer;
@@ -35,7 +33,6 @@ public class AvroReflectDatumBackedKafkaEncoder<T> extends AvroDatumSupport<T> i
}
@Override
@SuppressWarnings("unchecked")
public byte[] toBytes(final T source) {
return toBytes(source, writer);
}

6
spring-integration-kafka/src/main/java/org/springframework/integration/kafka/serializer/avro/AvroSpecificDatumBackedKafkaDecoder.java
View File

@@ -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<T> extends AvroDatumSupport<T> implements Decoder<T> {
private static final Log LOG = LogFactory.getLog(AvroSpecificDatumBackedKafkaDecoder.class);
private final DatumReader<T> reader;
public AvroSpecificDatumBackedKafkaDecoder(final Class<T> specificRecordBase) {
@@ -21,7 +18,6 @@ public class AvroSpecificDatumBackedKafkaDecoder<T> extends AvroDatumSupport<T>
}
@Override
@SuppressWarnings("unchecked")
public T fromBytes(final byte[] bytes) {
return fromBytes(bytes, reader);
}

6
spring-integration-kafka/src/main/java/org/springframework/integration/kafka/serializer/avro/AvroSpecificDatumBackedKafkaEncoder.java
View File

@@ -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<T> extends AvroDatumSupport<T> implements Encoder<T> {
private static final Log LOG = LogFactory.getLog(AvroSpecificDatumBackedKafkaEncoder.class);
private final DatumWriter<T> writer;
public AvroSpecificDatumBackedKafkaEncoder(final Class<T> specificRecordClazz) {
@@ -21,7 +18,6 @@ public class AvroSpecificDatumBackedKafkaEncoder<T> extends AvroDatumSupport<T>
}
@Override
@SuppressWarnings("unchecked")
public byte[] toBytes(final T source) {
return toBytes(source, writer);
}

139
spring-integration-kafka/src/main/java/org/springframework/integration/kafka/support/ConsumerConfiguration.java
View File

@@ -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<K,V> {
public class ConsumerConfiguration<K, V> {
private static final Log LOGGER = LogFactory.getLog(ConsumerConfiguration.class);
private final ConsumerMetadata<K,V> consumerMetadata;
private final ConsumerMetadata<K, V> consumerMetadata;
private final ConsumerConnectionProvider consumerConnectionProvider;
private final MessageLeftOverTracker<K,V> messageLeftOverTracker;
private final MessageLeftOverTracker<K, V> messageLeftOverTracker;
private ConsumerConnector consumerConnector;
private volatile int count = 0;
private int maxMessages = 1;
private Collection<List<KafkaStream<K, V>>> consumerMessageStreams;
private ExecutorService executorService = Executors.newCachedThreadPool();
private final ExecutorService executorService = Executors.newCachedThreadPool();
public ConsumerConfiguration(final ConsumerMetadata<K,V> consumerMetadata,
final ConsumerConnectionProvider consumerConnectionProvider,
final MessageLeftOverTracker<K,V> messageLeftOverTracker) {
public ConsumerConfiguration(final ConsumerMetadata<K, V> consumerMetadata,
final ConsumerConnectionProvider consumerConnectionProvider,
final MessageLeftOverTracker<K, V> messageLeftOverTracker) {
this.consumerMetadata = consumerMetadata;
this.consumerConnectionProvider = consumerConnectionProvider;
this.messageLeftOverTracker = messageLeftOverTracker;
}
public ConsumerMetadata<K,V> getConsumerMetadata() {
public ConsumerMetadata<K, V> getConsumerMetadata() {
return consumerMetadata;
}
@@ -65,23 +54,23 @@ public class ConsumerConfiguration<K,V> {
count = messageLeftOverTracker.getCurrentCount();
final Object lock = new Object();
final List<Callable<List<MessageAndMetadata<K,V>>>> tasks = new LinkedList<Callable<List<MessageAndMetadata<K,V>>>>();
final List<Callable<List<MessageAndMetadata<K, V>>>> tasks = new LinkedList<Callable<List<MessageAndMetadata<K, V>>>>();
final Map<String, List<KafkaStream<K, V>>> consumerMap = getConsumerMapWithMessageStreams();
for (final List<KafkaStream<K,V>> streams : consumerMap.values()) {
for (final KafkaStream<K,V> stream : streams) {
tasks.add(new Callable<List<MessageAndMetadata<K,V>>>() {
for (final List<KafkaStream<K, V>> streams : createConsumerMessageStreams()) {
for (final KafkaStream<K, V> stream : streams) {
tasks.add(new Callable<List<MessageAndMetadata<K, V>>>() {
@Override
public List<MessageAndMetadata<K,V>> call() throws Exception {
final List<MessageAndMetadata<K,V>> rawMessages = new ArrayList<MessageAndMetadata<K,V>>();
public List<MessageAndMetadata<K, V>> call() throws Exception {
final List<MessageAndMetadata<K, V>> rawMessages = new ArrayList<MessageAndMetadata<K, V>>();
try {
while (count < maxMessages) {
final MessageAndMetadata<K,V> messageAndMetadata = stream.iterator().next();
final MessageAndMetadata<K, V> 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<K,V> {
return executeTasks(tasks);
}
private Map<String, Map<Integer, List<Object>>> executeTasks(final List<Callable<List<MessageAndMetadata<K,V>>>> tasks) {
private Map<String, Map<Integer, List<Object>>> executeTasks(
final List<Callable<List<MessageAndMetadata<K, V>>>> tasks) {
final Map<String, Map<Integer, List<Object>>> messages = new ConcurrentHashMap<String, Map<Integer, List<Object>>>();
messages.putAll(getLeftOverMessageMap());
try {
for (final Future<List<MessageAndMetadata<K,V>>> result : executorService.invokeAll(tasks)) {
for (final Future<List<MessageAndMetadata<K, V>>> 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<Integer, List<Object>> existingPayloadMap = messages.get(topic);
getPayload(result.get(), existingPayloadMap);
@@ -126,21 +117,21 @@ public class ConsumerConfiguration<K,V> {
return messages;
}
@SuppressWarnings("unchecked")
private Map<String, Map<Integer, List<Object>>> getLeftOverMessageMap() {
final Map<String, Map<Integer, List<Object>>> messages = new ConcurrentHashMap<String, Map<Integer, List<Object>>>();
for (final MessageAndMetadata<K,V> mamd : messageLeftOverTracker.getMessageLeftOverFromPreviousPoll()) {
for (final MessageAndMetadata<K, V> mamd : messageLeftOverTracker.getMessageLeftOverFromPreviousPoll()) {
final String topic = mamd.topic();
if (!messages.containsKey(topic)) {
final List<MessageAndMetadata<K,V>> l = new ArrayList<MessageAndMetadata<K,V>>();
final List<MessageAndMetadata<K, V>> l = new ArrayList<MessageAndMetadata<K, V>>();
l.add(mamd);
messages.put(topic, getPayload(l));
} else {
}
else {
final Map<Integer, List<Object>> existingPayloadMap = messages.get(topic);
final List<MessageAndMetadata<K,V>> l = new ArrayList<MessageAndMetadata<K,V>>();
final List<MessageAndMetadata<K, V>> l = new ArrayList<MessageAndMetadata<K, V>>();
l.add(mamd);
getPayload(l, existingPayloadMap);
}
@@ -149,15 +140,16 @@ public class ConsumerConfiguration<K,V> {
return messages;
}
private Map<Integer, List<Object>> getPayload(final List<MessageAndMetadata<K,V>> messageAndMetadatas) {
private Map<Integer, List<Object>> getPayload(final List<MessageAndMetadata<K, V>> messageAndMetadatas) {
final Map<Integer, List<Object>> payloadMap = new ConcurrentHashMap<Integer, List<Object>>();
for (final MessageAndMetadata<K,V> messageAndMetadata : messageAndMetadatas) {
for (final MessageAndMetadata<K, V> messageAndMetadata : messageAndMetadatas) {
if (!payloadMap.containsKey(messageAndMetadata.partition())) {
final List<Object> payload = new ArrayList<Object>();
payload.add(messageAndMetadata.message());
payloadMap.put(messageAndMetadata.partition(), payload);
} else {
}
else {
final List<Object> payload = payloadMap.get(messageAndMetadata.partition());
payload.add(messageAndMetadata.message());
}
@@ -167,25 +159,52 @@ public class ConsumerConfiguration<K,V> {
return payloadMap;
}
private void getPayload(final List<MessageAndMetadata<K,V>> messageAndMetadatas, final Map<Integer, List<Object>> existingPayloadMap) {
for (final MessageAndMetadata<K,V> messageAndMetadata : messageAndMetadatas) {
private void getPayload(final List<MessageAndMetadata<K, V>> messageAndMetadatas,
final Map<Integer, List<Object>> existingPayloadMap) {
for (final MessageAndMetadata<K, V> messageAndMetadata : messageAndMetadatas) {
if (!existingPayloadMap.containsKey(messageAndMetadata.partition())) {
final List<Object> payload = new ArrayList<Object>();
payload.add(messageAndMetadata.message());
existingPayloadMap.put(messageAndMetadata.partition(), payload);
} else {
}
else {
final List<Object> payload = existingPayloadMap.get(messageAndMetadata.partition());
payload.add(messageAndMetadata.message());
}
}
}
@SuppressWarnings("unchecked")
public Map<String, List<KafkaStream<K,V>>> getConsumerMapWithMessageStreams() {
return getConsumerConnector().createMessageStreams(
consumerMetadata.getTopicStreamMap(),
consumerMetadata.getKeyDecoder(),
consumerMetadata.getValueDecoder());
private Collection<List<KafkaStream<K, V>>> createConsumerMessageStreams() {
if (consumerMessageStreams == null) {
if (!(consumerMetadata.getTopicStreamMap() == null || consumerMetadata.getTopicStreamMap().isEmpty())) {
consumerMessageStreams = createMessageStreamsForTopic().values();
}
else {
consumerMessageStreams = new ArrayList<List<KafkaStream<K, V>>>();
consumerMessageStreams.add(createMessageStreamsForTopicFilter());
}
}
return consumerMessageStreams;
}
public Map<String, List<KafkaStream<K, V>>> createMessageStreamsForTopic() {
return getConsumerConnector().createMessageStreams(consumerMetadata.getTopicStreamMap(),
consumerMetadata.getKeyDecoder(), consumerMetadata.getValueDecoder());
}
public List<KafkaStream<K, V>> createMessageStreamsForTopicFilter() {
List<KafkaStream<K, V>> messageStream = new ArrayList<KafkaStream<K, V>>();
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() {

16
spring-integration-kafka/src/main/java/org/springframework/integration/kafka/support/ConsumerMetadata.java
View File

@@ -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<K,V> implements InitializingBean {
@@ -46,6 +47,7 @@ public class ConsumerMetadata<K,V> implements InitializingBean {
private Decoder<V> valueDecoder;
private Decoder<K> keyDecoder;
private Map<String, Integer> topicStreamMap;
private TopicFilterConfiguration topicFilterConfiguration;
public String getGroupId() {
return groupId;
@@ -186,4 +188,14 @@ public class ConsumerMetadata<K,V> implements InitializingBean {
setKeyDecoder((Decoder<K>) getValueDecoder());
}
}
public TopicFilterConfiguration getTopicFilterConfiguration() {
return topicFilterConfiguration;
}
public void setTopicFilterConfiguration(
TopicFilterConfiguration topicFilterConfiguration) {
this.topicFilterConfiguration = topicFilterConfiguration;
}
}

21
spring-integration-kafka/src/main/java/org/springframework/integration/kafka/support/KafkaProducerContext.java
View File

@@ -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<K,V> implements BeanFactoryAware {
private static final Log LOGGER = LogFactory.getLog(KafkaProducerContext.class);
private Map<String, ProducerConfiguration<K,V>> topicsConfiguration;
@SuppressWarnings("unchecked")
public void send(final Message<?> message) throws Exception {
final ProducerConfiguration<K,V> producerConfiguration =
getTopicConfiguration(message.getHeaders().get("topic", String.class));
@@ -41,15 +42,15 @@ public class KafkaProducerContext<K,V> implements BeanFactoryAware {
}
}
private ProducerConfiguration<K,V> getTopicConfiguration(final String topic){
public ProducerConfiguration<K, V> getTopicConfiguration(final String topic) {
final Collection<ProducerConfiguration<K,V>> topics = topicsConfiguration.values();
for (final ProducerConfiguration<K,V> 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;
}

61
spring-integration-kafka/src/main/java/org/springframework/integration/kafka/support/ProducerConfiguration.java
View File

@@ -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<K,V> {
private final Producer<K,V> producer;
private final ProducerMetadata<K,V> producerMetadata;
public class ProducerConfiguration<K, V> {
private final Producer<K, V> producer;
private final ProducerMetadata<K, V> producerMetadata;
public ProducerConfiguration(final ProducerMetadata<K, V> producerMetadata, final Producer<K, V> producer){
public ProducerConfiguration(final ProducerMetadata<K, V> producerMetadata, final Producer<K, V> producer) {
this.producerMetadata = producerMetadata;
this.producer = producer;
}
@@ -46,10 +39,12 @@ public class ProducerConfiguration<K,V> {
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<K, V>(producerMetadata.getTopic(), getKey(message), v));
} else {
producer.send(new KeyedMessage<K, V>(producerMetadata.getTopic(), v));
producer.send(new KeyedMessage<K, V>(topic, getKey(message), v));
}
else {
producer.send(new KeyedMessage<K, V>(topic, v));
}
}
@@ -57,7 +52,8 @@ public class ProducerConfiguration<K,V> {
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<K,V> {
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<K,V> {
}
@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<K,V> {
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();
}
}

12
spring-integration-kafka/src/main/java/org/springframework/integration/kafka/support/ProducerMetadata.java
View File

@@ -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<K,V> implements InitializingBean {
@@ -137,4 +139,14 @@ public class ProducerMetadata<K,V> 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();
}
}

46
spring-integration-kafka/src/main/java/org/springframework/integration/kafka/support/TopicFilterConfiguration.java Normal file
View File

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

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

@@ -232,14 +232,56 @@
]]></xsd:documentation>
</xsd:annotation>
<xsd:complexType>
<xsd:sequence>
<xsd:choice>
<xsd:element name="topic" maxOccurs="unbounded">
<xsd:complexType>
<xsd:attribute name="id" type="xsd:string" use="required"/>
<xsd:attribute name="streams" type="xsd:string" use="required"/>
</xsd:complexType>
</xsd:element>
</xsd:sequence>
<xsd:element name="topic-filter" maxOccurs="1">
<xsd:complexType>
<xsd:attribute name="pattern" type="xsd:string" use="required">
<xsd:annotation>
<xsd:documentation><![CDATA[
Regex pattern to match topic
]]></xsd:documentation>
<xsd:appinfo>
<tool:annotation kind="direct">
<tool:expected-type type="java.lang.String"/>
</tool:annotation>
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="streams" type="xsd:string" use="required">
<xsd:annotation>
<xsd:documentation><![CDATA[
Number of streams (threads) to use to consume messages
]]></xsd:documentation>
<xsd:appinfo>
<tool:annotation kind="direct">
<tool:expected-type type="java.lang.String"/>
</tool:annotation>
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="exclude" type="xsd:boolean" use="optional" default="false">
<xsd:annotation>
<xsd:documentation><![CDATA[
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.
Default value is false.
]]></xsd:documentation>
<xsd:appinfo>
<tool:annotation kind="direct">
<tool:expected-type type="java.lang.String"/>
</tool:annotation>
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
</xsd:complexType>
</xsd:element>
</xsd:choice>
<xsd:attribute name="max-messages" use="optional">
<xsd:annotation>
<xsd:documentation><![CDATA[

3
spring-integration-kafka/src/test/java/org/springframework/integration/kafka/serializer/AvroReflectDatumBackedKafkaSerializerTest.java
View File

@@ -26,8 +26,8 @@ import org.springframework.integration.kafka.test.utils.TestObject;
* @since 0.5
*/
public class AvroReflectDatumBackedKafkaSerializerTest {
@Test
@SuppressWarnings("unchecked")
public void testDecodePlainSchema() {
final AvroReflectDatumBackedKafkaEncoder<TestObject> avroBackedKafkaEncoder = new AvroReflectDatumBackedKafkaEncoder<TestObject>(TestObject.class);
@@ -45,7 +45,6 @@ public class AvroReflectDatumBackedKafkaSerializerTest {
}
@Test
@SuppressWarnings("unchecked")
public void anotherTest() {
final AvroReflectDatumBackedKafkaEncoder<String> avroBackedKafkaEncoder = new AvroReflectDatumBackedKafkaEncoder<String>(java.lang.String.class);
final String testString = "Testing Avro";

1
spring-integration-kafka/src/test/java/org/springframework/integration/kafka/serializer/AvroSpecificDatumBackedKafkaSerializerTest.java
View File

@@ -13,7 +13,6 @@ import org.springframework.integration.kafka.test.utils.User;
public class AvroSpecificDatumBackedKafkaSerializerTest {
@Test
@SuppressWarnings("unchecked")
public void testEncodeDecodeFromSpecificDatumSchema() {
final AvroSpecificDatumBackedKafkaEncoder<User> avroBackedKafkaEncoder = new AvroSpecificDatumBackedKafkaEncoder<User>(User.class);

164
spring-integration-kafka/src/test/java/org/springframework/integration/kafka/support/ConsumerConfigurationTests.java
View File

@@ -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<K,V> {
@@ -55,6 +46,10 @@ public class ConsumerConfigurationTests<K,V> {
final MessageLeftOverTracker<K,V> messageLeftOverTracker = mock(MessageLeftOverTracker.class);
final ConsumerConnector consumerConnector = mock(ConsumerConnector.class);
Map<String, Integer> topicStreamMap = new HashMap<String, Integer>();
topicStreamMap.put("topic1", 1);
when(consumerMetadata.getTopicStreamMap()).thenReturn(topicStreamMap);
when(consumerConnectionProvider.getConsumerConnector()).thenReturn(consumerConnector);
final ConsumerConfiguration<K,V> consumerConfiguration = new ConsumerConfiguration<K,V>(consumerMetadata,
@@ -67,7 +62,7 @@ public class ConsumerConfigurationTests<K,V> {
final Map<String, List<KafkaStream<K,V>>> messageStreams = new HashMap<String, List<KafkaStream<K,V>>>();
messageStreams.put("topic", streams);
when(consumerConfiguration.getConsumerMapWithMessageStreams()).thenReturn(messageStreams);
when(consumerConfiguration.createMessageStreamsForTopic()).thenReturn(messageStreams);
final ConsumerIterator<K,V> iterator = mock(ConsumerIterator.class);
when(stream.iterator()).thenReturn(iterator);
final MessageAndMetadata<K,V> messageAndMetadata = mock(MessageAndMetadata.class);
@@ -95,6 +90,10 @@ public class ConsumerConfigurationTests<K,V> {
mock(ConsumerConnectionProvider.class);
final MessageLeftOverTracker<K,V> messageLeftOverTracker = mock(MessageLeftOverTracker.class);
Map<String, Integer> topicStreamMap = new HashMap<String, Integer>();
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<K,V> {
final Map<String, List<KafkaStream<K,V>>> messageStreams = new HashMap<String, List<KafkaStream<K,V>>>();
messageStreams.put("topic", streams);
when(consumerConfiguration.getConsumerMapWithMessageStreams()).thenReturn(messageStreams);
when(consumerConfiguration.createMessageStreamsForTopic()).thenReturn(messageStreams);
final ConsumerIterator<K,V> iterator1 = mock(ConsumerIterator.class);
final ConsumerIterator<K,V> iterator2 = mock(ConsumerIterator.class);
final ConsumerIterator<K,V> iterator3 = mock(ConsumerIterator.class);
@@ -160,7 +159,12 @@ public class ConsumerConfigurationTests<K,V> {
final ConsumerMetadata<K,V> consumerMetadata = mock(ConsumerMetadata.class);
final ConsumerConnectionProvider consumerConnectionProvider =
mock(ConsumerConnectionProvider.class);
final MessageLeftOverTracker<K,V> messageLeftOverTracker = mock(MessageLeftOverTracker.class);
final MessageLeftOverTracker<K, V> messageLeftOverTracker = mock(MessageLeftOverTracker.class);
Map<String, Integer> topicStreamMap = new HashMap<String, Integer>();
topicStreamMap.put("topic1", 1);
when(consumerMetadata.getTopicStreamMap()).thenReturn(topicStreamMap);
final ConsumerConnector consumerConnector = mock(ConsumerConnector.class);
@@ -182,7 +186,7 @@ public class ConsumerConfigurationTests<K,V> {
messageStreams.put("topic2", streams);
messageStreams.put("topic3", streams);
when(consumerConfiguration.getConsumerMapWithMessageStreams()).thenReturn(messageStreams);
when(consumerConfiguration.createMessageStreamsForTopic()).thenReturn(messageStreams);
final ConsumerIterator<K,V> iterator1 = mock(ConsumerIterator.class);
final ConsumerIterator<K,V> iterator2 = mock(ConsumerIterator.class);
final ConsumerIterator<K,V> iterator3 = mock(ConsumerIterator.class);
@@ -211,6 +215,7 @@ public class ConsumerConfigurationTests<K,V> {
when(messageAndMetadata1.partition()).thenAnswer(getAnswer());
final Map<String, Map<Integer, List<Object>>> messages = consumerConfiguration.receive();
int sum = 0;
final Collection<Map<Integer, List<Object>>> values = messages.values();
@@ -224,6 +229,8 @@ public class ConsumerConfigurationTests<K,V> {
Assert.assertEquals(sum, 9);
}
private Answer<Object> getAnswer() {
return new Answer<Object>() {
private int count = 0;
@@ -250,6 +257,9 @@ public class ConsumerConfigurationTests<K,V> {
final MessageLeftOverTracker<K,V> messageLeftOverTracker = mock(MessageLeftOverTracker.class);
final ConsumerConnector consumerConnector = mock(ConsumerConnector.class);
Map<String, Integer> topicStreamMap = new HashMap<String, Integer>();
topicStreamMap.put("topic1", 1);
when(consumerMetadata.getTopicStreamMap()).thenReturn(topicStreamMap);
when(messageLeftOverTracker.getCurrentCount()).thenReturn(3);
final MessageAndMetadata<String, String> m1 = new MessageAndMetadata<String, String>("key1", "value1", "topic1", 1, 1L);
final MessageAndMetadata<String, String> m2 = new MessageAndMetadata<String, String>("key2", "value2", "topic2", 1, 1L);
@@ -273,11 +283,10 @@ public class ConsumerConfigurationTests<K,V> {
streams.add(stream);
final Map<String, List<KafkaStream<K,V>>> messageStreams = new HashMap<String, List<KafkaStream<K,V>>>();
messageStreams.put("topic1", streams);
when(consumerConfiguration.getConsumerMapWithMessageStreams()).thenReturn(messageStreams);
final ConsumerIterator<String, String> iterator = mock(ConsumerIterator.class);
when(stream.iterator()).thenReturn((ConsumerIterator<K,V>) iterator);
final MessageAndMetadata<String, String> 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<K,V> {
final ConsumerConfiguration<K,V> consumerConfiguration = new ConsumerConfiguration<K,V>(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<K,V> {
final ConsumerConfiguration<String, String> consumerConfiguration = new ConsumerConfiguration<String, String>(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<KafkaStream<byte[], byte[]>> streams = new ArrayList<KafkaStream<byte[], byte[]>>();
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<String, Map<Integer, List<Object>>> 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<KafkaStream<byte[], byte[]>> streams = new ArrayList<KafkaStream<byte[], byte[]>>();
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<String, Map<Integer, List<Object>>> messages = consumerConfiguration.receive();
Assert.assertEquals(messages.size(), 1);
int sum = 0;
final Map<Integer, List<Object>> values = messages.get("topic");
for (final List<Object> l : values.values()) {
sum += l.size();
}
Assert.assertEquals(sum, 3);
}
private boolean valueFound(final List<Object> l, final String value){
for (final Object o : l){
if (value.equals(o)){

65
spring-integration-kafka/src/test/java/org/springframework/integration/kafka/support/KafkaProducerContextTests.java Normal file
View File

@@ -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<String, String> producerMetadata = Mockito.mock(ProducerMetadata.class);
String testRegex = "test.*";
Mockito.when(producerMetadata.getTopic()).thenReturn(testRegex);
final Producer<String, String> producer = Mockito.mock(Producer.class);
final ProducerConfiguration<String, String> producerConfiguration = new ProducerConfiguration<String, String>(producerMetadata, producer);
final Map<String, ProducerConfiguration> topicConfigurations = new HashMap<String, ProducerConfiguration>();
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"));
}
}

3
spring-integration-kafka/src/test/java/org/springframework/integration/kafka/test/utils/User.java
View File

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