Merge pull request #56 from sobychacko/INTEXT-84
* sobychacko-INTEXT-84: INTEXT-84 Kafka: Enhance Avro serialization support
This commit is contained in:
Gunnar Hillert
@@ -1,27 +1,62 @@
|
||||
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"
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
|
||||
apply plugin: 'base'
|
||||
apply plugin: 'java'
|
||||
apply plugin: 'eclipse'
|
||||
apply plugin: 'application'
|
||||
apply plugin: 'idea'
|
||||
apply plugin: 'maven'
|
||||
apply plugin: 'avro-gradle-plugin'
|
||||
|
||||
ext {
|
||||
avroTaskGroup = "Avro"
|
||||
avroSource = "schemas"
|
||||
avroDest = "target/generated-avro-sources/main/java"
|
||||
}
|
||||
|
||||
repositories {
|
||||
mavenLocal()
|
||||
mavenCentral()
|
||||
maven { url "http://repo.springsource.org/libs-snapshot" }
|
||||
maven { url 'http://repo.springsource.org/plugins-release' }
|
||||
|
||||
mavenLocal()
|
||||
maven {
|
||||
url 'https://repository.apache.org/content/groups/public'
|
||||
}
|
||||
maven { url 'https://repo.springsource.org/libs-milestone' }
|
||||
|
||||
}
|
||||
|
||||
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.springframework.integration:spring-integration-kafka:0.5.0.BUILD-SNAPSHOT") {
|
||||
exclude module: 'log4j'
|
||||
exclude module: 'jms'
|
||||
exclude module: 'jmxtools'
|
||||
exclude module: 'jmxri'
|
||||
}
|
||||
compile("log4j:log4j:1.2.15") {
|
||||
exclude module: 'mail'
|
||||
exclude module: 'jms'
|
||||
exclude module: 'jmx'
|
||||
exclude module: 'jmxtools'
|
||||
@@ -32,6 +67,30 @@ compile("log4j:log4j:1.2.15") {
|
||||
compile "commons-logging:commons-logging:1.1.1"
|
||||
}
|
||||
|
||||
compileAvro.group = avroTaskGroup
|
||||
compileAvro.description = "Generates Java code from avro schema"
|
||||
compileAvro.source = avroSource
|
||||
compileAvro.destinationDir = file(avroDest)
|
||||
|
||||
task cleanAvro(type: Delete) {
|
||||
group = avroTaskGroup
|
||||
description = "deletes generated avro code"
|
||||
delete avroDest
|
||||
}
|
||||
|
||||
compileJava.dependsOn compileAvro
|
||||
|
||||
sourceSets {
|
||||
main {
|
||||
java {
|
||||
srcDir avroDest
|
||||
}
|
||||
resources {
|
||||
srcDir avroSource
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
task wrapper(type: Wrapper) {
|
||||
description = 'Generates gradlew[.bat] scripts'
|
||||
gradleVersion = '1.3'
|
||||
|
||||
7
samples/kafka/schemas/user.avdl
Normal file
7
samples/kafka/schemas/user.avdl
Normal file
@@ -0,0 +1,7 @@
|
||||
@namespace("org.springframework.integration.samples.kafka.user")
|
||||
protocol UserProtocol{
|
||||
record User {
|
||||
string firstName;
|
||||
string lastName;
|
||||
}
|
||||
}
|
||||
@@ -19,6 +19,7 @@ import org.apache.commons.logging.Log;
|
||||
import org.apache.commons.logging.LogFactory;
|
||||
import org.springframework.context.support.ClassPathXmlApplicationContext;
|
||||
import org.springframework.integration.MessageChannel;
|
||||
import org.springframework.integration.samples.kafka.user.User;
|
||||
import org.springframework.integration.support.MessageBuilder;
|
||||
|
||||
public class OutboundRunner {
|
||||
@@ -34,8 +35,11 @@ public class OutboundRunner {
|
||||
|
||||
//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("hello Fom ob adapter test1 - " + i)
|
||||
MessageBuilder.withPayload(user)
|
||||
.setHeader("messageKey", String.valueOf(i))
|
||||
.setHeader("topic", "test1").build());
|
||||
|
||||
|
||||
@@ -27,22 +27,26 @@
|
||||
<int:poller fixed-delay="1" time-unit="MILLISECONDS"/>
|
||||
</int-kafka:inbound-channel-adapter>
|
||||
|
||||
<bean id="kafkaDecoder" class="org.springframework.integration.kafka.serializer.avro.AvroBackedKafkaDecoder">
|
||||
<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>
|
||||
|
||||
<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="kafkaDecoder"
|
||||
key-decoder="kafkaDecoder"
|
||||
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="5">
|
||||
max-messages="50">
|
||||
<int-kafka:topic id="test2" streams="4"/>
|
||||
</int-kafka:consumer-configuration>
|
||||
</int-kafka:consumer-configurations>
|
||||
|
||||
@@ -21,20 +21,24 @@
|
||||
|
||||
<task:executor id="taskExecutor" pool-size="5" keep-alive="120" queue-capacity="500"/>
|
||||
|
||||
<bean id="kafkaEncoder" class="org.springframework.integration.kafka.serializer.avro.AvroBackedKafkaEncoder">
|
||||
<bean id="kafkaReflectionEncoder" class="org.springframework.integration.kafka.serializer.avro.AvroReflectDatumBackedKafkaEncoder">
|
||||
<constructor-arg value="java.lang.String" />
|
||||
</bean>
|
||||
|
||||
<bean id="kafkaSpecificEncoder" class="org.springframework.integration.kafka.serializer.avro.AvroSpecificDatumBackedKafkaEncoder">
|
||||
<constructor-arg value="org.springframework.integration.samples.kafka.user.User" />
|
||||
</bean>
|
||||
|
||||
<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="java.lang.String"
|
||||
value-class-type="org.springframework.integration.samples.kafka.user.User"
|
||||
topic="test1"
|
||||
value-encoder="kafkaEncoder"
|
||||
key-encoder="kafkaEncoder"
|
||||
value-encoder="kafkaSpecificEncoder"
|
||||
key-encoder="kafkaReflectionEncoder"
|
||||
compression-codec="default"
|
||||
partitioner="customPartitioner"/>
|
||||
<int-kafka:producer-configuration broker-list="localhost:9092"
|
||||
|
||||
@@ -6,7 +6,7 @@ Welcome to the *Spring Integration Kafka adapter*. Apache Kafka is a distributed
|
||||
data at constant time. For more information on Kafka and its design goals, please see [Kafka main page](http://kafka.apache.org/)
|
||||
|
||||
Spring Integration Kafka adapters are built for Kafka 0.8 and since 0.8 is not backward compatible with any previous versions, Spring Integration will not
|
||||
support any Kafka versions prior to 0.8. As of this writing, Kafka 0.8 is still WIP.
|
||||
support any Kafka versions prior to 0.8. As of this writing, Kafka 0.8 is still WIP, however a beta release is available [here](http://http://kafka.apache.org/downloads.html).
|
||||
|
||||
Checking out and building
|
||||
-----------------------------
|
||||
@@ -16,11 +16,11 @@ Scala 2.9.2.
|
||||
|
||||
In order to build the project:
|
||||
|
||||
./gradlew build
|
||||
./gradlew build
|
||||
|
||||
In order to install this into your local maven cache:
|
||||
|
||||
./gradlew install
|
||||
./gradlew install
|
||||
|
||||
Spring Integration Kafka project currently supports the following two components. Please keep in mind that
|
||||
this is very early stage in development and do not yet fully make use of all the features that Kafka provides.
|
||||
@@ -40,12 +40,12 @@ you have to specify a message key and the topic as header values and the message
|
||||
Here is an example.
|
||||
|
||||
```java
|
||||
final MessageChannel channel = ctx.getBean("inputToKafka", MessageChannel.class);
|
||||
final MessageChannel channel = ctx.getBean("inputToKafka", MessageChannel.class);
|
||||
|
||||
channel.send(
|
||||
MessageBuilder.withPayload(payload).
|
||||
setHeader("messageKey", "key")
|
||||
.setHeader("topic", "test").build());
|
||||
channel.send(
|
||||
MessageBuilder.withPayload(payload)
|
||||
.setHeader("messageKey", "key")
|
||||
.setHeader("topic", "test").build());
|
||||
```
|
||||
|
||||
This would create a message with a payload. In addition to this, it also creates two header entries as key/value pairs - one for
|
||||
@@ -54,12 +54,12 @@ the message key and another for the topic that this message belongs to.
|
||||
Here is how kafka outbound channel adapter is configured:
|
||||
|
||||
```xml
|
||||
<int-kafka:outbound-channel-adapter id="kafkaOutboundChannelAdapter"
|
||||
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>
|
||||
<int-kafka:outbound-channel-adapter id="kafkaOutboundChannelAdapter"
|
||||
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>
|
||||
```
|
||||
|
||||
The key aspect in this configuration is the producer-context-ref. Producer context contains all the producer configuration for all the topics that this adapter is expected to handle.
|
||||
@@ -73,21 +73,21 @@ the receive-timeout configuration. Then it will poll again with a delay of 1 sec
|
||||
Producer context is at the heart of the kafka outbound adapter. Here is an example of how it is configured.
|
||||
|
||||
```xml
|
||||
<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="java.lang.String"
|
||||
topic="test1"
|
||||
value-encoder="kafkaEncoder"
|
||||
key-encoder="kafkaEncoder"
|
||||
compression-codec="default"/>
|
||||
<int-kafka:producer-configuration broker-list="localhost:9092"
|
||||
topic="test2"
|
||||
compression-codec="default"
|
||||
async="true"/>
|
||||
</int-kafka:producer-configurations>
|
||||
</int-kafka:producer-context>
|
||||
<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="java.lang.String"
|
||||
topic="test1"
|
||||
value-encoder="kafkaEncoder"
|
||||
key-encoder="kafkaEncoder"
|
||||
compression-codec="default"/>
|
||||
<int-kafka:producer-configuration broker-list="localhost:9092"
|
||||
topic="test2"
|
||||
compression-codec="default"
|
||||
async="true"/>
|
||||
</int-kafka:producer-configurations>
|
||||
</int-kafka:producer-context>
|
||||
```
|
||||
|
||||
There are a few things going on here. So, lets go one by one. First of all, producer context is simply a holder of, as the name
|
||||
@@ -96,18 +96,18 @@ is ultimately gets translated into a Kafka native producer. Each producer config
|
||||
If you go by the above example, there are two producers generated from this configuration - one for topic named
|
||||
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
|
||||
compression-codec any compression 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.
|
||||
key-encoder serializer to be used for encoding the partition key
|
||||
key-class-type Type of the key class. This will be ignored if no key-encoder is provided
|
||||
value-class-type The type of the value class. This will be ignored if no value-encoder is provided.
|
||||
partitioner custom implementation of a Kafka Partitioner interface.
|
||||
async true/false - default is false. Setting this to true would make the Kafka producer to use
|
||||
an async producer
|
||||
batch-num-messages number of messages to batch at the producer. If async is false, then this has no effect.
|
||||
broker-list List of comma separated brokers that this producer connects to
|
||||
topic 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.
|
||||
key-encoder Serializer to be used for encoding the partition key
|
||||
key-class-type Type of the key class. This will be ignored if no key-encoder is provided
|
||||
value-class-type Type of the value class. This will be ignored if no value-encoder is provided.
|
||||
partitioner Custom implementation of a Kafka Partitioner interface.
|
||||
async True/False - default is false. Setting this to true would make the Kafka producer to use
|
||||
an async producer
|
||||
batch-num-messages Number of messages to batch at the producer. If async is false, then this has no effect.
|
||||
|
||||
The value-encoder and key-encoder are referring to other spring beans. They are essentially implementations of an
|
||||
interface provided by Kafka, the Encoder interface. Similarly, partitioner also refers a Spring bean which implements
|
||||
@@ -116,9 +116,9 @@ the Kafka Partitioner interface.
|
||||
Here is an example of configuring an encoder.
|
||||
|
||||
```xml
|
||||
<bean id="kafkaEncoder" class="org.springframework.integration.kafka.serializer.avro.AvroBackedKafkaEncoder">
|
||||
<constructor-arg value="java.lang.String" />
|
||||
</bean>
|
||||
<bean id="kafkaEncoder" class="org.springframework.integration.kafka.serializer.avro.AvroSpecificDatumBackedKafkaEncoder">
|
||||
<constructor-arg value="com.company.AvroGeneratedSpecificRecord" />
|
||||
</bean>
|
||||
```
|
||||
|
||||
Spring Integration Kafaka adapter provides Apache Avro backed encoders out of the box, as this is a popular choice
|
||||
@@ -134,6 +134,23 @@ If the encoders are default and the objets sent are not serializalbe, then that
|
||||
it is totally up to the developer to configure how the objects are serialized. In that case, the objects may or may not implement
|
||||
the Serializable interface.
|
||||
|
||||
A bit more on the Avro support. There are two flavors of Avro encoders provided, one based on the Avro ReflectDatum and the other
|
||||
based on SpecificDatum. The encoding using reflection is fairly simple as you only have to configure your POJO or other class types
|
||||
along with the XML. Here is an example.
|
||||
|
||||
```xml
|
||||
<bean id="kafkaEncoder" class="org.springframework.integration.kafka.serializer.avro.AvroReflectDatumBackedKafkaEncoder">
|
||||
<constructor-arg value="java.lang.String" />
|
||||
</bean>
|
||||
```
|
||||
|
||||
Reflection based encoding may not be appropriate for large scale systems and Avro's SpecificDatum based encoders can be a better fit. In this case, you can
|
||||
generate a specific Avro object (a glorified POJO) from a schema definition. The generated object will store the schema as well. In order to
|
||||
do this, you need to generate the Avro object separately though. There are both maven and gradle plugins available to do code generation
|
||||
automatically. You have to provide the avdl or avsc files to specify your schema. Once you take care of these steps, you can simply configure
|
||||
a specific datum based Avro encoder (see the first example above) and pass along the fully qualified class name of the generated Avro object
|
||||
for which you want to encode instances. The samples project has examples of using both of these encoders.
|
||||
|
||||
Encoding String for key and value is a very common use case and Kafka provides a StringEncoder out of the box. It takes a Kafka specific VerifiableProperties object
|
||||
along with its
|
||||
constructor that wraps a regular Java.util.Properties object. The StringEncoder is great when writing a
|
||||
@@ -150,54 +167,54 @@ Inbound Channel Adapter:
|
||||
--------------------------------------------
|
||||
|
||||
The Inbound channel adapter is used to consume messages from Kafka. These messages will be placed into a channel as Spring Integration specific Messages.
|
||||
Kafka provides two types of consumer API's primarily. One is called the high level consumer and the other is the Simple Consumer. Highlevel consumer is
|
||||
Kafka provides two types of consumer API's primarily. One is called the High Level Consumer and the other is the Simple Consumer. High Level consumer is
|
||||
pretty complex inside. Nonetheless, for the client, using the high level API is straightforward. Although easy to use, High level consumer
|
||||
does not provide any offset management. So, if you want to rewind and re-fetch messages, it is not possible to do so using the
|
||||
high level consumer API. Offsets are managed by the Zookeeper internally in the high level consumer. If your use case does not require any offset management
|
||||
High Level Consumer API. Offsets are managed by the Zookeeper internally in the High Level Consumer. If your use case does not require any offset management
|
||||
or re-reading messages from the same consumer, then high level consumer is a perfect fit. Spring Integration Kafka inbound channel adapter
|
||||
currently supports only the high level consumer. Here are the details of configuring one.
|
||||
currently supports only the High Level Consumer. Here are the details of configuring one.
|
||||
|
||||
```xml
|
||||
<int-kafka:inbound-channel-adapter id="kafkaInboundChannelAdapter"
|
||||
kafka-consumer-context-ref="consumerContext"
|
||||
auto-startup="false"
|
||||
channel="inputFromKafka">
|
||||
<int:poller fixed-delay="10" time-unit="MILLISECONDS" max-messages-per-poll="5"/>
|
||||
</int-kafka:inbound-channel-adapter>
|
||||
kafka-consumer-context-ref="consumerContext"
|
||||
auto-startup="false"
|
||||
channel="inputFromKafka">
|
||||
<int:poller fixed-delay="10" time-unit="MILLISECONDS" max-messages-per-poll="5"/>
|
||||
</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
|
||||
between the poller configured with this inbound adapter and other pollers is that the receive-timeout specified here
|
||||
between the poller configured with this inbound adapter and other pollers used in Spring Integration is that the receive-timeout specified on this poller
|
||||
does not have any effect. The reason for this is because of the way Kafka implements iterators on the consumer stream.
|
||||
It is using a BlockingQueue internally and thus it would wait indefinitely. Instead of interrupting the underlying thread,
|
||||
we are leveraging on direct Kafka support for consumer time out. It is configured on the consumer context. Everything else
|
||||
is pretty much the same as in a regular inbound adapter. Any messages that it receives will be sent to the channel configured with it.
|
||||
we are leveraging a direct Kafka support for consumer time out. It is configured on the consumer context. Everything else
|
||||
is pretty much the same as in a regular inbound adapter. Any message that it receives will be sent to the channel configured with it.
|
||||
|
||||
Inbound Kafka Adapter must specify a kafka-consumer-context-ref element and here is how it is configured:
|
||||
|
||||
```xml
|
||||
<int-kafka:consumer-context id="consumerContext"
|
||||
consumer-timeout="4000"
|
||||
zookeeper-connect="zookeeperConnect">
|
||||
<int-kafka:consumer-configurations>
|
||||
<int-kafka:consumer-configuration group-id="default"
|
||||
value-decoder="valueDecoder"
|
||||
key-decoder="valueDecoder"
|
||||
max-messages="5000">
|
||||
<int-kafka:topic id="test1" streams="4"/>
|
||||
<int-kafka:topic id="test2" streams="4"/>
|
||||
</int-kafka:consumer-configuration>
|
||||
</int-kafka:consumer-configurations>
|
||||
</int-kafka:consumer-context>
|
||||
consumer-timeout="4000"
|
||||
zookeeper-connect="zookeeperConnect">
|
||||
<int-kafka:consumer-configurations>
|
||||
<int-kafka:consumer-configuration group-id="default"
|
||||
value-decoder="valueDecoder"
|
||||
key-decoder="valueDecoder"
|
||||
max-messages="5000">
|
||||
<int-kafka:topic id="test1" streams="4"/>
|
||||
<int-kafka:topic id="test2" streams="4"/>
|
||||
</int-kafka:consumer-configuration>
|
||||
</int-kafka:consumer-configurations>
|
||||
</int-kafka:consumer-context>
|
||||
```
|
||||
|
||||
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.
|
||||
|
||||
```xml
|
||||
<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" />
|
||||
```
|
||||
|
||||
zk-connect attribute is where you would specify the zookeeper connection. All the other attributes get translated into their
|
||||
@@ -207,19 +224,19 @@ In the above consumer context, you can also specify a consumer-timeout value whi
|
||||
timeout the consumer in case of no messages to consume.
|
||||
This timeout would be applicable to all the streams (threads) in the consumer.
|
||||
The default value for this in Kafka is -1 which would make it wait
|
||||
indefinitely. However, Sping Integration overrides it to be 5 seconds in order to make sure that no
|
||||
indefinitely. However, Sping Integration overrides it to be 5 seconds by default in order to make sure that no
|
||||
threads are blocking indefinitely in the lifecycle of the application and thereby
|
||||
giving them a chance to free up any resources or locks that they hold. It is recommended to
|
||||
override this value so as to meet any specific use case requirements.
|
||||
By providing a reasonable consumer-timeout and a fixed-delay value on the poller,
|
||||
By providing a reasonable consumer-timeout on the context and a fixed-delay value on the poller,
|
||||
this inbound adapter is capable of simulating a message driven behaviour.
|
||||
|
||||
consumer context takes consumer-configurations which are at the center piece of the inbound adapter. It is a group of one or more
|
||||
consumer context takes consumer-configurations which are at the core of the inbound adapter. It is a group of one or more
|
||||
consumer-configuration elements which consists of a consumer group dictated by the group-id. Each consumer-configuration
|
||||
can be configured with one or more kafka-topic.
|
||||
can be configured with one or more kafka-topics.
|
||||
|
||||
In the above example provided, we have a single consumer-configuration that consumes messages from two topics each having 4 streams.
|
||||
These streams are fundamentally same as the number of partitions that a topic is configured
|
||||
These streams are fundamentally equivalent to the number of partitions that a topic is configured
|
||||
with in the producer. For instance, if you configure your topic with
|
||||
4 partitions, then the maximum number of streams that you may have in the consumer is also 4.
|
||||
Any more than this would be a no-op.
|
||||
@@ -235,13 +252,24 @@ Consumer configuration can also be configured with optional decoders for key and
|
||||
The default ones provided by Kafka are basically no-ops and would consume as byte arrays.
|
||||
If you provide an encoder for key/value in the producer, then it is recommended to provide
|
||||
corresponding decoders.
|
||||
Spring Integration Kafka adapter gives Apache Avro based data serialization components
|
||||
out of the box. You can use any serialization component for this purpose.
|
||||
Here is how you would configure a kafka decoder bean that is Avro backed.
|
||||
As disussed already in the outbound adapter, Spring Integration Kafka adapter gives Apache Avro based data serialization components
|
||||
out of the box. You can use any serialization component for this purpose as long as you implement the required encoder/decoder interfaces from Kafka.
|
||||
As with the Avro encoder support, decoders provided also
|
||||
implement Reflection and Specific datum based de-serialization. Here is how you would configure kafka decoder beans that is Avro backed.
|
||||
|
||||
Using Avro Specific support:
|
||||
|
||||
```xml
|
||||
<bean id="kafkaDecoder" class="org.springframework.integration.kafka.serializer.avro.AvroBackedKafkaDecoder">
|
||||
<constructor-arg type="java.lang.Class" value="java.lang.String" />
|
||||
<bean id="kafkaDecoder" class="org.springframework.integration.kafka.serializer.avro.AvroSpecificDatumBackedKafkaDecoder">
|
||||
<constructor-arg value="com.domain.AvroGeneratedSpecificRecord" />
|
||||
</bean>
|
||||
```
|
||||
|
||||
Using Reflection support:
|
||||
|
||||
```xml
|
||||
<bean id="kafkaDecoder" class="org.springframework.integration.kafka.serializer.avro.AvroReflectDatumBackedKafkaDecoder">
|
||||
<constructor-arg value="java.lang.String" />
|
||||
</bean>
|
||||
```
|
||||
|
||||
@@ -249,13 +277,13 @@ Another important attribute for the consumer-configuration is the max-messages.
|
||||
Please note that this is different from the max-messages-per-poll configured on the inbound adapter
|
||||
element.
|
||||
There it means the number of times the receive method called on the adapter.
|
||||
The max-messages on consumer configuration is different. Kafka is used mainly for big data purposes
|
||||
and usually that means the influx of large amount of data constantly. Because of this,
|
||||
The max-messages on consumer configuration is different. When you use Kafka for ingesting messages,
|
||||
it usually means an influx of large amount of data constantly. Because of this,
|
||||
each time a receive is invoked on the adapter, you would basically get a collection of messages.
|
||||
The maximum number of messages to retrieve for a topic in each execution of the
|
||||
receive is what configured through the max-messages attribute on the consumer-configuration.
|
||||
Basically, if the use case is to receive a constant stream of
|
||||
large number of data, simply specifying a receive-timeout alone would not be enough.
|
||||
large number of data, simply specifying a consumer-timeout alone would not be enough.
|
||||
You would also need to specify the max number of messages to receive.
|
||||
|
||||
The type of the payload of the Message returned by the adapter is the following:
|
||||
|
||||
@@ -15,6 +15,7 @@
|
||||
*/
|
||||
package org.springframework.integration.kafka.config.xml;
|
||||
|
||||
import kafka.serializer.DefaultDecoder;
|
||||
import org.springframework.beans.factory.config.BeanDefinition;
|
||||
import org.springframework.beans.factory.config.BeanDefinitionHolder;
|
||||
import org.springframework.beans.factory.support.AbstractBeanDefinition;
|
||||
@@ -32,7 +33,9 @@ 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;
|
||||
|
||||
/**
|
||||
@@ -54,6 +57,7 @@ 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")) {
|
||||
|
||||
@@ -30,11 +30,15 @@ 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
|
||||
*/
|
||||
public class KafkaProducerContextParser extends AbstractSimpleBeanDefinitionParser {
|
||||
|
||||
@Override
|
||||
protected Class<?> getBeanClass(final Element element) {
|
||||
return KafkaProducerContext.class;
|
||||
@@ -48,6 +52,7 @@ 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);
|
||||
|
||||
@@ -28,11 +28,11 @@ import java.util.Map;
|
||||
* @since 0.5
|
||||
*
|
||||
*/
|
||||
public class KafkaHighLevelConsumerMessageSource extends IntegrationObjectSupport implements MessageSource<Map<String, Map<Integer, List<Object>>>> {
|
||||
public class KafkaHighLevelConsumerMessageSource<K,V> extends IntegrationObjectSupport implements MessageSource<Map<String, Map<Integer, List<Object>>>> {
|
||||
|
||||
private final KafkaConsumerContext kafkaConsumerContext;
|
||||
private final KafkaConsumerContext<K,V> kafkaConsumerContext;
|
||||
|
||||
public KafkaHighLevelConsumerMessageSource(final KafkaConsumerContext kafkaConsumerContext) {
|
||||
public KafkaHighLevelConsumerMessageSource(final KafkaConsumerContext<K,V> kafkaConsumerContext) {
|
||||
this.kafkaConsumerContext = kafkaConsumerContext;
|
||||
}
|
||||
|
||||
|
||||
@@ -23,15 +23,15 @@ import org.springframework.integration.kafka.support.KafkaProducerContext;
|
||||
* @author Soby Chacko
|
||||
* @since 0.5
|
||||
*/
|
||||
public class KafkaProducerMessageHandler extends AbstractMessageHandler {
|
||||
public class KafkaProducerMessageHandler<K,V> extends AbstractMessageHandler {
|
||||
|
||||
private final KafkaProducerContext kafkaProducerContext;
|
||||
private final KafkaProducerContext<K,V> kafkaProducerContext;
|
||||
|
||||
public KafkaProducerMessageHandler(final KafkaProducerContext kafkaProducerContext) {
|
||||
public KafkaProducerMessageHandler(final KafkaProducerContext<K,V> kafkaProducerContext) {
|
||||
this.kafkaProducerContext = kafkaProducerContext;
|
||||
}
|
||||
|
||||
public KafkaProducerContext getKafkaProducerContext() {
|
||||
public KafkaProducerContext<K,V> getKafkaProducerContext() {
|
||||
return kafkaProducerContext;
|
||||
}
|
||||
|
||||
|
||||
@@ -0,0 +1,43 @@
|
||||
package org.springframework.integration.kafka.serializer.avro;
|
||||
|
||||
import org.apache.avro.io.DatumReader;
|
||||
import org.apache.avro.io.DatumWriter;
|
||||
import org.apache.commons.logging.Log;
|
||||
import org.apache.commons.logging.LogFactory;
|
||||
|
||||
import java.io.IOException;
|
||||
|
||||
/**
|
||||
* @author Soby Chacko
|
||||
* @since 0.5
|
||||
*/
|
||||
public abstract class AvroDatumSupport<T> {
|
||||
|
||||
private static final Log LOG = LogFactory.getLog(AvroDatumSupport.class);
|
||||
|
||||
private final AvroSerializer<T> avroSerializer;
|
||||
|
||||
protected AvroDatumSupport() {
|
||||
this.avroSerializer = new AvroSerializer<T>();
|
||||
}
|
||||
|
||||
@SuppressWarnings("unchecked")
|
||||
public byte[] toBytes(final T source, final DatumWriter<T> writer) {
|
||||
try {
|
||||
return avroSerializer.serialize(source, writer);
|
||||
} catch (IOException e) {
|
||||
LOG.error("Failed to encode source: " + e);
|
||||
}
|
||||
return null;
|
||||
}
|
||||
|
||||
@SuppressWarnings("unchecked")
|
||||
public T fromBytes(final byte[] bytes, final DatumReader<T> reader) {
|
||||
try {
|
||||
return avroSerializer.deserialize(bytes, reader);
|
||||
} catch (IOException e) {
|
||||
LOG.error("Failed to decode byte array: " + e);
|
||||
}
|
||||
return null;
|
||||
}
|
||||
}
|
||||
@@ -15,12 +15,9 @@
|
||||
*/
|
||||
package org.springframework.integration.kafka.serializer.avro;
|
||||
|
||||
|
||||
import kafka.serializer.Decoder;
|
||||
import org.apache.avro.Schema;
|
||||
import org.apache.avro.reflect.ReflectData;
|
||||
|
||||
import java.io.IOException;
|
||||
import org.apache.avro.io.DatumReader;
|
||||
import org.apache.avro.reflect.ReflectDatumReader;
|
||||
import org.apache.commons.logging.Log;
|
||||
import org.apache.commons.logging.LogFactory;
|
||||
|
||||
@@ -28,28 +25,19 @@ import org.apache.commons.logging.LogFactory;
|
||||
* @author Soby Chacko
|
||||
* @since 0.5
|
||||
*/
|
||||
public class AvroBackedKafkaDecoder<T> implements Decoder<T> {
|
||||
private static final Log LOG = LogFactory.getLog(AvroBackedKafkaDecoder.class);
|
||||
public class AvroReflectDatumBackedKafkaDecoder<T> extends AvroDatumSupport<T> implements Decoder<T> {
|
||||
private static final Log LOG = LogFactory.getLog(AvroReflectDatumBackedKafkaDecoder.class);
|
||||
|
||||
private final Class clazz;
|
||||
private final DatumReader<T> reader;
|
||||
|
||||
public AvroBackedKafkaDecoder(final Class clazz) {
|
||||
this.clazz = clazz;
|
||||
public AvroReflectDatumBackedKafkaDecoder(final Class<T> clazz) {
|
||||
this.reader = new ReflectDatumReader<T>(clazz);
|
||||
}
|
||||
|
||||
@Override
|
||||
@SuppressWarnings("unchecked")
|
||||
public T fromBytes(final byte[] bytes) {
|
||||
final Schema schema = ReflectData.get().getSchema(clazz);
|
||||
final AvroSerializer avroSerializer = new AvroSerializer();
|
||||
|
||||
try {
|
||||
return (T) avroSerializer.deserialize(bytes, schema);
|
||||
} catch (IOException e) {
|
||||
LOG.error("Failed to decode byte array for schema: " + schema.getFullName(), e);
|
||||
}
|
||||
|
||||
return null;
|
||||
return fromBytes(bytes, reader);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -16,10 +16,8 @@
|
||||
package org.springframework.integration.kafka.serializer.avro;
|
||||
|
||||
import kafka.serializer.Encoder;
|
||||
import org.apache.avro.Schema;
|
||||
import org.apache.avro.reflect.ReflectData;
|
||||
|
||||
import java.io.IOException;
|
||||
import org.apache.avro.io.DatumWriter;
|
||||
import org.apache.avro.reflect.ReflectDatumWriter;
|
||||
import org.apache.commons.logging.Log;
|
||||
import org.apache.commons.logging.LogFactory;
|
||||
|
||||
@@ -27,27 +25,18 @@ import org.apache.commons.logging.LogFactory;
|
||||
* @author Soby Chacko
|
||||
* @since 0.5
|
||||
*/
|
||||
public class AvroBackedKafkaEncoder<T> implements Encoder<T> {
|
||||
private static final Log LOG = LogFactory.getLog(AvroBackedKafkaEncoder.class);
|
||||
public class AvroReflectDatumBackedKafkaEncoder<T> extends AvroDatumSupport<T> implements Encoder<T> {
|
||||
private static final Log LOG = LogFactory.getLog(AvroReflectDatumBackedKafkaEncoder.class);
|
||||
|
||||
private final Class clazz;
|
||||
private final DatumWriter<T> writer;
|
||||
|
||||
public AvroBackedKafkaEncoder(final Class clazz) {
|
||||
this.clazz = clazz;
|
||||
public AvroReflectDatumBackedKafkaEncoder(final Class<T> clazz) {
|
||||
this.writer = new ReflectDatumWriter<T>(clazz);
|
||||
}
|
||||
|
||||
@Override
|
||||
@SuppressWarnings("unchecked")
|
||||
public byte[] toBytes(final T source) {
|
||||
final Schema schema = ReflectData.get().getSchema(clazz);
|
||||
final AvroSerializer avroSerializer = new AvroSerializer();
|
||||
|
||||
try {
|
||||
return avroSerializer.serialize(source, schema);
|
||||
} catch (IOException e) {
|
||||
LOG.error("Failed to encode source for schema: " + schema.getFullName());
|
||||
}
|
||||
|
||||
return null;
|
||||
return toBytes(source, writer);
|
||||
}
|
||||
}
|
||||
@@ -15,15 +15,12 @@
|
||||
*/
|
||||
package org.springframework.integration.kafka.serializer.avro;
|
||||
|
||||
import org.apache.avro.Schema;
|
||||
import org.apache.avro.io.DatumReader;
|
||||
import org.apache.avro.io.DatumWriter;
|
||||
import org.apache.avro.io.Decoder;
|
||||
import org.apache.avro.io.DecoderFactory;
|
||||
import org.apache.avro.io.Encoder;
|
||||
import org.apache.avro.io.EncoderFactory;
|
||||
import org.apache.avro.reflect.ReflectDatumReader;
|
||||
import org.apache.avro.reflect.ReflectDatumWriter;
|
||||
|
||||
import java.io.ByteArrayOutputStream;
|
||||
import java.io.IOException;
|
||||
@@ -33,15 +30,13 @@ import java.io.IOException;
|
||||
* @since 0.5
|
||||
*/
|
||||
public class AvroSerializer<T> {
|
||||
public T deserialize(final byte[] bytes, final Schema schema) throws IOException {
|
||||
final Decoder decoder = DecoderFactory.get().binaryDecoder(bytes, null);
|
||||
final DatumReader<T> reader = new ReflectDatumReader<T>(schema);
|
||||
|
||||
public T deserialize(final byte[] bytes, final DatumReader<T> reader) throws IOException {
|
||||
final Decoder decoder = DecoderFactory.get().binaryDecoder(bytes, null);
|
||||
return reader.read(null, decoder);
|
||||
}
|
||||
|
||||
public byte[] serialize(final T input, final Schema schema) throws IOException {
|
||||
final DatumWriter<T> writer = new ReflectDatumWriter<T>(schema);
|
||||
public byte[] serialize(final T input, final DatumWriter<T> writer) throws IOException {
|
||||
final ByteArrayOutputStream stream = new ByteArrayOutputStream();
|
||||
|
||||
final Encoder encoder = EncoderFactory.get().binaryEncoder(stream, null);
|
||||
|
||||
@@ -0,0 +1,28 @@
|
||||
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
|
||||
* @since 0.5
|
||||
*/
|
||||
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) {
|
||||
this.reader = new SpecificDatumReader<T>(specificRecordBase);
|
||||
}
|
||||
|
||||
@Override
|
||||
@SuppressWarnings("unchecked")
|
||||
public T fromBytes(final byte[] bytes) {
|
||||
return fromBytes(bytes, reader);
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,28 @@
|
||||
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
|
||||
* @since 0.5
|
||||
*/
|
||||
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) {
|
||||
this.writer = new SpecificDatumWriter<T>(specificRecordClazz);
|
||||
}
|
||||
|
||||
@Override
|
||||
@SuppressWarnings("unchecked")
|
||||
public byte[] toBytes(final T source) {
|
||||
return toBytes(source, writer);
|
||||
}
|
||||
}
|
||||
@@ -1,53 +0,0 @@
|
||||
/*
|
||||
* 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.serializer.avro;
|
||||
|
||||
import org.apache.avro.Schema;
|
||||
import org.apache.avro.io.DatumReader;
|
||||
import org.apache.avro.io.DatumWriter;
|
||||
import org.apache.avro.io.Decoder;
|
||||
import org.apache.avro.io.DecoderFactory;
|
||||
import org.apache.avro.io.Encoder;
|
||||
import org.apache.avro.io.EncoderFactory;
|
||||
import org.apache.avro.specific.SpecificDatumReader;
|
||||
import org.apache.avro.specific.SpecificDatumWriter;
|
||||
|
||||
import java.io.ByteArrayOutputStream;
|
||||
import java.io.IOException;
|
||||
|
||||
/**
|
||||
* @author Soby Chacko
|
||||
* @since 0.5
|
||||
*/
|
||||
public class AvroSpecificDatumSerializer<T> {
|
||||
public T deserialize(final byte[] bytes, final Schema schema) throws IOException {
|
||||
final Decoder decoder = DecoderFactory.get().binaryDecoder(bytes, null);
|
||||
final DatumReader<T> reader = new SpecificDatumReader<T>(schema);
|
||||
|
||||
return reader.read(null, decoder);
|
||||
}
|
||||
|
||||
public byte[] serialize(final T input, final Schema schema) throws IOException {
|
||||
final DatumWriter<T> writer = new SpecificDatumWriter<T>(schema);
|
||||
final ByteArrayOutputStream stream = new ByteArrayOutputStream();
|
||||
|
||||
final Encoder encoder = EncoderFactory.get().binaryEncoder(stream, null);
|
||||
writer.write(input, encoder);
|
||||
encoder.flush();
|
||||
|
||||
return stream.toByteArray();
|
||||
}
|
||||
}
|
||||
@@ -24,7 +24,7 @@ import java.util.Properties;
|
||||
* @author Soby Chacko
|
||||
* @since 0.5
|
||||
*/
|
||||
public class StringEncoder implements Encoder {
|
||||
public class StringEncoder<T> implements Encoder<T> {
|
||||
private String encoding = "UTF8";
|
||||
|
||||
public void setEncoding(final String encoding){
|
||||
|
||||
@@ -24,12 +24,12 @@ import java.util.Properties;
|
||||
* @author Soby Chacko
|
||||
* @since 0.5
|
||||
*/
|
||||
public class ConsumerConfigFactoryBean implements FactoryBean<ConsumerConfig> {
|
||||
public class ConsumerConfigFactoryBean<K,V> implements FactoryBean<ConsumerConfig> {
|
||||
|
||||
private final ConsumerMetadata consumerMetadata;
|
||||
private final ConsumerMetadata<K,V> consumerMetadata;
|
||||
private final ZookeeperConnect zookeeperConnect;
|
||||
|
||||
public ConsumerConfigFactoryBean(final ConsumerMetadata consumerMetadata,
|
||||
public ConsumerConfigFactoryBean(final ConsumerMetadata<K,V> consumerMetadata,
|
||||
final ZookeeperConnect zookeeperConnect){
|
||||
this.consumerMetadata = consumerMetadata;
|
||||
this.zookeeperConnect = zookeeperConnect;
|
||||
|
||||
@@ -37,46 +37,46 @@ import java.util.concurrent.Future;
|
||||
* @author Soby Chacko
|
||||
* @since 0.5
|
||||
*/
|
||||
public class ConsumerConfiguration {
|
||||
public class ConsumerConfiguration<K,V> {
|
||||
private static final Log LOGGER = LogFactory.getLog(ConsumerConfiguration.class);
|
||||
|
||||
private final ConsumerMetadata consumerMetadata;
|
||||
private final ConsumerMetadata<K,V> consumerMetadata;
|
||||
private final ConsumerConnectionProvider consumerConnectionProvider;
|
||||
private final MessageLeftOverTracker messageLeftOverTracker;
|
||||
private final MessageLeftOverTracker<K,V> messageLeftOverTracker;
|
||||
private ConsumerConnector consumerConnector;
|
||||
private volatile int count = 0;
|
||||
private int maxMessages = 1;
|
||||
|
||||
private ExecutorService executorService = Executors.newCachedThreadPool();
|
||||
|
||||
public ConsumerConfiguration(final ConsumerMetadata consumerMetadata,
|
||||
public ConsumerConfiguration(final ConsumerMetadata<K,V> consumerMetadata,
|
||||
final ConsumerConnectionProvider consumerConnectionProvider,
|
||||
final MessageLeftOverTracker messageLeftOverTracker) {
|
||||
final MessageLeftOverTracker<K,V> messageLeftOverTracker) {
|
||||
this.consumerMetadata = consumerMetadata;
|
||||
this.consumerConnectionProvider = consumerConnectionProvider;
|
||||
this.messageLeftOverTracker = messageLeftOverTracker;
|
||||
}
|
||||
|
||||
public ConsumerMetadata getConsumerMetadata() {
|
||||
public ConsumerMetadata<K,V> getConsumerMetadata() {
|
||||
return consumerMetadata;
|
||||
}
|
||||
|
||||
public Map<String, Map<Integer, List<Object>>> receive() {
|
||||
count = messageLeftOverTracker.getCurrentCount();
|
||||
|
||||
final List<Callable<List<MessageAndMetadata>>> tasks = new LinkedList<Callable<List<MessageAndMetadata>>>();
|
||||
final Object lock = new Object();
|
||||
|
||||
final Map<String, List<KafkaStream<byte[], byte[]>>> consumerMap = getConsumerMapWithMessageStreams();
|
||||
for (final List<KafkaStream<byte[], byte[]>> streams : consumerMap.values()) {
|
||||
for (final KafkaStream<byte[], byte[]> stream : streams) {
|
||||
tasks.add(new Callable<List<MessageAndMetadata>>() {
|
||||
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>>>() {
|
||||
@Override
|
||||
public List<MessageAndMetadata> call() throws Exception {
|
||||
final List<MessageAndMetadata> rawMessages = new ArrayList<MessageAndMetadata>();
|
||||
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 messageAndMetadata = stream.iterator().next();
|
||||
final MessageAndMetadata<K,V> messageAndMetadata = stream.iterator().next();
|
||||
synchronized (lock) {
|
||||
if (count < maxMessages) {
|
||||
rawMessages.add(messageAndMetadata);
|
||||
@@ -94,17 +94,16 @@ public class ConsumerConfiguration {
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
return executeTasks(tasks);
|
||||
}
|
||||
|
||||
private Map<String, Map<Integer, List<Object>>> executeTasks(final List<Callable<List<MessageAndMetadata>>> 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>> 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)) {
|
||||
@@ -127,20 +126,21 @@ public class ConsumerConfiguration {
|
||||
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 mamd : messageLeftOverTracker.getMessageLeftOverFromPreviousPoll()) {
|
||||
for (final MessageAndMetadata<K,V> mamd : messageLeftOverTracker.getMessageLeftOverFromPreviousPoll()) {
|
||||
final String topic = mamd.topic();
|
||||
|
||||
if (!messages.containsKey(topic)) {
|
||||
final List<MessageAndMetadata> l = new ArrayList<MessageAndMetadata>();
|
||||
final List<MessageAndMetadata<K,V>> l = new ArrayList<MessageAndMetadata<K,V>>();
|
||||
l.add(mamd);
|
||||
messages.put(topic, getPayload(l));
|
||||
} else {
|
||||
final Map<Integer, List<Object>> existingPayloadMap = messages.get(topic);
|
||||
final List<MessageAndMetadata> l = new ArrayList<MessageAndMetadata>();
|
||||
final List<MessageAndMetadata<K,V>> l = new ArrayList<MessageAndMetadata<K,V>>();
|
||||
l.add(mamd);
|
||||
getPayload(l, existingPayloadMap);
|
||||
}
|
||||
@@ -149,10 +149,10 @@ public class ConsumerConfiguration {
|
||||
return messages;
|
||||
}
|
||||
|
||||
private Map<Integer, List<Object>> getPayload(final List<MessageAndMetadata> 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 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());
|
||||
@@ -167,8 +167,8 @@ public class ConsumerConfiguration {
|
||||
return payloadMap;
|
||||
}
|
||||
|
||||
private void getPayload(final List<MessageAndMetadata> messageAndMetadatas, final Map<Integer, List<Object>> existingPayloadMap) {
|
||||
for (final MessageAndMetadata 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());
|
||||
@@ -181,16 +181,11 @@ public class ConsumerConfiguration {
|
||||
}
|
||||
|
||||
@SuppressWarnings("unchecked")
|
||||
public Map<String, List<KafkaStream<byte[], byte[]>>> getConsumerMapWithMessageStreams() {
|
||||
if (consumerMetadata.getValueDecoder() != null &&
|
||||
consumerMetadata.getKeyDecoder() != null) {
|
||||
return getConsumerConnector().createMessageStreams(
|
||||
consumerMetadata.getTopicStreamMap(),
|
||||
consumerMetadata.getKeyDecoder(),
|
||||
consumerMetadata.getValueDecoder());
|
||||
}
|
||||
|
||||
return getConsumerConnector().createMessageStreams(consumerMetadata.getTopicStreamMap());
|
||||
public Map<String, List<KafkaStream<K,V>>> getConsumerMapWithMessageStreams() {
|
||||
return getConsumerConnector().createMessageStreams(
|
||||
consumerMetadata.getTopicStreamMap(),
|
||||
consumerMetadata.getKeyDecoder(),
|
||||
consumerMetadata.getValueDecoder());
|
||||
}
|
||||
|
||||
public int getMaxMessages() {
|
||||
@@ -205,7 +200,6 @@ public class ConsumerConfiguration {
|
||||
if (consumerConnector == null) {
|
||||
consumerConnector = consumerConnectionProvider.getConsumerConnector();
|
||||
}
|
||||
|
||||
return consumerConnector;
|
||||
}
|
||||
}
|
||||
|
||||
@@ -16,6 +16,8 @@
|
||||
package org.springframework.integration.kafka.support;
|
||||
|
||||
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;
|
||||
@@ -24,7 +26,7 @@ import java.util.Map;
|
||||
* @author Soby Chacko
|
||||
* @since 0.5
|
||||
*/
|
||||
public class ConsumerMetadata<K,V> {
|
||||
public class ConsumerMetadata<K,V> implements InitializingBean {
|
||||
|
||||
//High level consumer defaults
|
||||
private String groupId = KafkaConsumerDefaults.GROUP_ID;
|
||||
@@ -172,4 +174,16 @@ public class ConsumerMetadata<K,V> {
|
||||
public void setTopicStreamMap(final Map<String, Integer> topicStreamMap) {
|
||||
this.topicStreamMap = topicStreamMap;
|
||||
}
|
||||
|
||||
@Override
|
||||
@SuppressWarnings("unchecked")
|
||||
public void afterPropertiesSet() throws Exception {
|
||||
if (valueDecoder == null) {
|
||||
setValueDecoder((Decoder<V>) new DefaultDecoder(null));
|
||||
}
|
||||
|
||||
if (keyDecoder == null) {
|
||||
setKeyDecoder((Decoder<K>) getValueDecoder());
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -32,24 +32,26 @@ import java.util.Map;
|
||||
* @author Soby Chacko
|
||||
* @since 0.5
|
||||
*/
|
||||
public class KafkaConsumerContext implements BeanFactoryAware {
|
||||
private Map<String, ConsumerConfiguration> consumerConfigurations;
|
||||
public class KafkaConsumerContext<K,V> implements BeanFactoryAware {
|
||||
private Map<String, ConsumerConfiguration<K,V>> consumerConfigurations;
|
||||
private String consumerTimeout = KafkaConsumerDefaults.CONSUMER_TIMEOUT;
|
||||
private ZookeeperConnect zookeeperConnect;
|
||||
|
||||
public Collection<ConsumerConfiguration> getConsumerConfigurations() {
|
||||
public Collection<ConsumerConfiguration<K,V>> getConsumerConfigurations() {
|
||||
return consumerConfigurations.values();
|
||||
}
|
||||
|
||||
@Override
|
||||
@SuppressWarnings("unchecked")
|
||||
public void setBeanFactory(final BeanFactory beanFactory) throws BeansException {
|
||||
consumerConfigurations = ((ListableBeanFactory) beanFactory).getBeansOfType(ConsumerConfiguration.class);
|
||||
consumerConfigurations = (Map<String, ConsumerConfiguration<K,V>>)
|
||||
(Object) ((ListableBeanFactory) beanFactory).getBeansOfType(ConsumerConfiguration.class);
|
||||
}
|
||||
|
||||
public Message<Map<String, Map<Integer, List<Object>>>> receive() {
|
||||
final Map<String, Map<Integer, List<Object>>> consumedData = new HashMap<String, Map<Integer, List<Object>>>();
|
||||
|
||||
for (final ConsumerConfiguration consumerConfiguration : getConsumerConfigurations()) {
|
||||
for (final ConsumerConfiguration<K,V> consumerConfiguration : getConsumerConfigurations()) {
|
||||
final Map<String, Map<Integer, List<Object>>> messages = consumerConfiguration.receive();
|
||||
|
||||
if (messages != null){
|
||||
|
||||
@@ -28,12 +28,12 @@ import java.util.Map;
|
||||
* @author Soby Chacko
|
||||
* @since 0.5
|
||||
*/
|
||||
public class KafkaProducerContext implements BeanFactoryAware {
|
||||
private Map<String, ProducerConfiguration> topicsConfiguration;
|
||||
public class KafkaProducerContext<K,V> implements BeanFactoryAware {
|
||||
private Map<String, ProducerConfiguration<K,V>> topicsConfiguration;
|
||||
|
||||
@SuppressWarnings("unchecked")
|
||||
public void send(final Message<?> message) throws Exception {
|
||||
final ProducerConfiguration producerConfiguration =
|
||||
final ProducerConfiguration<K,V> producerConfiguration =
|
||||
getTopicConfiguration(message.getHeaders().get("topic", String.class));
|
||||
|
||||
if (producerConfiguration != null) {
|
||||
@@ -41,10 +41,10 @@ public class KafkaProducerContext implements BeanFactoryAware {
|
||||
}
|
||||
}
|
||||
|
||||
private ProducerConfiguration getTopicConfiguration(final String topic){
|
||||
final Collection<ProducerConfiguration> topics = topicsConfiguration.values();
|
||||
private ProducerConfiguration<K,V> getTopicConfiguration(final String topic){
|
||||
final Collection<ProducerConfiguration<K,V>> topics = topicsConfiguration.values();
|
||||
|
||||
for (final ProducerConfiguration producerConfiguration : topics){
|
||||
for (final ProducerConfiguration<K,V> producerConfiguration : topics){
|
||||
if (producerConfiguration.getProducerMetadata().getTopic().equals(topic)){
|
||||
return producerConfiguration;
|
||||
}
|
||||
@@ -53,12 +53,15 @@ public class KafkaProducerContext implements BeanFactoryAware {
|
||||
return null;
|
||||
}
|
||||
|
||||
public Map<String, ProducerConfiguration> getTopicsConfiguration() {
|
||||
public Map<String, ProducerConfiguration<K,V>> getTopicsConfiguration() {
|
||||
return topicsConfiguration;
|
||||
}
|
||||
|
||||
@Override
|
||||
@SuppressWarnings("unchecked")
|
||||
public void setBeanFactory(final BeanFactory beanFactory) throws BeansException {
|
||||
topicsConfiguration = ((ListableBeanFactory)beanFactory).getBeansOfType(ProducerConfiguration.class);
|
||||
topicsConfiguration =
|
||||
(Map<String, ProducerConfiguration<K,V>>) (Object)
|
||||
((ListableBeanFactory)beanFactory).getBeansOfType(ProducerConfiguration.class);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -24,14 +24,14 @@ import java.util.List;
|
||||
* @author Soby Chacko
|
||||
* @since 0.5
|
||||
*/
|
||||
public class MessageLeftOverTracker {
|
||||
private final List<MessageAndMetadata> messageLeftOverFromPreviousPoll = new ArrayList<MessageAndMetadata>();
|
||||
public class MessageLeftOverTracker<K,V> {
|
||||
private final List<MessageAndMetadata<K,V>> messageLeftOverFromPreviousPoll = new ArrayList<MessageAndMetadata<K,V>>();
|
||||
|
||||
public void addMessageAndMetadata(final MessageAndMetadata messageAndMetadata){
|
||||
public void addMessageAndMetadata(final MessageAndMetadata<K,V> messageAndMetadata){
|
||||
messageLeftOverFromPreviousPoll.add(messageAndMetadata);
|
||||
}
|
||||
|
||||
public List<MessageAndMetadata> getMessageLeftOverFromPreviousPoll(){
|
||||
public List<MessageAndMetadata<K,V>> getMessageLeftOverFromPreviousPoll(){
|
||||
return messageLeftOverFromPreviousPoll;
|
||||
}
|
||||
|
||||
|
||||
@@ -10,7 +10,7 @@
|
||||
zk-sync-time="2000"/>
|
||||
|
||||
|
||||
<bean id="valueDecoder" class="org.springframework.integration.kafka.serializer.avro.AvroBackedKafkaDecoder">
|
||||
<bean id="valueDecoder" class="org.springframework.integration.kafka.serializer.avro.AvroReflectDatumBackedKafkaDecoder">
|
||||
<constructor-arg type="java.lang.Class" value="java.lang.String"/>
|
||||
</bean>
|
||||
|
||||
|
||||
@@ -31,7 +31,7 @@ import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
|
||||
*/
|
||||
@RunWith(SpringJUnit4ClassRunner.class)
|
||||
@ContextConfiguration
|
||||
public class KafkaConsumerContextParserTests {
|
||||
public class KafkaConsumerContextParserTests<K,V> {
|
||||
|
||||
@Autowired
|
||||
private ApplicationContext appContext;
|
||||
@@ -39,10 +39,10 @@ public class KafkaConsumerContextParserTests {
|
||||
@Test
|
||||
@SuppressWarnings("unchecked")
|
||||
public void testConsumerContextConfiguration() {
|
||||
final KafkaConsumerContext consumerContext = appContext.getBean("consumerContext", KafkaConsumerContext.class);
|
||||
final KafkaConsumerContext<K,V> consumerContext = appContext.getBean("consumerContext", KafkaConsumerContext.class);
|
||||
Assert.assertNotNull(consumerContext);
|
||||
|
||||
final ConsumerMetadata cm = appContext.getBean("consumerMetadata_default1", ConsumerMetadata.class);
|
||||
final ConsumerMetadata<K,V> cm = appContext.getBean("consumerMetadata_default1", ConsumerMetadata.class);
|
||||
Assert.assertNotNull(cm);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -20,7 +20,7 @@
|
||||
<int:poller fixed-delay="10" time-unit="MILLISECONDS" max-messages-per-poll="5"/>
|
||||
</int-kafka:inbound-channel-adapter>
|
||||
|
||||
<bean id="valueDecoder" class="org.springframework.integration.kafka.serializer.avro.AvroBackedKafkaDecoder">
|
||||
<bean id="valueDecoder" class="org.springframework.integration.kafka.serializer.avro.AvroReflectDatumBackedKafkaDecoder">
|
||||
<constructor-arg type="java.lang.Class" value="java.lang.String" />
|
||||
</bean>
|
||||
|
||||
|
||||
@@ -23,7 +23,7 @@
|
||||
|
||||
<task:executor id="taskExecutor" pool-size="5" keep-alive="120" queue-capacity="500"/>
|
||||
|
||||
<bean id="kafkaEncoder" class="org.springframework.integration.kafka.serializer.avro.AvroBackedKafkaEncoder">
|
||||
<bean id="kafkaEncoder" class="org.springframework.integration.kafka.serializer.avro.AvroReflectDatumBackedKafkaEncoder">
|
||||
<constructor-arg value="java.lang.String" />
|
||||
</bean>
|
||||
|
||||
|
||||
@@ -32,18 +32,19 @@ import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
|
||||
*/
|
||||
@RunWith(SpringJUnit4ClassRunner.class)
|
||||
@ContextConfiguration
|
||||
public class KafkaOutboundAdapterParserTests {
|
||||
public class KafkaOutboundAdapterParserTests<K,V> {
|
||||
|
||||
@Autowired
|
||||
private ApplicationContext appContext;
|
||||
|
||||
@Test
|
||||
@SuppressWarnings("unchecked")
|
||||
public void testOutboundAdapterConfiguration(){
|
||||
final PollingConsumer pollingConsumer = appContext.getBean("kafkaOutboundChannelAdapter", PollingConsumer.class);
|
||||
final KafkaProducerMessageHandler messageHandler = appContext.getBean(KafkaProducerMessageHandler.class);
|
||||
final KafkaProducerMessageHandler<K,V> messageHandler = appContext.getBean(KafkaProducerMessageHandler.class);
|
||||
Assert.assertNotNull(pollingConsumer);
|
||||
Assert.assertNotNull(messageHandler);
|
||||
final KafkaProducerContext producerContext = messageHandler.getKafkaProducerContext();
|
||||
final KafkaProducerContext<K,V> producerContext = messageHandler.getKafkaProducerContext();
|
||||
Assert.assertNotNull(producerContext);
|
||||
Assert.assertEquals(producerContext.getTopicsConfiguration().size(), 2);
|
||||
}
|
||||
|
||||
@@ -20,7 +20,7 @@
|
||||
</int-kafka:producer-configurations>
|
||||
</int-kafka:producer-context>
|
||||
|
||||
<bean id="valueEncoder" class="org.springframework.integration.kafka.serializer.avro.AvroBackedKafkaEncoder">
|
||||
<bean id="valueEncoder" class="org.springframework.integration.kafka.serializer.avro.AvroReflectDatumBackedKafkaEncoder">
|
||||
<constructor-arg value="java.lang.String" />
|
||||
</bean>
|
||||
</beans>
|
||||
|
||||
@@ -36,7 +36,7 @@ import java.util.Map;
|
||||
*/
|
||||
@RunWith(SpringJUnit4ClassRunner.class)
|
||||
@ContextConfiguration
|
||||
public class KafkaProducerContextParserTests {
|
||||
public class KafkaProducerContextParserTests<K,V,T> {
|
||||
|
||||
@Autowired
|
||||
private ApplicationContext appContext;
|
||||
@@ -44,34 +44,34 @@ public class KafkaProducerContextParserTests {
|
||||
@Test
|
||||
@SuppressWarnings("unchecked")
|
||||
public void testProducerContextConfiguration(){
|
||||
final KafkaProducerContext producerContext = appContext.getBean("producerContext", KafkaProducerContext.class);
|
||||
final KafkaProducerContext<K,V> producerContext = appContext.getBean("producerContext", KafkaProducerContext.class);
|
||||
Assert.assertNotNull(producerContext);
|
||||
|
||||
final Map<String, ProducerConfiguration> topicConfigurations = producerContext.getTopicsConfiguration();
|
||||
final Map<String, ProducerConfiguration<K,V>> topicConfigurations = producerContext.getTopicsConfiguration();
|
||||
Assert.assertEquals(topicConfigurations.size(), 2);
|
||||
|
||||
final ProducerConfiguration producerConfigurationTest1 = topicConfigurations.get("producerConfiguration_test1");
|
||||
final ProducerConfiguration<K,V> producerConfigurationTest1 = topicConfigurations.get("producerConfiguration_test1");
|
||||
Assert.assertNotNull(producerConfigurationTest1);
|
||||
final ProducerMetadata producerMetadataTest1 = producerConfigurationTest1.getProducerMetadata();
|
||||
final ProducerMetadata<K,V> producerMetadataTest1 = producerConfigurationTest1.getProducerMetadata();
|
||||
Assert.assertEquals(producerMetadataTest1.getTopic(), "test1");
|
||||
Assert.assertEquals(producerMetadataTest1.getCompressionCodec(), "0");
|
||||
Assert.assertEquals(producerMetadataTest1.getKeyClassType(), java.lang.String.class);
|
||||
Assert.assertEquals(producerMetadataTest1.getValueClassType(), java.lang.String.class);
|
||||
|
||||
final Encoder valueEncoder = appContext.getBean("valueEncoder", Encoder.class);
|
||||
final Encoder<T> valueEncoder = appContext.getBean("valueEncoder", Encoder.class);
|
||||
Assert.assertEquals(producerMetadataTest1.getValueEncoder(), valueEncoder);
|
||||
Assert.assertEquals(producerMetadataTest1.getKeyEncoder(), valueEncoder);
|
||||
|
||||
final Producer producerTest1 = appContext.getBean("prodFactory_test1", Producer.class);
|
||||
Assert.assertEquals(producerConfigurationTest1, new ProducerConfiguration(producerMetadataTest1, producerTest1));
|
||||
final Producer<K,V> producerTest1 = appContext.getBean("prodFactory_test1", Producer.class);
|
||||
Assert.assertEquals(producerConfigurationTest1, new ProducerConfiguration<K,V>(producerMetadataTest1, producerTest1));
|
||||
|
||||
final ProducerConfiguration producerConfigurationTest2 = topicConfigurations.get("producerConfiguration_" + "test2");
|
||||
final ProducerConfiguration<K,V> producerConfigurationTest2 = topicConfigurations.get("producerConfiguration_" + "test2");
|
||||
Assert.assertNotNull(producerConfigurationTest2);
|
||||
final ProducerMetadata producerMetadataTest2 = producerConfigurationTest2.getProducerMetadata();
|
||||
final ProducerMetadata<K,V> producerMetadataTest2 = producerConfigurationTest2.getProducerMetadata();
|
||||
Assert.assertEquals(producerMetadataTest2.getTopic(), "test2");
|
||||
Assert.assertEquals(producerMetadataTest2.getCompressionCodec(), "0");
|
||||
|
||||
final Producer producerTest2 = appContext.getBean("prodFactory_test2", Producer.class);
|
||||
Assert.assertEquals(producerConfigurationTest2, new ProducerConfiguration(producerMetadataTest2, producerTest2));
|
||||
final Producer<K,V> producerTest2 = appContext.getBean("prodFactory_test2", Producer.class);
|
||||
Assert.assertEquals(producerConfigurationTest2, new ProducerConfiguration<K,V>(producerMetadataTest2, producerTest2));
|
||||
}
|
||||
}
|
||||
|
||||
@@ -17,19 +17,19 @@ package org.springframework.integration.kafka.serializer;
|
||||
|
||||
import org.junit.Assert;
|
||||
import org.junit.Test;
|
||||
import org.springframework.integration.kafka.serializer.avro.AvroBackedKafkaDecoder;
|
||||
import org.springframework.integration.kafka.serializer.avro.AvroBackedKafkaEncoder;
|
||||
import org.springframework.integration.kafka.serializer.avro.AvroReflectDatumBackedKafkaDecoder;
|
||||
import org.springframework.integration.kafka.serializer.avro.AvroReflectDatumBackedKafkaEncoder;
|
||||
import org.springframework.integration.kafka.test.utils.TestObject;
|
||||
|
||||
/**
|
||||
* @author Soby Chacko
|
||||
* @since 0.5
|
||||
*/
|
||||
public class AvroBackedKafkaSerializerTest {
|
||||
public class AvroReflectDatumBackedKafkaSerializerTest {
|
||||
@Test
|
||||
@SuppressWarnings("unchecked")
|
||||
public void testDecodePlainSchema() {
|
||||
final AvroBackedKafkaEncoder avroBackedKafkaEncoder = new AvroBackedKafkaEncoder(TestObject.class);
|
||||
final AvroReflectDatumBackedKafkaEncoder<TestObject> avroBackedKafkaEncoder = new AvroReflectDatumBackedKafkaEncoder<TestObject>(TestObject.class);
|
||||
|
||||
final TestObject testObject = new TestObject();
|
||||
testObject.setTestData1("\"Test Data1\"");
|
||||
@@ -37,8 +37,8 @@ public class AvroBackedKafkaSerializerTest {
|
||||
|
||||
final byte[] data = avroBackedKafkaEncoder.toBytes(testObject);
|
||||
|
||||
final AvroBackedKafkaDecoder avroBackedKafkaDecoder = new AvroBackedKafkaDecoder(TestObject.class);
|
||||
final TestObject decodedFbu = (TestObject) avroBackedKafkaDecoder.fromBytes(data);
|
||||
final AvroReflectDatumBackedKafkaDecoder<TestObject> avroReflectDatumBackedKafkaDecoder = new AvroReflectDatumBackedKafkaDecoder<TestObject>(TestObject.class);
|
||||
final TestObject decodedFbu = avroReflectDatumBackedKafkaDecoder.fromBytes(data);
|
||||
|
||||
Assert.assertEquals(testObject.getTestData1(), decodedFbu.getTestData1());
|
||||
Assert.assertEquals(testObject.getTestData2(), decodedFbu.getTestData2());
|
||||
@@ -47,12 +47,12 @@ public class AvroBackedKafkaSerializerTest {
|
||||
@Test
|
||||
@SuppressWarnings("unchecked")
|
||||
public void anotherTest() {
|
||||
final AvroBackedKafkaEncoder avroBackedKafkaEncoder = new AvroBackedKafkaEncoder(java.lang.String.class);
|
||||
final AvroReflectDatumBackedKafkaEncoder<String> avroBackedKafkaEncoder = new AvroReflectDatumBackedKafkaEncoder<String>(java.lang.String.class);
|
||||
final String testString = "Testing Avro";
|
||||
final byte[] data = avroBackedKafkaEncoder.toBytes(testString);
|
||||
|
||||
final AvroBackedKafkaDecoder avroBackedKafkaDecoder = new AvroBackedKafkaDecoder(java.lang.String.class);
|
||||
final String decodedS = (String) avroBackedKafkaDecoder.fromBytes(data);
|
||||
final AvroReflectDatumBackedKafkaDecoder<String> avroReflectDatumBackedKafkaDecoder = new AvroReflectDatumBackedKafkaDecoder<String>(java.lang.String.class);
|
||||
final String decodedS = avroReflectDatumBackedKafkaDecoder.fromBytes(data);
|
||||
|
||||
Assert.assertEquals(testString, decodedS);
|
||||
}
|
||||
@@ -0,0 +1,30 @@
|
||||
package org.springframework.integration.kafka.serializer;
|
||||
|
||||
import org.junit.Assert;
|
||||
import org.junit.Test;
|
||||
import org.springframework.integration.kafka.serializer.avro.AvroSpecificDatumBackedKafkaDecoder;
|
||||
import org.springframework.integration.kafka.serializer.avro.AvroSpecificDatumBackedKafkaEncoder;
|
||||
import org.springframework.integration.kafka.test.utils.User;
|
||||
|
||||
/**
|
||||
* @author Soby Chacko
|
||||
* @since 0.5
|
||||
*/
|
||||
public class AvroSpecificDatumBackedKafkaSerializerTest {
|
||||
|
||||
@Test
|
||||
@SuppressWarnings("unchecked")
|
||||
public void testEncodeDecodeFromSpecificDatumSchema() {
|
||||
final AvroSpecificDatumBackedKafkaEncoder<User> avroBackedKafkaEncoder = new AvroSpecificDatumBackedKafkaEncoder<User>(User.class);
|
||||
|
||||
final User user = new User("First", "Last");
|
||||
|
||||
final byte[] data = avroBackedKafkaEncoder.toBytes(user);
|
||||
|
||||
final AvroSpecificDatumBackedKafkaDecoder<User> avroSpecificDatumBackedKafkaDecoder = new AvroSpecificDatumBackedKafkaDecoder<User>(User.class);
|
||||
final User decodedUser = avroSpecificDatumBackedKafkaDecoder.fromBytes(data);
|
||||
|
||||
Assert.assertEquals(user.getFirstName(), decodedUser.getFirstName().toString());
|
||||
Assert.assertEquals(user.getLastName(), decodedUser.getLastName().toString());
|
||||
}
|
||||
}
|
||||
@@ -45,34 +45,34 @@ import org.mockito.stubbing.Answer;
|
||||
* @author Gunnar Hillert
|
||||
* @since 0.5
|
||||
*/
|
||||
public class ConsumerConfigurationTests {
|
||||
public class ConsumerConfigurationTests<K,V> {
|
||||
@Test
|
||||
@SuppressWarnings("unchecked")
|
||||
public void testReceiveMessageForSingleTopicFromSingleStream() {
|
||||
final ConsumerMetadata consumerMetadata = mock(ConsumerMetadata.class);
|
||||
final ConsumerMetadata<K,V> consumerMetadata = mock(ConsumerMetadata.class);
|
||||
final ConsumerConnectionProvider consumerConnectionProvider =
|
||||
mock(ConsumerConnectionProvider.class);
|
||||
final MessageLeftOverTracker messageLeftOverTracker = mock(MessageLeftOverTracker.class);
|
||||
final MessageLeftOverTracker<K,V> messageLeftOverTracker = mock(MessageLeftOverTracker.class);
|
||||
final ConsumerConnector consumerConnector = mock(ConsumerConnector.class);
|
||||
|
||||
when(consumerConnectionProvider.getConsumerConnector()).thenReturn(consumerConnector);
|
||||
|
||||
final ConsumerConfiguration consumerConfiguration = new ConsumerConfiguration(consumerMetadata,
|
||||
final ConsumerConfiguration<K,V> consumerConfiguration = new ConsumerConfiguration<K,V>(consumerMetadata,
|
||||
consumerConnectionProvider, messageLeftOverTracker);
|
||||
consumerConfiguration.setMaxMessages(1);
|
||||
|
||||
final KafkaStream stream = mock(KafkaStream.class);
|
||||
final List<KafkaStream<byte[], byte[]>> streams = new ArrayList<KafkaStream<byte[], byte[]>>();
|
||||
final KafkaStream<K,V> stream = mock(KafkaStream.class);
|
||||
final List<KafkaStream<K,V>> streams = new ArrayList<KafkaStream<K,V>>();
|
||||
streams.add(stream);
|
||||
final Map<String, List<KafkaStream<byte[], byte[]>>> messageStreams = new HashMap<String, List<KafkaStream<byte[], byte[]>>>();
|
||||
final Map<String, List<KafkaStream<K,V>>> messageStreams = new HashMap<String, List<KafkaStream<K,V>>>();
|
||||
messageStreams.put("topic", streams);
|
||||
|
||||
when(consumerConfiguration.getConsumerMapWithMessageStreams()).thenReturn(messageStreams);
|
||||
final ConsumerIterator iterator = mock(ConsumerIterator.class);
|
||||
final ConsumerIterator<K,V> iterator = mock(ConsumerIterator.class);
|
||||
when(stream.iterator()).thenReturn(iterator);
|
||||
final MessageAndMetadata messageAndMetadata = mock(MessageAndMetadata.class);
|
||||
final MessageAndMetadata<K,V> messageAndMetadata = mock(MessageAndMetadata.class);
|
||||
when(iterator.next()).thenReturn(messageAndMetadata);
|
||||
when(messageAndMetadata.message()).thenReturn("got message");
|
||||
when(messageAndMetadata.message()).thenReturn((V) "got message");
|
||||
when(messageAndMetadata.topic()).thenReturn("topic");
|
||||
when(messageAndMetadata.partition()).thenReturn(1);
|
||||
|
||||
@@ -90,54 +90,54 @@ public class ConsumerConfigurationTests {
|
||||
@Test
|
||||
@SuppressWarnings("unchecked")
|
||||
public void testReceiveMessageForSingleTopicFromMultipleStreams() {
|
||||
final ConsumerMetadata consumerMetadata = mock(ConsumerMetadata.class);
|
||||
final ConsumerMetadata<K,V> consumerMetadata = mock(ConsumerMetadata.class);
|
||||
final ConsumerConnectionProvider consumerConnectionProvider =
|
||||
mock(ConsumerConnectionProvider.class);
|
||||
final MessageLeftOverTracker messageLeftOverTracker = mock(MessageLeftOverTracker.class);
|
||||
final MessageLeftOverTracker<K,V> messageLeftOverTracker = mock(MessageLeftOverTracker.class);
|
||||
|
||||
final ConsumerConnector consumerConnector = mock(ConsumerConnector.class);
|
||||
|
||||
when(consumerConnectionProvider.getConsumerConnector()).thenReturn(consumerConnector);
|
||||
|
||||
final ConsumerConfiguration consumerConfiguration = new ConsumerConfiguration(consumerMetadata,
|
||||
final ConsumerConfiguration<K,V> consumerConfiguration = new ConsumerConfiguration<K,V>(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[]>>();
|
||||
final KafkaStream<K,V> stream1 = mock(KafkaStream.class);
|
||||
final KafkaStream<K,V> stream2 = mock(KafkaStream.class);
|
||||
final KafkaStream<K,V> stream3 = mock(KafkaStream.class);
|
||||
final List<KafkaStream<K,V>> streams = new ArrayList<KafkaStream<K,V>>();
|
||||
streams.add(stream1);
|
||||
streams.add(stream2);
|
||||
streams.add(stream3);
|
||||
final Map<String, List<KafkaStream<byte[], byte[]>>> messageStreams = new HashMap<String, List<KafkaStream<byte[], byte[]>>>();
|
||||
final Map<String, List<KafkaStream<K,V>>> messageStreams = new HashMap<String, List<KafkaStream<K,V>>>();
|
||||
messageStreams.put("topic", streams);
|
||||
|
||||
when(consumerConfiguration.getConsumerMapWithMessageStreams()).thenReturn(messageStreams);
|
||||
final ConsumerIterator iterator1 = mock(ConsumerIterator.class);
|
||||
final ConsumerIterator iterator2 = mock(ConsumerIterator.class);
|
||||
final ConsumerIterator iterator3 = mock(ConsumerIterator.class);
|
||||
final ConsumerIterator<K,V> iterator1 = mock(ConsumerIterator.class);
|
||||
final ConsumerIterator<K,V> iterator2 = mock(ConsumerIterator.class);
|
||||
final ConsumerIterator<K,V> 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);
|
||||
final MessageAndMetadata<K,V> messageAndMetadata1 = mock(MessageAndMetadata.class);
|
||||
final MessageAndMetadata<K,V> messageAndMetadata2 = mock(MessageAndMetadata.class);
|
||||
final MessageAndMetadata<K,V> 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.message()).thenReturn((V)"got message");
|
||||
when(messageAndMetadata1.topic()).thenReturn("topic");
|
||||
when(messageAndMetadata1.partition()).thenReturn(1);
|
||||
|
||||
when(messageAndMetadata2.message()).thenReturn("got message");
|
||||
when(messageAndMetadata2.message()).thenReturn((V)"got message");
|
||||
when(messageAndMetadata2.topic()).thenReturn("topic");
|
||||
when(messageAndMetadata2.partition()).thenReturn(2);
|
||||
|
||||
when(messageAndMetadata3.message()).thenReturn("got message");
|
||||
when(messageAndMetadata3.message()).thenReturn((V)"got message");
|
||||
when(messageAndMetadata3.topic()).thenReturn("topic");
|
||||
when(messageAndMetadata3.partition()).thenReturn(3);
|
||||
|
||||
@@ -157,56 +157,56 @@ public class ConsumerConfigurationTests {
|
||||
@Test
|
||||
@SuppressWarnings("unchecked")
|
||||
public void testReceiveMessageForMultipleTopicsFromMultipleStreams() {
|
||||
final ConsumerMetadata consumerMetadata = mock(ConsumerMetadata.class);
|
||||
final ConsumerMetadata<K,V> consumerMetadata = mock(ConsumerMetadata.class);
|
||||
final ConsumerConnectionProvider consumerConnectionProvider =
|
||||
mock(ConsumerConnectionProvider.class);
|
||||
final MessageLeftOverTracker messageLeftOverTracker = mock(MessageLeftOverTracker.class);
|
||||
final MessageLeftOverTracker<K,V> messageLeftOverTracker = mock(MessageLeftOverTracker.class);
|
||||
|
||||
final ConsumerConnector consumerConnector = mock(ConsumerConnector.class);
|
||||
|
||||
when(consumerConnectionProvider.getConsumerConnector()).thenReturn(consumerConnector);
|
||||
|
||||
final ConsumerConfiguration consumerConfiguration = new ConsumerConfiguration(consumerMetadata,
|
||||
final ConsumerConfiguration<K,V> consumerConfiguration = new ConsumerConfiguration<K,V>(consumerMetadata,
|
||||
consumerConnectionProvider, messageLeftOverTracker);
|
||||
consumerConfiguration.setMaxMessages(9);
|
||||
|
||||
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[]>>();
|
||||
final KafkaStream<K,V> stream1 = mock(KafkaStream.class);
|
||||
final KafkaStream<K,V> stream2 = mock(KafkaStream.class);
|
||||
final KafkaStream<K,V> stream3 = mock(KafkaStream.class);
|
||||
final List<KafkaStream<K,V>> streams = new ArrayList<KafkaStream<K,V>>();
|
||||
streams.add(stream1);
|
||||
streams.add(stream2);
|
||||
streams.add(stream3);
|
||||
final Map<String, List<KafkaStream<byte[], byte[]>>> messageStreams = new HashMap<String, List<KafkaStream<byte[], byte[]>>>();
|
||||
final Map<String, List<KafkaStream<K,V>>> messageStreams = new HashMap<String, List<KafkaStream<K,V>>>();
|
||||
messageStreams.put("topic1", streams);
|
||||
messageStreams.put("topic2", streams);
|
||||
messageStreams.put("topic3", streams);
|
||||
|
||||
when(consumerConfiguration.getConsumerMapWithMessageStreams()).thenReturn(messageStreams);
|
||||
final ConsumerIterator iterator1 = mock(ConsumerIterator.class);
|
||||
final ConsumerIterator iterator2 = mock(ConsumerIterator.class);
|
||||
final ConsumerIterator iterator3 = mock(ConsumerIterator.class);
|
||||
final ConsumerIterator<K,V> iterator1 = mock(ConsumerIterator.class);
|
||||
final ConsumerIterator<K,V> iterator2 = mock(ConsumerIterator.class);
|
||||
final ConsumerIterator<K,V> 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);
|
||||
final MessageAndMetadata<K,V> messageAndMetadata1 = mock(MessageAndMetadata.class);
|
||||
final MessageAndMetadata<K,V> messageAndMetadata2 = mock(MessageAndMetadata.class);
|
||||
final MessageAndMetadata<K,V> messageAndMetadata3 = mock(MessageAndMetadata.class);
|
||||
|
||||
when(iterator1.next()).thenReturn(messageAndMetadata1);
|
||||
when(iterator2.next()).thenReturn(messageAndMetadata2);
|
||||
when(iterator3.next()).thenReturn(messageAndMetadata3);
|
||||
|
||||
when(messageAndMetadata1.message()).thenReturn("got message1");
|
||||
when(messageAndMetadata1.message()).thenReturn((V)"got message1");
|
||||
when(messageAndMetadata1.topic()).thenReturn("topic1");
|
||||
when(messageAndMetadata1.partition()).thenAnswer(getAnswer());
|
||||
|
||||
when(messageAndMetadata2.message()).thenReturn("got message2");
|
||||
when(messageAndMetadata2.message()).thenReturn((V)"got message2");
|
||||
when(messageAndMetadata2.topic()).thenReturn("topic2");
|
||||
when(messageAndMetadata1.partition()).thenAnswer(getAnswer());
|
||||
|
||||
when(messageAndMetadata3.message()).thenReturn("got message3");
|
||||
when(messageAndMetadata3.message()).thenReturn((V)"got message3");
|
||||
when(messageAndMetadata3.topic()).thenReturn("topic3");
|
||||
when(messageAndMetadata1.partition()).thenAnswer(getAnswer());
|
||||
|
||||
@@ -244,39 +244,39 @@ public class ConsumerConfigurationTests {
|
||||
@Test
|
||||
@SuppressWarnings("unchecked")
|
||||
public void testReceiveMessageAndVerifyMessageLeftoverFromPreviousPollAreTakenFirst() {
|
||||
final ConsumerMetadata consumerMetadata = mock(ConsumerMetadata.class);
|
||||
final ConsumerMetadata<K,V> consumerMetadata = mock(ConsumerMetadata.class);
|
||||
final ConsumerConnectionProvider consumerConnectionProvider =
|
||||
mock(ConsumerConnectionProvider.class);
|
||||
final MessageLeftOverTracker messageLeftOverTracker = mock(MessageLeftOverTracker.class);
|
||||
final MessageLeftOverTracker<K,V> messageLeftOverTracker = mock(MessageLeftOverTracker.class);
|
||||
final ConsumerConnector consumerConnector = mock(ConsumerConnector.class);
|
||||
|
||||
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);
|
||||
final MessageAndMetadata m3 = new MessageAndMetadata("key1", "value3", "topic3", 1, 1L);
|
||||
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);
|
||||
final MessageAndMetadata<String, String> m3 = new MessageAndMetadata<String, String>("key1", "value3", "topic3", 1, 1L);
|
||||
|
||||
final List<MessageAndMetadata> mList = new ArrayList<MessageAndMetadata>();
|
||||
final List<MessageAndMetadata<String, String>> mList = new ArrayList<MessageAndMetadata<String, String>>();
|
||||
mList.add(m1);
|
||||
mList.add(m2);
|
||||
mList.add(m3);
|
||||
|
||||
when(messageLeftOverTracker.getMessageLeftOverFromPreviousPoll()).thenReturn(mList);
|
||||
when((List<MessageAndMetadata<String, String>>) (Object) messageLeftOverTracker.getMessageLeftOverFromPreviousPoll()).thenReturn(mList);
|
||||
|
||||
when(consumerConnectionProvider.getConsumerConnector()).thenReturn(consumerConnector);
|
||||
|
||||
final ConsumerConfiguration consumerConfiguration = new ConsumerConfiguration(consumerMetadata,
|
||||
final ConsumerConfiguration<K,V> consumerConfiguration = new ConsumerConfiguration<K,V>(consumerMetadata,
|
||||
consumerConnectionProvider, messageLeftOverTracker);
|
||||
consumerConfiguration.setMaxMessages(5);
|
||||
|
||||
final KafkaStream stream = mock(KafkaStream.class);
|
||||
final List<KafkaStream<byte[], byte[]>> streams = new ArrayList<KafkaStream<byte[], byte[]>>();
|
||||
final KafkaStream<K,V> stream = mock(KafkaStream.class);
|
||||
final List<KafkaStream<K,V>> streams = new ArrayList<KafkaStream<K,V>>();
|
||||
streams.add(stream);
|
||||
final Map<String, List<KafkaStream<byte[], byte[]>>> messageStreams = new HashMap<String, List<KafkaStream<byte[], byte[]>>>();
|
||||
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(iterator);
|
||||
when(stream.iterator()).thenReturn((ConsumerIterator<K,V>) iterator);
|
||||
final MessageAndMetadata<String, String> messageAndMetadata = mock(MessageAndMetadata.class);
|
||||
when(iterator.next()).thenReturn(messageAndMetadata);
|
||||
when(messageAndMetadata.message()).thenReturn("got message");
|
||||
@@ -306,9 +306,10 @@ public class ConsumerConfigurationTests {
|
||||
}
|
||||
|
||||
@Test
|
||||
@SuppressWarnings("unchecked")
|
||||
public void testGetConsumerMapWithMessageStreamsWithNullDecoders() {
|
||||
|
||||
final ConsumerMetadata<?,?> mockedConsumerMetadata = mock(ConsumerMetadata.class);
|
||||
final ConsumerMetadata<K,V> mockedConsumerMetadata = mock(ConsumerMetadata.class);
|
||||
|
||||
assertNull(mockedConsumerMetadata.getKeyDecoder());
|
||||
assertNull(mockedConsumerMetadata.getValueDecoder());
|
||||
@@ -317,25 +318,26 @@ public class ConsumerConfigurationTests {
|
||||
when(mockedConsumerMetadata.getTopicStreamMap()).thenReturn(topicsStreamMap);
|
||||
|
||||
final ConsumerConnectionProvider mockedConsumerConnectionProvider = mock(ConsumerConnectionProvider.class);
|
||||
final MessageLeftOverTracker mockedMessageLeftOverTracker = mock(MessageLeftOverTracker.class);
|
||||
final MessageLeftOverTracker<K,V> mockedMessageLeftOverTracker = mock(MessageLeftOverTracker.class);
|
||||
final ConsumerConnector mockedConsumerConnector = mock(ConsumerConnector.class);
|
||||
|
||||
when(mockedConsumerConnectionProvider.getConsumerConnector()).thenReturn(mockedConsumerConnector);
|
||||
|
||||
final Map<String, List<KafkaStream<byte[], byte[]>>> messageStreams = new HashMap<String, List<KafkaStream<byte[],byte[]>>>();
|
||||
when(mockedConsumerConnector.createMessageStreams(topicsStreamMap)).thenReturn(messageStreams);
|
||||
final Map<String, List<KafkaStream<K,V>>> messageStreams = new HashMap<String, List<KafkaStream<K,V>>>();
|
||||
when((Map<String, List<KafkaStream<K,V>>>) (Object) mockedConsumerConnector.createMessageStreams(topicsStreamMap)).thenReturn(messageStreams);
|
||||
|
||||
final ConsumerConfiguration consumerConfiguration = new ConsumerConfiguration(mockedConsumerMetadata,
|
||||
final ConsumerConfiguration<K,V> consumerConfiguration = new ConsumerConfiguration<K,V>(mockedConsumerMetadata,
|
||||
mockedConsumerConnectionProvider, mockedMessageLeftOverTracker);
|
||||
|
||||
consumerConfiguration.getConsumerMapWithMessageStreams();
|
||||
|
||||
verify(mockedConsumerMetadata, atLeast(1)).getTopicStreamMap();
|
||||
verify(mockedConsumerConnector, atLeast(1)).createMessageStreams(topicsStreamMap);
|
||||
verify(mockedConsumerConnector, atMost(0)).createMessageStreams(topicsStreamMap, null, null);
|
||||
verify(mockedConsumerConnector, atLeast(1)).createMessageStreams(topicsStreamMap, null, null);
|
||||
//verify(mockedConsumerConnector, atMost(0)).createMessageStreams(topicsStreamMap, null, null);
|
||||
}
|
||||
|
||||
@Test
|
||||
@SuppressWarnings("unchecked")
|
||||
public void testGetConsumerMapWithMessageStreamsWithDecoders() {
|
||||
|
||||
@SuppressWarnings("unchecked")
|
||||
@@ -355,7 +357,7 @@ public class ConsumerConfigurationTests {
|
||||
|
||||
final ConsumerConnectionProvider mockedConsumerConnectionProvider =
|
||||
mock(ConsumerConnectionProvider.class);
|
||||
final MessageLeftOverTracker mockedMessageLeftOverTracker = mock(MessageLeftOverTracker.class);
|
||||
final MessageLeftOverTracker<String, String> mockedMessageLeftOverTracker = mock(MessageLeftOverTracker.class);
|
||||
final ConsumerConnector mockedConsumerConnector = mock(ConsumerConnector.class);
|
||||
|
||||
when(mockedConsumerConnectionProvider.getConsumerConnector()).thenReturn(mockedConsumerConnector);
|
||||
@@ -363,7 +365,7 @@ public class ConsumerConfigurationTests {
|
||||
final Map<String, List<KafkaStream<byte[], byte[]>>> messageStreams = new HashMap<String, List<KafkaStream<byte[],byte[]>>>();
|
||||
when(mockedConsumerConnector.createMessageStreams(topicsStreamMap)).thenReturn(messageStreams);
|
||||
|
||||
final ConsumerConfiguration consumerConfiguration = new ConsumerConfiguration(mockedConsumerMetadata,
|
||||
final ConsumerConfiguration<String, String> consumerConfiguration = new ConsumerConfiguration<String, String>(mockedConsumerMetadata,
|
||||
mockedConsumerConnectionProvider, mockedMessageLeftOverTracker);
|
||||
|
||||
consumerConfiguration.getConsumerMapWithMessageStreams();
|
||||
|
||||
@@ -30,20 +30,22 @@ import java.util.Map;
|
||||
* @author Soby Chacko
|
||||
* @since 0.5
|
||||
*/
|
||||
public class KafkaConsumerContextTest {
|
||||
public class KafkaConsumerContextTest<K,V> {
|
||||
|
||||
@Test
|
||||
@SuppressWarnings("unchecked")
|
||||
public void testMergeResultsFromMultipleConsumerConfiguration() {
|
||||
final KafkaConsumerContext kafkaConsumerContext = new KafkaConsumerContext();
|
||||
final KafkaConsumerContext<K,V> kafkaConsumerContext = new KafkaConsumerContext<K,V>();
|
||||
final ListableBeanFactory beanFactory = Mockito.mock(ListableBeanFactory.class);
|
||||
final ConsumerConfiguration consumerConfiguration1 = Mockito.mock(ConsumerConfiguration.class);
|
||||
final ConsumerConfiguration consumerConfiguration2 = Mockito.mock(ConsumerConfiguration.class);
|
||||
final ConsumerConfiguration<K,V> consumerConfiguration1 = Mockito.mock(ConsumerConfiguration.class);
|
||||
final ConsumerConfiguration<K,V> consumerConfiguration2 = Mockito.mock(ConsumerConfiguration.class);
|
||||
|
||||
final Map<String, ConsumerConfiguration> map = new HashMap<String, ConsumerConfiguration>();
|
||||
final Map<String, ConsumerConfiguration<K,V>> map = new HashMap<String, ConsumerConfiguration<K,V>>();
|
||||
map.put("config1", consumerConfiguration1);
|
||||
map.put("config2", consumerConfiguration2);
|
||||
|
||||
Mockito.when(beanFactory.getBeansOfType(ConsumerConfiguration.class)).thenReturn(map);
|
||||
Mockito.when((Map<String, ConsumerConfiguration<K,V>>) (Object) beanFactory.getBeansOfType(ConsumerConfiguration.class)).thenReturn(
|
||||
map);
|
||||
kafkaConsumerContext.setBeanFactory(beanFactory);
|
||||
|
||||
final Map<String, Map<Integer, List<Object>>> result1 = new HashMap<String, Map<Integer, List<Object>>>();
|
||||
|
||||
@@ -24,7 +24,7 @@ import org.junit.Test;
|
||||
import org.mockito.ArgumentCaptor;
|
||||
import org.mockito.Mockito;
|
||||
import org.springframework.integration.Message;
|
||||
import org.springframework.integration.kafka.serializer.avro.AvroBackedKafkaEncoder;
|
||||
import org.springframework.integration.kafka.serializer.avro.AvroReflectDatumBackedKafkaEncoder;
|
||||
import org.springframework.integration.kafka.test.utils.NonSerializableTestKey;
|
||||
import org.springframework.integration.kafka.test.utils.NonSerializableTestPayload;
|
||||
import org.springframework.integration.kafka.test.utils.TestKey;
|
||||
@@ -39,7 +39,7 @@ import java.io.ObjectInputStream;
|
||||
* @author Soby Chacko
|
||||
* @since 0.5
|
||||
*/
|
||||
public class ProducerConfigurationTests {
|
||||
public class ProducerConfigurationTests<K,V> {
|
||||
@Test
|
||||
@SuppressWarnings("unchecked")
|
||||
public void testSendMessageWithNonDefaultKeyAndValueEncoders() throws Exception {
|
||||
@@ -61,10 +61,12 @@ public class ProducerConfigurationTests {
|
||||
|
||||
Mockito.verify(producer, Mockito.times(1)).send(Mockito.any(KeyedMessage.class));
|
||||
|
||||
final ArgumentCaptor<KeyedMessage> argument = ArgumentCaptor.forClass(KeyedMessage.class);
|
||||
final ArgumentCaptor<KeyedMessage<String, String>> argument =
|
||||
(ArgumentCaptor<KeyedMessage<String, String>>) (Object)
|
||||
ArgumentCaptor.forClass(KeyedMessage.class);
|
||||
Mockito.verify(producer).send(argument.capture());
|
||||
|
||||
final KeyedMessage capturedKeyMessage = argument.getValue();
|
||||
final KeyedMessage<String, String> capturedKeyMessage = argument.getValue();
|
||||
|
||||
Assert.assertEquals(capturedKeyMessage.key(), "key");
|
||||
Assert.assertEquals(capturedKeyMessage.message(), "test message");
|
||||
@@ -93,12 +95,14 @@ public class ProducerConfigurationTests {
|
||||
|
||||
Mockito.verify(producer, Mockito.times(1)).send(Mockito.any(KeyedMessage.class));
|
||||
|
||||
final ArgumentCaptor<KeyedMessage> argument = ArgumentCaptor.forClass(KeyedMessage.class);
|
||||
final ArgumentCaptor<KeyedMessage<byte[], byte[]>> argument =
|
||||
(ArgumentCaptor<KeyedMessage<byte[], byte[]>>) (Object)
|
||||
ArgumentCaptor.forClass(KeyedMessage.class);
|
||||
Mockito.verify(producer).send(argument.capture());
|
||||
|
||||
final KeyedMessage capturedKeyMessage = argument.getValue();
|
||||
final KeyedMessage<byte[], byte[]> capturedKeyMessage = argument.getValue();
|
||||
|
||||
final byte[] keyBytes = (byte[])capturedKeyMessage.key();
|
||||
final byte[] keyBytes = capturedKeyMessage.key();
|
||||
|
||||
final ByteArrayInputStream keyInputStream = new ByteArrayInputStream (keyBytes);
|
||||
final ObjectInputStream keyObjectInputStream = new ObjectInputStream (keyInputStream);
|
||||
@@ -109,7 +113,7 @@ public class ProducerConfigurationTests {
|
||||
Assert.assertEquals(tk.getKeyPart1(), "compositePart1");
|
||||
Assert.assertEquals(tk.getKeyPart2(), "compositePart2");
|
||||
|
||||
final byte[] messageBytes = (byte[])capturedKeyMessage.message();
|
||||
final byte[] messageBytes = capturedKeyMessage.message();
|
||||
|
||||
final ByteArrayInputStream messageInputStream = new ByteArrayInputStream (messageBytes);
|
||||
final ObjectInputStream messageObjectInputStream = new ObjectInputStream (messageInputStream);
|
||||
@@ -130,7 +134,7 @@ public class ProducerConfigurationTests {
|
||||
@SuppressWarnings("unchecked")
|
||||
public void testSendMessageWithDefaultKeyEncoderAndNonDefaultValueEncoderAndCorrespondingData() throws Exception {
|
||||
final ProducerMetadata<byte[], TestPayload> producerMetadata = new ProducerMetadata<byte[], TestPayload>("test");
|
||||
final AvroBackedKafkaEncoder<TestPayload> encoder = new AvroBackedKafkaEncoder<TestPayload>(TestPayload.class);
|
||||
final AvroReflectDatumBackedKafkaEncoder<TestPayload> encoder = new AvroReflectDatumBackedKafkaEncoder<TestPayload>(TestPayload.class);
|
||||
producerMetadata.setValueEncoder(encoder);
|
||||
producerMetadata.setKeyEncoder(new DefaultEncoder(null));
|
||||
producerMetadata.setValueClassType(TestPayload.class);
|
||||
@@ -147,12 +151,14 @@ public class ProducerConfigurationTests {
|
||||
|
||||
Mockito.verify(producer, Mockito.times(1)).send(Mockito.any(KeyedMessage.class));
|
||||
|
||||
final ArgumentCaptor<KeyedMessage> argument = ArgumentCaptor.forClass(KeyedMessage.class);
|
||||
final ArgumentCaptor<KeyedMessage<byte[], TestPayload>> argument =
|
||||
(ArgumentCaptor<KeyedMessage<byte[], TestPayload>>) (Object)
|
||||
ArgumentCaptor.forClass(KeyedMessage.class);
|
||||
Mockito.verify(producer).send(argument.capture());
|
||||
|
||||
final KeyedMessage capturedKeyMessage = argument.getValue();
|
||||
final KeyedMessage<byte[], TestPayload> capturedKeyMessage = argument.getValue();
|
||||
|
||||
final byte[] keyBytes = (byte[])capturedKeyMessage.key();
|
||||
final byte[] keyBytes = capturedKeyMessage.key();
|
||||
|
||||
final ByteArrayInputStream keyInputStream = new ByteArrayInputStream (keyBytes);
|
||||
final ObjectInputStream keyObjectInputStream = new ObjectInputStream (keyInputStream);
|
||||
@@ -171,7 +177,7 @@ public class ProducerConfigurationTests {
|
||||
@SuppressWarnings("unchecked")
|
||||
public void testSendMessageWithNonDefaultKeyEncoderAndDefaultValueEncoderAndCorrespondingData() throws Exception {
|
||||
final ProducerMetadata<TestKey, byte[]> producerMetadata = new ProducerMetadata<TestKey, byte[]>("test");
|
||||
final AvroBackedKafkaEncoder<TestKey> encoder = new AvroBackedKafkaEncoder<TestKey>(TestKey.class);
|
||||
final AvroReflectDatumBackedKafkaEncoder<TestKey> encoder = new AvroReflectDatumBackedKafkaEncoder<TestKey>(TestKey.class);
|
||||
producerMetadata.setKeyEncoder(encoder);
|
||||
producerMetadata.setValueEncoder(new DefaultEncoder(null));
|
||||
producerMetadata.setKeyClassType(TestKey.class);
|
||||
@@ -187,14 +193,16 @@ public class ProducerConfigurationTests {
|
||||
|
||||
Mockito.verify(producer, Mockito.times(1)).send(Mockito.any(KeyedMessage.class));
|
||||
|
||||
final ArgumentCaptor<KeyedMessage> argument = ArgumentCaptor.forClass(KeyedMessage.class);
|
||||
final ArgumentCaptor<KeyedMessage<TestKey, byte[]>> argument =
|
||||
(ArgumentCaptor<KeyedMessage<TestKey, byte[]>>) (Object)
|
||||
ArgumentCaptor.forClass(KeyedMessage.class);
|
||||
Mockito.verify(producer).send(argument.capture());
|
||||
|
||||
final KeyedMessage capturedKeyMessage = argument.getValue();
|
||||
final KeyedMessage<TestKey, byte[]> capturedKeyMessage = argument.getValue();
|
||||
|
||||
Assert.assertEquals(capturedKeyMessage.key(), tk);
|
||||
|
||||
final byte[] payloadBytes = (byte[])capturedKeyMessage.message();
|
||||
final byte[] payloadBytes = capturedKeyMessage.message();
|
||||
|
||||
final ByteArrayInputStream payloadBis = new ByteArrayInputStream (payloadBytes);
|
||||
final ObjectInputStream payloadOis = new ObjectInputStream (payloadBis);
|
||||
@@ -226,11 +234,13 @@ public class ProducerConfigurationTests {
|
||||
|
||||
Mockito.verify(producer, Mockito.times(1)).send(Mockito.any(KeyedMessage.class));
|
||||
|
||||
final ArgumentCaptor<KeyedMessage> argument = ArgumentCaptor.forClass(KeyedMessage.class);
|
||||
final ArgumentCaptor<KeyedMessage<byte[], byte[]>> argument =
|
||||
(ArgumentCaptor<KeyedMessage<byte[], byte[]>>) (Object)
|
||||
ArgumentCaptor.forClass(KeyedMessage.class);
|
||||
Mockito.verify(producer).send(argument.capture());
|
||||
|
||||
final KeyedMessage capturedKeyMessage = argument.getValue();
|
||||
final byte[] keyBytes = (byte[])capturedKeyMessage.key();
|
||||
final KeyedMessage<byte[], byte[]> capturedKeyMessage = argument.getValue();
|
||||
final byte[] keyBytes = capturedKeyMessage.key();
|
||||
|
||||
final ByteArrayInputStream keyBis = new ByteArrayInputStream (keyBytes);
|
||||
final ObjectInputStream keyOis = new ObjectInputStream (keyBis);
|
||||
@@ -238,7 +248,7 @@ public class ProducerConfigurationTests {
|
||||
|
||||
Assert.assertEquals("key", keyObj);
|
||||
|
||||
final byte[] payloadBytes = (byte[])capturedKeyMessage.message();
|
||||
final byte[] payloadBytes = capturedKeyMessage.message();
|
||||
|
||||
final ByteArrayInputStream payloadBis = new ByteArrayInputStream (payloadBytes);
|
||||
final ObjectInputStream payloadOis = new ObjectInputStream (payloadBis);
|
||||
|
||||
@@ -24,14 +24,14 @@ import org.mockito.Mockito;
|
||||
* @author Soby Chacko
|
||||
* @since 0.5
|
||||
*/
|
||||
public class ProducerFactoryBeanTests {
|
||||
public class ProducerFactoryBeanTests<K,V> {
|
||||
|
||||
@Test
|
||||
public void createProducerWithDefaultMetadata() throws Exception {
|
||||
final ProducerMetadata<byte[], byte[]> producerMetadata = new ProducerMetadata<byte[], byte[]>("test");
|
||||
final ProducerMetadata<byte[], byte[]> tm = Mockito.spy(producerMetadata);
|
||||
final ProducerFactoryBean<byte[], byte[]> producerFactoryBean = new ProducerFactoryBean<byte[], byte[]>(tm, "localhost:9092");
|
||||
final Producer producer = producerFactoryBean.getObject();
|
||||
final Producer<byte[], byte[]> producer = producerFactoryBean.getObject();
|
||||
|
||||
Assert.assertTrue(producer != null);
|
||||
|
||||
@@ -50,7 +50,7 @@ public class ProducerFactoryBeanTests {
|
||||
producerMetadata.setBatchNumMessages("300");
|
||||
final ProducerMetadata<byte[], byte[]> tm = Mockito.spy(producerMetadata);
|
||||
final ProducerFactoryBean<byte[], byte[]> producerFactoryBean = new ProducerFactoryBean<byte[], byte[]>(tm, "localhost:9092");
|
||||
final Producer producer = producerFactoryBean.getObject();
|
||||
final Producer<byte[], byte[]> producer = producerFactoryBean.getObject();
|
||||
|
||||
Assert.assertTrue(producer != null);
|
||||
|
||||
|
||||
@@ -0,0 +1,96 @@
|
||||
package org.springframework.integration.kafka.test.utils;
|
||||
|
||||
import org.apache.avro.specific.SpecificRecord;
|
||||
|
||||
/**
|
||||
* @author Soby Chacko
|
||||
* @since 0.5
|
||||
* <p/>
|
||||
* This class is copied (partly) from an Avro generated class for necessary testing.
|
||||
* Please use caution when modify.
|
||||
*/
|
||||
public class User extends org.apache.avro.specific.SpecificRecordBase implements SpecificRecord {
|
||||
|
||||
public static final org.apache.avro.Schema SCHEMA$ = new org.apache.avro.Schema.Parser().parse("{\"type\":\"record\",\"name\":\"User\",\"namespace\":\"org.springframework.integration.samples.kafka.user\",\"fields\":[{\"name\":\"firstName\",\"type\":\"string\"},{\"name\":\"lastName\",\"type\":\"string\"}]}");
|
||||
public java.lang.CharSequence firstName;
|
||||
public java.lang.CharSequence lastName;
|
||||
|
||||
/**
|
||||
* Default constructor.
|
||||
*/
|
||||
public User() {
|
||||
}
|
||||
|
||||
/**
|
||||
* All-args constructor.
|
||||
*/
|
||||
public User(java.lang.CharSequence firstName, java.lang.CharSequence lastName) {
|
||||
this.firstName = firstName;
|
||||
this.lastName = lastName;
|
||||
}
|
||||
|
||||
public org.apache.avro.Schema getSchema() {
|
||||
return SCHEMA$;
|
||||
}
|
||||
|
||||
// Used by DatumWriter. Applications should not call.
|
||||
public java.lang.Object get(int field$) {
|
||||
switch (field$) {
|
||||
case 0:
|
||||
return firstName;
|
||||
case 1:
|
||||
return lastName;
|
||||
default:
|
||||
throw new org.apache.avro.AvroRuntimeException("Bad index");
|
||||
}
|
||||
}
|
||||
|
||||
// Used by DatumReader. Applications should not call.
|
||||
@SuppressWarnings(value = "unchecked")
|
||||
public void put(int field$, java.lang.Object value$) {
|
||||
switch (field$) {
|
||||
case 0:
|
||||
firstName = (java.lang.CharSequence) value$;
|
||||
break;
|
||||
case 1:
|
||||
lastName = (java.lang.CharSequence) value$;
|
||||
break;
|
||||
default:
|
||||
throw new org.apache.avro.AvroRuntimeException("Bad index");
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Gets the value of the 'firstName' field.
|
||||
*/
|
||||
public java.lang.CharSequence getFirstName() {
|
||||
return firstName;
|
||||
}
|
||||
|
||||
/**
|
||||
* Sets the value of the 'firstName' field.
|
||||
*
|
||||
* @param value the value to set.
|
||||
*/
|
||||
public void setFirstName(java.lang.CharSequence value) {
|
||||
this.firstName = value;
|
||||
}
|
||||
|
||||
/**
|
||||
* Gets the value of the 'lastName' field.
|
||||
*/
|
||||
public java.lang.CharSequence getLastName() {
|
||||
return lastName;
|
||||
}
|
||||
|
||||
/**
|
||||
* Sets the value of the 'lastName' field.
|
||||
*
|
||||
* @param value the value to set.
|
||||
*/
|
||||
public void setLastName(java.lang.CharSequence value) {
|
||||
this.lastName = value;
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
Reference in New Issue
Block a user