INTEXT-168 Upgrade to Kafka 0.8.2.1

JIRA: https://jira.spring.io/browse/INTEXT-168

- Replace producers with the new producer API
- Change KafkaProducerContext and ProducerConfiguration to match the new Producer's behaviour (i.e. providing partition as argument)
- support using the `partitionId` message header for targetting specific partitions;
- allow using a SpEL expression for partitioning;
- remove configurations that do not apply anymore, i.e. async (new producer is always async), batch size as message count, etc.
- add new configuration options wherever necessary;
- tweak schema - allow reuse of Encoders and Partitioners if available;

TODO: Add missing unit tests;

Fixes and polishing
This commit is contained in:
Marius Bogoevici
2015-05-01 16:59:24 -04:00
committed by Artem Bilan
parent fbe2975ed4
commit d8df11c2aa
50 changed files with 976 additions and 723 deletions

View File

@@ -60,6 +60,13 @@ public class KafkaOutboundChannelAdapterParser extends AbstractOutboundChannelAd
kafkaProducerMessageHandlerBuilder.addPropertyValue("messageKeyExpression", messageKeyExpressionDef);
}
BeanDefinition partitionIdExpressionDef =
IntegrationNamespaceUtils.createExpressionDefinitionFromValueOrExpression("partition-id",
"partition-id-expression", parserContext, element, false);
if (partitionIdExpressionDef != null) {
kafkaProducerMessageHandlerBuilder.addPropertyValue("partitionIdExpression", partitionIdExpressionDef);
}
return kafkaProducerMessageHandlerBuilder.getBeanDefinition();
}

View File

@@ -18,6 +18,8 @@ package org.springframework.integration.kafka.config.xml;
import java.util.Map;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.w3c.dom.Element;
import org.springframework.beans.BeanMetadataElement;
@@ -31,6 +33,8 @@ import org.springframework.integration.kafka.support.KafkaProducerContext;
import org.springframework.integration.kafka.support.ProducerConfiguration;
import org.springframework.integration.kafka.support.ProducerFactoryBean;
import org.springframework.integration.kafka.support.ProducerMetadata;
import org.springframework.integration.kafka.util.EncoderAdaptingSerializer;
import org.springframework.util.Assert;
import org.springframework.util.StringUtils;
import org.springframework.util.xml.DomUtils;
@@ -42,6 +46,8 @@ import org.springframework.util.xml.DomUtils;
*/
public class KafkaProducerContextParser extends AbstractSimpleBeanDefinitionParser {
private static final Log log = LogFactory.getLog(KafkaProducerContextParser.class);
@Override
protected Class<?> getBeanClass(final Element element) {
return KafkaProducerContext.class;
@@ -73,22 +79,54 @@ public class KafkaProducerContextParser extends AbstractSimpleBeanDefinitionPars
BeanDefinitionBuilder producerMetadataBuilder =
BeanDefinitionBuilder.genericBeanDefinition(ProducerMetadata.class);
producerMetadataBuilder.addConstructorArgValue(producerConfiguration.getAttribute("topic"));
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(producerMetadataBuilder, producerConfiguration,
"value-encoder");
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(producerMetadataBuilder, producerConfiguration,
"key-encoder");
IntegrationNamespaceUtils.setValueIfAttributeDefined(producerMetadataBuilder, producerConfiguration,
"key-class-type");
IntegrationNamespaceUtils.setValueIfAttributeDefined(producerMetadataBuilder, producerConfiguration,
"value-class-type");
producerMetadataBuilder.addConstructorArgValue(producerConfiguration.getAttribute("key-class-type"));
producerMetadataBuilder.addConstructorArgValue(producerConfiguration.getAttribute("value-class-type"));
String keySerializer = producerConfiguration.getAttribute("key-serializer");
String keyEncoder = producerConfiguration.getAttribute("key-encoder");
Assert.isTrue((StringUtils.hasText(keySerializer) ^ StringUtils.hasText(keyEncoder)),
"Exactly one of 'key-serializer' or 'key-encoder' must be specified");
if (StringUtils.hasText(keyEncoder)) {
if (log.isWarnEnabled()) {
log.warn("'key-encoder' is a deprecated option, use 'key-serializer' instead.");
}
BeanDefinitionBuilder encoderAdaptingSerializerBean = BeanDefinitionBuilder.genericBeanDefinition(EncoderAdaptingSerializer.class);
encoderAdaptingSerializerBean.addConstructorArgReference(keyEncoder);
producerMetadataBuilder.addConstructorArgValue(encoderAdaptingSerializerBean.getBeanDefinition());
}
else {
producerMetadataBuilder.addConstructorArgReference(keySerializer);
}
String valueSerializer = producerConfiguration.getAttribute("value-serializer");
String valueEncoder = producerConfiguration.getAttribute("value-encoder");
Assert.isTrue((StringUtils.hasText(valueSerializer) ^ StringUtils.hasText(valueEncoder)),
"Exactly one of 'value-serializer' or 'value-encoder' must be specified");
if (StringUtils.hasText(valueEncoder)) {
if (log.isWarnEnabled()) {
log.warn("'value-encoder' is a deprecated option, use 'value-serializer' instead.");
}
BeanDefinitionBuilder encoderAdaptingSerializerBean =
BeanDefinitionBuilder.genericBeanDefinition(EncoderAdaptingSerializer.class);
encoderAdaptingSerializerBean.addConstructorArgReference(valueEncoder);
producerMetadataBuilder.addConstructorArgValue(encoderAdaptingSerializerBean.getBeanDefinition());
}
else {
producerMetadataBuilder.addConstructorArgReference(valueSerializer);
}
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(producerMetadataBuilder, producerConfiguration,
"partitioner");
if (StringUtils.hasText(producerConfiguration.getAttribute("partitioner"))) {
if (log.isWarnEnabled()) {
log.warn("'partitioner' is a deprecated option. Use the 'kafka_partitionId' message header or " +
"the partition argument in the send() or convertAndSend() methods");
}
}
IntegrationNamespaceUtils.setValueIfAttributeDefined(producerMetadataBuilder, producerConfiguration,
"compression-codec");
"compression-type");
IntegrationNamespaceUtils.setValueIfAttributeDefined(producerMetadataBuilder, producerConfiguration,
"async");
IntegrationNamespaceUtils.setValueIfAttributeDefined(producerMetadataBuilder, producerConfiguration,
"batch-num-messages");
"batch-bytes");
AbstractBeanDefinition producerMetadataBeanDefinition = producerMetadataBuilder.getBeanDefinition();
String producerPropertiesBean = parentElem.getAttribute("producer-properties");
@@ -108,13 +146,14 @@ public class KafkaProducerContextParser extends AbstractSimpleBeanDefinitionPars
AbstractBeanDefinition producerFactoryBeanDefinition = producerFactoryBuilder.getBeanDefinition();
AbstractBeanDefinition producerConfigurationBeanDefinition =
BeanDefinitionBuilder producerConfigurationBuilder =
BeanDefinitionBuilder.genericBeanDefinition(ProducerConfiguration.class)
.addConstructorArgValue(producerMetadataBeanDefinition)
.addConstructorArgValue(producerFactoryBeanDefinition)
.getBeanDefinition();
.addConstructorArgValue(producerFactoryBeanDefinition);
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(producerConfigurationBuilder, producerConfiguration,
"conversion-service");
producerConfigurationsMap.put(producerConfiguration.getAttribute("topic"),
producerConfigurationBeanDefinition);
producerConfigurationBuilder.getBeanDefinition());
}
builder.addPropertyValue("producerConfigurations", producerConfigurationsMap);

View File

@@ -24,17 +24,11 @@ import java.util.Map;
import java.util.Random;
import java.util.concurrent.atomic.AtomicInteger;
import com.gs.collections.api.block.function.Function;
import com.gs.collections.api.block.function.Function2;
import com.gs.collections.api.tuple.Pair;
import com.gs.collections.impl.list.mutable.FastList;
import com.gs.collections.impl.tuple.Tuples;
import com.gs.collections.impl.utility.LazyIterate;
import com.gs.collections.impl.utility.MapIterate;
import kafka.api.FetchRequestBuilder;
import kafka.api.PartitionOffsetRequestInfo;
import kafka.cluster.Broker;
import kafka.common.ErrorMapping;
import kafka.common.OffsetAndMetadata;
import kafka.common.OffsetMetadataAndError;
import kafka.common.TopicAndPartition;
import kafka.javaapi.FetchResponse;
@@ -51,11 +45,19 @@ import kafka.javaapi.TopicMetadataResponse;
import kafka.javaapi.consumer.SimpleConsumer;
import kafka.javaapi.message.ByteBufferMessageSet;
import kafka.message.MessageAndOffset;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.util.Assert;
import com.gs.collections.api.block.function.Function;
import com.gs.collections.api.block.function.Function2;
import com.gs.collections.api.tuple.Pair;
import com.gs.collections.impl.list.mutable.FastList;
import com.gs.collections.impl.tuple.Tuples;
import com.gs.collections.impl.utility.LazyIterate;
import com.gs.collections.impl.utility.MapIterate;
/**
* A connection to a Kafka broker.
*
@@ -213,13 +215,13 @@ public class DefaultConnection implements Connection {
@Override
public Result<Void> commitOffsetsForConsumer(String consumerId, Map<Partition, Long> offsets)
throws ConsumerException {
Map<TopicAndPartition, OffsetMetadataAndError> requestInfo =
Map<TopicAndPartition, OffsetAndMetadata> requestInfo =
MapIterate.collect(offsets, new CreateRequestInfoMapEntryFunction());
OffsetCommitResponse offsetCommitResponse = null;
try {
offsetCommitResponse = simpleConsumer.commitOffsets(
new OffsetCommitRequest(consumerId, requestInfo, kafka.api.OffsetCommitRequest.CurrentVersion(),
createCorrelationId(), simpleConsumer.clientId()));
new OffsetCommitRequest(consumerId, requestInfo, createCorrelationId(),
simpleConsumer.clientId(), kafka.api.OffsetCommitRequest.CurrentVersion()));
}
catch (Exception e) {
throw new ConsumerException(e);
@@ -309,12 +311,12 @@ public class DefaultConnection implements Connection {
@SuppressWarnings("serial")
private static class CreateRequestInfoMapEntryFunction
implements Function2<Partition, Long, Pair<TopicAndPartition, OffsetMetadataAndError>> {
implements Function2<Partition, Long, Pair<TopicAndPartition, OffsetAndMetadata>> {
@Override
public Pair<TopicAndPartition, OffsetMetadataAndError> value(Partition partition, Long offset) {
public Pair<TopicAndPartition, OffsetAndMetadata> value(Partition partition, Long offset) {
return Tuples.pair(new TopicAndPartition(partition.getTopic(), partition.getId()),
new OffsetMetadataAndError(offset, OffsetMetadataAndError.NoMetadata(), ErrorMapping.NoError()));
new OffsetAndMetadata(offset, OffsetAndMetadata.NoMetadata(), ErrorMapping.NoError()));
}
}

View File

@@ -135,7 +135,11 @@ public abstract class AbstractOffsetManager implements OffsetManager, Disposable
Connection connection = this.connectionFactory.connect(leader);
Result<Long> offsetResult = connection.fetchInitialOffset(this.referenceTimestamp, partition);
if (offsetResult.getErrors().size() > 0) {
throw new ConsumerException(ErrorMapping.exceptionFor(offsetResult.getError(partition)));
short errorCode = offsetResult.getError(partition);
if (log.isWarnEnabled()) {
log.warn("Error code while retrieving offset for partition " + partition.toString() + " : " + errorCode);
}
throw new ConsumerException(ErrorMapping.exceptionFor(errorCode));
}
if (!offsetResult.getResults().containsKey(partition)) {
throw new IllegalStateException("Result does not contain an expected value");

View File

@@ -22,6 +22,9 @@ import java.util.List;
import java.util.concurrent.Executor;
import java.util.concurrent.Executors;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.integration.kafka.core.KafkaMessage;
import org.springframework.integration.kafka.core.Partition;
import org.springframework.scheduling.concurrent.CustomizableThreadFactory;
@@ -42,6 +45,8 @@ class ConcurrentMessageListenerDispatcher {
public static final CustomizableThreadFactory THREAD_FACTORY = new CustomizableThreadFactory("dispatcher-");
private static final Log log = LogFactory.getLog(ConcurrentMessageListenerDispatcher.class);
private static final StartDelegateProcedure startDelegateProcedure = new StartDelegateProcedure();
private static final StopDelegateProcedure stopDelegateProcedure = new StopDelegateProcedure();
@@ -67,7 +72,7 @@ class ConcurrentMessageListenerDispatcher {
private Executor taskExecutor;
public ConcurrentMessageListenerDispatcher(Object delegateListener, ErrorHandler errorHandler,
Collection<Partition> partitions, OffsetManager offsetManager, int consumers, int queueSize) {
Collection<Partition> partitions, OffsetManager offsetManager, int consumers, int queueSize, Executor taskExecutor) {
Assert.isTrue
(delegateListener instanceof MessageListener
|| delegateListener instanceof AcknowledgingMessageListener,
@@ -83,6 +88,7 @@ class ConcurrentMessageListenerDispatcher {
this.offsetManager = offsetManager;
this.consumers = consumers;
this.queueSize = queueSize;
this.taskExecutor = taskExecutor;
}
public void start() {
@@ -100,6 +106,7 @@ class ConcurrentMessageListenerDispatcher {
this.running = false;
delegates.flip().keyBag().toSet().forEachWith(stopDelegateProcedure, stopTimeout);
}
this.delegates = null;
}
}
@@ -136,7 +143,14 @@ class ConcurrentMessageListenerDispatcher {
@Override
public void value(QueueingMessageListenerInvoker delegate, Integer stopTimeout) {
delegate.stop(stopTimeout);
try {
delegate.stop(stopTimeout);
} catch (Exception e) {
// ignore the exception, but log it
if(log.isInfoEnabled()) {
log.info("Exception thrown while stopping dispatcher:", e);
}
}
}
}

View File

@@ -100,6 +100,8 @@ public class KafkaMessageListenerContainer implements SmartLifecycle {
private Executor adminTaskExecutor = Executors.newSingleThreadExecutor();
private Executor dispatcherTaskExecutor;
private int concurrency = 1;
private volatile boolean running = false;
@@ -206,17 +208,27 @@ public class KafkaMessageListenerContainer implements SmartLifecycle {
this.fetchTaskExecutor = fetchTaskExecutor;
}
/**
* @return the task executor for leader and offset updates.
*/
public Executor getAdminTaskExecutor() {
return adminTaskExecutor;
}
/**
* The task executor for leader and offset updates
* @param adminTaskExecutor the task executor for leader and offset updates.
*/
public void setAdminTaskExecutor(Executor adminTaskExecutor) {
this.adminTaskExecutor = adminTaskExecutor;
}
/**
* The task executor for invoking the MessageListener
* @param dispatcherTaskExecutor the task executor for invoking the MessageListener
*/
public void setDispatcherTaskExecutor(Executor dispatcherTaskExecutor) {
this.dispatcherTaskExecutor = dispatcherTaskExecutor;
}
/**
* @return the maximum amount of data (in bytes) that pollers will fetch in one round
*/
@@ -284,7 +296,7 @@ public class KafkaMessageListenerContainer implements SmartLifecycle {
ImmutableList<Partition> partitionsAsList = Lists.immutable.with(partitions);
this.fetchOffsets = new ConcurrentHashMap<Partition, Long>(partitionsAsList.toMap(passThru, getOffset));
this.messageDispatcher = new ConcurrentMessageListenerDispatcher(messageListener, errorHandler,
Arrays.asList(partitions), offsetManager, concurrency, queueSize);
Arrays.asList(partitions), offsetManager, concurrency, queueSize, dispatcherTaskExecutor);
this.messageDispatcher.start();
partitionsByBrokerMap.clear();
partitionsByBrokerMap.putAll(partitionsAsList.groupBy(getLeader));

View File

@@ -25,10 +25,20 @@ import java.util.Properties;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import kafka.admin.AdminUtils$;
import kafka.api.OffsetRequest;
import kafka.common.ErrorMapping$;
import kafka.common.TopicExistsException;
import kafka.serializer.Decoder;
import kafka.utils.ZKStringSerializer$;
import org.I0Itec.zkclient.ZkClient;
import org.I0Itec.zkclient.exception.ZkInterruptedException;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.Serializer;
import org.springframework.beans.factory.BeanInitializationException;
import org.springframework.beans.factory.InitializingBean;
@@ -55,17 +65,6 @@ import org.springframework.retry.support.RetryTemplate;
import org.springframework.util.Assert;
import org.springframework.util.ObjectUtils;
import kafka.admin.AdminUtils$;
import kafka.api.OffsetRequest;
import kafka.common.ErrorMapping$;
import kafka.common.TopicExistsException;
import kafka.javaapi.producer.Producer;
import kafka.producer.DefaultPartitioner;
import kafka.producer.KeyedMessage;
import kafka.serializer.Decoder;
import kafka.serializer.Encoder;
import kafka.utils.ZKStringSerializer$;
/**
* Implementation of an {@link OffsetManager} that uses a Kafka topic as the underlying support.
* For its proper functioning, the Kafka server(s) must set {@code log.cleaner.enable=true}. It relies on the property
@@ -77,9 +76,9 @@ import kafka.utils.ZKStringSerializer$;
*/
public class KafkaTopicOffsetManager extends AbstractOffsetManager implements InitializingBean {
private static final KeyEncoderDecoder KEY_CODEC = new KeyEncoderDecoder();
private static final KeySerializerDecoder KEY_CODEC = new KeySerializerDecoder();
private static final LongEncoderDecoder VALUE_CODEC = new LongEncoderDecoder();
private static final LongSerializerDecoder VALUE_CODEC = new LongSerializerDecoder();
public static final String CLEANUP_POLICY = "cleanup.policy";
@@ -97,26 +96,24 @@ public class KafkaTopicOffsetManager extends AbstractOffsetManager implements In
private final ConcurrentMap<Partition, Long> data = new ConcurrentHashMap<Partition, Long>();
private String compressionCodec = "default";
private ProducerMetadata.CompressionType compressionType = ProducerMetadata.CompressionType.none;
private Producer<Key, Long> producer;
private int maxSize = 10 * 1024;
private int maxQueueBufferingTime = 1000;
private int segmentSize = 25 * 1024;
private int retentionTime = 60000;
private int replicationFactor;
private int maxBatchSize = 200;
private boolean batchWrites = true;
private int batchBytes = 200;
private int requiredAcks = 1;
private int maxQueueBufferingTime;
public KafkaTopicOffsetManager(ZookeeperConnect zookeeperConnect, String topic) {
this(zookeeperConnect, topic, new HashMap<Partition, Long>());
}
@@ -140,12 +137,12 @@ public class KafkaTopicOffsetManager extends AbstractOffsetManager implements In
}
/**
* The compression codec for writing to the offset topic
* The compression type for writing to the offset topic
*
* @param compressionCodec the compression codec
* @param compressionType the compression type
*/
public void setCompressionCodec(String compressionCodec) {
this.compressionCodec = compressionCodec;
public void setCompressionCodec(ProducerMetadata.CompressionType compressionType) {
this.compressionType = compressionType;
}
/**
@@ -185,21 +182,12 @@ public class KafkaTopicOffsetManager extends AbstractOffsetManager implements In
}
/**
* The maximum batch size for offset writes
* The maximum batch size in bytes for offset writes
*
* @param maxBatchSize maximum batching window
* @param batchBytes maximum batching window
*/
public void setMaxBatchSize(int maxBatchSize) {
this.maxBatchSize = maxBatchSize;
}
/**
* Whether offset writes should be batched or not
*
* @param batchWrites true if writes are batched
*/
public void setBatchWrites(boolean batchWrites) {
this.batchWrites = batchWrites;
public void setBatchBytes(int batchBytes) {
this.batchBytes = batchBytes;
}
/**
@@ -218,7 +206,6 @@ public class KafkaTopicOffsetManager extends AbstractOffsetManager implements In
Integer.parseInt(this.zookeeperConnect.getZkSessionTimeout()),
Integer.parseInt(this.zookeeperConnect.getZkConnectionTimeout()),
ZKStringSerializer$.MODULE$);
try {
createCompactedTopicIfNotFound(zkClient);
validateOffsetTopic(zkClient);
@@ -240,13 +227,13 @@ public class KafkaTopicOffsetManager extends AbstractOffsetManager implements In
@Override
protected void doUpdateOffset(Partition partition, long offset) {
this.data.put(partition, offset);
this.producer.send(new KeyedMessage<Key, Long>(this.topic, new Key(this.consumerId, partition), offset));
this.producer.send(new ProducerRecord<>(this.topic, new Key(this.consumerId, partition), offset));
}
@Override
protected void doRemoveOffset(Partition partition) {
this.data.remove(partition);
this.producer.send(new KeyedMessage<Key, Long>(this.topic, new Key(this.consumerId, partition), null));
this.producer.send(new ProducerRecord<Key, Long>(this.topic, new Key(this.consumerId, partition), null));
}
@Override
@@ -385,28 +372,16 @@ public class KafkaTopicOffsetManager extends AbstractOffsetManager implements In
private void initializeProducer(BrokerAddress leader) throws Exception {
ProducerMetadata<Key, Long> producerMetadata = new ProducerMetadata<Key, Long>(this.topic);
producerMetadata.setValueEncoder(VALUE_CODEC);
producerMetadata.setValueClassType(Long.class);
producerMetadata.setKeyEncoder(KEY_CODEC);
producerMetadata.setKeyClassType(Key.class);
producerMetadata.setPartitioner(new DefaultPartitioner(null));
ProducerMetadata<Key,Long> producerMetadata = new ProducerMetadata<>(this.topic, Key.class, Long.class,
KEY_CODEC, VALUE_CODEC);
producerMetadata.setBatchBytes(batchBytes);
producerMetadata.setCompressionType(compressionType);
Properties additionalProps = new Properties();
producerMetadata.setAsync(this.batchWrites);
if (this.batchWrites) {
producerMetadata.setBatchNumMessages(String.valueOf(this.maxBatchSize));
producerMetadata.setCompressionCodec(this.compressionCodec);
additionalProps.put("request.required.acks", String.valueOf(this.requiredAcks));
additionalProps.put("queue.buffering.max.ms", String.valueOf(this.maxQueueBufferingTime));
}
ProducerFactoryBean<Key, Long> producerFB
= new ProducerFactoryBean<Key, Long>(producerMetadata, leader.toString(), additionalProps);
this.producer = producerFB.getObject();
additionalProps.setProperty(ProducerConfig.LINGER_MS_CONFIG, Integer.toString(maxQueueBufferingTime));
additionalProps.setProperty(ProducerConfig.ACKS_CONFIG, Integer.toString(requiredAcks));
ProducerFactoryBean<Key,Long> producerFactoryBean = new ProducerFactoryBean<>(producerMetadata,
leader.toString(), additionalProps);
this.producer = producerFactoryBean.getObject();
}
private static int intFromBytes(byte[] bytes, int start) {
@@ -468,9 +443,9 @@ public class KafkaTopicOffsetManager extends AbstractOffsetManager implements In
}
public static class KeyEncoderDecoder implements Encoder<Key>, Decoder<Key> {
public static class KeySerializerDecoder implements Serializer<Key>, Decoder<Key> {
private static final Log log = LogFactory.getLog(KeyEncoderDecoder.class);
private static final Log log = LogFactory.getLog(KeySerializerDecoder.class);
@Override
public Key fromBytes(byte[] bytes) {
@@ -495,14 +470,19 @@ public class KafkaTopicOffsetManager extends AbstractOffsetManager implements In
}
@Override
public byte[] toBytes(Key key) {
if (key == null) {
public void configure(Map<String, ?> configs, boolean isKey) {
// no-op
}
@Override
public byte[] serialize(String topic, Key data) {
if (data == null) {
return null;
}
try {
byte[] consumerIdBytes = key.consumerId.getBytes("UTF-8");
byte[] topicNameBytes = key.partition.getTopic().getBytes("UTF-8");
byte[] partitionIdBytes = intToBytes(key.partition.getId());
byte[] consumerIdBytes = data.consumerId.getBytes("UTF-8");
byte[] topicNameBytes = data.partition.getTopic().getBytes("UTF-8");
byte[] partitionIdBytes = intToBytes(data.partition.getId());
byte[] result = new byte[4 + consumerIdBytes.length + 4 + topicNameBytes.length + 4];
System.arraycopy(intToBytes(consumerIdBytes.length), 0, result, 0, 4);
System.arraycopy(consumerIdBytes, 0, result, 4, consumerIdBytes.length);
@@ -516,6 +496,11 @@ public class KafkaTopicOffsetManager extends AbstractOffsetManager implements In
}
}
@Override
public void close() {
//no-op
}
}
}

View File

@@ -17,23 +17,24 @@
package org.springframework.integration.kafka.listener;
import java.nio.ByteBuffer;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.integration.kafka.util.LoggingUtils;
import java.util.Map;
import kafka.serializer.Decoder;
import kafka.serializer.Encoder;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.kafka.common.serialization.Serializer;
import org.springframework.integration.kafka.util.LoggingUtils;
/**
* Kafka {@link Encoder} and {@link Decoder} for Long values.
*
* @author Marius Bogoevici
*/
public class LongEncoderDecoder implements Encoder<Long>, Decoder<Long> {
public class LongSerializerDecoder implements Serializer<Long>, Decoder<Long> {
private Log log = LogFactory.getLog(LongEncoderDecoder.class);
private Log log = LogFactory.getLog(LongSerializerDecoder.class);
@Override
public Long fromBytes(byte[] bytes) {
@@ -54,13 +55,22 @@ public class LongEncoderDecoder implements Encoder<Long>, Decoder<Long> {
}
@Override
public byte[] toBytes(Long value) {
if (value == null) {
public void configure(Map<String, ?> configs, boolean isKey) {
// no-op
}
@Override
public byte[] serialize(String topic, Long data) {
if (data == null) {
return null;
}
else {
return ByteBuffer.allocate(8).putLong(value).array();
return ByteBuffer.allocate(8).putLong(data).array();
}
}
@Override
public void close() {
// no-op
}
}

View File

@@ -31,10 +31,10 @@ import org.springframework.util.Assert;
* @author Gary Russell
* @since 0.5
*/
public class KafkaProducerMessageHandler<K,V> extends AbstractMessageHandler
public class KafkaProducerMessageHandler extends AbstractMessageHandler
implements IntegrationEvaluationContextAware {
private final KafkaProducerContext<K,V> kafkaProducerContext;
private final KafkaProducerContext kafkaProducerContext;
private EvaluationContext evaluationContext;
@@ -42,7 +42,10 @@ public class KafkaProducerMessageHandler<K,V> extends AbstractMessageHandler
private volatile Expression messageKeyExpression;
public KafkaProducerMessageHandler(final KafkaProducerContext<K,V> kafkaProducerContext) {
private volatile Expression partitionExpression;
@SuppressWarnings("unchecked")
public KafkaProducerMessageHandler(final KafkaProducerContext kafkaProducerContext) {
this.kafkaProducerContext = kafkaProducerContext;
}
@@ -59,7 +62,11 @@ public class KafkaProducerMessageHandler<K,V> extends AbstractMessageHandler
this.messageKeyExpression = messageKeyExpression;
}
public KafkaProducerContext<K,V> getKafkaProducerContext() {
public void setPartitionExpression(Expression partitionExpression) {
this.partitionExpression = partitionExpression;
}
public KafkaProducerContext getKafkaProducerContext() {
return this.kafkaProducerContext;
}
@@ -70,15 +77,19 @@ public class KafkaProducerMessageHandler<K,V> extends AbstractMessageHandler
@Override
protected void handleMessageInternal(final Message<?> message) throws Exception {
String topic = this.topicExpression != null
? this.topicExpression.getValue(this.evaluationContext, message, String.class)
String topic = this.topicExpression != null ?
this.topicExpression.getValue(this.evaluationContext, message, String.class)
: message.getHeaders().get(KafkaHeaders.TOPIC, String.class);
Integer partitionId = this.partitionExpression != null ?
this.partitionExpression.getValue(this.evaluationContext, message, Integer.class)
: message.getHeaders().get(KafkaHeaders.PARTITION_ID, Integer.class);
Object messageKey = this.messageKeyExpression != null
? this.messageKeyExpression.getValue(this.evaluationContext, message)
: message.getHeaders().get(KafkaHeaders.MESSAGE_KEY);
this.kafkaProducerContext.send(topic, messageKey, message);
this.kafkaProducerContext.send(topic, partitionId, messageKey, message.getPayload());
}
@Override

View File

@@ -15,28 +15,33 @@
*/
package org.springframework.integration.kafka.serializer.common;
import java.util.Properties;
import kafka.serializer.Encoder;
import kafka.utils.VerifiableProperties;
import java.util.Properties;
/**
* @author Soby Chacko
* @author Marius Bogoevici
* @since 0.5
*/
public class StringEncoder<T> implements Encoder<T> {
private String encoding = "UTF8";
public class StringEncoder implements Encoder<String> {
public void setEncoding(final String encoding){
this.encoding = encoding;
private kafka.serializer.StringEncoder stringEncoder;
public StringEncoder() {
this("UTF-8");
}
public StringEncoder(String encoding) {
final Properties props = new Properties();
props.put("serializer.encoding", encoding);
final VerifiableProperties verifiableProperties = new VerifiableProperties(props);
stringEncoder = new kafka.serializer.StringEncoder(verifiableProperties);
}
@Override
public byte[] toBytes(final Object o) {
final Properties props = new Properties();
props.put("serializer.encoding", encoding);
final VerifiableProperties verifiableProperties = new VerifiableProperties(props);
return new kafka.serializer.StringEncoder(verifiableProperties).toBytes((String)o);
public byte[] toBytes(final String value) {
return stringEncoder.toBytes(value);
}
}

View File

@@ -18,10 +18,13 @@ package org.springframework.integration.kafka.support;
import java.util.Collection;
import java.util.Map;
import java.util.concurrent.Future;
import java.util.concurrent.atomic.AtomicBoolean;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.springframework.beans.factory.BeanNameAware;
import org.springframework.context.SmartLifecycle;
@@ -38,15 +41,15 @@ import org.springframework.util.StringUtils;
* @author Marius Bogoevici
* @since 0.5
*/
public class KafkaProducerContext<K, V> implements SmartLifecycle, NamedComponent, BeanNameAware {
public class KafkaProducerContext implements SmartLifecycle, NamedComponent, BeanNameAware {
private static final Log logger = LogFactory.getLog(KafkaProducerContext.class);
private final AtomicBoolean running = new AtomicBoolean();
private volatile Map<String, ProducerConfiguration<K, V>> producerConfigurations;
private volatile Map<String, ProducerConfiguration<?, ?>> producerConfigurations;
private volatile ProducerConfiguration<K, V> theProducerConfiguration;
private volatile ProducerConfiguration<?,?> theProducerConfiguration;
private String beanName = "not_specified";
@@ -54,16 +57,16 @@ public class KafkaProducerContext<K, V> implements SmartLifecycle, NamedComponen
private boolean autoStartup = true;
public ProducerConfiguration<K, V> getTopicConfiguration(final String topic) {
public ProducerConfiguration<?, ?> getTopicConfiguration(final String topic) {
if (this.theProducerConfiguration != null) {
if (topic.matches(this.theProducerConfiguration.getProducerMetadata().getTopic())) {
return this.theProducerConfiguration;
}
}
Collection<ProducerConfiguration<K, V>> topics = this.producerConfigurations.values();
Collection<ProducerConfiguration<?, ?>> topics = this.producerConfigurations.values();
for (final ProducerConfiguration<K, V> producerConfiguration : topics) {
for (final ProducerConfiguration<?, ?> producerConfiguration : topics) {
if (topic.matches(producerConfiguration.getProducerMetadata().getTopic())) {
return producerConfiguration;
}
@@ -71,11 +74,11 @@ public class KafkaProducerContext<K, V> implements SmartLifecycle, NamedComponen
return null;
}
public Map<String, ProducerConfiguration<K, V>> getProducerConfigurations() {
public Map<String, ProducerConfiguration<?, ?>> getProducerConfigurations() {
return this.producerConfigurations;
}
public void setProducerConfigurations(Map<String, ProducerConfiguration<K, V>> producerConfigurations) {
public void setProducerConfigurations(Map<String, ProducerConfiguration<?, ?>> producerConfigurations) {
this.producerConfigurations = producerConfigurations;
if (this.producerConfigurations.size() == 1) {
this.theProducerConfiguration = this.producerConfigurations.values().iterator().next();
@@ -183,7 +186,11 @@ public class KafkaProducerContext<K, V> implements SmartLifecycle, NamedComponen
callback.run();
}
public void send(String topic, Object messageKey, final Message<?> message) throws Exception {
public Future<RecordMetadata> send(String topic, Object messageKey, Object messagePayload) {
return this.send(topic, null, messageKey, messagePayload);
}
public Future<RecordMetadata> send(String topic, Integer partition, Object messageKey, Object messagePayload) {
if (!running.get()) {
start();
}
@@ -191,14 +198,14 @@ public class KafkaProducerContext<K, V> implements SmartLifecycle, NamedComponen
// only try to look up for a producer configuration if the topic is passed as argument
// if no topic is configured, then we'll fall back to the default if a single
// producer configuration is available
ProducerConfiguration<K, V> producerConfiguration = StringUtils.hasText(topic) ? getTopicConfiguration(topic) : null;
ProducerConfiguration<?, ?> producerConfiguration = StringUtils.hasText(topic) ? getTopicConfiguration(topic) : null;
if (producerConfiguration != null) {
producerConfiguration.send(topic, messageKey, message);
return producerConfiguration.convertAndSend(topic, partition, messageKey, messagePayload);
}
// if there is a single producer configuration then use that config to send message.
else if (this.theProducerConfiguration != null) {
this.theProducerConfiguration.send(topic, messageKey, message);
return this.theProducerConfiguration.convertAndSend(topic, partition, messageKey, messagePayload);
}
else {
throw new IllegalStateException("Could not send messages as there are multiple producer configurations " +

View File

@@ -16,27 +16,26 @@
package org.springframework.integration.kafka.support;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.io.ObjectOutputStream;
import java.util.concurrent.Future;
import org.apache.commons.lang.builder.EqualsBuilder;
import org.apache.commons.lang.builder.HashCodeBuilder;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHandlingException;
import org.springframework.core.convert.ConversionService;
import org.springframework.core.convert.support.GenericConversionService;
import org.springframework.core.serializer.support.SerializingConverter;
import org.springframework.util.Assert;
import org.springframework.util.StringUtils;
import kafka.javaapi.producer.Producer;
import kafka.producer.KeyedMessage;
import kafka.serializer.DefaultEncoder;
/**
* @author Soby Chacko
* @author Rajasekar Elango
* @author Ilayaperumal Gopinathan
* @author Gary Russell
* @author Marius Bogoevici
* @since 0.5
*/
public class ProducerConfiguration<K, V> {
@@ -45,63 +44,76 @@ public class ProducerConfiguration<K, V> {
private final ProducerMetadata<K, V> producerMetadata;
private ConversionService conversionService;
public ProducerConfiguration(ProducerMetadata<K, V> producerMetadata, Producer<K, V> producer) {
Assert.notNull(producerMetadata);
Assert.notNull(producer);
this.producerMetadata = producerMetadata;
this.producer = producer;
GenericConversionService genericConversionService = new GenericConversionService();
genericConversionService.addConverter(Object.class, byte[].class, new SerializingConverter());
conversionService = genericConversionService;
}
public void setConversionService(ConversionService conversionService) {
Assert.notNull(conversionService, "Conversion service must not be null");
this.conversionService = conversionService;
}
public ProducerMetadata<K, V> getProducerMetadata() {
return this.producerMetadata;
}
public Producer<K, V> getProducer() {
return this.producer;
public Future<RecordMetadata> send(String topic, K messageKey, V messagePayload) {
if (this.getProducerMetadata().getPartitioner() != null) {
String targetTopic = StringUtils.hasText(topic) ? topic : this.producerMetadata.getTopic();
int partition = this.getProducerMetadata().getPartitioner().partition(messageKey,
this.producer.partitionsFor(targetTopic).size());
return this.send(targetTopic, partition, messageKey, messagePayload);
}
return this.send(topic, null, messageKey, messagePayload);
}
public void send(String topic, Object messageKey, final Message<?> message) throws Exception {
final V v = getPayload(message);
if (!StringUtils.hasText(topic)) {
topic = this.producerMetadata.getTopic();
}
this.producer.send(new KeyedMessage<K, V>(topic, (messageKey != null ? getKey(messageKey) : null), v));
public Future<RecordMetadata> send(String topic, Integer partition, K messageKey, V messagePayload) {
String targetTopic = StringUtils.hasText(topic) ? topic : this.producerMetadata.getTopic();
return this.producer.send(new ProducerRecord<>(targetTopic, partition, messageKey, messagePayload));
}
@SuppressWarnings("unchecked")
private V getPayload(final Message<?> message) throws Exception {
if (this.producerMetadata.getValueEncoder() instanceof DefaultEncoder) {
return (V) getByteStream(message.getPayload());
}
else if (producerMetadata.getValueClassType().isAssignableFrom(message.getPayload().getClass())) {
return producerMetadata.getValueClassType().cast(message.getPayload());
}
throw new MessageHandlingException(message, "Message payload type is not matching with what is configured");
public Future<RecordMetadata> convertAndSend(String topic, Integer partition, Object messageKey, Object messagePayload) {
return this.send(topic, partition, convertKeyIfNecessary(messageKey), convertPayloadIfNecessary(messagePayload));
}
@SuppressWarnings("unchecked")
private K getKey(Object messageKey) throws Exception {
if (this.producerMetadata.getKeyEncoder() instanceof DefaultEncoder) {
return (K) getByteStream(messageKey);
}
return (K) messageKey;
public Future<RecordMetadata> convertAndSend(String topic, Object messageKey, Object messagePayload) {
return this.send(topic, convertKeyIfNecessary(messageKey), convertPayloadIfNecessary(messagePayload));
}
private static boolean isRawByteArray(final Object obj) {
return obj instanceof byte[];
private K convertKeyIfNecessary(Object messageKey) {
if (messageKey != null) {
if (getProducerMetadata().getKeyClassType().isAssignableFrom(
messageKey.getClass())) {
return getProducerMetadata().getKeyClassType().cast(messageKey);
}
return conversionService.convert(messageKey,
producerMetadata.getKeyClassType());
}
else {
return null;
}
}
private static byte[] getByteStream(final Object obj) throws IOException {
if (isRawByteArray(obj)) {
return (byte[]) obj;
private V convertPayloadIfNecessary(Object messagePayload) {
if (messagePayload != null) {
if (getProducerMetadata().getKeyClassType().isAssignableFrom(
messagePayload.getClass())) {
return getProducerMetadata().getValueClassType().cast(messagePayload);
}
return conversionService.convert(messagePayload,
producerMetadata.getValueClassType());
}
else {
return null;
}
final ByteArrayOutputStream out = new ByteArrayOutputStream();
final ObjectOutputStream os = new ObjectOutputStream(out);
os.writeObject(obj);
return out.toByteArray();
}
@Override

View File

@@ -15,69 +15,55 @@
*/
package org.springframework.integration.kafka.support;
import kafka.javaapi.producer.Producer;
import kafka.producer.ProducerConfig;
import kafka.producer.ProducerPool;
import kafka.producer.async.DefaultEventHandler;
import kafka.producer.async.EventHandler;
import java.util.Properties;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.springframework.beans.factory.FactoryBean;
import scala.collection.mutable.HashMap;
import java.util.Properties;
/**
* @author Soby Chacko
* @author Marius Bogoevici
*
* @since 0.5
*/
public class ProducerFactoryBean<K,V> implements FactoryBean<Producer<K,V>> {
public class ProducerFactoryBean<K, V> implements FactoryBean<Producer<K, V>> {
private static final Log LOGGER = LogFactory.getLog(ProducerFactoryBean.class);
private static final Log LOGGER = LogFactory.getLog(ProducerFactoryBean.class);
private final String brokerList;
private final ProducerMetadata<K,V> producerMetadata;
private Properties producerProperties = new Properties();
public ProducerFactoryBean(final ProducerMetadata<K, V> producerMetadata, final String brokerList,
final Properties producerProperties) {
private final ProducerMetadata<K, V> producerMetadata;
private Properties producerProperties = new Properties();
public ProducerFactoryBean(final ProducerMetadata<K, V> producerMetadata, final String brokerList,
final Properties producerProperties) {
this.producerMetadata = producerMetadata;
this.brokerList = brokerList;
if (producerProperties != null) {
this.producerProperties = producerProperties;
}
}
}
public ProducerFactoryBean(final ProducerMetadata<K, V> producerMetadata, final String brokerList) {
this(producerMetadata, brokerList, null);
public ProducerFactoryBean(final ProducerMetadata<K, V> producerMetadata, final String brokerList) {
this(producerMetadata, brokerList, null);
}
@Override
public Producer<K, V> getObject() throws Exception {
final Properties props = new Properties();
props.putAll(producerProperties);
props.put("metadata.broker.list", brokerList);
props.put("compression.codec", producerMetadata.getCompressionCodec());
if (producerMetadata.isAsync()){
props.put("producer.type", "async");
if (producerMetadata.getBatchNumMessages() != null){
props.put("batch.num.messages", producerMetadata.getBatchNumMessages());
}
}
LOGGER.info("Using producer properties => " + props);
final ProducerConfig config = new ProducerConfig(props);
final EventHandler<K, V> eventHandler = new DefaultEventHandler<K, V>(config,
producerMetadata.getPartitioner() == null ? new DefaultPartitioner() : producerMetadata.getPartitioner(),
producerMetadata.getValueEncoder(), producerMetadata.getKeyEncoder(),
new ProducerPool(config), new HashMap<String, kafka.api.TopicMetadata>());
final kafka.producer.Producer<K, V> prod = new kafka.producer.Producer<K, V>(config,
eventHandler);
return new Producer<K, V>(prod);
props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, brokerList);
props.put(ProducerConfig.COMPRESSION_TYPE_CONFIG, producerMetadata.getCompressionType().name());
LOGGER.info("Using producer properties => " + props);
return new KafkaProducer<>(props,
producerMetadata.getKeySerializer(),
producerMetadata.getValueSerializer());
}
@Override

View File

@@ -16,118 +16,95 @@
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;
import org.apache.kafka.common.serialization.Serializer;
import org.springframework.util.Assert;
/**
*
* @author Soby Chacko
* @author Rajasekar Elango
* @author Marius Bogoevici
*
* @since 0.5
*/
public class ProducerMetadata<K,V> implements InitializingBean {
private Encoder<K> keyEncoder;
private Encoder<V> valueEncoder;
private Class<K> keyClassType;
private Class<V> valueClassType;
private final String topic;
private String compressionCodec = "default";
private Partitioner partitioner;
private boolean async = false;
private String batchNumMessages;
public class ProducerMetadata<K,V> {
public ProducerMetadata(final String topic) {
private final Class<K> keyClassType;
private final Class<V> valueClassType;
private Serializer<K> keySerializer;
private Serializer<V> valueSerializer;
private final String topic;
private Partitioner partitioner;
private CompressionType compressionType = CompressionType.none;
private int batchBytes = 16384;
public ProducerMetadata(final String topic, Class<K> keyClassType, Class<V> valueClassType, Serializer<K> keySerializer, Serializer<V> valueSerializer) {
Assert.notNull(topic, "Topic cannot be null");
Assert.notNull(keyClassType, "Key class type serializer cannot be null");
Assert.notNull(valueClassType, "Value class type cannot be null");
Assert.notNull(keySerializer, "Value serializer cannot be null");
Assert.notNull(valueSerializer, "Value serializer cannot be null");
this.topic = topic;
this.keyClassType = keyClassType;
this.valueClassType = valueClassType;
this.keySerializer = keySerializer;
this.valueSerializer = valueSerializer;
}
public String getTopic() {
return topic;
}
public Encoder<K> getKeyEncoder() {
return keyEncoder;
public Serializer<K> getKeySerializer() {
return keySerializer;
}
public void setKeyEncoder(final Encoder<K> keyEncoder) {
this.keyEncoder = keyEncoder;
public Serializer<V> getValueSerializer() {
return valueSerializer;
}
public Encoder<V> getValueEncoder() {
return valueEncoder;
public CompressionType getCompressionType() {
return compressionType;
}
public void setValueEncoder(final Encoder<V> valueEncoder) {
this.valueEncoder = valueEncoder;
public void setCompressionType(CompressionType compressionType) {
Assert.notNull(compressionType, "Compression type cannot be null");
this.compressionType = compressionType;
}
public Class<K> getKeyClassType() {
return keyClassType;
public int getBatchBytes() {
return batchBytes;
}
public void setKeyClassType(final Class<K> keyClassType) {
this.keyClassType = keyClassType;
}
public Class<V> getValueClassType() {
return valueClassType;
}
public void setValueClassType(final Class<V> valueClassType) {
this.valueClassType = valueClassType;
}
//TODO: Use an enum
public String getCompressionCodec() {
if (compressionCodec.equalsIgnoreCase("gzip")) {
return "1";
} else if (compressionCodec.equalsIgnoreCase("snappy")) {
return "2";
}
return "0";
}
public void setCompressionCodec(final String compressionCodec) {
this.compressionCodec = compressionCodec;
public void setBatchBytes(int batchBytes) {
Assert.isTrue(batchBytes > 0, "Buffer size must be greater than zero");
this.batchBytes = batchBytes;
}
public Partitioner getPartitioner() {
return partitioner;
}
public void setPartitioner(final Partitioner partitioner) {
public void setPartitioner(Partitioner partitioner) {
this.partitioner = partitioner;
}
@Override
@SuppressWarnings("unchecked")
public void afterPropertiesSet() throws Exception {
if (valueEncoder == null) {
setValueEncoder((Encoder<V>) new DefaultEncoder(null));
}
if (keyEncoder == null) {
setKeyEncoder((Encoder<K>) getValueEncoder());
}
public Class<K> getKeyClassType() {
return keyClassType;
}
public boolean isAsync() {
return async;
}
public void setAsync(final boolean async) {
this.async = async;
}
public String getBatchNumMessages() {
return batchNumMessages;
}
public void setBatchNumMessages(final String batchNumMessages) {
this.batchNumMessages = batchNumMessages;
public Class<V> getValueClassType() {
return valueClassType;
}
@Override
@@ -143,10 +120,17 @@ public class ProducerMetadata<K,V> implements InitializingBean {
@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("]");
builder.append("ProducerMetadata [keyEncoder=").append(keySerializer)
.append(", valueEncoder=").append(valueSerializer)
.append(", topic=").append(topic)
.append(", compressionType=").append(compressionType)
.append("batchBytes").append(batchBytes).append("]");
return builder.toString();
}
public enum CompressionType {
none,
gzip,
snappy
}
}

View File

@@ -0,0 +1,57 @@
/*
* Copyright 2015 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.util;
import java.util.Map;
import kafka.serializer.DefaultEncoder;
import kafka.serializer.Encoder;
import org.apache.kafka.common.serialization.Serializer;
/**
* An adapter from the pre-0.8.2 Kafka {@link Encoder} to the {@link Serializer} interface used
* by the new client.
*
* @author Marius Bogoevici
*/
public class EncoderAdaptingSerializer<T> implements Serializer<T> {
private final Encoder<T> encoder;
public EncoderAdaptingSerializer(Encoder<T> encoder) {
this.encoder = encoder;
}
public Encoder<T> getEncoder() {
return encoder;
}
@Override
public void configure(Map<String, ?> configs, boolean isKey) {
// no-op
}
@Override
public byte[] serialize(String topic, T data) {
return encoder.toBytes(data);
}
@Override
public void close() {
// no-op;
}
}

View File

@@ -1,2 +1,2 @@
http\://www.springframework.org/schema/integration/kafka/spring-integration-kafka-1.1.xsd=org/springframework/integration/config/xml/spring-integration-kafka-1.1.xsd
http\://www.springframework.org/schema/integration/kafka/spring-integration-kafka.xsd=org/springframework/integration/config/xml/spring-integration-kafka-1.1.xsd
http\://www.springframework.org/schema/integration/kafka/spring-integration-kafka-1.2.xsd=org/springframework/integration/config/xml/spring-integration-kafka-1.2.xsd
http\://www.springframework.org/schema/integration/kafka/spring-integration-kafka.xsd=org/springframework/integration/config/xml/spring-integration-kafka-1.2.xsd

View File

@@ -105,11 +105,35 @@
]]></xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="key-class-type" use="optional" type="xsd:string" default="[B">
<xsd:annotation>
<xsd:appinfo>
<xsd:documentation>
Class type used for the key
</xsd:documentation>
<tool:annotation kind="direct">
<tool:expected-type type="java.lang.Class"/>
</tool:annotation>
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="value-class-type" use="optional" type="xsd:string" default="[B">
<xsd:annotation>
<xsd:appinfo>
<xsd:documentation>
Class type used for the value
</xsd:documentation>
<tool:annotation kind="direct">
<tool:expected-type type="java.lang.Class"/>
</tool:annotation>
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="value-encoder" use="optional" type="xsd:string">
<xsd:annotation>
<xsd:appinfo>
<xsd:documentation>
Custom implementation of a Kafka Encoder for encoding message values.
Custom implementation of a Kafka Encoder for encoding message values. This option is deprecated, 'value-serializer' is the recommended option.
</xsd:documentation>
<tool:annotation kind="ref">
<tool:expected-type type="kafka.serializer.Encoder"/>
@@ -117,6 +141,18 @@
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="value-serializer" use="optional" type="xsd:string">
<xsd:annotation>
<xsd:appinfo>
<xsd:documentation>
Custom implementation of a Kafka Serializer for message values.
</xsd:documentation>
<tool:annotation kind="ref">
<tool:expected-type type="org.apache.kafka.common.serialization.Serializer"/>
</tool:annotation>
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="key-encoder" use="optional" type="xsd:string">
<xsd:annotation>
<xsd:appinfo>
@@ -129,31 +165,31 @@
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="key-class-type" use="optional" type="xsd:string">
<xsd:attribute name="key-serializer" use="optional" type="xsd:string">
<xsd:annotation>
<xsd:appinfo>
<xsd:documentation>
Class type used for the key
Custom implementation of a Kafka Serializer for message keys.
</xsd:documentation>
<tool:annotation kind="direct">
<tool:expected-type type="java.lang.Class"/>
<tool:annotation kind="ref">
<tool:expected-type type="kafka.serializer.Encoder"/>
</tool:annotation>
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="value-class-type" use="optional" type="xsd:string">
<xsd:attribute name="conversion-service" use="optional" type="xsd:string">
<xsd:annotation>
<xsd:appinfo>
<xsd:documentation>
Class type used for the value
Conversion service for transforming inbound objects into the target key and value
</xsd:documentation>
<tool:annotation kind="direct">
<tool:expected-type type="java.lang.Class"/>
<tool:annotation kind="ref">
<tool:expected-type type="org.springframework.core.convert.ConversionService"/>
</tool:annotation>
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="compression-codec" use="optional" type="xsd:string">
<xsd:attribute name="compression-type" use="optional" type="xsd:string">
<xsd:annotation>
<xsd:documentation><![CDATA[
Indicates the type of compression codec used for message compression.
@@ -172,17 +208,7 @@
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="async" use="optional">
<xsd:annotation>
<xsd:documentation>
Indicates if this producer is async or not.
</xsd:documentation>
</xsd:annotation>
<xsd:simpleType>
<xsd:union memberTypes="xsd:boolean xsd:string"/>
</xsd:simpleType>
</xsd:attribute>
<xsd:attribute name="batch-num-messages" use="optional" type="xsd:string">
<xsd:attribute name="batch-bytes" use="optional" type="xsd:string">
<xsd:annotation>
<xsd:documentation>
number of messages to batch at this producer.
@@ -470,6 +496,23 @@
]]></xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="partition-id" type="xsd:integer">
<xsd:annotation>
<xsd:documentation><![CDATA[
Specifies the target partition for the Kafka message.
This attribute is mutually exclusive with 'partition-id-expression' attribute.
]]></xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="partition-id-expression" type="xsd:integer">
<xsd:annotation>
<xsd:documentation><![CDATA[
Specifies the expression to determine the partition for Kafka message
against the Message at runtime.
This attribute is mutually exclusive with 'partition-id' attribute.
]]></xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="order">
<xsd:annotation>
<xsd:documentation>

View File

@@ -38,10 +38,12 @@
topic="test1"
value-encoder="kafkaEncoder"
key-encoder="kafkaEncoder"
compression-codec="default"/>
compression-type="none"/>
<int-kafka:producer-configuration broker-list="localhost:9092"
topic="test2"
compression-codec="default"/>
value-encoder="kafkaEncoder"
key-encoder="kafkaEncoder"
compression-type="none"/>
</int-kafka:producer-configurations>
</int-kafka:producer-context>

View File

@@ -41,7 +41,7 @@ import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
*/
@RunWith(SpringJUnit4ClassRunner.class)
@ContextConfiguration
public class KafkaOutboundAdapterParserTests<K, V> {
public class KafkaOutboundAdapterParserTests {
@ClassRule
public static KafkaRunning kafkaRunning = KafkaRunning.isRunning();
@@ -53,13 +53,13 @@ public class KafkaOutboundAdapterParserTests<K, V> {
@SuppressWarnings("unchecked")
public void testOutboundAdapterConfiguration() {
PollingConsumer pollingConsumer = this.appContext.getBean("kafkaOutboundChannelAdapter", PollingConsumer.class);
KafkaProducerMessageHandler<K, V> messageHandler = this.appContext.getBean(KafkaProducerMessageHandler.class);
KafkaProducerMessageHandler messageHandler = this.appContext.getBean(KafkaProducerMessageHandler.class);
assertNotNull(pollingConsumer);
assertNotNull(messageHandler);
assertEquals(messageHandler.getOrder(), 3);
assertEquals("foo", TestUtils.getPropertyValue(messageHandler, "topicExpression.literalValue"));
assertEquals("'bar'", TestUtils.getPropertyValue(messageHandler, "messageKeyExpression.expression"));
KafkaProducerContext<K, V> producerContext = messageHandler.getKafkaProducerContext();
KafkaProducerContext producerContext = messageHandler.getKafkaProducerContext();
assertNotNull(producerContext);
assertEquals(producerContext.getProducerConfigurations().size(), 2);
}

View File

@@ -11,8 +11,8 @@
<bean id="producerProperties" class="org.springframework.beans.factory.config.PropertiesFactoryBean">
<property name="properties">
<props>
<prop key="topic.metadata.refresh.interval.ms">3600000</prop>
<prop key="message.send.max.retries">5</prop>
<prop key="metadata.max.age.ms">3600000</prop>
<prop key="retries">5</prop>
<prop key="send.buffer.bytes">5242880</prop>
</props>
</property>
@@ -36,14 +36,27 @@
key-encoder="valueEncoder"
value-encoder="valueEncoder"
topic="${topic1}"
compression-codec="default"/>
compression-type="none"/>
<int-kafka:producer-configuration broker-list="${brokerList2}"
topic="${topic2}"
compression-codec="default"/>
key-class-type="java.lang.String"
value-class-type="java.lang.String"
key-serializer="stringSerializer"
value-serializer="stringSerializer"
batch-bytes="9876"
partitioner="partitioner"
conversion-service="conversionService"
compression-type="none"/>
</int-kafka:producer-configurations>
</int-kafka:producer-context>
<bean id="valueEncoder" class="org.springframework.integration.kafka.serializer.avro.AvroReflectDatumBackedKafkaEncoder">
<constructor-arg value="java.lang.String" />
</bean>
<bean id="stringSerializer" class="org.apache.kafka.common.serialization.StringSerializer"/>
<bean id="partitioner" class="org.springframework.integration.kafka.support.DefaultPartitioner"/>
<bean id="conversionService" class="org.springframework.integration.kafka.config.xml.KafkaProducerContextParserTests.StubConversionService"/>
</beans>

View File

@@ -15,25 +15,32 @@
*/
package org.springframework.integration.kafka.config.xml;
import static org.hamcrest.Matchers.equalTo;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertSame;
import java.util.Map;
import kafka.javaapi.producer.Producer;
import kafka.producer.Partitioner;
import kafka.serializer.Encoder;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.common.serialization.Serializer;
import org.junit.Assert;
import org.junit.ClassRule;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.DirectFieldAccessor;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ApplicationContext;
import org.springframework.core.convert.ConversionService;
import org.springframework.core.convert.TypeDescriptor;
import org.springframework.integration.kafka.rule.KafkaRunning;
import org.springframework.integration.kafka.support.KafkaProducerContext;
import org.springframework.integration.kafka.support.ProducerConfiguration;
import org.springframework.integration.kafka.support.ProducerMetadata;
import org.springframework.integration.kafka.util.EncoderAdaptingSerializer;
import org.springframework.integration.test.util.TestUtils;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
@@ -45,7 +52,7 @@ import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
*/
@RunWith(SpringJUnit4ClassRunner.class)
@ContextConfiguration
public class KafkaProducerContextParserTests<K,V,T> {
public class KafkaProducerContextParserTests {
@ClassRule
public static KafkaRunning kafkaRunning = KafkaRunning.isRunning();
@@ -54,39 +61,78 @@ public class KafkaProducerContextParserTests<K,V,T> {
private ApplicationContext appContext;
@Test
@SuppressWarnings("unchecked")
@SuppressWarnings({"unchecked","rawtypes"})
public void testProducerContextConfiguration(){
final KafkaProducerContext<K,V> producerContext = appContext.getBean("producerContext", KafkaProducerContext.class);
final KafkaProducerContext producerContext = appContext.getBean("producerContext", KafkaProducerContext.class);
Assert.assertNotNull(producerContext);
final Map<String, ProducerConfiguration<K,V>> producerConfigurations = producerContext.getProducerConfigurations();
Assert.assertEquals(producerConfigurations.size(), 2);
final Map<String, ProducerConfiguration<?,?>> producerConfigurations = producerContext.getProducerConfigurations();
assertEquals(producerConfigurations.size(), 2);
final ProducerConfiguration<K,V> producerConfigurationTest1 = producerConfigurations.get("test1");
final ProducerConfiguration<?,?> producerConfigurationTest1 = producerConfigurations.get("test1");
Assert.assertNotNull(producerConfigurationTest1);
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 ProducerMetadata<?,?> producerMetadataTest1 = producerConfigurationTest1.getProducerMetadata();
assertEquals(producerMetadataTest1.getTopic(), "test1");
assertEquals(producerMetadataTest1.getCompressionType(), ProducerMetadata.CompressionType.none);
assertEquals(producerMetadataTest1.getKeyClassType(), java.lang.String.class);
assertEquals(producerMetadataTest1.getValueClassType(), java.lang.String.class);
final Encoder<T> valueEncoder = appContext.getBean("valueEncoder", Encoder.class);
Assert.assertEquals(producerMetadataTest1.getValueEncoder(), valueEncoder);
Assert.assertEquals(producerMetadataTest1.getKeyEncoder(), valueEncoder);
final Encoder<?> valueEncoder = appContext.getBean("valueEncoder", Encoder.class);
Assert.assertThat((Class) producerMetadataTest1.getKeySerializer().getClass(), equalTo((Class) EncoderAdaptingSerializer.class));
Assert.assertThat(((EncoderAdaptingSerializer) producerMetadataTest1.getKeySerializer()).getEncoder(), equalTo((Encoder) valueEncoder));
Assert.assertThat((Class)producerMetadataTest1.getValueSerializer().getClass(), equalTo((Class)EncoderAdaptingSerializer.class));
Assert.assertThat(((EncoderAdaptingSerializer)producerMetadataTest1.getValueSerializer()).getEncoder(), equalTo((Encoder)valueEncoder));
final Producer<K,V> producerTest1 = producerConfigurationTest1.getProducer();
Assert.assertEquals(producerConfigurationTest1, new ProducerConfiguration<K,V>(producerMetadataTest1, producerTest1));
DirectFieldAccessor directFieldAccessor = new DirectFieldAccessor(producerConfigurationTest1);
Producer<?,?> producerTest1 = (Producer<?, ?>) directFieldAccessor.getPropertyValue("producer");
assertEquals(producerConfigurationTest1.getProducerMetadata(), producerMetadataTest1);
final ProducerConfiguration<K,V> producerConfigurationTest2 = producerConfigurations.get("test2");
final ProducerConfiguration<?,?> producerConfigurationTest2 = producerConfigurations.get("test2");
Assert.assertNotNull(producerConfigurationTest2);
final ProducerMetadata<K,V> producerMetadataTest2 = producerConfigurationTest2.getProducerMetadata();
Assert.assertEquals(producerMetadataTest2.getTopic(), "test2");
Assert.assertEquals(producerMetadataTest2.getCompressionCodec(), "0");
final ProducerMetadata<?,?> producerMetadataTest2 = producerConfigurationTest2.getProducerMetadata();
assertEquals(producerMetadataTest2.getTopic(), "test2");
assertEquals(producerMetadataTest2.getCompressionType(), ProducerMetadata.CompressionType.none);
final Producer<K,V> producerTest2 = producerConfigurationTest2.getProducer();
Assert.assertEquals(producerConfigurationTest2, new ProducerConfiguration<K,V>(producerMetadataTest2, producerTest2));
DirectFieldAccessor directFieldAccessor2 = new DirectFieldAccessor(producerConfigurationTest2);
Producer<?,?> producerTest2 = (Producer<?, ?>) directFieldAccessor2.getPropertyValue("producer");
assertEquals(producerConfigurationTest2.getProducerMetadata(), producerMetadataTest2);
final Serializer<?> stringSerializer = appContext.getBean("stringSerializer", Serializer.class);
assertSame(stringSerializer, producerConfigurationTest2.getProducerMetadata().getKeySerializer());
assertSame(stringSerializer, producerConfigurationTest2.getProducerMetadata().getValueSerializer());
final Partitioner partitioner = appContext.getBean("partitioner", Partitioner.class);
assertSame(partitioner, producerConfigurationTest2.getProducerMetadata().getPartitioner());
final ConversionService conversionService = appContext.getBean("conversionService", ConversionService.class);
ConversionService configuredConversionService = (ConversionService) directFieldAccessor2.getPropertyValue("conversionService");
assertSame(conversionService, configuredConversionService);
assertEquals(9876,producerConfigurationTest2.getProducerMetadata().getBatchBytes());
assertFalse(TestUtils.getPropertyValue(producerContext, "autoStartup", Boolean.class));
assertEquals(123, TestUtils.getPropertyValue(producerContext, "phase"));
}
public static class StubConversionService implements ConversionService {
@Override
public boolean canConvert(Class<?> sourceType, Class<?> targetType) {
return false;
}
@Override
public boolean canConvert(TypeDescriptor sourceType, TypeDescriptor targetType) {
return false;
}
@Override
public <T> T convert(Object source, Class<T> targetType) {
return null;
}
@Override
public Object convert(Object source, TypeDescriptor sourceType, TypeDescriptor targetType) {
return null;
}
}
}

View File

@@ -20,19 +20,11 @@ import static org.springframework.integration.kafka.util.TopicUtils.ensureTopicC
import static scala.collection.JavaConversions.asScalaBuffer;
import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
import java.util.Properties;
import org.I0Itec.zkclient.ZkClient;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.junit.After;
import org.springframework.integration.kafka.core.BrokerAddressListConfiguration;
import org.springframework.integration.kafka.core.Configuration;
import org.springframework.integration.kafka.core.ConnectionFactory;
import org.springframework.integration.kafka.core.DefaultConnectionFactory;
import org.springframework.integration.kafka.rule.KafkaRule;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
import com.gs.collections.api.RichIterable;
import com.gs.collections.api.block.function.Function2;
@@ -41,19 +33,30 @@ import com.gs.collections.api.multimap.MutableMultimap;
import com.gs.collections.api.tuple.Pair;
import com.gs.collections.impl.factory.Multimaps;
import com.gs.collections.impl.tuple.Tuples;
import kafka.admin.AdminUtils;
import kafka.producer.KeyedMessage;
import kafka.producer.Producer;
import kafka.producer.ProducerConfig;
import kafka.serializer.StringEncoder;
import kafka.utils.TestUtils;
import org.I0Itec.zkclient.ZkClient;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.junit.After;
import scala.collection.JavaConversions;
import scala.collection.Map;
import scala.collection.immutable.List$;
import scala.collection.immutable.Map$;
import scala.collection.immutable.Seq;
import org.springframework.integration.kafka.core.BrokerAddressListConfiguration;
import org.springframework.integration.kafka.core.Configuration;
import org.springframework.integration.kafka.core.ConnectionFactory;
import org.springframework.integration.kafka.core.DefaultConnectionFactory;
import org.springframework.integration.kafka.rule.KafkaRule;
import org.springframework.integration.kafka.serializer.common.StringEncoder;
import org.springframework.integration.kafka.util.EncoderAdaptingSerializer;
/**
* @author Marius Bogoevici
*/
@@ -110,26 +113,27 @@ public abstract class AbstractBrokerTests {
return configuration;
}
public static scala.collection.Seq<KeyedMessage<String, String>> createMessages(int count, String topic) {
return createMessagesInRange(0, count - 1, topic);
public static Collection<ProducerRecord<String, String>> createMessages(int count, String topic, int partitionCount) {
return createMessagesInRange(0, count - 1, topic, partitionCount);
}
public static scala.collection.Seq<KeyedMessage<String, String>> createMessagesInRange(int start, int end,
String topic) {
List<KeyedMessage<String, String>> messages = new ArrayList<KeyedMessage<String, String>>();
public static Collection<ProducerRecord<String, String>> createMessagesInRange(int start, int end,
String topic, int partitionCount) {
List<ProducerRecord<String, String>> messages = new ArrayList<>();
for (int i = start; i <= end; i++) {
messages.add(new KeyedMessage<String, String>(topic, "Key " + i, i, "Message " + i));
messages.add(new ProducerRecord<>(topic, i % partitionCount, "Key " + i, "Message " + i));
}
return asScalaBuffer(messages).toSeq();
return messages;
}
public Producer<String, String> createStringProducer(int compression) {
Properties producerConfig = TestUtils.getProducerConfig(getKafkaRule().getBrokersAsString(),
TestPartitioner.class.getCanonicalName());
producerConfig.put("serializer.class", StringEncoder.class.getCanonicalName());
producerConfig.put("key.serializer.class", StringEncoder.class.getCanonicalName());
producerConfig.put("compression.codec", Integer.toString(compression));
return new Producer<String, String>(new ProducerConfig(producerConfig));
public Sender<String, String> createMessageSender(String compression) {
Properties producerConfig = new Properties();
producerConfig.setProperty("bootstrap.servers", getKafkaRule().getBrokersAsString());
producerConfig.setProperty("compression.type", compression);
KafkaProducer<String, String> producer = new KafkaProducer<>(producerConfig,
new EncoderAdaptingSerializer<>(new StringEncoder()),
new EncoderAdaptingSerializer<>(new StringEncoder()));
return new Sender<>(producer);
}
public ConnectionFactory getKafkaBrokerConnectionFactory() throws Exception {
@@ -162,4 +166,32 @@ public abstract class AbstractBrokerTests {
}
}
public class Sender<K,V> {
private Producer<K,V> producer;
public Sender(Producer<K, V> producer) {
this.producer = producer;
}
public void send(Collection<ProducerRecord<K,V>> records) {
Future<RecordMetadata> lastFuture = null;
for (ProducerRecord<K, V> record : records) {
lastFuture = producer.send(record);
}
// only block if there is at least one message to be sent
if (lastFuture != null) {
try {
// block until the last message has been sent, so we make this deterministic
lastFuture.get();
} catch (InterruptedException e) {
// not being able to confirm that all messages have been sent, fail the test
throw new RuntimeException(e);
} catch (ExecutionException e) {
throw new RuntimeException(e);
}
}
}
}
}

View File

@@ -56,7 +56,8 @@ import org.springframework.integration.kafka.core.KafkaMessage;
*/
public abstract class AbstractMessageListenerContainerTests extends AbstractBrokerTests {
public void runMessageListenerTest(int maxReceiveSize, int concurrency, int partitionCount, int testMessageCount, int divisionFactor, int compressionCodec, String topic) throws Exception {
public void runMessageListenerTest(int maxReceiveSize, int concurrency, int partitionCount, int testMessageCount,
int divisionFactor, String compressionCodec, String topic) throws Exception {
ConnectionFactory connectionFactory = getKafkaBrokerConnectionFactory();
ArrayList<Partition> readPartitions = new ArrayList<Partition>();
@@ -65,7 +66,8 @@ public abstract class AbstractMessageListenerContainerTests extends AbstractBrok
readPartitions.add(new Partition(topic, i));
}
}
final KafkaMessageListenerContainer kafkaMessageListenerContainer = new KafkaMessageListenerContainer(connectionFactory, readPartitions.toArray(new Partition[readPartitions.size()]));
final KafkaMessageListenerContainer kafkaMessageListenerContainer =
new KafkaMessageListenerContainer(connectionFactory, readPartitions.toArray(new Partition[readPartitions.size()]));
kafkaMessageListenerContainer.setMaxFetch(maxReceiveSize);
kafkaMessageListenerContainer.setConcurrency(concurrency);
@@ -77,14 +79,18 @@ public abstract class AbstractMessageListenerContainerTests extends AbstractBrok
@Override
public void onMessage(KafkaMessage message) {
StringDecoder decoder = new StringDecoder(new VerifiableProperties());
receivedData.put(message.getMetadata().getPartition().getId(),new KeyedMessageWithOffset(decodeKey(message, decoder), decodePayload(message, decoder), message.getMetadata().getOffset(), Thread.currentThread().getName(), message.getMetadata().getPartition().getId()));
receivedData.put(message.getMetadata().getPartition().getId(),
new KeyedMessageWithOffset(decodeKey(message, decoder),
decodePayload(message, decoder), message.getMetadata().getOffset(),
Thread.currentThread().getName(),
message.getMetadata().getPartition().getId()));
latch.countDown();
}
});
kafkaMessageListenerContainer.start();
createStringProducer(compressionCodec).send(createMessages(testMessageCount, topic));
createMessageSender(compressionCodec).send(createMessages(testMessageCount, topic, partitionCount));
latch.await((expectedMessageCount/5000) + 1, TimeUnit.MINUTES);
kafkaMessageListenerContainer.stop();
@@ -93,12 +99,14 @@ public abstract class AbstractMessageListenerContainerTests extends AbstractBrok
assertThat(latch.getCount(), equalTo(0L));
System.out.println("All messages received ... checking ");
validateMessageReceipt(receivedData, concurrency, partitionCount, testMessageCount, expectedMessageCount, readPartitions, divisionFactor);
validateMessageReceipt(receivedData, concurrency, partitionCount, expectedMessageCount,
divisionFactor);
}
@SuppressWarnings("serial")
public void validateMessageReceipt(MutableListMultimap<Integer, KeyedMessageWithOffset> receivedData, int concurrency, int partitionCount, int testMessageCount, int expectedMessageCount, ArrayList<Partition> readPartitions, int divisionFactor) {
public void validateMessageReceipt(MutableListMultimap<Integer, KeyedMessageWithOffset> receivedData,
int concurrency, int partitionCount, int expectedMessageCount, int divisionFactor) {
// Group messages received by processing thread
MutableListMultimap<String, KeyedMessageWithOffset> messagesByThread = receivedData.valuesView().toList().groupBy(new Function<KeyedMessageWithOffset, String>() {
@Override
@@ -108,6 +116,12 @@ public abstract class AbstractMessageListenerContainerTests extends AbstractBrok
});
// Execution has taken place on as many distinct threads as configured
messagesByThread.keysView().forEach(new Procedure<String>() {
@Override
public void value(String s) {
System.out.println(s);
}
});
assertThat(messagesByThread.keysView().size(), Matchers.equalTo(concurrency));
// Group partitions by thread

View File

@@ -20,8 +20,6 @@ package org.springframework.integration.kafka.listener;
import static org.hamcrest.Matchers.notNullValue;
import kafka.message.NoCompressionCodec$;
import org.junit.Assert;
import org.junit.Rule;
import org.junit.Test;
@@ -50,9 +48,11 @@ public class ChannelAdapterWithXmlConfigurationTests extends AbstractMessageList
System.setProperty("kafka.test.topic", TEST_TOPIC);
createTopic(TEST_TOPIC, 1, 1, 1);
int partitionCount = 1;
createStringProducer(NoCompressionCodec$.MODULE$.codec()).send(createMessages(100, TEST_TOPIC));
createTopic(TEST_TOPIC, partitionCount, 1, 1);
createMessageSender("none").send(createMessages(100, TEST_TOPIC, partitionCount));
ClassPathXmlApplicationContext context =
new ClassPathXmlApplicationContext("ChannelAdapterWithXmlConfigurationTests-context.xml",
@@ -65,5 +65,4 @@ public class ChannelAdapterWithXmlConfigurationTests extends AbstractMessageList
Assert.assertThat(received, notNullValue());
}
}
}

View File

@@ -17,7 +17,7 @@
package org.springframework.integration.kafka.listener;
import kafka.producer.Producer;
import org.apache.kafka.clients.producer.Producer;
import org.junit.Assert;
import org.junit.Ignore;
import org.junit.Rule;
@@ -62,8 +62,7 @@ public class DefaultConnectionTests extends AbstractBrokerTests {
@Test
public void testReceiveMessages() throws Exception {
createTopic(TEST_TOPIC, 1, 1, 1);
Producer<String, String> producer = createStringProducer(0);
producer.send( createMessages(10, TEST_TOPIC));
createMessageSender("none").send(createMessages(10, TEST_TOPIC, 1));
Connection brokerConnection =
new DefaultConnection(getKafkaRule().getBrokerAddresses()[0], "client", 64*1024, 3000, 1, 10000);
Partition partition = new Partition(TEST_TOPIC, 0);
@@ -85,8 +84,7 @@ public class DefaultConnectionTests extends AbstractBrokerTests {
@Test
public void testReceiveMessagesWithGZipCompression() throws Exception {
createTopic(TEST_TOPIC, 1, 1, 1);
Producer<String, String> producer = createStringProducer(1);
producer.send( createMessages(10, TEST_TOPIC));
createMessageSender("gzip").send(createMessages(10, TEST_TOPIC, 1));
Connection brokerConnection =
new DefaultConnection(getKafkaRule().getBrokerAddresses()[0], "client", 64*1024, 3000, 1, 10000);
Partition partition = new Partition(TEST_TOPIC, 0);
@@ -108,15 +106,12 @@ public class DefaultConnectionTests extends AbstractBrokerTests {
@Test
@Ignore
/**
* The compression codec '2' is for Snappy:
* {@code producerConfig.put("compression.codec", 2);}
* Since it relies on the native library we can't test it on all environment,
* Since the test relies on the native library we can't test it on all environments,
* because we may not have permission to load dll(so).
*/
public void testReceiveMessagesWithSnappyCompression() throws Exception {
createTopic(TEST_TOPIC, 1, 1, 1);
Producer<String, String> producer = createStringProducer(2);
producer.send( createMessages(10, TEST_TOPIC));
createMessageSender("snappy").send( createMessages(10, TEST_TOPIC,1));
Connection brokerConnection =
new DefaultConnection(getKafkaRule().getBrokerAddresses()[0], "client", 64*1024, 3000, 1, 10000);
Partition partition = new Partition(TEST_TOPIC, 0);

View File

@@ -32,7 +32,6 @@ import java.util.concurrent.TimeUnit;
import com.gs.collections.api.multimap.list.MutableListMultimap;
import com.gs.collections.impl.list.mutable.FastList;
import com.gs.collections.impl.multimap.list.SynchronizedPutFastListMultimap;
import kafka.message.NoCompressionCodec$;
import org.junit.Rule;
import org.junit.Test;
@@ -64,11 +63,13 @@ public class KafkaMessageDrivenChannelAdapterTests extends AbstractMessageListen
@Test
@SuppressWarnings("serial")
public void testLowVolumeLowConcurrency() throws Exception {
createTopic(TEST_TOPIC, 5, 1, 1);
int partitionCount = 5;
createTopic(TEST_TOPIC, partitionCount, 1, 1);
ConnectionFactory connectionFactory = getKafkaBrokerConnectionFactory();
ArrayList<Partition> readPartitions = new ArrayList<Partition>();
for (int i = 0; i < 5; i++) {
for (int i = 0; i < partitionCount; i++) {
readPartitions.add(new Partition(TEST_TOPIC, i));
}
@@ -116,7 +117,7 @@ public class KafkaMessageDrivenChannelAdapterTests extends AbstractMessageListen
kafkaMessageDrivenChannelAdapter.afterPropertiesSet();
kafkaMessageDrivenChannelAdapter.start();
createStringProducer(NoCompressionCodec$.MODULE$.codec()).send(createMessages(100, TEST_TOPIC));
createMessageSender("none").send(createMessages(100, TEST_TOPIC, partitionCount));
latch.await((expectedMessageCount / 5000) + 1, TimeUnit.MINUTES);
kafkaMessageListenerContainer.stop();
@@ -125,18 +126,20 @@ public class KafkaMessageDrivenChannelAdapterTests extends AbstractMessageListen
assertThat(latch.getCount(), equalTo(0L));
System.out.println("All messages received ... checking ");
validateMessageReceipt(receivedData, 2, 5, 100, expectedMessageCount, readPartitions, 1);
validateMessageReceipt(receivedData, 2, partitionCount, expectedMessageCount, 1);
}
@Test
@SuppressWarnings("serial")
public void testManualAck() throws Exception {
createTopic(TEST_TOPIC, 5, 1, 1);
int partitionCount = 5;
createTopic(TEST_TOPIC, partitionCount, 1, 1);
ConnectionFactory connectionFactory = getKafkaBrokerConnectionFactory();
ArrayList<Partition> readPartitions = new ArrayList<Partition>();
for (int i = 0; i < 5; i++) {
for (int i = 0; i < partitionCount; i++) {
readPartitions.add(new Partition(TEST_TOPIC, i));
}
@@ -191,7 +194,7 @@ public class KafkaMessageDrivenChannelAdapterTests extends AbstractMessageListen
kafkaMessageDrivenChannelAdapter.afterPropertiesSet();
kafkaMessageDrivenChannelAdapter.start();
createStringProducer(NoCompressionCodec$.MODULE$.codec()).send(createMessages(100, TEST_TOPIC));
createMessageSender("none").send(createMessages(100, TEST_TOPIC, partitionCount));
latch.await((expectedMessageCount / 5000) + 1, TimeUnit.MINUTES);
kafkaMessageListenerContainer.stop();
@@ -200,7 +203,7 @@ public class KafkaMessageDrivenChannelAdapterTests extends AbstractMessageListen
assertThat(latch.getCount(), equalTo(0L));
System.out.println("All messages received ... checking ");
validateMessageReceipt(receivedData, 2, 5, 100, expectedMessageCount, readPartitions, 1);
validateMessageReceipt(receivedData, 2, partitionCount, expectedMessageCount, 1);
// at this point, all messages have been processed but not acknowledged
for (Partition readPartition : readPartitions) {

View File

@@ -45,8 +45,6 @@ import org.springframework.messaging.MessagingException;
import com.gs.collections.api.multimap.list.MutableListMultimap;
import com.gs.collections.impl.multimap.list.SynchronizedPutFastListMultimap;
import kafka.message.NoCompressionCodec$;
/**
* @author Marius Bogoevici
*/
@@ -76,8 +74,7 @@ public class KafkaMessageDrivenChannelAdapterWithKafkaOffsetManagerTests extends
KafkaTopicOffsetManager offsetManager = new KafkaTopicOffsetManager(
new ZookeeperConnect(getKafkaRule().getZookeeperConnectionString()), "si-offsets");
offsetManager.afterPropertiesSet();
kafkaMessageListenerContainer.setOffsetManager(
offsetManager);
kafkaMessageListenerContainer.setOffsetManager(offsetManager);
kafkaMessageListenerContainer.setMaxFetch(100);
kafkaMessageListenerContainer.setConcurrency(2);
@@ -114,12 +111,7 @@ public class KafkaMessageDrivenChannelAdapterWithKafkaOffsetManagerTests extends
kafkaMessageDrivenChannelAdapter.afterPropertiesSet();
kafkaMessageDrivenChannelAdapter.start();
createStringProducer(NoCompressionCodec$.MODULE$.codec()).send(createMessages(expectedMessageCount, TEST_TOPIC));
Thread.sleep(100);
kafkaMessageDrivenChannelAdapter.stop();
kafkaMessageDrivenChannelAdapter.start();
createMessageSender("none").send(createMessages(expectedMessageCount, TEST_TOPIC, 5));
latch.await((expectedMessageCount / 5000) + 1, TimeUnit.MINUTES);
kafkaMessageListenerContainer.stop();
@@ -128,7 +120,19 @@ public class KafkaMessageDrivenChannelAdapterWithKafkaOffsetManagerTests extends
assertThat(latch.getCount(), equalTo(0L));
System.out.println("All messages received ... checking ");
validateMessageReceipt(receivedData, 2, 5, 100, expectedMessageCount, readPartitions, 1);
validateMessageReceipt(receivedData, 2, 5, expectedMessageCount, 1);
offsetManager.close();
KafkaTopicOffsetManager offsetManager2 =
new KafkaTopicOffsetManager(new ZookeeperConnect(getKafkaRule().getZookeeperConnectionString()), "si-offsets");
offsetManager2.afterPropertiesSet();
assertThat(offsetManager2.getOffset(new Partition(TEST_TOPIC,0)),equalTo(20L));
assertThat(offsetManager2.getOffset(new Partition(TEST_TOPIC,1)),equalTo(20L));
assertThat(offsetManager2.getOffset(new Partition(TEST_TOPIC,2)),equalTo(20L));
assertThat(offsetManager2.getOffset(new Partition(TEST_TOPIC,3)),equalTo(20L));
assertThat(offsetManager2.getOffset(new Partition(TEST_TOPIC,4)),equalTo(20L));
}

View File

@@ -37,8 +37,6 @@ import com.gs.collections.impl.list.mutable.FastList;
import com.gs.collections.impl.multimap.list.SynchronizedPutFastListMultimap;
import com.gs.collections.impl.utility.Iterate;
import kafka.message.NoCompressionCodec$;
import org.junit.Rule;
import org.junit.Test;
@@ -73,13 +71,14 @@ public class KafkaMessageDrivenChannelAdapterWithSpecialOffsetTests extends Abst
// we will send 300 messages: first 200, then another 100
// we will start reading from all partitions at offset 100
int expectedMessageCount = 200;
int partitionCount = 5;
createTopic(TEST_TOPIC, 5, 1, 1);
createTopic(TEST_TOPIC, partitionCount, 1, 1);
ConnectionFactory connectionFactory = getKafkaBrokerConnectionFactory();
ArrayList<Partition> readPartitions = new ArrayList<Partition>();
Map<Partition, Long> startingOffsets = new HashMap<Partition, Long>();
for (int i = 0; i < 5; i++) {
for (int i = 0; i < partitionCount; i++) {
Partition partition = new Partition(TEST_TOPIC, i);
readPartitions.add(partition);
startingOffsets.put(partition, 20L);
@@ -93,8 +92,8 @@ public class KafkaMessageDrivenChannelAdapterWithSpecialOffsetTests extends Abst
MetadataStoreOffsetManager offsetManager = new MetadataStoreOffsetManager(connectionFactory, startingOffsets);
kafkaMessageListenerContainer.setOffsetManager(offsetManager);
// we send 100 messages
createStringProducer(NoCompressionCodec$.MODULE$.codec()).send(createMessagesInRange(0, 199, TEST_TOPIC));
// we send 200 messages
createMessageSender("none").send(createMessagesInRange(0, 199, TEST_TOPIC, partitionCount));
final MutableListMultimap<Integer, KeyedMessageWithOffset> receivedData =
new SynchronizedPutFastListMultimap<Integer, KeyedMessageWithOffset>();
@@ -132,7 +131,7 @@ public class KafkaMessageDrivenChannelAdapterWithSpecialOffsetTests extends Abst
kafkaMessageDrivenChannelAdapter.afterPropertiesSet();
kafkaMessageDrivenChannelAdapter.start();
createStringProducer(NoCompressionCodec$.MODULE$.codec()).send(createMessagesInRange(200, 299, TEST_TOPIC));
createMessageSender("none").send(createMessagesInRange(200, 299, TEST_TOPIC, partitionCount));
latch.await((expectedMessageCount / 5000) + 1, TimeUnit.MINUTES);
kafkaMessageListenerContainer.stop();
@@ -141,7 +140,7 @@ public class KafkaMessageDrivenChannelAdapterWithSpecialOffsetTests extends Abst
assertThat(latch.getCount(), equalTo(0L));
System.out.println("All messages received ... checking ");
validateMessageReceipt(receivedData, 2, 5, 100, expectedMessageCount, readPartitions, 1);
validateMessageReceipt(receivedData, 2, partitionCount, expectedMessageCount, 1);
// For all received messages
Collection<KeyedMessageWithOffset> allReceivedMessages =
@@ -166,7 +165,7 @@ public class KafkaMessageDrivenChannelAdapterWithSpecialOffsetTests extends Abst
}).min();
// The lowest received value is 100. That is correct, because messages are evenly distributed across partitions
// and we start reading only at partition 200
// and we start reading only at offset 20
assertThat(minValueInMessage, equalTo(100));
}

View File

@@ -38,8 +38,6 @@ import com.gs.collections.impl.list.mutable.FastList;
import com.gs.collections.impl.multimap.list.SynchronizedPutFastListMultimap;
import com.gs.collections.impl.utility.Iterate;
import kafka.message.NoCompressionCodec$;
import org.junit.Rule;
import org.junit.Test;
@@ -74,14 +72,15 @@ public class KafkaMessageDrivenChannelAdapterWithWrongOffsetTests extends Abstra
// we will send 300 messages: first 200, then another 100
// we will start reading from all partitions at offset 100
int expectedMessageCount = 200;
int expectedMessageCount = 300;
createTopic(TEST_TOPIC, 5, 1, 1);
int partitionCount = 5;
createTopic(TEST_TOPIC, partitionCount, 1, 1);
ConnectionFactory connectionFactory = getKafkaBrokerConnectionFactory();
ArrayList<Partition> readPartitions = new ArrayList<Partition>();
Map<Partition, Long> startingOffsets = new HashMap<Partition, Long>();
for (int i = 0; i < 5; i++) {
for (int i = 0; i < partitionCount; i++) {
Partition partition = new Partition(TEST_TOPIC, i);
readPartitions.add(partition);
startingOffsets.put(partition, 900L);
@@ -95,8 +94,8 @@ public class KafkaMessageDrivenChannelAdapterWithWrongOffsetTests extends Abstra
MetadataStoreOffsetManager offsetManager = new MetadataStoreOffsetManager(connectionFactory, startingOffsets);
kafkaMessageListenerContainer.setOffsetManager(offsetManager);
// we send 100 messages
createStringProducer(NoCompressionCodec$.MODULE$.codec()).send(createMessagesInRange(0, 199, TEST_TOPIC));
// we send 200 messages
createMessageSender("none").send(createMessagesInRange(0, 199, TEST_TOPIC, partitionCount));
final MutableListMultimap<Integer, KeyedMessageWithOffset> receivedData =
new SynchronizedPutFastListMultimap<Integer, KeyedMessageWithOffset>();
@@ -135,7 +134,7 @@ public class KafkaMessageDrivenChannelAdapterWithWrongOffsetTests extends Abstra
kafkaMessageDrivenChannelAdapter.afterPropertiesSet();
kafkaMessageDrivenChannelAdapter.start();
createStringProducer(NoCompressionCodec$.MODULE$.codec()).send(createMessagesInRange(200, 299, TEST_TOPIC));
createMessageSender("none").send(createMessagesInRange(200, 299, TEST_TOPIC, partitionCount));
latch.await((expectedMessageCount / 5000) + 1, TimeUnit.MINUTES);
kafkaMessageListenerContainer.stop();
@@ -144,7 +143,7 @@ public class KafkaMessageDrivenChannelAdapterWithWrongOffsetTests extends Abstra
assertThat(latch.getCount(), equalTo(0L));
System.out.println("All messages received ... checking ");
validateMessageReceipt(receivedData, 2, 5, 100, expectedMessageCount, readPartitions, 1);
validateMessageReceipt(receivedData, 2, partitionCount, expectedMessageCount, 1);
// For all received messages
Collection<KeyedMessageWithOffset> allReceivedMessages =
@@ -178,13 +177,14 @@ public class KafkaMessageDrivenChannelAdapterWithWrongOffsetTests extends Abstra
// we will send 300 messages: first 200, then another 100
// we will start reading from all partitions at offset 900, but since that is reset
int expectedMessageCount = 100;
int partitionCount = 5;
createTopic(TEST_TOPIC, 5, 1, 1);
createTopic(TEST_TOPIC, partitionCount, 1, 1);
ConnectionFactory connectionFactory = getKafkaBrokerConnectionFactory();
ArrayList<Partition> readPartitions = new ArrayList<Partition>();
Map<Partition, Long> startingOffsets = new HashMap<Partition, Long>();
for (int i = 0; i < 5; i++) {
for (int i = 0; i < partitionCount; i++) {
Partition partition = new Partition(TEST_TOPIC, i);
readPartitions.add(partition);
startingOffsets.put(partition, 900L);
@@ -244,7 +244,7 @@ public class KafkaMessageDrivenChannelAdapterWithWrongOffsetTests extends Abstra
kafkaMessageListenerContainer.setOffsetManager(offsetManagerWrapper);
createStringProducer(NoCompressionCodec$.MODULE$.codec()).send(createMessagesInRange(0, 199, TEST_TOPIC));
createMessageSender("none").send(createMessagesInRange(0, 199, TEST_TOPIC, partitionCount));
final MutableListMultimap<Integer, KeyedMessageWithOffset> receivedData =
new SynchronizedPutFastListMultimap<Integer, KeyedMessageWithOffset>();
@@ -284,7 +284,7 @@ public class KafkaMessageDrivenChannelAdapterWithWrongOffsetTests extends Abstra
// we wait for the reset event first
offsetResetLatch.await(1000, TimeUnit.MILLISECONDS);
Thread.sleep(1000);
createStringProducer(NoCompressionCodec$.MODULE$.codec()).send(createMessagesInRange(200, 299, TEST_TOPIC));
createMessageSender("none").send(createMessagesInRange(200, 299, TEST_TOPIC, partitionCount));
latch.await((expectedMessageCount / 5000) + 1, TimeUnit.MINUTES);
kafkaMessageListenerContainer.stop();
@@ -293,7 +293,7 @@ public class KafkaMessageDrivenChannelAdapterWithWrongOffsetTests extends Abstra
assertThat(latch.getCount(), equalTo(0L));
System.out.println("All messages received ... checking ");
validateMessageReceipt(receivedData, 2, 5, 100, expectedMessageCount, readPartitions, 1);
validateMessageReceipt(receivedData, 2, partitionCount, expectedMessageCount, 1);
// For all received messages
Collection<KeyedMessageWithOffset> allReceivedMessages =
@@ -319,7 +319,8 @@ public class KafkaMessageDrivenChannelAdapterWithWrongOffsetTests extends Abstra
})
.min();
// The lowest received value is 0. That is correct, because reading started at 0 after the reset
// The lowest received value is 0. That is correct, because reading started at offset 40 (last after the first
// 200 messages have been sent) after the reset
assertThat(minValueInMessage, equalTo(200));
}

View File

@@ -40,21 +40,21 @@ public class MultiBroker20Tests extends AbstractMessageListenerContainerTests {
@Ignore
public void testLowVolumeHighConcurrency() throws Exception {
createTopic(TEST_TOPIC, 100, 20, 1);
runMessageListenerTest(100, 20, 100, 1000, 1, 0, TEST_TOPIC);
runMessageListenerTest(100, 20, 100, 1000, 1, "none", TEST_TOPIC);
}
@Test
@Ignore
public void testMediumVolumeHighConcurrency() throws Exception {
createTopic(TEST_TOPIC, 100, 20, 1);
runMessageListenerTest(100, 20, 100, 10000, 1, 0, TEST_TOPIC);
runMessageListenerTest(100, 20, 100, 10000, 1, "none", TEST_TOPIC);
}
@Test
@Ignore
public void testHighVolumeHighConcurrency() throws Exception {
createTopic(TEST_TOPIC, 100, 20, 1);
runMessageListenerTest(100, 20, 100, 100000, 1, 0, TEST_TOPIC);
runMessageListenerTest(100, 20, 100, 100000, 1, "none", TEST_TOPIC);
}
}

View File

@@ -40,42 +40,42 @@ public class MultiBroker2Tests extends AbstractMessageListenerContainerTests {
@Test
public void testLowVolumeLowConcurrency() throws Exception {
createTopic(TEST_TOPIC, 5, 2, 1);
runMessageListenerTest(100, 2, 5, 100, 1, 0, TEST_TOPIC);
runMessageListenerTest(100, 2, 5, 100, 1, "none", TEST_TOPIC);
}
@Test
@Ignore
public void testMediumVolumeLowConcurrency() throws Exception {
createTopic(TEST_TOPIC, 5, 2, 1);
runMessageListenerTest(100, 2, 5, 1000, 1, 0, TEST_TOPIC);
runMessageListenerTest(100, 2, 5, 1000, 1, "none", TEST_TOPIC);
}
@Test
@Ignore
public void testHighVolumeLowConcurrency() throws Exception {
createTopic(TEST_TOPIC, 5, 2, 1);
runMessageListenerTest(100, 2, 5, 10000, 1, 0, TEST_TOPIC);
runMessageListenerTest(100, 2, 5, 10000, 1, "none", TEST_TOPIC);
}
@Test
@Ignore
public void testLowVolumeMediumConcurrency() throws Exception {
createTopic(TEST_TOPIC, 5, 2, 1);
runMessageListenerTest(100, 5, 5, 100, 1, 0, TEST_TOPIC);
runMessageListenerTest(100, 5, 5, 100, 1, "none", TEST_TOPIC);
}
@Test
@Ignore
public void testMediumVolumeMediumConcurrency() throws Exception {
createTopic(TEST_TOPIC, 5, 2, 1);
runMessageListenerTest(100, 5, 5, 1000, 1, 0, TEST_TOPIC);
runMessageListenerTest(100, 5, 5, 1000, 1, "none", TEST_TOPIC);
}
@Test
@Ignore
public void testHighVolumeMediumConcurrency() throws Exception {
createTopic(TEST_TOPIC, 5, 2, 1);
runMessageListenerTest(100, 5, 5, 100000, 1, 0, TEST_TOPIC);
runMessageListenerTest(100, 5, 5, 100000, 1, "none", TEST_TOPIC);
}
@@ -83,21 +83,21 @@ public class MultiBroker2Tests extends AbstractMessageListenerContainerTests {
@Ignore
public void testLowVolumeHighConcurrency() throws Exception {
createTopic(TEST_TOPIC, 100, 2, 1);
runMessageListenerTest(100, 20, 100, 1000, 1, 0, TEST_TOPIC);
runMessageListenerTest(100, 20, 100, 1000, 1, "none", TEST_TOPIC);
}
@Test
@Ignore
public void testMediumVolumeHighConcurrency() throws Exception {
createTopic(TEST_TOPIC, 100, 2, 1);
runMessageListenerTest(100, 20, 100, 10000, 1, 0, TEST_TOPIC);
runMessageListenerTest(100, 20, 100, 10000, 1, "none", TEST_TOPIC);
}
@Test
@Ignore
public void testHighVolumeHighConcurrency() throws Exception {
createTopic(TEST_TOPIC, 100, 2, 1);
runMessageListenerTest(100, 20, 100, 100000, 1, 0, TEST_TOPIC);
runMessageListenerTest(100, 20, 100, 100000, 1, "none", TEST_TOPIC);
}
}

View File

@@ -39,42 +39,42 @@ public class MultiBroker5ReplicatedTests extends AbstractMessageListenerContaine
@Test
public void testLowVolumeLowConcurrency() throws Exception {
createTopic(TEST_TOPIC, 5, 5, 3);
runMessageListenerTest(100, 2, 5, 100, 1, 0, TEST_TOPIC);
runMessageListenerTest(100, 2, 5, 100, 1, "none", TEST_TOPIC);
}
@Test
@Ignore
public void testMediumVolumeLowConcurrency() throws Exception {
createTopic(TEST_TOPIC, 5, 5, 3);
runMessageListenerTest(100, 2, 5, 1000, 1, 0, TEST_TOPIC);
runMessageListenerTest(100, 2, 5, 1000, 1, "none", TEST_TOPIC);
}
@Test
@Ignore
public void testHighVolumeLowConcurrency() throws Exception {
createTopic(TEST_TOPIC, 5, 5, 3);
runMessageListenerTest(100, 2, 5, 10000, 1, 0, TEST_TOPIC);
runMessageListenerTest(100, 2, 5, 10000, 1, "none", TEST_TOPIC);
}
@Test
@Ignore
public void testLowVolumeMediumConcurrency() throws Exception {
createTopic(TEST_TOPIC, 5, 5, 3);
runMessageListenerTest(100, 5, 5, 100, 1, 0, TEST_TOPIC);
runMessageListenerTest(100, 5, 5, 100, 1, "none", TEST_TOPIC);
}
@Test
@Ignore
public void testMediumVolumeMediumConcurrency() throws Exception {
createTopic(TEST_TOPIC, 5, 5, 3);
runMessageListenerTest(100, 5, 5, 1000, 1, 0, TEST_TOPIC);
runMessageListenerTest(100, 5, 5, 1000, 1, "none", TEST_TOPIC);
}
@Test
@Ignore
public void testHighVolumeMediumConcurrency() throws Exception {
createTopic(TEST_TOPIC, 5, 5, 1);
runMessageListenerTest(100, 5, 5, 100000, 1, 0, TEST_TOPIC);
runMessageListenerTest(100, 5, 5, 100000, 1, "none", TEST_TOPIC);
}
@@ -82,21 +82,21 @@ public class MultiBroker5ReplicatedTests extends AbstractMessageListenerContaine
@Ignore
public void testLowVolumeHighConcurrency() throws Exception {
createTopic(TEST_TOPIC, 100, 5, 3);
runMessageListenerTest(100, 20, 100, 1000, 1, 0, TEST_TOPIC);
runMessageListenerTest(100, 20, 100, 1000, 1, "none", TEST_TOPIC);
}
@Test
@Ignore
public void testMediumVolumeHighConcurrency() throws Exception {
createTopic(TEST_TOPIC, 100, 5, 3);
runMessageListenerTest(100, 20, 100, 10000, 1, 0, TEST_TOPIC);
runMessageListenerTest(100, 20, 100, 10000, 1, "none", TEST_TOPIC);
}
@Test
@Ignore
public void testHighVolumeHighConcurrency() throws Exception {
createTopic(TEST_TOPIC, 100, 5, 3);
runMessageListenerTest(100, 20, 100, 100000, 1, 0, TEST_TOPIC);
runMessageListenerTest(100, 20, 100, 100000, 1, "none", TEST_TOPIC);
}
}

View File

@@ -57,11 +57,15 @@ public class MultiBroker5ReplicatedWithBounceTests extends AbstractMessageListen
@Test
public void testLowVolumeLowConcurrency() throws Exception {
createTopic(TEST_TOPIC, 5, 5, 3);
int partitionCount = 5;
createTopic(TEST_TOPIC, partitionCount, 5, 3);
Thread.sleep(1000);
ConnectionFactory connectionFactory = getKafkaBrokerConnectionFactory();
ArrayList<Partition> readPartitions = new ArrayList<Partition>();
for (int i = 0; i < 5; i++) {
for (int i = 0; i < partitionCount; i++) {
readPartitions.add(new Partition(TEST_TOPIC, i));
}
final KafkaMessageListenerContainer kafkaMessageListenerContainer =
@@ -72,7 +76,7 @@ public class MultiBroker5ReplicatedWithBounceTests extends AbstractMessageListen
final int expectedMessageCount = 100;
createStringProducer(0).send(createMessages(100, TEST_TOPIC));
createMessageSender("none").send(createMessages(100, TEST_TOPIC, partitionCount));
final MutableListMultimap<Integer, KeyedMessageWithOffset> receivedData =
@@ -100,14 +104,14 @@ public class MultiBroker5ReplicatedWithBounceTests extends AbstractMessageListen
}
}
latch.await(50, TimeUnit.SECONDS);
latch.await(120, TimeUnit.SECONDS);
kafkaMessageListenerContainer.stop();
assertThat(receivedData.valuesView().toList(), hasSize(expectedMessageCount));
assertThat(latch.getCount(), equalTo(0L));
System.out.println("All messages received ... checking ");
validateMessageReceipt(receivedData, 2, 5, 100, expectedMessageCount, readPartitions, 1);
validateMessageReceipt(receivedData, 2, partitionCount, expectedMessageCount, 1);
}

View File

@@ -39,42 +39,42 @@ public class MultiBroker5Tests extends AbstractMessageListenerContainerTests {
@Test
public void testLowVolumeLowConcurrency() throws Exception {
createTopic(TEST_TOPIC, 5, 5, 1);
runMessageListenerTest(100, 2, 5, 100, 1, 0, TEST_TOPIC);
runMessageListenerTest(100, 2, 5, 100, 1, "none", TEST_TOPIC);
}
@Test
@Ignore
public void testMediumVolumeLowConcurrency() throws Exception {
createTopic(TEST_TOPIC, 5, 5, 1);
runMessageListenerTest(100, 2, 5, 1000, 1, 0, TEST_TOPIC);
runMessageListenerTest(100, 2, 5, 1000, 1, "none", TEST_TOPIC);
}
@Test
@Ignore
public void testHighVolumeLowConcurrency() throws Exception {
createTopic(TEST_TOPIC, 5, 5, 1);
runMessageListenerTest(100, 2, 5, 10000, 1, 0, TEST_TOPIC);
runMessageListenerTest(100, 2, 5, 10000, 1, "none", TEST_TOPIC);
}
@Test
@Ignore
public void testLowVolumeMediumConcurrency() throws Exception {
createTopic(TEST_TOPIC, 5, 5, 1);
runMessageListenerTest(100, 5, 5, 100, 1, 0, TEST_TOPIC);
runMessageListenerTest(100, 5, 5, 100, 1, "none", TEST_TOPIC);
}
@Test
@Ignore
public void testMediumVolumeMediumConcurrency() throws Exception {
createTopic(TEST_TOPIC, 5, 5, 1);
runMessageListenerTest(100, 5, 5, 1000, 1, 0, TEST_TOPIC);
runMessageListenerTest(100, 5, 5, 1000, 1, "none", TEST_TOPIC);
}
@Test
@Ignore
public void testHighVolumeMediumConcurrency() throws Exception {
createTopic(TEST_TOPIC, 5, 5, 1);
runMessageListenerTest(100, 5, 5, 100000, 1, 0, TEST_TOPIC);
runMessageListenerTest(100, 5, 5, 100000, 1, "none", TEST_TOPIC);
}
@@ -82,21 +82,21 @@ public class MultiBroker5Tests extends AbstractMessageListenerContainerTests {
@Ignore
public void testLowVolumeHighConcurrency() throws Exception {
createTopic(TEST_TOPIC, 100, 5, 1);
runMessageListenerTest(100, 20, 100, 1000, 1, 0, TEST_TOPIC);
runMessageListenerTest(100, 20, 100, 1000, 1, "none", TEST_TOPIC);
}
@Test
@Ignore
public void testMediumVolumeHighConcurrency() throws Exception {
createTopic(TEST_TOPIC, 100, 5, 1);
runMessageListenerTest(100, 20, 100, 10000, 1, 0, TEST_TOPIC);
runMessageListenerTest(100, 20, 100, 10000, 1, "none", TEST_TOPIC);
}
@Test
@Ignore
public void testHighVolumeHighConcurrency() throws Exception {
createTopic(TEST_TOPIC, 100, 5, 1);
runMessageListenerTest(100, 20, 100, 100000, 1, 0, TEST_TOPIC);
runMessageListenerTest(100, 20, 100, 100000, 1, "none", TEST_TOPIC);
}
}

View File

@@ -55,7 +55,7 @@ public class SingleBrokerExternalTests extends AbstractMessageListenerContainerT
public void testLowVolumeLowConcurrency() throws Exception {
String topicName = generateTopicName();
createTopic(topicName, 5, 1, 1);
runMessageListenerTest(100, 2, 5, 100, 1, 0, topicName);
runMessageListenerTest(100, 2, 5, 100, 1, "none", topicName);
deleteTopic(topicName);
}
@@ -63,7 +63,7 @@ public class SingleBrokerExternalTests extends AbstractMessageListenerContainerT
public void testMediumVolumeLowConcurrency() throws Exception {
String topicName = generateTopicName();
createTopic(topicName, 5, 1, 1);
runMessageListenerTest(100, 2, 5, 1000, 1, 0, topicName);
runMessageListenerTest(100, 2, 5, 1000, 1, "none", topicName);
deleteTopic(topicName);
}
@@ -72,7 +72,7 @@ public class SingleBrokerExternalTests extends AbstractMessageListenerContainerT
public void testHighVolumeLowConcurrency() throws Exception {
String topicName = generateTopicName();
createTopic(topicName, 5, 1, 1);
runMessageListenerTest(100, 2, 5, 10000, 1, 0, topicName);
runMessageListenerTest(100, 2, 5, 10000, 1, "none", topicName);
deleteTopic(topicName);
}
@@ -80,7 +80,7 @@ public class SingleBrokerExternalTests extends AbstractMessageListenerContainerT
public void testLowVolumeMediumConcurrency() throws Exception {
String topicName = generateTopicName();
createTopic(topicName, 5, 1, 1);
runMessageListenerTest(100, 5, 5, 100, 1, 0, topicName);
runMessageListenerTest(100, 5, 5, 100, 1, "none", topicName);
deleteTopic(topicName);
}
@@ -88,7 +88,7 @@ public class SingleBrokerExternalTests extends AbstractMessageListenerContainerT
public void testMediumVolumeMediumConcurrency() throws Exception {
String topicName = generateTopicName();
createTopic(topicName, 5, 1, 1);
runMessageListenerTest(100, 5, 5, 1000, 1, 0, topicName);
runMessageListenerTest(100, 5, 5, 1000, 1, "none", topicName);
deleteTopic(topicName);
}
@@ -97,7 +97,7 @@ public class SingleBrokerExternalTests extends AbstractMessageListenerContainerT
public void testHighVolumeMediumConcurrency() throws Exception {
String topicName = generateTopicName();
createTopic(topicName, 5, 1, 1);
runMessageListenerTest(100, 5, 5, 100000, 1, 0, topicName);
runMessageListenerTest(100, 5, 5, 100000, 1, "none", topicName);
deleteTopic(topicName);
}
@@ -106,7 +106,7 @@ public class SingleBrokerExternalTests extends AbstractMessageListenerContainerT
public void testLowVolumeHighConcurrency() throws Exception {
String topicName = generateTopicName();
createTopic(topicName, 100, 1, 1);
runMessageListenerTest(100, 20, 100, 1000, 1, 0, topicName);
runMessageListenerTest(100, 20, 100, 1000, 1, "none", topicName);
deleteTopic(topicName);
}
@@ -115,7 +115,7 @@ public class SingleBrokerExternalTests extends AbstractMessageListenerContainerT
public void testMediumVolumeHighConcurrency() throws Exception {
String topicName = generateTopicName();
createTopic(topicName, 100, 1, 1);
runMessageListenerTest(100, 20, 100, 10000, 1, 0, topicName);
runMessageListenerTest(100, 20, 100, 10000, 1, "none", topicName);
deleteTopic(topicName);
}
@@ -124,7 +124,7 @@ public class SingleBrokerExternalTests extends AbstractMessageListenerContainerT
public void testHighVolumeHighConcurrency() throws Exception {
String topicName = generateTopicName();
createTopic(topicName, 100, 1, 1);
runMessageListenerTest(100, 20, 100, 100000, 1, 0, topicName);
runMessageListenerTest(100, 20, 100, 100000, 1, "none", topicName);
deleteTopic(topicName);
}

View File

@@ -55,7 +55,9 @@ public class SingleBrokerRecoveryTests extends AbstractMessageListenerContainerT
@Test
public void testCompleteShutdown() throws Exception {
createTopic(TEST_TOPIC, 1, 1, 1);
int partitionCount = 1;
createTopic(TEST_TOPIC, partitionCount, 1, 1);
ConnectionFactory connectionFactory = getKafkaBrokerConnectionFactory();
ArrayList<Partition> readPartitions = new ArrayList<Partition>();
@@ -68,7 +70,7 @@ public class SingleBrokerRecoveryTests extends AbstractMessageListenerContainerT
final int expectedMessageCount = 100;
createStringProducer(0).send(createMessages(10, TEST_TOPIC));
createMessageSender("none").send(createMessages(10, TEST_TOPIC,partitionCount));
final MutableListMultimap<Integer, KeyedMessageWithOffset> receivedData =
new SynchronizedPutFastListMultimap<Integer, KeyedMessageWithOffset>();
@@ -100,7 +102,7 @@ public class SingleBrokerRecoveryTests extends AbstractMessageListenerContainerT
kafkaEmbeddedBrokerRule.restart(0);
// now start sending messages again
createStringProducer(0).send(createMessages(90, TEST_TOPIC));
createMessageSender("none").send(createMessages(90, TEST_TOPIC,partitionCount));
latch.await(50, TimeUnit.SECONDS);
kafkaMessageListenerContainer.stop();
@@ -109,7 +111,7 @@ public class SingleBrokerRecoveryTests extends AbstractMessageListenerContainerT
assertThat(latch.getCount(), equalTo(0L));
System.out.println("All messages received ... checking ");
validateMessageReceipt(receivedData, 1, 1, 100, expectedMessageCount, readPartitions, 1);
validateMessageReceipt(receivedData, 1, partitionCount, expectedMessageCount, 1);
}

View File

@@ -39,60 +39,60 @@ public class SingleBrokerTests extends AbstractMessageListenerContainerTests {
@Test
public void testLowVolumeLowConcurrency() throws Exception {
createTopic(TEST_TOPIC, 5, 1, 1);
runMessageListenerTest(100, 2, 5, 100, 1, 0, TEST_TOPIC);
runMessageListenerTest(100, 2, 5, 100, 1, "none", TEST_TOPIC);
}
@Test
public void testMediumVolumeLowConcurrency() throws Exception {
createTopic(TEST_TOPIC, 5, 1, 1);
runMessageListenerTest(100, 2, 5, 1000, 1, 0, TEST_TOPIC);
runMessageListenerTest(100, 2, 5, 1000, 1, "none", TEST_TOPIC);
}
@Test
@Ignore
public void testHighVolumeLowConcurrency() throws Exception {
createTopic(TEST_TOPIC, 5, 1, 1);
runMessageListenerTest(100, 2, 5, 10000, 1, 0, TEST_TOPIC);
runMessageListenerTest(100, 2, 5, 10000, 1, "none", TEST_TOPIC);
}
@Test
public void testLowVolumeMediumConcurrency() throws Exception {
createTopic(TEST_TOPIC, 5, 1, 1);
runMessageListenerTest(100, 5, 5, 100, 1, 0, TEST_TOPIC);
runMessageListenerTest(100, 5, 5, 100, 1, "none", TEST_TOPIC);
}
@Test
public void testMediumVolumeMediumConcurrency() throws Exception {
createTopic(TEST_TOPIC, 5, 1, 1);
runMessageListenerTest(100, 5, 5, 1000, 1, 0, TEST_TOPIC);
runMessageListenerTest(100, 5, 5, 1000, 1, "none", TEST_TOPIC);
}
@Test
@Ignore
public void testHighVolumeMediumConcurrency() throws Exception {
createTopic(TEST_TOPIC, 5, 1, 1);
runMessageListenerTest(100, 5, 5, 100000, 1, 0, TEST_TOPIC);
runMessageListenerTest(100, 5, 5, 100000, 1, "none", TEST_TOPIC);
}
@Test
@Ignore
public void testLowVolumeHighConcurrency() throws Exception {
createTopic(TEST_TOPIC, 100, 1, 1);
runMessageListenerTest(100, 20, 100, 1000, 1, 0, TEST_TOPIC);
runMessageListenerTest(100, 20, 100, 1000, 1, "none", TEST_TOPIC);
}
@Test
@Ignore
public void testMediumVolumeHighConcurrency() throws Exception {
createTopic(TEST_TOPIC, 100, 1, 1);
runMessageListenerTest(100, 20, 100, 10000, 1, 0, TEST_TOPIC);
runMessageListenerTest(100, 20, 100, 10000, 1, "none", TEST_TOPIC);
}
@Test
@Ignore
public void testHighVolumeHighConcurrency() throws Exception {
createTopic(TEST_TOPIC, 100, 1, 1);
runMessageListenerTest(100, 20, 100, 100000, 1, 0, TEST_TOPIC);
runMessageListenerTest(100, 20, 100, 100000, 1, "none", TEST_TOPIC);
}
}

View File

@@ -39,60 +39,60 @@ public class SingleBrokerWithCompressionTests extends AbstractMessageListenerCon
@Test
public void testLowVolumeLowConcurrency() throws Exception {
createTopic(TEST_TOPIC, 5, 1, 1);
runMessageListenerTest(10000, 2, 5, 100, 1, 1, TEST_TOPIC);
runMessageListenerTest(10000, 2, 5, 100, 1, "gzip", TEST_TOPIC);
}
@Test
public void testMediumVolumeLowConcurrency() throws Exception {
createTopic(TEST_TOPIC, 5, 1, 1);
runMessageListenerTest(10000, 2, 5, 1000, 1, 1, TEST_TOPIC);
runMessageListenerTest(10000, 2, 5, 1000, 1, "gzip", TEST_TOPIC);
}
@Test
@Ignore
public void testHighVolumeLowConcurrency() throws Exception {
createTopic(TEST_TOPIC, 5, 1, 1);
runMessageListenerTest(10000, 2, 5, 10000, 1, 1, TEST_TOPIC);
runMessageListenerTest(10000, 2, 5, 10000, 1, "gzip", TEST_TOPIC);
}
@Test
public void testLowVolumeMediumConcurrency() throws Exception {
createTopic(TEST_TOPIC, 5, 1, 1);
runMessageListenerTest(10000, 5, 5, 100, 1, 1, TEST_TOPIC);
runMessageListenerTest(10000, 5, 5, 100, 1, "gzip", TEST_TOPIC);
}
@Test
public void testMediumVolumeMediumConcurrency() throws Exception {
createTopic(TEST_TOPIC, 5, 1, 1);
runMessageListenerTest(10000, 5, 5, 1000, 1, 1, TEST_TOPIC);
runMessageListenerTest(10000, 5, 5, 1000, 1, "gzip", TEST_TOPIC);
}
@Test
@Ignore
public void testHighVolumeMediumConcurrency() throws Exception {
createTopic(TEST_TOPIC, 5, 1, 1);
runMessageListenerTest(10000, 5, 5, 100000, 1, 1, TEST_TOPIC);
runMessageListenerTest(10000, 5, 5, 100000, 1, "gzip", TEST_TOPIC);
}
@Test
@Ignore
public void testLowVolumeHighConcurrency() throws Exception {
createTopic(TEST_TOPIC, 100, 1, 1);
runMessageListenerTest(10000, 20, 100, 1000, 1, 1, TEST_TOPIC);
runMessageListenerTest(10000, 20, 100, 1000, 1, "gzip", TEST_TOPIC);
}
@Test
@Ignore
public void testMediumVolumeHighConcurrency() throws Exception {
createTopic(TEST_TOPIC, 100, 1, 1);
runMessageListenerTest(10000, 20, 100, 10000, 1, 1, TEST_TOPIC);
runMessageListenerTest(10000, 20, 100, 10000, 1, "gzip", TEST_TOPIC);
}
@Test
@Ignore
public void testHighVolumeHighConcurrency() throws Exception {
createTopic(TEST_TOPIC, 100, 1, 1);
runMessageListenerTest(10000, 20, 100, 100000, 1, 1, TEST_TOPIC);
runMessageListenerTest(10000, 20, 100, 100000, 1, "gzip", TEST_TOPIC);
}
}

View File

@@ -64,9 +64,7 @@ public class SingleBrokerWithManualAckTests extends AbstractMessageListenerConta
ConnectionFactory connectionFactory = getKafkaBrokerConnectionFactory();
ArrayList<Partition> readPartitions = new ArrayList<Partition>();
for (int i = 0; i < 5; i++) {
if (i % 1 == 0) {
readPartitions.add(new Partition(TEST_TOPIC, i));
}
readPartitions.add(new Partition(TEST_TOPIC, i));
}
final KafkaMessageListenerContainer kafkaMessageListenerContainer = new KafkaMessageListenerContainer(
connectionFactory, readPartitions.toArray(new Partition[readPartitions.size()]));
@@ -78,11 +76,13 @@ public class SingleBrokerWithManualAckTests extends AbstractMessageListenerConta
@Test
public void testLowVolumeLowConcurrency() throws Exception {
createTopic(TEST_TOPIC, 5, 1, 1);
int partitionCount = 5;
createTopic(TEST_TOPIC, partitionCount, 1, 1);
ConnectionFactory connectionFactory = getKafkaBrokerConnectionFactory();
ArrayList<Partition> readPartitions = new ArrayList<Partition>();
for (int i = 0; i < 5; i++) {
for (int i = 0; i < partitionCount; i++) {
if (i % 1 == 0) {
readPartitions.add(new Partition(TEST_TOPIC, i));
}
@@ -118,7 +118,7 @@ public class SingleBrokerWithManualAckTests extends AbstractMessageListenerConta
kafkaMessageListenerContainer.start();
createStringProducer(0).send(createMessages(100, TEST_TOPIC));
createMessageSender("none").send(createMessages(100, TEST_TOPIC, partitionCount));
latch.await((expectedMessageCount / 5000) + 1, TimeUnit.MINUTES);
kafkaMessageListenerContainer.stop();
@@ -127,7 +127,7 @@ public class SingleBrokerWithManualAckTests extends AbstractMessageListenerConta
assertThat(latch.getCount(), equalTo(0L));
System.out.println("All messages received ... checking ");
validateMessageReceipt(receivedData, 2, 5, 100, expectedMessageCount, readPartitions, 1);
validateMessageReceipt(receivedData, 2, partitionCount, expectedMessageCount, 1);
// at this point, all messages have been processed but not acknowledged
for (Partition readPartition : readPartitions) {

View File

@@ -39,42 +39,42 @@ public class SingleBrokerWithPartitionSubsetTests extends AbstractMessageListene
@Test
public void testLowVolumeLowConcurrency() throws Exception {
createTopic(TEST_TOPIC, 4, 1, 1);
runMessageListenerTest(100, 2, 4, 100, 2, 0, TEST_TOPIC);
runMessageListenerTest(100, 2, 4, 100, 2, "none", TEST_TOPIC);
}
@Test
@Ignore
public void testMediumVolumeLowConcurrency() throws Exception {
createTopic(TEST_TOPIC, 4, 1, 1);
runMessageListenerTest(100, 2, 4, 1000, 2, 0, TEST_TOPIC);
runMessageListenerTest(100, 2, 4, 1000, 2, "none", TEST_TOPIC);
}
@Test
@Ignore
public void testHighVolumeLowConcurrency() throws Exception {
createTopic(TEST_TOPIC, 4, 1, 1);
runMessageListenerTest(100, 2, 4, 10000, 2, 0, TEST_TOPIC);
runMessageListenerTest(100, 2, 4, 10000, 2, "none", TEST_TOPIC);
}
@Test
@Ignore
public void testLowVolumeMediumConcurrency() throws Exception {
createTopic(TEST_TOPIC, 10, 1, 1);
runMessageListenerTest(100, 5, 10, 100, 2, 0, TEST_TOPIC);
runMessageListenerTest(100, 5, 10, 100, 2, "none", TEST_TOPIC);
}
@Test
@Ignore
public void testMediumVolumeMediumConcurrency() throws Exception {
createTopic(TEST_TOPIC, 10, 1, 1);
runMessageListenerTest(100, 5, 10, 1000, 2, 0, TEST_TOPIC);
runMessageListenerTest(100, 5, 10, 1000, 2, "none", TEST_TOPIC);
}
@Test
@Ignore
public void testHighVolumeMediumConcurrency() throws Exception {
createTopic(TEST_TOPIC, 10, 1, 1);
runMessageListenerTest(100, 5, 10, 100000, 2, 0, TEST_TOPIC);
runMessageListenerTest(100, 5, 10, 100000, 2, "none", TEST_TOPIC);
}
@@ -82,21 +82,21 @@ public class SingleBrokerWithPartitionSubsetTests extends AbstractMessageListene
@Ignore
public void testLowVolumeHighConcurrency() throws Exception {
createTopic(TEST_TOPIC, 100, 1, 1);
runMessageListenerTest(100, 20, 100, 1000, 2, 0, TEST_TOPIC);
runMessageListenerTest(100, 20, 100, 1000, 2, "none", TEST_TOPIC);
}
@Test
@Ignore
public void testMediumVolumeHighConcurrency() throws Exception {
createTopic(TEST_TOPIC, 100, 1, 1);
runMessageListenerTest(100, 20, 100, 10000, 2, 0, TEST_TOPIC);
runMessageListenerTest(100, 20, 100, 10000, 2, "none", TEST_TOPIC);
}
@Test
@Ignore
public void testHighVolumeHighConcurrency() throws Exception {
createTopic(TEST_TOPIC, 100, 1, 1);
runMessageListenerTest(100, 20, 100, 100000, 2, 0, TEST_TOPIC);
runMessageListenerTest(100, 20, 100, 100000, 2, "none", TEST_TOPIC);
}

View File

@@ -64,11 +64,14 @@ public class ZookeeperConfigurationTests extends AbstractMessageListenerContaine
@Test
public void testLowVolumeLowConcurrency() throws Exception {
createTopic(TEST_TOPIC, 5, 2, 1);
int partitionCount = 5;
createTopic(TEST_TOPIC, partitionCount, 2, 1);
ConnectionFactory connectionFactory = getKafkaBrokerConnectionFactory();
ArrayList<Partition> readPartitions = new ArrayList<Partition>();
for (int i = 0; i < 5; i++) {
for (int i = 0; i < partitionCount; i++) {
readPartitions.add(new Partition(TEST_TOPIC, i));
}
final KafkaMessageListenerContainer kafkaMessageListenerContainer = new KafkaMessageListenerContainer(connectionFactory, readPartitions.toArray(new Partition[readPartitions.size()]));
@@ -90,16 +93,16 @@ public class ZookeeperConfigurationTests extends AbstractMessageListenerContaine
kafkaMessageListenerContainer.start();
createStringProducer(0).send(createMessages(100, TEST_TOPIC));
createMessageSender("none").send(createMessages(100, TEST_TOPIC, partitionCount));
latch.await((expectedMessageCount/5000) + 1, TimeUnit.MINUTES);
latch.await((expectedMessageCount / 5000) + 1, TimeUnit.MINUTES);
kafkaMessageListenerContainer.stop();
assertThat(receivedData.valuesView().toList(), hasSize(expectedMessageCount));
assertThat(latch.getCount(), equalTo(0L));
System.out.println("All messages received ... checking ");
validateMessageReceipt(receivedData, 2, 5, 100, expectedMessageCount, readPartitions, 1);
validateMessageReceipt(receivedData, 2, partitionCount, expectedMessageCount, 1);
}

View File

@@ -24,6 +24,10 @@ import java.util.HashMap;
import java.util.Map;
import java.util.Properties;
import kafka.api.OffsetRequest;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.junit.Rule;
import org.junit.Test;
@@ -32,18 +36,12 @@ import org.springframework.integration.kafka.core.Partition;
import org.springframework.integration.kafka.core.PartitionNotFoundException;
import org.springframework.integration.kafka.core.ZookeeperConfiguration;
import org.springframework.integration.kafka.listener.OffsetManager;
import org.springframework.integration.kafka.listener.TestPartitioner;
import org.springframework.integration.kafka.rule.KafkaEmbedded;
import org.springframework.integration.kafka.rule.KafkaRule;
import org.springframework.integration.kafka.serializer.common.StringEncoder;
import org.springframework.integration.kafka.util.EncoderAdaptingSerializer;
import org.springframework.integration.kafka.util.TopicUtils;
import kafka.api.OffsetRequest;
import kafka.javaapi.producer.Producer;
import kafka.producer.KeyedMessage;
import kafka.producer.ProducerConfig;
import kafka.serializer.StringEncoder;
import kafka.utils.TestUtils;
/**
* @author Marius Bogoevici
*/
@@ -81,7 +79,7 @@ public abstract class AbstractOffsetManagerTests {
// send data to increase offsets on topics
Producer<String, String> producer = createProducer();
for (int i = 0; i < 10; i++) {
producer.send(new KeyedMessage<String, String>(TEST_TOPIC, String.valueOf(i), i, String.valueOf(i)));
producer.send(new ProducerRecord<String, String>(TEST_TOPIC, i%3, String.valueOf(i), String.valueOf(i))).get();
}
// earliest time resets at the start of the queue
@@ -90,6 +88,8 @@ public abstract class AbstractOffsetManagerTests {
assertThat(offsetManager2.getOffset(partitions[1]), equalTo(0L));
assertThat(offsetManager2.getOffset(partitions[2]), equalTo(0L));
offsetManager1.close();
// latest time resets at the end of the queue
OffsetManager offsetManager3 = createOffsetManager(OffsetRequest.LatestTime(), "offset3");
assertThat(offsetManager3.getOffset(partitions[0]), equalTo(4L));
@@ -118,7 +118,8 @@ public abstract class AbstractOffsetManagerTests {
// send data to increase offsets on topics
Producer<String, String> producer = createProducer();
for (int i = 0; i < 10; i++) {
producer.send(new KeyedMessage<String, String>(TEST_TOPIC, String.valueOf(i), i, String.valueOf(i)));
producer.send(new ProducerRecord<String, String>(TEST_TOPIC,
i%numPartitions, String.valueOf(i), String.valueOf(i))).get();
}
assertThat(offsetManager1.getOffset(partitions[0]), equalTo(0L));
@@ -139,6 +140,8 @@ public abstract class AbstractOffsetManagerTests {
assertThat(newOffsetManager1.getOffset(partitions[1]), equalTo(4L));
assertThat(newOffsetManager1.getOffset(partitions[2]), equalTo(3L));
offsetManager1.close();
// a new offset manager with a different consumerId will not observe the updates
OffsetManager offsetManager2 = createOffsetManager(OffsetRequest.EarliestTime(), "offset2");
assertThat(offsetManager2.getOffset(partitions[0]), equalTo(0L));
@@ -167,7 +170,7 @@ public abstract class AbstractOffsetManagerTests {
// send data to increase offsets on topics
Producer<String, String> producer = createProducer();
for (int i = 0; i < 6; i++) {
producer.send(new KeyedMessage<String, String>(TEST_TOPIC, String.valueOf(i), i, String.valueOf(i)));
producer.send(new ProducerRecord<String, String>(TEST_TOPIC, i%numPartitions, String.valueOf(i), String.valueOf(i))).get();
}
assertThat(offsetManager1.getOffset(partitions[0]), equalTo(0L));
@@ -192,6 +195,8 @@ public abstract class AbstractOffsetManagerTests {
assertThat(newOffsetManager1.getOffset(partitions[1]), equalTo(0L));
assertThat(newOffsetManager1.getOffset(partitions[2]), equalTo(0L));
offsetManager1.close();
// a new offset manager with a different consumerId will not observe the updates
OffsetManager offsetManager2 = createOffsetManager(OffsetRequest.LatestTime(), "offset2");
assertThat(offsetManager2.getOffset(partitions[0]), equalTo(2L));
@@ -215,12 +220,14 @@ public abstract class AbstractOffsetManagerTests {
initialOffsets.put(partitions[1], 9L);
initialOffsets.put(partitions[2], 10L);
OffsetManager offsetManager1 = createOffsetManager(OffsetRequest.EarliestTime(), "offset1", initialOffsets);
// send data to increase offsets on topics
Producer<String, String> producer = createProducer();
for (int i = 0; i < 6; i++) {
producer.send(new KeyedMessage<String, String>(TEST_TOPIC, String.valueOf(i), i, String.valueOf(i)));
producer.send(new ProducerRecord<String, String>(TEST_TOPIC, i%numPartitions, String.valueOf(i), String.valueOf(i))).get();
}
// The offset manager starts at the configured offsets
@@ -262,6 +269,8 @@ public abstract class AbstractOffsetManagerTests {
assertThat(offsetManager1.getOffset(partitions[1]), equalTo(0L));
assertThat(offsetManager1.getOffset(partitions[2]), equalTo(0L));
offsetManager1.close();
// a new offset manager with a different consumerId will not observe the updates
OffsetManager offsetManager2 = createOffsetManager(OffsetRequest.LatestTime(), "offset2");
assertThat(offsetManager2.getOffset(partitions[0]), equalTo(2L));
@@ -277,13 +286,11 @@ public abstract class AbstractOffsetManagerTests {
}
protected Producer<String, String> createProducer() {
Properties properties = TestUtils.getProducerConfig(kafkaRule.getBrokersAsString(),
TestPartitioner.class.getCanonicalName());
properties.put("serializer.class", StringEncoder.class.getCanonicalName());
properties.put("key.serializer.class", StringEncoder.class.getCanonicalName());
ProducerConfig producerConfig = new ProducerConfig(properties);
return new Producer<String, String>(new kafka.producer.Producer<String, String>(producerConfig));
}
Properties properties = new Properties();
properties.setProperty("bootstrap.servers", kafkaRule.getBrokersAsString());
EncoderAdaptingSerializer<String> serializer = new EncoderAdaptingSerializer<>(new StringEncoder());
return new KafkaProducer<String, String>(properties, serializer, serializer);
}
protected OffsetManager createOffsetManager(long referenceTimestamp, String consumerId) throws Exception {
return createOffsetManager(referenceTimestamp, consumerId, new HashMap<Partition, Long>());

View File

@@ -37,9 +37,9 @@ public class CodecTests {
@Test
public void testKeyCodec() {
KafkaTopicOffsetManager.KeyEncoderDecoder codec = new KafkaTopicOffsetManager.KeyEncoderDecoder();
KafkaTopicOffsetManager.KeySerializerDecoder codec = new KafkaTopicOffsetManager.KeySerializerDecoder();
byte[] encodedValue = codec.toBytes(new KafkaTopicOffsetManager.Key(SOME_CONSUMER,
byte[] encodedValue = codec.serialize("#unused", new KafkaTopicOffsetManager.Key(SOME_CONSUMER,
new Partition(SOME_TOPIC, SOME_PARTITION_ID)));
KafkaTopicOffsetManager.Key key = codec.fromBytes(encodedValue);

View File

@@ -16,7 +16,6 @@
package org.springframework.integration.kafka.offset;
import java.util.HashMap;
import java.util.Map;
import org.springframework.integration.kafka.core.Partition;
@@ -37,8 +36,6 @@ public class KafkaOffsetManagerTests extends AbstractOffsetManagerTests {
"offsets-spring-test",
initialOffsets);
kafkaTopicOffsetManager.setConsumerId(consumerId);
// do not batch writes during tests for deterministic behaviour
kafkaTopicOffsetManager.setBatchWrites(false);
kafkaTopicOffsetManager.afterPropertiesSet();
kafkaTopicOffsetManager.setReferenceTimestamp(referenceTimestamp);
return kafkaTopicOffsetManager;

View File

@@ -26,11 +26,19 @@ import static org.junit.Assert.assertThat;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.Properties;
import java.util.UUID;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import com.gs.collections.api.multimap.MutableMultimap;
import com.gs.collections.impl.factory.Multimaps;
import kafka.admin.AdminUtils;
import kafka.api.OffsetRequest;
import kafka.common.TopicExistsException;
import kafka.serializer.Decoder;
import kafka.serializer.Encoder;
import org.junit.After;
import org.junit.Rule;
import org.junit.Test;
@@ -54,19 +62,11 @@ import org.springframework.integration.kafka.support.ProducerConfiguration;
import org.springframework.integration.kafka.support.ProducerFactoryBean;
import org.springframework.integration.kafka.support.ProducerMetadata;
import org.springframework.integration.kafka.support.ZookeeperConnect;
import org.springframework.integration.kafka.util.EncoderAdaptingSerializer;
import org.springframework.integration.kafka.util.MessageUtils;
import org.springframework.integration.kafka.util.TopicUtils;
import org.springframework.messaging.support.MessageBuilder;
import com.gs.collections.api.multimap.MutableMultimap;
import com.gs.collections.impl.factory.Multimaps;
import kafka.admin.AdminUtils;
import kafka.api.OffsetRequest;
import kafka.common.TopicExistsException;
import kafka.serializer.Decoder;
import kafka.serializer.Encoder;
/**
* @author Gary Russell
* @author Marius Bogoevici
@@ -124,9 +124,9 @@ public class OutboundTests {
kafkaMessageListenerContainer.start();
KafkaProducerContext<String, String> producerContext = createProducerContext();
KafkaProducerMessageHandler<String, String> handler =
new KafkaProducerMessageHandler<String, String>(producerContext);
KafkaProducerContext producerContext = createProducerContext();
KafkaProducerMessageHandler handler =
new KafkaProducerMessageHandler(producerContext);
handler.handleMessage(MessageBuilder.withPayload("foo" + suffix)
.setHeader(KafkaHeaders.MESSAGE_KEY, "3")
@@ -195,9 +195,9 @@ public class OutboundTests {
kafkaMessageListenerContainer.start();
KafkaProducerContext<String, String> producerContext = createProducerContext();
KafkaProducerMessageHandler<String, String> handler
= new KafkaProducerMessageHandler<String, String>(producerContext);
KafkaProducerContext producerContext = createProducerContext();
KafkaProducerMessageHandler handler
= new KafkaProducerMessageHandler(producerContext);
handler.handleMessage(MessageBuilder.withPayload("fooTopic1" + suffix)
.setHeader(KafkaHeaders.MESSAGE_KEY, "3")
@@ -273,9 +273,8 @@ public class OutboundTests {
kafkaMessageListenerContainer.start();
KafkaProducerContext<String, String> producerContext = createProducerContext();
KafkaProducerMessageHandler<String, String> handler
= new KafkaProducerMessageHandler<String, String>(producerContext);
KafkaProducerContext producerContext = createProducerContext();
KafkaProducerMessageHandler handler = new KafkaProducerMessageHandler(producerContext);
handler.handleMessage(MessageBuilder.withPayload("fooTopic1" + suffix).build());
@@ -309,22 +308,18 @@ public class OutboundTests {
return kafkaMessageListenerContainer;
}
private KafkaProducerContext<String, String> createProducerContext() throws Exception {
KafkaProducerContext<String, String> kafkaProducerContext = new KafkaProducerContext<String, String>();
ProducerMetadata<String, String> producerMetadata = new ProducerMetadata<String, String>(TOPIC);
producerMetadata.setValueClassType(String.class);
producerMetadata.setKeyClassType(String.class);
Encoder<String> encoder = new StringEncoder<String>();
producerMetadata.setValueEncoder(encoder);
producerMetadata.setKeyEncoder(encoder);
producerMetadata.setAsync(true);
private KafkaProducerContext createProducerContext() throws Exception {
KafkaProducerContext kafkaProducerContext = new KafkaProducerContext();
Encoder<String> encoder = new StringEncoder();
ProducerMetadata<String, String> producerMetadata = new ProducerMetadata<String, String>(TOPIC, String.class, String.class, new EncoderAdaptingSerializer<String>(encoder), new EncoderAdaptingSerializer<String>(encoder));
Properties props = new Properties();
props.put("queue.buffering.max.ms", "15000");
props.put("linger.ms", "15000");
ProducerFactoryBean<String, String> producer =
new ProducerFactoryBean<String, String>(producerMetadata, kafkaRule.getBrokersAsString(), props);
ProducerConfiguration<String, String> config =
new ProducerConfiguration<String, String>(producerMetadata, producer.getObject());
kafkaProducerContext.setProducerConfigurations(Collections.singletonMap(TOPIC, config));
Map<String, ProducerConfiguration<?, ?>> producerConfigurationMap = Collections.<String, ProducerConfiguration<?, ?>>singletonMap(TOPIC, config);
kafkaProducerContext.setProducerConfigurations(producerConfigurationMap);
return kafkaProducerContext;
}

View File

@@ -25,13 +25,9 @@ import java.util.Collections;
import java.util.List;
import java.util.Properties;
import com.gs.collections.api.block.function.Function;
import com.gs.collections.impl.list.mutable.FastList;
import com.gs.collections.impl.utility.ListIterate;
import kafka.Kafka;
import kafka.server.KafkaConfig;
import kafka.server.KafkaServer;
import kafka.server.NotRunning;
import kafka.utils.SystemTime$;
import kafka.utils.TestUtils;
import kafka.utils.TestZKUtils;
@@ -42,18 +38,19 @@ import kafka.zk.EmbeddedZookeeper;
import org.I0Itec.zkclient.ZkClient;
import org.I0Itec.zkclient.exception.ZkInterruptedException;
import org.junit.rules.ExternalResource;
import scala.collection.JavaConversions;
import org.springframework.integration.kafka.core.BrokerAddress;
import org.springframework.retry.RetryCallback;
import org.springframework.retry.RetryContext;
import org.springframework.retry.RetryPolicy;
import org.springframework.retry.backoff.BackOffPolicy;
import org.springframework.retry.backoff.ExponentialBackOffPolicy;
import org.springframework.retry.policy.SimpleRetryPolicy;
import org.springframework.retry.support.RetryTemplate;
import scala.collection.JavaConversions;
import com.gs.collections.api.block.function.Function;
import com.gs.collections.impl.list.mutable.FastList;
import com.gs.collections.impl.utility.ListIterate;
/**
* @author Marius Bogoevici
* @author Artem Bilan
@@ -93,8 +90,9 @@ public class KafkaEmbedded extends ExternalResource implements KafkaRule {
ZKStringSerializer$.MODULE$);
kafkaServers = new ArrayList<KafkaServer>();
for (int i = 0; i < count; i++) {
Properties brokerConfigProperties = TestUtils.createBrokerConfig(i, kafkaPorts.get(i));
brokerConfigProperties.put("controlled.shutdown.enable", Boolean.toString(controlledShutdown));
Properties brokerConfigProperties = TestUtils.createBrokerConfig(i, kafkaPorts.get(i),controlledShutdown);
brokerConfigProperties.setProperty("replica.socket.timeout.ms","1000");
brokerConfigProperties.setProperty("controller.socket.timeout.ms","1000");
KafkaServer server = TestUtils.createServer(new KafkaConfig(brokerConfigProperties), SystemTime$.MODULE$);
kafkaServers.add(server);
}
@@ -104,7 +102,10 @@ public class KafkaEmbedded extends ExternalResource implements KafkaRule {
protected void after() {
for (KafkaServer kafkaServer : kafkaServers) {
try {
kafkaServer.shutdown();
if (kafkaServer.brokerState().currentState() != (NotRunning.state())) {
kafkaServer.shutdown();
kafkaServer.awaitShutdown();
}
}
catch (Exception e) {
// do nothing
@@ -170,6 +171,7 @@ public class KafkaEmbedded extends ExternalResource implements KafkaRule {
for (KafkaServer kafkaServer : getKafkaServers()) {
if (brokerAddress.equals(new BrokerAddress(kafkaServer.config().hostName(), kafkaServer.config().port()))) {
kafkaServer.shutdown();
kafkaServer.awaitShutdown();
}
}
}

View File

@@ -18,13 +18,11 @@ package org.springframework.integration.kafka.support;
import java.util.HashMap;
import java.util.Map;
import org.apache.kafka.clients.producer.Producer;
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
@@ -47,7 +45,7 @@ public class KafkaProducerContextTests {
final ProducerConfiguration<String, String> producerConfiguration = new ProducerConfiguration<String, String>(producerMetadata, producer);
final Map<String, ProducerConfiguration> topicConfigurations = new HashMap<String, ProducerConfiguration>();
final Map<String, ProducerConfiguration<?,?>> topicConfigurations = new HashMap<String, ProducerConfiguration<?,?>>();
topicConfigurations.put(testRegex, producerConfiguration);
kafkaProducerContext.setProducerConfigurations(topicConfigurations);

View File

@@ -17,24 +17,27 @@
package org.springframework.integration.kafka.support;
import java.io.ByteArrayInputStream;
import java.io.NotSerializableException;
import java.io.ObjectInputStream;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.ByteArraySerializer;
import org.apache.kafka.common.serialization.StringSerializer;
import org.junit.Assert;
import org.junit.Test;
import org.mockito.ArgumentCaptor;
import org.mockito.Mockito;
import org.springframework.core.convert.ConversionFailedException;
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;
import org.springframework.integration.kafka.test.utils.TestPayload;
import org.springframework.integration.kafka.util.EncoderAdaptingSerializer;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.GenericMessage;
import kafka.javaapi.producer.Producer;
import kafka.producer.KeyedMessage;
import kafka.serializer.DefaultEncoder;
import kafka.serializer.StringEncoder;
@@ -43,34 +46,30 @@ import kafka.serializer.StringEncoder;
* @author Artem Bilan
* @since 0.5
*/
public class ProducerConfigurationTests<K, V> {
public class ProducerConfigurationTests {
@Test
@SuppressWarnings("unchecked")
public void testSendMessageWithNonDefaultKeyAndValueEncoders() throws Exception {
final ProducerMetadata<String, String> producerMetadata = new ProducerMetadata<String, String>("test");
producerMetadata.setValueEncoder(new StringEncoder(null));
producerMetadata.setKeyEncoder(new StringEncoder(null));
producerMetadata.setKeyClassType(String.class);
producerMetadata.setValueClassType(String.class);
final ProducerMetadata<String, String> producerMetadata = new ProducerMetadata<String, String>("test", String.class, String.class, new StringSerializer(), new StringSerializer());
final Producer<String, String> producer = Mockito.mock(Producer.class);
final ProducerConfiguration<String, String> configuration =
new ProducerConfiguration<String, String>(producerMetadata, producer);
configuration.send("test", "key", new GenericMessage<String>("test message"));
configuration.send("test", "key", "test message");
Mockito.verify(producer, Mockito.times(1)).send(Mockito.any(KeyedMessage.class));
Mockito.verify(producer, Mockito.times(1)).send(Mockito.any(ProducerRecord.class));
final ArgumentCaptor<KeyedMessage<String, String>> argument =
(ArgumentCaptor<KeyedMessage<String, String>>) (Object)
ArgumentCaptor.forClass(KeyedMessage.class);
final ArgumentCaptor<ProducerRecord<String, String>> argument =
(ArgumentCaptor<ProducerRecord<String, String>>) (Object)
ArgumentCaptor.forClass(ProducerRecord.class);
Mockito.verify(producer).send(argument.capture());
final KeyedMessage<String, String> capturedKeyMessage = argument.getValue();
final ProducerRecord<String, String> capturedKeyMessage = argument.getValue();
Assert.assertEquals(capturedKeyMessage.key(), "key");
Assert.assertEquals(capturedKeyMessage.message(), "test message");
Assert.assertEquals(capturedKeyMessage.value(), "test message");
Assert.assertEquals(capturedKeyMessage.topic(), "test");
}
@@ -81,26 +80,23 @@ public class ProducerConfigurationTests<K, V> {
@SuppressWarnings("unchecked")
public void testSendMessageWithDefaultKeyAndValueEncodersAndCustomSerializableKeyAndPayloadObject()
throws Exception {
final ProducerMetadata<byte[], byte[]> producerMetadata = new ProducerMetadata<byte[], byte[]>("test");
producerMetadata.setValueEncoder(new DefaultEncoder(null));
producerMetadata.setKeyEncoder(new DefaultEncoder(null));
final ProducerMetadata<byte[], byte[]> producerMetadata = new ProducerMetadata<byte[], byte[]>("test",
byte[].class,byte[].class, new ByteArraySerializer(), new ByteArraySerializer());
final Producer<byte[], byte[]> producer = Mockito.mock(Producer.class);
final ProducerConfiguration<byte[], byte[]> configuration =
new ProducerConfiguration<byte[], byte[]>(producerMetadata, producer);
Message<TestPayload> message = new GenericMessage<TestPayload>(new TestPayload("part1", "part2"));
configuration.convertAndSend("test", new TestKey("compositePart1", "compositePart2"), new TestPayload("part1", "part2"));
configuration.send("test", new TestKey("compositePart1", "compositePart2"), message);
Mockito.verify(producer, Mockito.times(1)).send(Mockito.any(ProducerRecord.class));
Mockito.verify(producer, Mockito.times(1)).send(Mockito.any(KeyedMessage.class));
final ArgumentCaptor<KeyedMessage<byte[], byte[]>> argument =
(ArgumentCaptor<KeyedMessage<byte[], byte[]>>) (Object)
ArgumentCaptor.forClass(KeyedMessage.class);
final ArgumentCaptor<ProducerRecord<byte[], byte[]>> argument =
(ArgumentCaptor<ProducerRecord<byte[], byte[]>>) (Object)
ArgumentCaptor.forClass(ProducerRecord.class);
Mockito.verify(producer).send(argument.capture());
final KeyedMessage<byte[], byte[]> capturedKeyMessage = argument.getValue();
final ProducerRecord<byte[], byte[]> capturedKeyMessage = argument.getValue();
final byte[] keyBytes = capturedKeyMessage.key();
@@ -113,7 +109,7 @@ public class ProducerConfigurationTests<K, V> {
Assert.assertEquals(tk.getKeyPart1(), "compositePart1");
Assert.assertEquals(tk.getKeyPart2(), "compositePart2");
final byte[] messageBytes = capturedKeyMessage.message();
final byte[] messageBytes = capturedKeyMessage.value();
final ByteArrayInputStream messageInputStream = new ByteArrayInputStream(messageBytes);
final ObjectInputStream messageObjectInputStream = new ObjectInputStream(messageInputStream);
@@ -133,28 +129,26 @@ public class ProducerConfigurationTests<K, V> {
@Test
@SuppressWarnings("unchecked")
public void testSendMessageWithDefaultKeyEncoderAndNonDefaultValueEncoderAndCorrespondingData() throws Exception {
final ProducerMetadata<byte[], TestPayload> producerMetadata = new ProducerMetadata<byte[], TestPayload>("test");
final AvroReflectDatumBackedKafkaEncoder<TestPayload> encoder =
new AvroReflectDatumBackedKafkaEncoder<TestPayload>(TestPayload.class);
producerMetadata.setValueEncoder(encoder);
producerMetadata.setKeyEncoder(new DefaultEncoder(null));
producerMetadata.setValueClassType(TestPayload.class);
final ProducerMetadata<byte[], TestPayload> producerMetadata = new ProducerMetadata<byte[], TestPayload>("test", byte[].class,TestPayload.class, new ByteArraySerializer(), new EncoderAdaptingSerializer<TestPayload>(encoder));
final Producer<byte[], TestPayload> producer = Mockito.mock(Producer.class);
final ProducerConfiguration<byte[], TestPayload> configuration =
new ProducerConfiguration<byte[], TestPayload>(producerMetadata, producer);
new ProducerConfiguration<>(producerMetadata, producer);
TestPayload tp = new TestPayload("part1", "part2");
configuration.send("test", "key", new GenericMessage<TestPayload>(tp));
configuration.convertAndSend("test", "key", tp);
Mockito.verify(producer, Mockito.times(1)).send(Mockito.any(KeyedMessage.class));
Mockito.verify(producer, Mockito.times(1)).send(Mockito.any(ProducerRecord.class));
final ArgumentCaptor<KeyedMessage<byte[], TestPayload>> argument =
(ArgumentCaptor<KeyedMessage<byte[], TestPayload>>) (Object)
ArgumentCaptor.forClass(KeyedMessage.class);
final ArgumentCaptor<ProducerRecord<byte[], TestPayload>> argument =
(ArgumentCaptor<ProducerRecord<byte[], TestPayload>>) (Object)
ArgumentCaptor.forClass(ProducerRecord.class);
Mockito.verify(producer).send(argument.capture());
final KeyedMessage<byte[], TestPayload> capturedKeyMessage = argument.getValue();
final ProducerRecord<byte[], TestPayload> capturedKeyMessage = argument.getValue();
final byte[] keyBytes = capturedKeyMessage.key();
@@ -163,7 +157,7 @@ public class ProducerConfigurationTests<K, V> {
final Object keyObj = keyObjectInputStream.readObject();
Assert.assertEquals("key", keyObj);
Assert.assertEquals(capturedKeyMessage.message(), tp);
Assert.assertEquals(capturedKeyMessage.value(), tp);
Assert.assertEquals(capturedKeyMessage.topic(), "test");
}
@@ -174,11 +168,9 @@ public class ProducerConfigurationTests<K, V> {
@Test
@SuppressWarnings("unchecked")
public void testSendMessageWithNonDefaultKeyEncoderAndDefaultValueEncoderAndCorrespondingData() throws Exception {
final ProducerMetadata<TestKey, byte[]> producerMetadata = new ProducerMetadata<TestKey, byte[]>("test");
final AvroReflectDatumBackedKafkaEncoder<TestKey> encoder = new AvroReflectDatumBackedKafkaEncoder<TestKey>(TestKey.class);
producerMetadata.setKeyEncoder(encoder);
producerMetadata.setValueEncoder(new DefaultEncoder(null));
producerMetadata.setKeyClassType(TestKey.class);
final ProducerMetadata<TestKey, byte[]> producerMetadata = new ProducerMetadata<TestKey, byte[]>("test", TestKey.class, byte[].class, new EncoderAdaptingSerializer<TestKey>(encoder), new ByteArraySerializer());
final Producer<TestKey, byte[]> producer = Mockito.mock(Producer.class);
final ProducerConfiguration<TestKey, byte[]> configuration =
@@ -186,20 +178,20 @@ public class ProducerConfigurationTests<K, V> {
final TestKey tk = new TestKey("part1", "part2");
configuration.send("test", tk, new GenericMessage<String>("test message"));
configuration.convertAndSend("test", tk, "test message");
Mockito.verify(producer, Mockito.times(1)).send(Mockito.any(KeyedMessage.class));
Mockito.verify(producer, Mockito.times(1)).send(Mockito.any(ProducerRecord.class));
final ArgumentCaptor<KeyedMessage<TestKey, byte[]>> argument =
(ArgumentCaptor<KeyedMessage<TestKey, byte[]>>) (Object)
ArgumentCaptor.forClass(KeyedMessage.class);
final ArgumentCaptor<ProducerRecord<TestKey, byte[]>> argument =
(ArgumentCaptor<ProducerRecord<TestKey, byte[]>>) (Object)
ArgumentCaptor.forClass(ProducerRecord.class);
Mockito.verify(producer).send(argument.capture());
final KeyedMessage<TestKey, byte[]> capturedKeyMessage = argument.getValue();
final ProducerRecord<TestKey, byte[]> capturedKeyMessage = argument.getValue();
Assert.assertEquals(capturedKeyMessage.key(), tk);
final byte[] payloadBytes = capturedKeyMessage.message();
final byte[] payloadBytes = capturedKeyMessage.value();
final ByteArrayInputStream payloadBis = new ByteArrayInputStream(payloadBytes);
final ObjectInputStream payloadOis = new ObjectInputStream(payloadBis);
@@ -216,24 +208,22 @@ public class ProducerConfigurationTests<K, V> {
@Test
@SuppressWarnings("unchecked")
public void testSendMessageWithDefaultKeyAndValueEncodersAndStringKeyAndValue() throws Exception {
final ProducerMetadata<byte[], byte[]> producerMetadata = new ProducerMetadata<byte[], byte[]>("test");
producerMetadata.setValueEncoder(new DefaultEncoder(null));
producerMetadata.setKeyEncoder(new DefaultEncoder(null));
final ProducerMetadata<byte[], byte[]> producerMetadata = new ProducerMetadata<byte[], byte[]>("test", byte[].class, byte[].class, new ByteArraySerializer(), new ByteArraySerializer());
final Producer<byte[], byte[]> producer = Mockito.mock(Producer.class);
final ProducerConfiguration<byte[], byte[]> configuration =
new ProducerConfiguration<byte[], byte[]>(producerMetadata, producer);
configuration.send("test", "key", new GenericMessage<String>("test message"));
configuration.convertAndSend("test", "key", "test message");
Mockito.verify(producer, Mockito.times(1)).send(Mockito.any(KeyedMessage.class));
Mockito.verify(producer, Mockito.times(1)).send(Mockito.any(ProducerRecord.class));
final ArgumentCaptor<KeyedMessage<byte[], byte[]>> argument =
(ArgumentCaptor<KeyedMessage<byte[], byte[]>>) (Object)
ArgumentCaptor.forClass(KeyedMessage.class);
final ArgumentCaptor<ProducerRecord<byte[], byte[]>> argument =
(ArgumentCaptor<ProducerRecord<byte[], byte[]>>) (Object)
ArgumentCaptor.forClass(ProducerRecord.class);
Mockito.verify(producer).send(argument.capture());
final KeyedMessage<byte[], byte[]> capturedKeyMessage = argument.getValue();
final ProducerRecord<byte[], byte[]> capturedKeyMessage = argument.getValue();
final byte[] keyBytes = capturedKeyMessage.key();
final ByteArrayInputStream keyBis = new ByteArrayInputStream(keyBytes);
@@ -242,7 +232,7 @@ public class ProducerConfigurationTests<K, V> {
Assert.assertEquals("key", keyObj);
final byte[] payloadBytes = capturedKeyMessage.message();
final byte[] payloadBytes = capturedKeyMessage.value();
final ByteArrayInputStream payloadBis = new ByteArrayInputStream(payloadBytes);
final ObjectInputStream payloadOis = new ObjectInputStream(payloadBis);
@@ -255,12 +245,10 @@ public class ProducerConfigurationTests<K, V> {
/**
* User does not set an explicit key/value encoder, but send non-serializable object for both key/value
*/
@Test(expected = NotSerializableException.class)
@Test(expected = ConversionFailedException.class)
@SuppressWarnings("unchecked")
public void testSendMessageWithDefaultKeyAndValueEncodersButNonSerializableKeyAndValue() throws Exception {
final ProducerMetadata<byte[], byte[]> producerMetadata = new ProducerMetadata<byte[], byte[]>("test");
producerMetadata.setValueEncoder(new DefaultEncoder(null));
producerMetadata.setKeyEncoder(new DefaultEncoder(null));
final ProducerMetadata<byte[], byte[]> producerMetadata = new ProducerMetadata<byte[], byte[]>("test", byte[].class, byte[].class, new ByteArraySerializer(), new ByteArraySerializer());
final Producer<byte[], byte[]> producer = Mockito.mock(Producer.class);
final ProducerConfiguration<byte[], byte[]> configuration =
@@ -269,48 +257,39 @@ public class ProducerConfigurationTests<K, V> {
Message<NonSerializableTestPayload> message =
new GenericMessage<NonSerializableTestPayload>(new NonSerializableTestPayload("part1", "part2"));
configuration.send("test", new NonSerializableTestKey("compositePart1", "compositePart2"), message);
configuration.convertAndSend("test", new NonSerializableTestKey("compositePart1", "compositePart2"), message);
}
/**
* User does not set an explicit key/value encoder, but send non-serializable key and serializable value
*/
@Test(expected = NotSerializableException.class)
@Test(expected = ConversionFailedException.class)
@SuppressWarnings("unchecked")
public void testSendMessageWithDefaultKeyAndValueEncodersButNonSerializableKeyAndSerializableValue()
throws Exception {
final ProducerMetadata<byte[], byte[]> producerMetadata = new ProducerMetadata<byte[], byte[]>("test");
producerMetadata.setValueEncoder(new DefaultEncoder(null));
producerMetadata.setKeyEncoder(new DefaultEncoder(null));
final ProducerMetadata<byte[], byte[]> producerMetadata = new ProducerMetadata<byte[], byte[]>("test", byte[].class, byte[].class, new ByteArraySerializer(), new ByteArraySerializer());
final Producer<byte[], byte[]> producer = Mockito.mock(Producer.class);
final ProducerConfiguration<byte[], byte[]> configuration =
new ProducerConfiguration<byte[], byte[]>(producerMetadata, producer);
Message<TestPayload> message = new GenericMessage<TestPayload>(new TestPayload("part1", "part2"));
configuration.send("test", new NonSerializableTestKey("compositePart1", "compositePart2"), message);
configuration.convertAndSend("test", new NonSerializableTestKey("compositePart1", "compositePart2"), new TestPayload("part1", "part2"));
}
/**
* User does not set an explicit key/value encoder, but send serializable key and non-serializable value
*/
@Test(expected = NotSerializableException.class)
@Test(expected = ConversionFailedException.class)
@SuppressWarnings("unchecked")
public void testSendMessageWithDefaultKeyAndValueEncodersButSerializableKeyAndNonSerializableValue()
throws Exception {
final ProducerMetadata<byte[], byte[]> producerMetadata = new ProducerMetadata<byte[], byte[]>("test");
producerMetadata.setValueEncoder(new DefaultEncoder(null));
producerMetadata.setKeyEncoder(new DefaultEncoder(null));
final ProducerMetadata<byte[], byte[]> producerMetadata = new ProducerMetadata<byte[], byte[]>("test", byte[].class, byte[].class, new EncoderAdaptingSerializer<byte[]>(new DefaultEncoder(null)), new EncoderAdaptingSerializer<byte[]>(new DefaultEncoder(null)));
final Producer<byte[], byte[]> producer = Mockito.mock(Producer.class);
final ProducerConfiguration<byte[], byte[]> configuration =
new ProducerConfiguration<byte[], byte[]>(producerMetadata, producer);
Message<NonSerializableTestPayload> message =
new GenericMessage<NonSerializableTestPayload>(new NonSerializableTestPayload("part1", "part2"));
configuration.send("test", new TestKey("compositePart1", "compositePart2"), message);
configuration.convertAndSend("test", new TestKey("compositePart1", "compositePart2"), new NonSerializableTestPayload("part1", "part2"));
}
}

View File

@@ -15,8 +15,9 @@
*/
package org.springframework.integration.kafka.support;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.common.serialization.ByteArraySerializer;
import org.junit.Assert;
import kafka.javaapi.producer.Producer;
import org.junit.Test;
import org.mockito.Mockito;
@@ -24,41 +25,19 @@ import org.mockito.Mockito;
* @author Soby Chacko
* @since 0.5
*/
public class ProducerFactoryBeanTests<K,V> {
public class ProducerFactoryBeanTests {
@Test
public void createProducerWithDefaultMetadata() throws Exception {
final ProducerMetadata<byte[], byte[]> producerMetadata = new ProducerMetadata<byte[], byte[]>("test");
final ProducerMetadata<byte[], byte[]> producerMetadata = new ProducerMetadata<byte[], byte[]>("test", byte[].class, byte[].class, new ByteArraySerializer(), new ByteArraySerializer());
final ProducerMetadata<byte[], byte[]> tm = Mockito.spy(producerMetadata);
final ProducerFactoryBean<byte[], byte[]> producerFactoryBean = new ProducerFactoryBean<byte[], byte[]>(tm, "localhost:9092");
final Producer<byte[], byte[]> producer = producerFactoryBean.getObject();
Assert.assertTrue(producer != null);
Mockito.verify(tm, Mockito.times(1)).getPartitioner();
Mockito.verify(tm, Mockito.times(1)).getCompressionCodec();
Mockito.verify(tm, Mockito.times(1)).getValueEncoder();
Mockito.verify(tm, Mockito.times(1)).getKeyEncoder();
Mockito.verify(tm, Mockito.times(1)).isAsync();
Mockito.verify(tm, Mockito.times(0)).getBatchNumMessages();
}
@Test
public void createProducerWithAsyncFeatures() throws Exception {
final ProducerMetadata<byte[], byte[]> producerMetadata = new ProducerMetadata<byte[], byte[]>("test");
producerMetadata.setAsync(true);
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<byte[], byte[]> producer = producerFactoryBean.getObject();
Assert.assertTrue(producer != null);
Mockito.verify(tm, Mockito.times(1)).getPartitioner();
Mockito.verify(tm, Mockito.times(1)).getCompressionCodec();
Mockito.verify(tm, Mockito.times(1)).getValueEncoder();
Mockito.verify(tm, Mockito.times(1)).getKeyEncoder();
Mockito.verify(tm, Mockito.times(1)).isAsync();
Mockito.verify(tm, Mockito.times(2)).getBatchNumMessages();
Mockito.verify(tm, Mockito.times(1)).getCompressionType();
Mockito.verify(tm, Mockito.times(1)).getValueSerializer();
Mockito.verify(tm, Mockito.times(1)).getKeySerializer();
}
}

View File

@@ -6,3 +6,4 @@ log4j.appender.stdout.layout.ConversionPattern=%d{HH:mm:ss.SSS} %-5p [%t][%c] %m
log4j.category.org.springframework.integration=WARN
log4j.category.org.springframework.integration.kafka=INFO
log4j.category.org.apache.kafka.common.network.Selector=ERROR