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

INTEXT-77 Add Kafka topic filter support for producer + consumer

Browse Source 
* Add tests
* Add samples
* Update readme

Jira: https://jira.springsource.org/browse/INTEXT-77
This commit is contained in:
Rajasekar Elango
2013-07-19 11:32:31 -07:00
committed by Artem Bilan
parent 94734e4ca3
commit d226b7c3c5
19 changed files with 471 additions and 170 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -15,7 +15,10 @@
*/
package org.springframework.integration.kafka.config.xml;
import kafka.serializer.DefaultDecoder;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import org.springframework.beans.factory.config.BeanDefinition;
import org.springframework.beans.factory.config.BeanDefinitionHolder;
import org.springframework.beans.factory.support.AbstractBeanDefinition;
@@ -29,17 +32,14 @@ import org.springframework.integration.kafka.support.ConsumerConnectionProvider;
import org.springframework.integration.kafka.support.ConsumerMetadata;
import org.springframework.integration.kafka.support.KafkaConsumerContext;
import org.springframework.integration.kafka.support.MessageLeftOverTracker;
import org.springframework.integration.kafka.support.TopicFilterConfiguration;
import org.springframework.util.StringUtils;
import org.springframework.util.xml.DomUtils;
import org.w3c.dom.Element;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
/**
* @author Soby Chacko
* @author Rajasekar Elango
* @since 0.5
*/
public class KafkaConsumerContextParser extends AbstractSingleBeanDefinitionParser {
@@ -57,7 +57,6 @@ public class KafkaConsumerContextParser extends AbstractSingleBeanDefinitionPars
parseConsumerConfigurations(consumerConfigurations, parserContext, builder, element);
}
@SuppressWarnings("unchecked")
private void parseConsumerConfigurations(final Element consumerConfigurations, final ParserContext parserContext,
final BeanDefinitionBuilder builder, final Element parentElem) {
for (final Element consumerConfiguration : DomUtils.getChildElementsByTagName(consumerConfigurations, "consumer-configuration")) {
@@ -75,14 +74,24 @@ public class KafkaConsumerContextParser extends AbstractSingleBeanDefinitionPars
final Map<String, Integer> topicStreamsMap = new HashMap<String, Integer>();
for (final Element topicConfiguration : DomUtils.getChildElementsByTagName(consumerConfiguration, "topic")) {
final String topic = topicConfiguration.getAttribute("id");
final String streams = topicConfiguration.getAttribute("streams");
final Integer streamsInt = Integer.valueOf(streams);
topicStreamsMap.put(topic, streamsInt);
final List<Element> topicConfigurations = DomUtils.getChildElementsByTagName(consumerConfiguration, "topic");
if (topicConfigurations != null){
for (final Element topicConfiguration : topicConfigurations) {
final String topic = topicConfiguration.getAttribute("id");
final String streams = topicConfiguration.getAttribute("streams");
final Integer streamsInt = Integer.valueOf(streams);
topicStreamsMap.put(topic, streamsInt);
}
consumerMetadataBuilder.addPropertyValue("topicStreamMap", topicStreamsMap);
}
consumerMetadataBuilder.addPropertyValue("topicStreamMap", topicStreamsMap);
final Element topicFilter = DomUtils.getChildElementByTagName(consumerConfiguration, "topic-filter");
if (topicFilter != null){
final TopicFilterConfiguration topicFilterConfiguration = new TopicFilterConfiguration(topicFilter.getAttribute("pattern"),Integer.valueOf(topicFilter.getAttribute("streams")), Boolean.valueOf(topicFilter.getAttribute("exclude")));
consumerMetadataBuilder.addPropertyValue("topicFilterConfiguration", topicFilterConfiguration);
}
final BeanDefinition consumerMetadataBeanDef = consumerMetadataBuilder.getBeanDefinition();
registerBeanDefinition(new BeanDefinitionHolder(consumerMetadataBeanDef, "consumerMetadata_" + consumerConfiguration.getAttribute("group-id")),

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

@@ -30,9 +30,6 @@ import org.springframework.util.StringUtils;
import org.springframework.util.xml.DomUtils;
import org.w3c.dom.Element;
import java.util.HashMap;
import java.util.Map;
/**
* @author Soby Chacko
* @since 0.5
@@ -52,7 +49,6 @@ public class KafkaProducerContextParser extends AbstractSimpleBeanDefinitionPars
parseProducerConfigurations(topics, parserContext);
}
@SuppressWarnings("unchecked")
private void parseProducerConfigurations(final Element topics, final ParserContext parserContext) {
for (final Element producerConfiguration : DomUtils.getChildElementsByTagName(topics, "producer-configuration")){
final BeanDefinitionBuilder producerConfigurationBuilder = BeanDefinitionBuilder.genericBeanDefinition(ProducerConfiguration.class);

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

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

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

@@ -16,17 +16,15 @@
package org.springframework.integration.kafka.serializer.avro;
import kafka.serializer.Decoder;
import org.apache.avro.io.DatumReader;
import org.apache.avro.reflect.ReflectDatumReader;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
/**
* @author Soby Chacko
* @since 0.5
*/
public class AvroReflectDatumBackedKafkaDecoder<T> extends AvroDatumSupport<T> implements Decoder<T> {
private static final Log LOG = LogFactory.getLog(AvroReflectDatumBackedKafkaDecoder.class);
private final DatumReader<T> reader;
@@ -35,9 +33,7 @@ public class AvroReflectDatumBackedKafkaDecoder<T> extends AvroDatumSupport<T> i
}
@Override
@SuppressWarnings("unchecked")
public T fromBytes(final byte[] bytes) {
return fromBytes(bytes, reader);
}
}

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

@@ -16,17 +16,15 @@
package org.springframework.integration.kafka.serializer.avro;
import kafka.serializer.Encoder;
import org.apache.avro.io.DatumWriter;
import org.apache.avro.reflect.ReflectDatumWriter;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
/**
* @author Soby Chacko
* @since 0.5
*/
public class AvroReflectDatumBackedKafkaEncoder<T> extends AvroDatumSupport<T> implements Encoder<T> {
private static final Log LOG = LogFactory.getLog(AvroReflectDatumBackedKafkaEncoder.class);
private final DatumWriter<T> writer;
@@ -35,7 +33,6 @@ public class AvroReflectDatumBackedKafkaEncoder<T> extends AvroDatumSupport<T> i
}
@Override
@SuppressWarnings("unchecked")
public byte[] toBytes(final T source) {
return toBytes(source, writer);
}

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

@@ -1,10 +1,9 @@
package org.springframework.integration.kafka.serializer.avro;
import kafka.serializer.Decoder;
import org.apache.avro.io.DatumReader;
import org.apache.avro.specific.SpecificDatumReader;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
/**
* @author Soby Chacko
@@ -12,8 +11,6 @@ import org.apache.commons.logging.LogFactory;
*/
public class AvroSpecificDatumBackedKafkaDecoder<T> extends AvroDatumSupport<T> implements Decoder<T> {
private static final Log LOG = LogFactory.getLog(AvroSpecificDatumBackedKafkaDecoder.class);
private final DatumReader<T> reader;
public AvroSpecificDatumBackedKafkaDecoder(final Class<T> specificRecordBase) {
@@ -21,7 +18,6 @@ public class AvroSpecificDatumBackedKafkaDecoder<T> extends AvroDatumSupport<T>
}
@Override
@SuppressWarnings("unchecked")
public T fromBytes(final byte[] bytes) {
return fromBytes(bytes, reader);
}

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

@@ -1,10 +1,9 @@
package org.springframework.integration.kafka.serializer.avro;
import kafka.serializer.Encoder;
import org.apache.avro.io.DatumWriter;
import org.apache.avro.specific.SpecificDatumWriter;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
/**
* @author Soby Chacko
@@ -12,8 +11,6 @@ import org.apache.commons.logging.LogFactory;
*/
public class AvroSpecificDatumBackedKafkaEncoder<T> extends AvroDatumSupport<T> implements Encoder<T> {
private static final Log LOG = LogFactory.getLog(AvroSpecificDatumBackedKafkaEncoder.class);
private final DatumWriter<T> writer;
public AvroSpecificDatumBackedKafkaEncoder(final Class<T> specificRecordClazz) {
@@ -21,7 +18,6 @@ public class AvroSpecificDatumBackedKafkaEncoder<T> extends AvroDatumSupport<T>
}
@Override
@SuppressWarnings("unchecked")
public byte[] toBytes(final T source) {
return toBytes(source, writer);
}

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

@@ -1,63 +1,52 @@
/*
* Copyright 2002-2013 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
* Copyright 2002-2013 the original author or authors. Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License. You may obtain a copy of the License at
* http://www.apache.org/licenses/LICENSE-2.0 Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND,
* either express or implied. See the License for the specific language governing permissions and limitations under the
* License.
*/
package org.springframework.integration.kafka.support;
import java.util.*;
import java.util.concurrent.*;
import kafka.consumer.ConsumerTimeoutException;
import kafka.consumer.KafkaStream;
import kafka.javaapi.consumer.ConsumerConnector;
import kafka.message.MessageAndMetadata;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.integration.MessagingException;
import java.util.ArrayList;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.Callable;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
/**
* @author Soby Chacko
* @author Rajasekar Elango
* @since 0.5
*/
public class ConsumerConfiguration<K,V> {
public class ConsumerConfiguration<K, V> {
private static final Log LOGGER = LogFactory.getLog(ConsumerConfiguration.class);
private final ConsumerMetadata<K,V> consumerMetadata;
private final ConsumerMetadata<K, V> consumerMetadata;
private final ConsumerConnectionProvider consumerConnectionProvider;
private final MessageLeftOverTracker<K,V> messageLeftOverTracker;
private final MessageLeftOverTracker<K, V> messageLeftOverTracker;
private ConsumerConnector consumerConnector;
private volatile int count = 0;
private int maxMessages = 1;
private Collection<List<KafkaStream<K, V>>> consumerMessageStreams;
private ExecutorService executorService = Executors.newCachedThreadPool();
private final ExecutorService executorService = Executors.newCachedThreadPool();
public ConsumerConfiguration(final ConsumerMetadata<K,V> consumerMetadata,
final ConsumerConnectionProvider consumerConnectionProvider,
final MessageLeftOverTracker<K,V> messageLeftOverTracker) {
public ConsumerConfiguration(final ConsumerMetadata<K, V> consumerMetadata,
final ConsumerConnectionProvider consumerConnectionProvider,
final MessageLeftOverTracker<K, V> messageLeftOverTracker) {
this.consumerMetadata = consumerMetadata;
this.consumerConnectionProvider = consumerConnectionProvider;
this.messageLeftOverTracker = messageLeftOverTracker;
}
public ConsumerMetadata<K,V> getConsumerMetadata() {
public ConsumerMetadata<K, V> getConsumerMetadata() {
return consumerMetadata;
}
@@ -65,23 +54,23 @@ public class ConsumerConfiguration<K,V> {
count = messageLeftOverTracker.getCurrentCount();
final Object lock = new Object();
final List<Callable<List<MessageAndMetadata<K,V>>>> tasks = new LinkedList<Callable<List<MessageAndMetadata<K,V>>>>();
final List<Callable<List<MessageAndMetadata<K, V>>>> tasks = new LinkedList<Callable<List<MessageAndMetadata<K, V>>>>();
final Map<String, List<KafkaStream<K, V>>> consumerMap = getConsumerMapWithMessageStreams();
for (final List<KafkaStream<K,V>> streams : consumerMap.values()) {
for (final KafkaStream<K,V> stream : streams) {
tasks.add(new Callable<List<MessageAndMetadata<K,V>>>() {
for (final List<KafkaStream<K, V>> streams : createConsumerMessageStreams()) {
for (final KafkaStream<K, V> stream : streams) {
tasks.add(new Callable<List<MessageAndMetadata<K, V>>>() {
@Override
public List<MessageAndMetadata<K,V>> call() throws Exception {
final List<MessageAndMetadata<K,V>> rawMessages = new ArrayList<MessageAndMetadata<K,V>>();
public List<MessageAndMetadata<K, V>> call() throws Exception {
final List<MessageAndMetadata<K, V>> rawMessages = new ArrayList<MessageAndMetadata<K, V>>();
try {
while (count < maxMessages) {
final MessageAndMetadata<K,V> messageAndMetadata = stream.iterator().next();
final MessageAndMetadata<K, V> messageAndMetadata = stream.iterator().next();
synchronized (lock) {
if (count < maxMessages) {
rawMessages.add(messageAndMetadata);
count++;
} else {
}
else {
messageLeftOverTracker.addMessageAndMetadata(messageAndMetadata);
}
}
@@ -97,18 +86,20 @@ public class ConsumerConfiguration<K,V> {
return executeTasks(tasks);
}
private Map<String, Map<Integer, List<Object>>> executeTasks(final List<Callable<List<MessageAndMetadata<K,V>>>> tasks) {
private Map<String, Map<Integer, List<Object>>> executeTasks(
final List<Callable<List<MessageAndMetadata<K, V>>>> tasks) {
final Map<String, Map<Integer, List<Object>>> messages = new ConcurrentHashMap<String, Map<Integer, List<Object>>>();
messages.putAll(getLeftOverMessageMap());
try {
for (final Future<List<MessageAndMetadata<K,V>>> result : executorService.invokeAll(tasks)) {
for (final Future<List<MessageAndMetadata<K, V>>> result : executorService.invokeAll(tasks)) {
if (!result.get().isEmpty()) {
final String topic = result.get().get(0).topic();
if (!messages.containsKey(topic)) {
messages.put(topic, getPayload(result.get()));
} else {
}
else {
final Map<Integer, List<Object>> existingPayloadMap = messages.get(topic);
getPayload(result.get(), existingPayloadMap);
@@ -126,21 +117,21 @@ public class ConsumerConfiguration<K,V> {
return messages;
}
@SuppressWarnings("unchecked")
private Map<String, Map<Integer, List<Object>>> getLeftOverMessageMap() {
final Map<String, Map<Integer, List<Object>>> messages = new ConcurrentHashMap<String, Map<Integer, List<Object>>>();
for (final MessageAndMetadata<K,V> mamd : messageLeftOverTracker.getMessageLeftOverFromPreviousPoll()) {
for (final MessageAndMetadata<K, V> mamd : messageLeftOverTracker.getMessageLeftOverFromPreviousPoll()) {
final String topic = mamd.topic();
if (!messages.containsKey(topic)) {
final List<MessageAndMetadata<K,V>> l = new ArrayList<MessageAndMetadata<K,V>>();
final List<MessageAndMetadata<K, V>> l = new ArrayList<MessageAndMetadata<K, V>>();
l.add(mamd);
messages.put(topic, getPayload(l));
} else {
}
else {
final Map<Integer, List<Object>> existingPayloadMap = messages.get(topic);
final List<MessageAndMetadata<K,V>> l = new ArrayList<MessageAndMetadata<K,V>>();
final List<MessageAndMetadata<K, V>> l = new ArrayList<MessageAndMetadata<K, V>>();
l.add(mamd);
getPayload(l, existingPayloadMap);
}
@@ -149,15 +140,16 @@ public class ConsumerConfiguration<K,V> {
return messages;
}
private Map<Integer, List<Object>> getPayload(final List<MessageAndMetadata<K,V>> messageAndMetadatas) {
private Map<Integer, List<Object>> getPayload(final List<MessageAndMetadata<K, V>> messageAndMetadatas) {
final Map<Integer, List<Object>> payloadMap = new ConcurrentHashMap<Integer, List<Object>>();
for (final MessageAndMetadata<K,V> messageAndMetadata : messageAndMetadatas) {
for (final MessageAndMetadata<K, V> messageAndMetadata : messageAndMetadatas) {
if (!payloadMap.containsKey(messageAndMetadata.partition())) {
final List<Object> payload = new ArrayList<Object>();
payload.add(messageAndMetadata.message());
payloadMap.put(messageAndMetadata.partition(), payload);
} else {
}
else {
final List<Object> payload = payloadMap.get(messageAndMetadata.partition());
payload.add(messageAndMetadata.message());
}
@@ -167,25 +159,52 @@ public class ConsumerConfiguration<K,V> {
return payloadMap;
}
private void getPayload(final List<MessageAndMetadata<K,V>> messageAndMetadatas, final Map<Integer, List<Object>> existingPayloadMap) {
for (final MessageAndMetadata<K,V> messageAndMetadata : messageAndMetadatas) {
private void getPayload(final List<MessageAndMetadata<K, V>> messageAndMetadatas,
final Map<Integer, List<Object>> existingPayloadMap) {
for (final MessageAndMetadata<K, V> messageAndMetadata : messageAndMetadatas) {
if (!existingPayloadMap.containsKey(messageAndMetadata.partition())) {
final List<Object> payload = new ArrayList<Object>();
payload.add(messageAndMetadata.message());
existingPayloadMap.put(messageAndMetadata.partition(), payload);
} else {
}
else {
final List<Object> payload = existingPayloadMap.get(messageAndMetadata.partition());
payload.add(messageAndMetadata.message());
}
}
}
@SuppressWarnings("unchecked")
public Map<String, List<KafkaStream<K,V>>> getConsumerMapWithMessageStreams() {
return getConsumerConnector().createMessageStreams(
consumerMetadata.getTopicStreamMap(),
consumerMetadata.getKeyDecoder(),
consumerMetadata.getValueDecoder());
private Collection<List<KafkaStream<K, V>>> createConsumerMessageStreams() {
if (consumerMessageStreams == null) {
if (!(consumerMetadata.getTopicStreamMap() == null || consumerMetadata.getTopicStreamMap().isEmpty())) {
consumerMessageStreams = createMessageStreamsForTopic().values();
}
else {
consumerMessageStreams = new ArrayList<List<KafkaStream<K, V>>>();
consumerMessageStreams.add(createMessageStreamsForTopicFilter());
}
}
return consumerMessageStreams;
}
public Map<String, List<KafkaStream<K, V>>> createMessageStreamsForTopic() {
return getConsumerConnector().createMessageStreams(consumerMetadata.getTopicStreamMap(),
consumerMetadata.getKeyDecoder(), consumerMetadata.getValueDecoder());
}
public List<KafkaStream<K, V>> createMessageStreamsForTopicFilter() {
List<KafkaStream<K, V>> messageStream = new ArrayList<KafkaStream<K, V>>();
TopicFilterConfiguration topicFilterConfiguration = consumerMetadata.getTopicFilterConfiguration();
if (topicFilterConfiguration != null) {
messageStream = getConsumerConnector().createMessageStreamsByFilter(
topicFilterConfiguration.getTopicFilter(), topicFilterConfiguration.getNumberOfStreams(),
consumerMetadata.getKeyDecoder(), consumerMetadata.getValueDecoder());
}
else {
LOGGER.warn("No Topic Filter Configuration defined");
}
return messageStream;
}
public int getMaxMessages() {

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

@@ -15,15 +15,16 @@
*/
package org.springframework.integration.kafka.support;
import java.util.Map;
import kafka.serializer.Decoder;
import kafka.serializer.DefaultDecoder;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.integration.kafka.core.KafkaConsumerDefaults;
import java.util.Map;
/**
* @author Soby Chacko
* @author Rajasekar Elango
* @since 0.5
*/
public class ConsumerMetadata<K,V> implements InitializingBean {
@@ -46,6 +47,7 @@ public class ConsumerMetadata<K,V> implements InitializingBean {
private Decoder<V> valueDecoder;
private Decoder<K> keyDecoder;
private Map<String, Integer> topicStreamMap;
private TopicFilterConfiguration topicFilterConfiguration;
public String getGroupId() {
return groupId;
@@ -186,4 +188,14 @@ public class ConsumerMetadata<K,V> implements InitializingBean {
setKeyDecoder((Decoder<K>) getValueDecoder());
}
}
public TopicFilterConfiguration getTopicFilterConfiguration() {
return topicFilterConfiguration;
}
public void setTopicFilterConfiguration(
TopicFilterConfiguration topicFilterConfiguration) {
this.topicFilterConfiguration = topicFilterConfiguration;
}
}

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

@@ -15,23 +15,24 @@
*/
package org.springframework.integration.kafka.support;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.beans.factory.BeanFactoryAware;
import org.springframework.beans.factory.ListableBeanFactory;
import org.springframework.integration.Message;
import java.util.Collection;
import java.util.Map;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.*;
import org.springframework.integration.Message;
/**
* @author Soby Chacko
* @author Rajasekar Elango
* @since 0.5
*/
public class KafkaProducerContext<K,V> implements BeanFactoryAware {
private static final Log LOGGER = LogFactory.getLog(KafkaProducerContext.class);
private Map<String, ProducerConfiguration<K,V>> topicsConfiguration;
@SuppressWarnings("unchecked")
public void send(final Message<?> message) throws Exception {
final ProducerConfiguration<K,V> producerConfiguration =
getTopicConfiguration(message.getHeaders().get("topic", String.class));
@@ -41,15 +42,15 @@ public class KafkaProducerContext<K,V> implements BeanFactoryAware {
}
}
private ProducerConfiguration<K,V> getTopicConfiguration(final String topic){
public ProducerConfiguration<K, V> getTopicConfiguration(final String topic) {
final Collection<ProducerConfiguration<K,V>> topics = topicsConfiguration.values();
for (final ProducerConfiguration<K,V> producerConfiguration : topics){
if (producerConfiguration.getProducerMetadata().getTopic().equals(topic)){
if (topic.matches(producerConfiguration.getProducerMetadata().getTopic())){
return producerConfiguration;
}
}
LOGGER.error("No is producer-configuration defined for topic " + topic + ". cannot send message");
return null;
}

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

@@ -1,40 +1,33 @@
/*
* Copyright 2002-2013 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
* Copyright 2002-2013 the original author or authors. Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License. You may obtain a copy of the License at
* http://www.apache.org/licenses/LICENSE-2.0 Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND,
* either express or implied. See the License for the specific language governing permissions and limitations under the
* License.
*/
package org.springframework.integration.kafka.support;
import java.io.*;
import kafka.javaapi.producer.Producer;
import kafka.producer.KeyedMessage;
import kafka.serializer.DefaultEncoder;
import org.apache.commons.lang.builder.EqualsBuilder;
import org.apache.commons.lang.builder.HashCodeBuilder;
import org.springframework.integration.Message;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.io.ObjectOutputStream;
/**
* @author Soby Chacko
* @author Rajasekar Elango
* @since 0.5
*/
public class ProducerConfiguration<K,V> {
private final Producer<K,V> producer;
private final ProducerMetadata<K,V> producerMetadata;
public class ProducerConfiguration<K, V> {
private final Producer<K, V> producer;
private final ProducerMetadata<K, V> producerMetadata;
public ProducerConfiguration(final ProducerMetadata<K, V> producerMetadata, final Producer<K, V> producer){
public ProducerConfiguration(final ProducerMetadata<K, V> producerMetadata, final Producer<K, V> producer) {
this.producerMetadata = producerMetadata;
this.producer = producer;
}
@@ -46,10 +39,12 @@ public class ProducerConfiguration<K,V> {
public void send(final Message<?> message) throws Exception {
final V v = getPayload(message);
String topic = message.getHeaders().get("topic", String.class);
if (message.getHeaders().containsKey("messageKey")) {
producer.send(new KeyedMessage<K, V>(producerMetadata.getTopic(), getKey(message), v));
} else {
producer.send(new KeyedMessage<K, V>(producerMetadata.getTopic(), v));
producer.send(new KeyedMessage<K, V>(topic, getKey(message), v));
}
else {
producer.send(new KeyedMessage<K, V>(topic, v));
}
}
@@ -57,7 +52,8 @@ public class ProducerConfiguration<K,V> {
private V getPayload(final Message<?> message) throws Exception {
if (producerMetadata.getValueEncoder().getClass().isAssignableFrom(DefaultEncoder.class)) {
return (V) getByteStream(message.getPayload());
} else if (message.getPayload().getClass().isAssignableFrom(producerMetadata.getValueClassType())) {
}
else if (message.getPayload().getClass().isAssignableFrom(producerMetadata.getValueClassType())) {
return producerMetadata.getValueClassType().cast(message.getPayload());
}
@@ -75,13 +71,13 @@ public class ProducerConfiguration<K,V> {
return message.getHeaders().get("messageKey", producerMetadata.getKeyClassType());
}
private static boolean isRawByteArray(final Object obj){
private static boolean isRawByteArray(final Object obj) {
return obj instanceof byte[];
}
private static byte[] getByteStream(final Object obj) throws IOException {
if (isRawByteArray(obj)){
return (byte[])obj;
if (isRawByteArray(obj)) {
return (byte[]) obj;
}
final ByteArrayOutputStream out = new ByteArrayOutputStream();
@@ -92,7 +88,7 @@ public class ProducerConfiguration<K,V> {
}
@Override
public boolean equals(final Object obj){
public boolean equals(final Object obj) {
return EqualsBuilder.reflectionEquals(this, obj);
}
@@ -100,4 +96,11 @@ public class ProducerConfiguration<K,V> {
public int hashCode() {
return HashCodeBuilder.reflectionHashCode(this);
}
@Override
public String toString() {
StringBuilder builder = new StringBuilder();
builder.append("ProducerConfiguration [producerMetadata=").append(producerMetadata).append("]");
return builder.toString();
}
}

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

@@ -18,12 +18,14 @@ package org.springframework.integration.kafka.support;
import kafka.producer.Partitioner;
import kafka.serializer.DefaultEncoder;
import kafka.serializer.Encoder;
import org.apache.commons.lang.builder.EqualsBuilder;
import org.apache.commons.lang.builder.HashCodeBuilder;
import org.springframework.beans.factory.InitializingBean;
/**
* @author Soby Chacko
* @author Rajasekar Elango
* @since 0.5
*/
public class ProducerMetadata<K,V> implements InitializingBean {
@@ -137,4 +139,14 @@ public class ProducerMetadata<K,V> implements InitializingBean {
public int hashCode() {
return HashCodeBuilder.reflectionHashCode(this);
}
@Override
public String toString() {
StringBuilder builder = new StringBuilder();
builder.append("ProducerMetadata [keyEncoder=").append(keyEncoder).append(", valueEncoder=")
.append(valueEncoder).append(", topic=").append(topic).append(", compressionCodec=")
.append(compressionCodec).append(", partitioner=").append(partitioner).append(", async=").append(async)
.append(", batchNumMessages=").append(batchNumMessages).append("]");
return builder.toString();
}
}

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

@@ -0,0 +1,46 @@
/*
* Copyright 2002-2013 the original author or authors. Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License. You may obtain a copy of the License at
* http://www.apache.org/licenses/LICENSE-2.0 Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND,
* either express or implied. See the License for the specific language governing permissions and limitations under the
* License.
*/
package org.springframework.integration.kafka.support;
import kafka.consumer.Blacklist;
import kafka.consumer.TopicFilter;
import kafka.consumer.Whitelist;
/**
* @author Rajasekar Elango
* @since 0.5
*/
public class TopicFilterConfiguration {
private final int numberOfStreams;
private TopicFilter topicFilter;
public TopicFilterConfiguration(final String pattern, final int numberOfStreams, final boolean exclude) {
this.numberOfStreams = numberOfStreams;
if (exclude) {
topicFilter = new Blacklist(pattern);
}
else {
topicFilter = new Whitelist(pattern);
}
}
public TopicFilter getTopicFilter() {
return topicFilter;
}
public int getNumberOfStreams() {
return numberOfStreams;
}
@Override
public String toString() {
return new StringBuilder(topicFilter.toString()).append(" : ").append(numberOfStreams).toString();
}
}

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

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

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

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

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

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

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

@@ -16,18 +16,9 @@
package org.springframework.integration.kafka.support;
import static org.junit.Assert.assertNull;
import static org.mockito.Mockito.atLeast;
import static org.mockito.Mockito.atMost;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
import static org.mockito.Mockito.*;
import java.util.ArrayList;
import java.util.Collection;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.*;
import kafka.consumer.ConsumerIterator;
import kafka.consumer.KafkaStream;
@@ -42,7 +33,7 @@ import org.mockito.stubbing.Answer;
/**
* @author Soby Chacko
* @author Gunnar Hillert
* @author Rajasekar Elango
* @since 0.5
*/
public class ConsumerConfigurationTests<K,V> {
@@ -55,6 +46,10 @@ public class ConsumerConfigurationTests<K,V> {
final MessageLeftOverTracker<K,V> messageLeftOverTracker = mock(MessageLeftOverTracker.class);
final ConsumerConnector consumerConnector = mock(ConsumerConnector.class);
Map<String, Integer> topicStreamMap = new HashMap<String, Integer>();
topicStreamMap.put("topic1", 1);
when(consumerMetadata.getTopicStreamMap()).thenReturn(topicStreamMap);
when(consumerConnectionProvider.getConsumerConnector()).thenReturn(consumerConnector);
final ConsumerConfiguration<K,V> consumerConfiguration = new ConsumerConfiguration<K,V>(consumerMetadata,
@@ -67,7 +62,7 @@ public class ConsumerConfigurationTests<K,V> {
final Map<String, List<KafkaStream<K,V>>> messageStreams = new HashMap<String, List<KafkaStream<K,V>>>();
messageStreams.put("topic", streams);
when(consumerConfiguration.getConsumerMapWithMessageStreams()).thenReturn(messageStreams);
when(consumerConfiguration.createMessageStreamsForTopic()).thenReturn(messageStreams);
final ConsumerIterator<K,V> iterator = mock(ConsumerIterator.class);
when(stream.iterator()).thenReturn(iterator);
final MessageAndMetadata<K,V> messageAndMetadata = mock(MessageAndMetadata.class);
@@ -95,6 +90,10 @@ public class ConsumerConfigurationTests<K,V> {
mock(ConsumerConnectionProvider.class);
final MessageLeftOverTracker<K,V> messageLeftOverTracker = mock(MessageLeftOverTracker.class);
Map<String, Integer> topicStreamMap = new HashMap<String, Integer>();
topicStreamMap.put("topic1", 1);
when(consumerMetadata.getTopicStreamMap()).thenReturn(topicStreamMap);
final ConsumerConnector consumerConnector = mock(ConsumerConnector.class);
when(consumerConnectionProvider.getConsumerConnector()).thenReturn(consumerConnector);
@@ -113,7 +112,7 @@ public class ConsumerConfigurationTests<K,V> {
final Map<String, List<KafkaStream<K,V>>> messageStreams = new HashMap<String, List<KafkaStream<K,V>>>();
messageStreams.put("topic", streams);
when(consumerConfiguration.getConsumerMapWithMessageStreams()).thenReturn(messageStreams);
when(consumerConfiguration.createMessageStreamsForTopic()).thenReturn(messageStreams);
final ConsumerIterator<K,V> iterator1 = mock(ConsumerIterator.class);
final ConsumerIterator<K,V> iterator2 = mock(ConsumerIterator.class);
final ConsumerIterator<K,V> iterator3 = mock(ConsumerIterator.class);
@@ -160,7 +159,12 @@ public class ConsumerConfigurationTests<K,V> {
final ConsumerMetadata<K,V> consumerMetadata = mock(ConsumerMetadata.class);
final ConsumerConnectionProvider consumerConnectionProvider =
mock(ConsumerConnectionProvider.class);
final MessageLeftOverTracker<K,V> messageLeftOverTracker = mock(MessageLeftOverTracker.class);
final MessageLeftOverTracker<K, V> messageLeftOverTracker = mock(MessageLeftOverTracker.class);
Map<String, Integer> topicStreamMap = new HashMap<String, Integer>();
topicStreamMap.put("topic1", 1);
when(consumerMetadata.getTopicStreamMap()).thenReturn(topicStreamMap);
final ConsumerConnector consumerConnector = mock(ConsumerConnector.class);
@@ -182,7 +186,7 @@ public class ConsumerConfigurationTests<K,V> {
messageStreams.put("topic2", streams);
messageStreams.put("topic3", streams);
when(consumerConfiguration.getConsumerMapWithMessageStreams()).thenReturn(messageStreams);
when(consumerConfiguration.createMessageStreamsForTopic()).thenReturn(messageStreams);
final ConsumerIterator<K,V> iterator1 = mock(ConsumerIterator.class);
final ConsumerIterator<K,V> iterator2 = mock(ConsumerIterator.class);
final ConsumerIterator<K,V> iterator3 = mock(ConsumerIterator.class);
@@ -211,6 +215,7 @@ public class ConsumerConfigurationTests<K,V> {
when(messageAndMetadata1.partition()).thenAnswer(getAnswer());
final Map<String, Map<Integer, List<Object>>> messages = consumerConfiguration.receive();
int sum = 0;
final Collection<Map<Integer, List<Object>>> values = messages.values();
@@ -224,6 +229,8 @@ public class ConsumerConfigurationTests<K,V> {
Assert.assertEquals(sum, 9);
}
private Answer<Object> getAnswer() {
return new Answer<Object>() {
private int count = 0;
@@ -250,6 +257,9 @@ public class ConsumerConfigurationTests<K,V> {
final MessageLeftOverTracker<K,V> messageLeftOverTracker = mock(MessageLeftOverTracker.class);
final ConsumerConnector consumerConnector = mock(ConsumerConnector.class);
Map<String, Integer> topicStreamMap = new HashMap<String, Integer>();
topicStreamMap.put("topic1", 1);
when(consumerMetadata.getTopicStreamMap()).thenReturn(topicStreamMap);
when(messageLeftOverTracker.getCurrentCount()).thenReturn(3);
final MessageAndMetadata<String, String> m1 = new MessageAndMetadata<String, String>("key1", "value1", "topic1", 1, 1L);
final MessageAndMetadata<String, String> m2 = new MessageAndMetadata<String, String>("key2", "value2", "topic2", 1, 1L);
@@ -273,11 +283,10 @@ public class ConsumerConfigurationTests<K,V> {
streams.add(stream);
final Map<String, List<KafkaStream<K,V>>> messageStreams = new HashMap<String, List<KafkaStream<K,V>>>();
messageStreams.put("topic1", streams);
when(consumerConfiguration.getConsumerMapWithMessageStreams()).thenReturn(messageStreams);
final ConsumerIterator<String, String> iterator = mock(ConsumerIterator.class);
when(stream.iterator()).thenReturn((ConsumerIterator<K,V>) iterator);
final MessageAndMetadata<String, String> messageAndMetadata = mock(MessageAndMetadata.class);
when(consumerConfiguration.createMessageStreamsForTopic()).thenReturn(messageStreams);
final ConsumerIterator iterator = mock(ConsumerIterator.class);
when(stream.iterator()).thenReturn(iterator);
final MessageAndMetadata messageAndMetadata = mock(MessageAndMetadata.class);
when(iterator.next()).thenReturn(messageAndMetadata);
when(messageAndMetadata.message()).thenReturn("got message");
when(messageAndMetadata.topic()).thenReturn("topic1");
@@ -329,7 +338,7 @@ public class ConsumerConfigurationTests<K,V> {
final ConsumerConfiguration<K,V> consumerConfiguration = new ConsumerConfiguration<K,V>(mockedConsumerMetadata,
mockedConsumerConnectionProvider, mockedMessageLeftOverTracker);
consumerConfiguration.getConsumerMapWithMessageStreams();
consumerConfiguration.createMessageStreamsForTopic();
verify(mockedConsumerMetadata, atLeast(1)).getTopicStreamMap();
verify(mockedConsumerConnector, atLeast(1)).createMessageStreams(topicsStreamMap, null, null);
@@ -368,13 +377,122 @@ public class ConsumerConfigurationTests<K,V> {
final ConsumerConfiguration<String, String> consumerConfiguration = new ConsumerConfiguration<String, String>(mockedConsumerMetadata,
mockedConsumerConnectionProvider, mockedMessageLeftOverTracker);
consumerConfiguration.getConsumerMapWithMessageStreams();
consumerConfiguration.createMessageStreamsForTopic();
verify(mockedConsumerMetadata, atLeast(1)).getTopicStreamMap();
verify(mockedConsumerConnector, atMost(0)).createMessageStreams(topicsStreamMap);
verify(mockedConsumerConnector, atLeast(1)).createMessageStreams(topicsStreamMap, mockedKeyDecoder, mockedValueDecoder);
}
@Test
@SuppressWarnings("unchecked")
public void testReceiveMessageForTopicFilterFromSingleStream() {
final ConsumerMetadata consumerMetadata = mock(ConsumerMetadata.class);
final ConsumerConnectionProvider consumerConnectionProvider =
mock(ConsumerConnectionProvider.class);
final MessageLeftOverTracker messageLeftOverTracker = mock(MessageLeftOverTracker.class);
final ConsumerConnector consumerConnector = mock(ConsumerConnector.class);
when(consumerMetadata.getTopicFilterConfiguration()).thenReturn(new TopicFilterConfiguration(".*", 1, false));
when(consumerConnectionProvider.getConsumerConnector()).thenReturn(consumerConnector);
final ConsumerConfiguration consumerConfiguration = new ConsumerConfiguration(consumerMetadata,
consumerConnectionProvider, messageLeftOverTracker);
consumerConfiguration.setMaxMessages(1);
final KafkaStream stream = mock(KafkaStream.class);
final List<KafkaStream<byte[], byte[]>> streams = new ArrayList<KafkaStream<byte[], byte[]>>();
streams.add(stream);
when(consumerConfiguration.createMessageStreamsForTopicFilter()).thenReturn(streams);
final ConsumerIterator iterator = mock(ConsumerIterator.class);
when(stream.iterator()).thenReturn(iterator);
final MessageAndMetadata messageAndMetadata = mock(MessageAndMetadata.class);
when(iterator.next()).thenReturn(messageAndMetadata);
when(messageAndMetadata.message()).thenReturn("got message");
when(messageAndMetadata.topic()).thenReturn("topic");
when(messageAndMetadata.partition()).thenReturn(1);
final Map<String, Map<Integer, List<Object>>> messages = consumerConfiguration.receive();
Assert.assertEquals(messages.size(), 1);
Assert.assertEquals(messages.get("topic").size(), 1);
Assert.assertEquals(messages.get("topic").get(1).get(0), "got message");
verify(stream, times(1)).iterator();
verify(iterator, times(1)).next();
verify(messageAndMetadata, times(1)).message();
verify(messageAndMetadata, times(1)).topic();
}
@Test
@SuppressWarnings("unchecked")
public void testReceiveMessageForTopicFilterFromMultipleStreams() {
final ConsumerMetadata consumerMetadata = mock(ConsumerMetadata.class);
final ConsumerConnectionProvider consumerConnectionProvider =
mock(ConsumerConnectionProvider.class);
final MessageLeftOverTracker messageLeftOverTracker = mock(MessageLeftOverTracker.class);
when(consumerMetadata.getTopicFilterConfiguration()).thenReturn(new TopicFilterConfiguration(".*", 1, false));
final ConsumerConnector consumerConnector = mock(ConsumerConnector.class);
when(consumerConnectionProvider.getConsumerConnector()).thenReturn(consumerConnector);
final ConsumerConfiguration consumerConfiguration = new ConsumerConfiguration(consumerMetadata,
consumerConnectionProvider, messageLeftOverTracker);
consumerConfiguration.setMaxMessages(3);
final KafkaStream stream1 = mock(KafkaStream.class);
final KafkaStream stream2 = mock(KafkaStream.class);
final KafkaStream stream3 = mock(KafkaStream.class);
final List<KafkaStream<byte[], byte[]>> streams = new ArrayList<KafkaStream<byte[], byte[]>>();
streams.add(stream1);
streams.add(stream2);
streams.add(stream3);
when(consumerConfiguration.createMessageStreamsForTopicFilter()).thenReturn(streams);
final ConsumerIterator iterator1 = mock(ConsumerIterator.class);
final ConsumerIterator iterator2 = mock(ConsumerIterator.class);
final ConsumerIterator iterator3 = mock(ConsumerIterator.class);
when(stream1.iterator()).thenReturn(iterator1);
when(stream2.iterator()).thenReturn(iterator2);
when(stream3.iterator()).thenReturn(iterator3);
final MessageAndMetadata messageAndMetadata1 = mock(MessageAndMetadata.class);
final MessageAndMetadata messageAndMetadata2 = mock(MessageAndMetadata.class);
final MessageAndMetadata messageAndMetadata3 = mock(MessageAndMetadata.class);
when(iterator1.next()).thenReturn(messageAndMetadata1);
when(iterator2.next()).thenReturn(messageAndMetadata2);
when(iterator3.next()).thenReturn(messageAndMetadata3);
when(messageAndMetadata1.message()).thenReturn("got message");
when(messageAndMetadata1.topic()).thenReturn("topic");
when(messageAndMetadata1.partition()).thenReturn(1);
when(messageAndMetadata2.message()).thenReturn("got message");
when(messageAndMetadata2.topic()).thenReturn("topic");
when(messageAndMetadata2.partition()).thenReturn(2);
when(messageAndMetadata3.message()).thenReturn("got message");
when(messageAndMetadata3.topic()).thenReturn("topic");
when(messageAndMetadata3.partition()).thenReturn(3);
final Map<String, Map<Integer, List<Object>>> messages = consumerConfiguration.receive();
Assert.assertEquals(messages.size(), 1);
int sum = 0;
final Map<Integer, List<Object>> values = messages.get("topic");
for (final List<Object> l : values.values()) {
sum += l.size();
}
Assert.assertEquals(sum, 3);
}
private boolean valueFound(final List<Object> l, final String value){
for (final Object o : l){
if (value.equals(o)){

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

@@ -0,0 +1,65 @@
/*
* Copyright 2002-2013 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.integration.kafka.support;
import java.util.HashMap;
import java.util.Map;
import org.junit.Assert;
import kafka.javaapi.producer.Producer;
import org.junit.Test;
import org.mockito.Mockito;
import org.springframework.beans.factory.ListableBeanFactory;
/**
* @author Rajasekar Elango
* @since 0.5
*/
public class KafkaProducerContextTests {
@SuppressWarnings({ "rawtypes", "unchecked" })
@Test
public void testTopicRegexForProducerConfiguration(){
final KafkaProducerContext kafkaProducerContext = new KafkaProducerContext();
final ListableBeanFactory beanFactory = Mockito.mock(ListableBeanFactory.class);
final ProducerMetadata<String, String> producerMetadata = Mockito.mock(ProducerMetadata.class);
String testRegex = "test.*";
Mockito.when(producerMetadata.getTopic()).thenReturn(testRegex);
final Producer<String, String> producer = Mockito.mock(Producer.class);
final ProducerConfiguration<String, String> producerConfiguration = new ProducerConfiguration<String, String>(producerMetadata, producer);
final Map<String, ProducerConfiguration> topicConfigurations = new HashMap<String, ProducerConfiguration>();
topicConfigurations.put(testRegex, producerConfiguration);
Mockito.when(beanFactory.getBeansOfType(ProducerConfiguration.class)).thenReturn(topicConfigurations);
kafkaProducerContext.setBeanFactory(beanFactory);
Assert.assertNotNull(kafkaProducerContext.getTopicConfiguration("test1"));
Assert.assertNotNull(kafkaProducerContext.getTopicConfiguration("test2"));
Assert.assertNotNull(kafkaProducerContext.getTopicConfiguration("testabc"));
Assert.assertNull(kafkaProducerContext.getTopicConfiguration("dontmatch_testRegex"));
}
}

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

@@ -46,7 +46,6 @@ public class User extends org.apache.avro.specific.SpecificRecordBase implements
}
// Used by DatumReader. Applications should not call.
@SuppressWarnings(value = "unchecked")
public void put(int field$, java.lang.Object value$) {
switch (field$) {
case 0:
@@ -92,5 +91,3 @@ public class User extends org.apache.avro.specific.SpecificRecordBase implements
this.lastName = value;
}
}