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

INTEXT-84 Kafka: Enhance Avro serialization support

Browse Source
This commit is contained in:
Soby Chacko
2013-07-16 22:40:28 -04:00
committed by Artem Bilan
parent af9440315c
commit 94734e4ca3
32 changed files with 471 additions and 290 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

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

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

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

6
spring-integration-kafka/src/main/java/org/springframework/integration/kafka/inbound/KafkaHighLevelConsumerMessageSource.java
View File

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

8
spring-integration-kafka/src/main/java/org/springframework/integration/kafka/outbound/KafkaProducerMessageHandler.java
View File

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

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

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

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

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

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

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

11
spring-integration-kafka/src/main/java/org/springframework/integration/kafka/serializer/avro/AvroSerializer.java
View File

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

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

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

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

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

53
spring-integration-kafka/src/main/java/org/springframework/integration/kafka/serializer/avro/AvroSpecificDatumSerializer.java
View File

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

2
spring-integration-kafka/src/main/java/org/springframework/integration/kafka/serializer/common/StringEncoder.java
View File

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

6
spring-integration-kafka/src/main/java/org/springframework/integration/kafka/support/ConsumerConfigFactoryBean.java
View File

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

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

@@ -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
spring-integration-kafka/src/main/java/org/springframework/integration/kafka/support/ConsumerMetadata.java
View File

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

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

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

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

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

8
spring-integration-kafka/src/main/java/org/springframework/integration/kafka/support/MessageLeftOverTracker.java
View File

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

18
spring-integration-kafka/src/test/java/org/springframework/integration/kafka/serializer/AvroBackedKafkaSerializerTest.java → spring-integration-kafka/src/test/java/org/springframework/integration/kafka/serializer/AvroReflectDatumBackedKafkaSerializerTest.java
View File

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

30
spring-integration-kafka/src/test/java/org/springframework/integration/kafka/serializer/AvroSpecificDatumBackedKafkaSerializerTest.java Normal file
View File

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

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

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

14
spring-integration-kafka/src/test/java/org/springframework/integration/kafka/support/KafkaConsumerContextTest.java
View File

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

50
spring-integration-kafka/src/test/java/org/springframework/integration/kafka/support/ProducerConfigurationTests.java
View File

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

6
spring-integration-kafka/src/test/java/org/springframework/integration/kafka/support/ProducerFactoryBeanTests.java
View File

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

96
spring-integration-kafka/src/test/java/org/springframework/integration/kafka/test/utils/User.java Normal file
View File

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