diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/config/xml/KafkaConsumerContextParser.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/config/xml/KafkaConsumerContextParser.java index e116eec6ed..c21f7f19e3 100644 --- a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/config/xml/KafkaConsumerContextParser.java +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/config/xml/KafkaConsumerContextParser.java @@ -15,7 +15,10 @@ */ package org.springframework.integration.kafka.config.xml; -import kafka.serializer.DefaultDecoder; +import java.util.HashMap; +import java.util.List; +import java.util.Map; + import org.springframework.beans.factory.config.BeanDefinition; import org.springframework.beans.factory.config.BeanDefinitionHolder; import org.springframework.beans.factory.support.AbstractBeanDefinition; @@ -29,17 +32,14 @@ import org.springframework.integration.kafka.support.ConsumerConnectionProvider; import org.springframework.integration.kafka.support.ConsumerMetadata; import org.springframework.integration.kafka.support.KafkaConsumerContext; import org.springframework.integration.kafka.support.MessageLeftOverTracker; +import org.springframework.integration.kafka.support.TopicFilterConfiguration; import org.springframework.util.StringUtils; import org.springframework.util.xml.DomUtils; import org.w3c.dom.Element; -import java.util.ArrayList; -import java.util.HashMap; -import java.util.List; -import java.util.Map; - /** * @author Soby Chacko + * @author Rajasekar Elango * @since 0.5 */ public class KafkaConsumerContextParser extends AbstractSingleBeanDefinitionParser { @@ -57,7 +57,6 @@ public class KafkaConsumerContextParser extends AbstractSingleBeanDefinitionPars parseConsumerConfigurations(consumerConfigurations, parserContext, builder, element); } - @SuppressWarnings("unchecked") private void parseConsumerConfigurations(final Element consumerConfigurations, final ParserContext parserContext, final BeanDefinitionBuilder builder, final Element parentElem) { for (final Element consumerConfiguration : DomUtils.getChildElementsByTagName(consumerConfigurations, "consumer-configuration")) { @@ -75,14 +74,24 @@ public class KafkaConsumerContextParser extends AbstractSingleBeanDefinitionPars final Map topicStreamsMap = new HashMap(); - for (final Element topicConfiguration : DomUtils.getChildElementsByTagName(consumerConfiguration, "topic")) { - final String topic = topicConfiguration.getAttribute("id"); - final String streams = topicConfiguration.getAttribute("streams"); - final Integer streamsInt = Integer.valueOf(streams); - topicStreamsMap.put(topic, streamsInt); + final List topicConfigurations = DomUtils.getChildElementsByTagName(consumerConfiguration, "topic"); + + if (topicConfigurations != null){ + for (final Element topicConfiguration : topicConfigurations) { + final String topic = topicConfiguration.getAttribute("id"); + final String streams = topicConfiguration.getAttribute("streams"); + final Integer streamsInt = Integer.valueOf(streams); + topicStreamsMap.put(topic, streamsInt); + } + consumerMetadataBuilder.addPropertyValue("topicStreamMap", topicStreamsMap); } - consumerMetadataBuilder.addPropertyValue("topicStreamMap", topicStreamsMap); + final Element topicFilter = DomUtils.getChildElementByTagName(consumerConfiguration, "topic-filter"); + + if (topicFilter != null){ + final TopicFilterConfiguration topicFilterConfiguration = new TopicFilterConfiguration(topicFilter.getAttribute("pattern"),Integer.valueOf(topicFilter.getAttribute("streams")), Boolean.valueOf(topicFilter.getAttribute("exclude"))); + consumerMetadataBuilder.addPropertyValue("topicFilterConfiguration", topicFilterConfiguration); + } final BeanDefinition consumerMetadataBeanDef = consumerMetadataBuilder.getBeanDefinition(); registerBeanDefinition(new BeanDefinitionHolder(consumerMetadataBeanDef, "consumerMetadata_" + consumerConfiguration.getAttribute("group-id")), diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/config/xml/KafkaProducerContextParser.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/config/xml/KafkaProducerContextParser.java index dadfe47771..ee2a8bb941 100644 --- a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/config/xml/KafkaProducerContextParser.java +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/config/xml/KafkaProducerContextParser.java @@ -30,9 +30,6 @@ import org.springframework.util.StringUtils; import org.springframework.util.xml.DomUtils; import org.w3c.dom.Element; -import java.util.HashMap; -import java.util.Map; - /** * @author Soby Chacko * @since 0.5 @@ -52,7 +49,6 @@ public class KafkaProducerContextParser extends AbstractSimpleBeanDefinitionPars parseProducerConfigurations(topics, parserContext); } - @SuppressWarnings("unchecked") private void parseProducerConfigurations(final Element topics, final ParserContext parserContext) { for (final Element producerConfiguration : DomUtils.getChildElementsByTagName(topics, "producer-configuration")){ final BeanDefinitionBuilder producerConfigurationBuilder = BeanDefinitionBuilder.genericBeanDefinition(ProducerConfiguration.class); diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/serializer/avro/AvroDatumSupport.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/serializer/avro/AvroDatumSupport.java index 6165b1d39b..808c5a0bce 100644 --- a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/serializer/avro/AvroDatumSupport.java +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/serializer/avro/AvroDatumSupport.java @@ -21,7 +21,6 @@ public abstract class AvroDatumSupport { this.avroSerializer = new AvroSerializer(); } - @SuppressWarnings("unchecked") public byte[] toBytes(final T source, final DatumWriter writer) { try { return avroSerializer.serialize(source, writer); @@ -31,7 +30,6 @@ public abstract class AvroDatumSupport { return null; } - @SuppressWarnings("unchecked") public T fromBytes(final byte[] bytes, final DatumReader reader) { try { return avroSerializer.deserialize(bytes, reader); diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/serializer/avro/AvroReflectDatumBackedKafkaDecoder.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/serializer/avro/AvroReflectDatumBackedKafkaDecoder.java index 7e9b7c95c4..68a6663cca 100644 --- a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/serializer/avro/AvroReflectDatumBackedKafkaDecoder.java +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/serializer/avro/AvroReflectDatumBackedKafkaDecoder.java @@ -16,17 +16,15 @@ package org.springframework.integration.kafka.serializer.avro; import kafka.serializer.Decoder; + import org.apache.avro.io.DatumReader; import org.apache.avro.reflect.ReflectDatumReader; -import org.apache.commons.logging.Log; -import org.apache.commons.logging.LogFactory; /** * @author Soby Chacko * @since 0.5 */ public class AvroReflectDatumBackedKafkaDecoder extends AvroDatumSupport implements Decoder { - private static final Log LOG = LogFactory.getLog(AvroReflectDatumBackedKafkaDecoder.class); private final DatumReader reader; @@ -35,9 +33,7 @@ public class AvroReflectDatumBackedKafkaDecoder extends AvroDatumSupport i } @Override - @SuppressWarnings("unchecked") public T fromBytes(final byte[] bytes) { return fromBytes(bytes, reader); } } - diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/serializer/avro/AvroReflectDatumBackedKafkaEncoder.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/serializer/avro/AvroReflectDatumBackedKafkaEncoder.java index 6f1c0cc4df..4d51d096d7 100644 --- a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/serializer/avro/AvroReflectDatumBackedKafkaEncoder.java +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/serializer/avro/AvroReflectDatumBackedKafkaEncoder.java @@ -16,17 +16,15 @@ package org.springframework.integration.kafka.serializer.avro; import kafka.serializer.Encoder; + import org.apache.avro.io.DatumWriter; import org.apache.avro.reflect.ReflectDatumWriter; -import org.apache.commons.logging.Log; -import org.apache.commons.logging.LogFactory; /** * @author Soby Chacko * @since 0.5 */ public class AvroReflectDatumBackedKafkaEncoder extends AvroDatumSupport implements Encoder { - private static final Log LOG = LogFactory.getLog(AvroReflectDatumBackedKafkaEncoder.class); private final DatumWriter writer; @@ -35,7 +33,6 @@ public class AvroReflectDatumBackedKafkaEncoder extends AvroDatumSupport i } @Override - @SuppressWarnings("unchecked") public byte[] toBytes(final T source) { return toBytes(source, writer); } diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/serializer/avro/AvroSpecificDatumBackedKafkaDecoder.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/serializer/avro/AvroSpecificDatumBackedKafkaDecoder.java index 622a9e38ca..9ad7ee6b54 100644 --- a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/serializer/avro/AvroSpecificDatumBackedKafkaDecoder.java +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/serializer/avro/AvroSpecificDatumBackedKafkaDecoder.java @@ -1,10 +1,9 @@ package org.springframework.integration.kafka.serializer.avro; import kafka.serializer.Decoder; + import org.apache.avro.io.DatumReader; import org.apache.avro.specific.SpecificDatumReader; -import org.apache.commons.logging.Log; -import org.apache.commons.logging.LogFactory; /** * @author Soby Chacko @@ -12,8 +11,6 @@ import org.apache.commons.logging.LogFactory; */ public class AvroSpecificDatumBackedKafkaDecoder extends AvroDatumSupport implements Decoder { - private static final Log LOG = LogFactory.getLog(AvroSpecificDatumBackedKafkaDecoder.class); - private final DatumReader reader; public AvroSpecificDatumBackedKafkaDecoder(final Class specificRecordBase) { @@ -21,7 +18,6 @@ public class AvroSpecificDatumBackedKafkaDecoder extends AvroDatumSupport } @Override - @SuppressWarnings("unchecked") public T fromBytes(final byte[] bytes) { return fromBytes(bytes, reader); } diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/serializer/avro/AvroSpecificDatumBackedKafkaEncoder.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/serializer/avro/AvroSpecificDatumBackedKafkaEncoder.java index 410dfd6047..92fb4acd21 100644 --- a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/serializer/avro/AvroSpecificDatumBackedKafkaEncoder.java +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/serializer/avro/AvroSpecificDatumBackedKafkaEncoder.java @@ -1,10 +1,9 @@ package org.springframework.integration.kafka.serializer.avro; import kafka.serializer.Encoder; + import org.apache.avro.io.DatumWriter; import org.apache.avro.specific.SpecificDatumWriter; -import org.apache.commons.logging.Log; -import org.apache.commons.logging.LogFactory; /** * @author Soby Chacko @@ -12,8 +11,6 @@ import org.apache.commons.logging.LogFactory; */ public class AvroSpecificDatumBackedKafkaEncoder extends AvroDatumSupport implements Encoder { - private static final Log LOG = LogFactory.getLog(AvroSpecificDatumBackedKafkaEncoder.class); - private final DatumWriter writer; public AvroSpecificDatumBackedKafkaEncoder(final Class specificRecordClazz) { @@ -21,7 +18,6 @@ public class AvroSpecificDatumBackedKafkaEncoder extends AvroDatumSupport } @Override - @SuppressWarnings("unchecked") public byte[] toBytes(final T source) { return toBytes(source, writer); } diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/support/ConsumerConfiguration.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/support/ConsumerConfiguration.java index 25fe67fe58..8376effbc0 100644 --- a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/support/ConsumerConfiguration.java +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/support/ConsumerConfiguration.java @@ -1,63 +1,52 @@ /* - * Copyright 2002-2013 the original author or authors. - * - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. + * Copyright 2002-2013 the original author or authors. Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. You may obtain a copy of the License at + * http://www.apache.org/licenses/LICENSE-2.0 Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, + * either express or implied. See the License for the specific language governing permissions and limitations under the + * License. */ package org.springframework.integration.kafka.support; +import java.util.*; +import java.util.concurrent.*; + import kafka.consumer.ConsumerTimeoutException; import kafka.consumer.KafkaStream; import kafka.javaapi.consumer.ConsumerConnector; import kafka.message.MessageAndMetadata; + import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; import org.springframework.integration.MessagingException; -import java.util.ArrayList; -import java.util.LinkedList; -import java.util.List; -import java.util.Map; -import java.util.concurrent.Callable; -import java.util.concurrent.ConcurrentHashMap; -import java.util.concurrent.ExecutorService; -import java.util.concurrent.Executors; -import java.util.concurrent.Future; - /** * @author Soby Chacko + * @author Rajasekar Elango * @since 0.5 */ -public class ConsumerConfiguration { +public class ConsumerConfiguration { private static final Log LOGGER = LogFactory.getLog(ConsumerConfiguration.class); - private final ConsumerMetadata consumerMetadata; + private final ConsumerMetadata consumerMetadata; private final ConsumerConnectionProvider consumerConnectionProvider; - private final MessageLeftOverTracker messageLeftOverTracker; + private final MessageLeftOverTracker messageLeftOverTracker; private ConsumerConnector consumerConnector; private volatile int count = 0; private int maxMessages = 1; + private Collection>> consumerMessageStreams; - private ExecutorService executorService = Executors.newCachedThreadPool(); + private final ExecutorService executorService = Executors.newCachedThreadPool(); - public ConsumerConfiguration(final ConsumerMetadata consumerMetadata, - final ConsumerConnectionProvider consumerConnectionProvider, - final MessageLeftOverTracker messageLeftOverTracker) { + public ConsumerConfiguration(final ConsumerMetadata consumerMetadata, + final ConsumerConnectionProvider consumerConnectionProvider, + final MessageLeftOverTracker messageLeftOverTracker) { this.consumerMetadata = consumerMetadata; this.consumerConnectionProvider = consumerConnectionProvider; this.messageLeftOverTracker = messageLeftOverTracker; } - public ConsumerMetadata getConsumerMetadata() { + public ConsumerMetadata getConsumerMetadata() { return consumerMetadata; } @@ -65,23 +54,23 @@ public class ConsumerConfiguration { count = messageLeftOverTracker.getCurrentCount(); final Object lock = new Object(); - final List>>> tasks = new LinkedList>>>(); + final List>>> tasks = new LinkedList>>>(); - final Map>> consumerMap = getConsumerMapWithMessageStreams(); - for (final List> streams : consumerMap.values()) { - for (final KafkaStream stream : streams) { - tasks.add(new Callable>>() { + for (final List> streams : createConsumerMessageStreams()) { + for (final KafkaStream stream : streams) { + tasks.add(new Callable>>() { @Override - public List> call() throws Exception { - final List> rawMessages = new ArrayList>(); + public List> call() throws Exception { + final List> rawMessages = new ArrayList>(); try { while (count < maxMessages) { - final MessageAndMetadata messageAndMetadata = stream.iterator().next(); + final MessageAndMetadata messageAndMetadata = stream.iterator().next(); synchronized (lock) { if (count < maxMessages) { rawMessages.add(messageAndMetadata); count++; - } else { + } + else { messageLeftOverTracker.addMessageAndMetadata(messageAndMetadata); } } @@ -97,18 +86,20 @@ public class ConsumerConfiguration { return executeTasks(tasks); } - private Map>> executeTasks(final List>>> tasks) { + private Map>> executeTasks( + final List>>> tasks) { final Map>> messages = new ConcurrentHashMap>>(); messages.putAll(getLeftOverMessageMap()); try { - for (final Future>> result : executorService.invokeAll(tasks)) { + for (final Future>> result : executorService.invokeAll(tasks)) { if (!result.get().isEmpty()) { final String topic = result.get().get(0).topic(); if (!messages.containsKey(topic)) { messages.put(topic, getPayload(result.get())); - } else { + } + else { final Map> existingPayloadMap = messages.get(topic); getPayload(result.get(), existingPayloadMap); @@ -126,21 +117,21 @@ public class ConsumerConfiguration { return messages; } - @SuppressWarnings("unchecked") private Map>> getLeftOverMessageMap() { final Map>> messages = new ConcurrentHashMap>>(); - for (final MessageAndMetadata mamd : messageLeftOverTracker.getMessageLeftOverFromPreviousPoll()) { + for (final MessageAndMetadata mamd : messageLeftOverTracker.getMessageLeftOverFromPreviousPoll()) { final String topic = mamd.topic(); if (!messages.containsKey(topic)) { - final List> l = new ArrayList>(); + final List> l = new ArrayList>(); l.add(mamd); messages.put(topic, getPayload(l)); - } else { + } + else { final Map> existingPayloadMap = messages.get(topic); - final List> l = new ArrayList>(); + final List> l = new ArrayList>(); l.add(mamd); getPayload(l, existingPayloadMap); } @@ -149,15 +140,16 @@ public class ConsumerConfiguration { return messages; } - private Map> getPayload(final List> messageAndMetadatas) { + private Map> getPayload(final List> messageAndMetadatas) { final Map> payloadMap = new ConcurrentHashMap>(); - for (final MessageAndMetadata messageAndMetadata : messageAndMetadatas) { + for (final MessageAndMetadata messageAndMetadata : messageAndMetadatas) { if (!payloadMap.containsKey(messageAndMetadata.partition())) { final List payload = new ArrayList(); payload.add(messageAndMetadata.message()); payloadMap.put(messageAndMetadata.partition(), payload); - } else { + } + else { final List payload = payloadMap.get(messageAndMetadata.partition()); payload.add(messageAndMetadata.message()); } @@ -167,25 +159,52 @@ public class ConsumerConfiguration { return payloadMap; } - private void getPayload(final List> messageAndMetadatas, final Map> existingPayloadMap) { - for (final MessageAndMetadata messageAndMetadata : messageAndMetadatas) { + private void getPayload(final List> messageAndMetadatas, + final Map> existingPayloadMap) { + for (final MessageAndMetadata messageAndMetadata : messageAndMetadatas) { if (!existingPayloadMap.containsKey(messageAndMetadata.partition())) { final List payload = new ArrayList(); payload.add(messageAndMetadata.message()); existingPayloadMap.put(messageAndMetadata.partition(), payload); - } else { + } + else { final List payload = existingPayloadMap.get(messageAndMetadata.partition()); payload.add(messageAndMetadata.message()); } } } - @SuppressWarnings("unchecked") - public Map>> getConsumerMapWithMessageStreams() { - return getConsumerConnector().createMessageStreams( - consumerMetadata.getTopicStreamMap(), - consumerMetadata.getKeyDecoder(), - consumerMetadata.getValueDecoder()); + private Collection>> createConsumerMessageStreams() { + if (consumerMessageStreams == null) { + if (!(consumerMetadata.getTopicStreamMap() == null || consumerMetadata.getTopicStreamMap().isEmpty())) { + consumerMessageStreams = createMessageStreamsForTopic().values(); + } + else { + consumerMessageStreams = new ArrayList>>(); + consumerMessageStreams.add(createMessageStreamsForTopicFilter()); + } + } + return consumerMessageStreams; + } + + public Map>> createMessageStreamsForTopic() { + return getConsumerConnector().createMessageStreams(consumerMetadata.getTopicStreamMap(), + consumerMetadata.getKeyDecoder(), consumerMetadata.getValueDecoder()); + } + + public List> createMessageStreamsForTopicFilter() { + List> messageStream = new ArrayList>(); + TopicFilterConfiguration topicFilterConfiguration = consumerMetadata.getTopicFilterConfiguration(); + if (topicFilterConfiguration != null) { + messageStream = getConsumerConnector().createMessageStreamsByFilter( + topicFilterConfiguration.getTopicFilter(), topicFilterConfiguration.getNumberOfStreams(), + consumerMetadata.getKeyDecoder(), consumerMetadata.getValueDecoder()); + } + else { + LOGGER.warn("No Topic Filter Configuration defined"); + } + + return messageStream; } public int getMaxMessages() { diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/support/ConsumerMetadata.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/support/ConsumerMetadata.java index 476e66a461..55bb6d5e63 100644 --- a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/support/ConsumerMetadata.java +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/support/ConsumerMetadata.java @@ -15,15 +15,16 @@ */ package org.springframework.integration.kafka.support; +import java.util.Map; + import kafka.serializer.Decoder; import kafka.serializer.DefaultDecoder; import org.springframework.beans.factory.InitializingBean; import org.springframework.integration.kafka.core.KafkaConsumerDefaults; -import java.util.Map; - /** * @author Soby Chacko + * @author Rajasekar Elango * @since 0.5 */ public class ConsumerMetadata implements InitializingBean { @@ -46,6 +47,7 @@ public class ConsumerMetadata implements InitializingBean { private Decoder valueDecoder; private Decoder keyDecoder; private Map topicStreamMap; + private TopicFilterConfiguration topicFilterConfiguration; public String getGroupId() { return groupId; @@ -186,4 +188,14 @@ public class ConsumerMetadata implements InitializingBean { setKeyDecoder((Decoder) getValueDecoder()); } } + + public TopicFilterConfiguration getTopicFilterConfiguration() { + return topicFilterConfiguration; + } + + public void setTopicFilterConfiguration( + TopicFilterConfiguration topicFilterConfiguration) { + this.topicFilterConfiguration = topicFilterConfiguration; + } + } diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/support/KafkaProducerContext.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/support/KafkaProducerContext.java index bdb7e1183f..0fdb3c2078 100644 --- a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/support/KafkaProducerContext.java +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/support/KafkaProducerContext.java @@ -15,23 +15,24 @@ */ package org.springframework.integration.kafka.support; -import org.springframework.beans.BeansException; -import org.springframework.beans.factory.BeanFactory; -import org.springframework.beans.factory.BeanFactoryAware; -import org.springframework.beans.factory.ListableBeanFactory; -import org.springframework.integration.Message; - import java.util.Collection; import java.util.Map; +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; +import org.springframework.beans.BeansException; +import org.springframework.beans.factory.*; +import org.springframework.integration.Message; + /** * @author Soby Chacko + * @author Rajasekar Elango * @since 0.5 */ public class KafkaProducerContext implements BeanFactoryAware { + private static final Log LOGGER = LogFactory.getLog(KafkaProducerContext.class); private Map> topicsConfiguration; - @SuppressWarnings("unchecked") public void send(final Message message) throws Exception { final ProducerConfiguration producerConfiguration = getTopicConfiguration(message.getHeaders().get("topic", String.class)); @@ -41,15 +42,15 @@ public class KafkaProducerContext implements BeanFactoryAware { } } - private ProducerConfiguration getTopicConfiguration(final String topic){ + public ProducerConfiguration getTopicConfiguration(final String topic) { final Collection> topics = topicsConfiguration.values(); for (final ProducerConfiguration producerConfiguration : topics){ - if (producerConfiguration.getProducerMetadata().getTopic().equals(topic)){ + if (topic.matches(producerConfiguration.getProducerMetadata().getTopic())){ return producerConfiguration; } } - + LOGGER.error("No is producer-configuration defined for topic " + topic + ". cannot send message"); return null; } diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/support/ProducerConfiguration.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/support/ProducerConfiguration.java index 71026aeb4e..e7e325eda9 100644 --- a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/support/ProducerConfiguration.java +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/support/ProducerConfiguration.java @@ -1,40 +1,33 @@ /* - * Copyright 2002-2013 the original author or authors. - * - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. + * Copyright 2002-2013 the original author or authors. Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. You may obtain a copy of the License at + * http://www.apache.org/licenses/LICENSE-2.0 Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, + * either express or implied. See the License for the specific language governing permissions and limitations under the + * License. */ package org.springframework.integration.kafka.support; +import java.io.*; + import kafka.javaapi.producer.Producer; import kafka.producer.KeyedMessage; import kafka.serializer.DefaultEncoder; + import org.apache.commons.lang.builder.EqualsBuilder; import org.apache.commons.lang.builder.HashCodeBuilder; import org.springframework.integration.Message; -import java.io.ByteArrayOutputStream; -import java.io.IOException; -import java.io.ObjectOutputStream; - /** * @author Soby Chacko + * @author Rajasekar Elango * @since 0.5 */ -public class ProducerConfiguration { - private final Producer producer; - private final ProducerMetadata producerMetadata; +public class ProducerConfiguration { + private final Producer producer; + private final ProducerMetadata producerMetadata; - public ProducerConfiguration(final ProducerMetadata producerMetadata, final Producer producer){ + public ProducerConfiguration(final ProducerMetadata producerMetadata, final Producer producer) { this.producerMetadata = producerMetadata; this.producer = producer; } @@ -46,10 +39,12 @@ public class ProducerConfiguration { public void send(final Message message) throws Exception { final V v = getPayload(message); + String topic = message.getHeaders().get("topic", String.class); if (message.getHeaders().containsKey("messageKey")) { - producer.send(new KeyedMessage(producerMetadata.getTopic(), getKey(message), v)); - } else { - producer.send(new KeyedMessage(producerMetadata.getTopic(), v)); + producer.send(new KeyedMessage(topic, getKey(message), v)); + } + else { + producer.send(new KeyedMessage(topic, v)); } } @@ -57,7 +52,8 @@ public class ProducerConfiguration { private V getPayload(final Message message) throws Exception { if (producerMetadata.getValueEncoder().getClass().isAssignableFrom(DefaultEncoder.class)) { return (V) getByteStream(message.getPayload()); - } else if (message.getPayload().getClass().isAssignableFrom(producerMetadata.getValueClassType())) { + } + else if (message.getPayload().getClass().isAssignableFrom(producerMetadata.getValueClassType())) { return producerMetadata.getValueClassType().cast(message.getPayload()); } @@ -75,13 +71,13 @@ public class ProducerConfiguration { return message.getHeaders().get("messageKey", producerMetadata.getKeyClassType()); } - private static boolean isRawByteArray(final Object obj){ + private static boolean isRawByteArray(final Object obj) { return obj instanceof byte[]; } private static byte[] getByteStream(final Object obj) throws IOException { - if (isRawByteArray(obj)){ - return (byte[])obj; + if (isRawByteArray(obj)) { + return (byte[]) obj; } final ByteArrayOutputStream out = new ByteArrayOutputStream(); @@ -92,7 +88,7 @@ public class ProducerConfiguration { } @Override - public boolean equals(final Object obj){ + public boolean equals(final Object obj) { return EqualsBuilder.reflectionEquals(this, obj); } @@ -100,4 +96,11 @@ public class ProducerConfiguration { public int hashCode() { return HashCodeBuilder.reflectionHashCode(this); } + + @Override + public String toString() { + StringBuilder builder = new StringBuilder(); + builder.append("ProducerConfiguration [producerMetadata=").append(producerMetadata).append("]"); + return builder.toString(); + } } diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/support/ProducerMetadata.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/support/ProducerMetadata.java index 69b432e102..e6dca255c7 100644 --- a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/support/ProducerMetadata.java +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/support/ProducerMetadata.java @@ -18,12 +18,14 @@ package org.springframework.integration.kafka.support; import kafka.producer.Partitioner; import kafka.serializer.DefaultEncoder; import kafka.serializer.Encoder; + import org.apache.commons.lang.builder.EqualsBuilder; import org.apache.commons.lang.builder.HashCodeBuilder; import org.springframework.beans.factory.InitializingBean; /** * @author Soby Chacko + * @author Rajasekar Elango * @since 0.5 */ public class ProducerMetadata implements InitializingBean { @@ -137,4 +139,14 @@ public class ProducerMetadata implements InitializingBean { public int hashCode() { return HashCodeBuilder.reflectionHashCode(this); } + + @Override + public String toString() { + StringBuilder builder = new StringBuilder(); + builder.append("ProducerMetadata [keyEncoder=").append(keyEncoder).append(", valueEncoder=") + .append(valueEncoder).append(", topic=").append(topic).append(", compressionCodec=") + .append(compressionCodec).append(", partitioner=").append(partitioner).append(", async=").append(async) + .append(", batchNumMessages=").append(batchNumMessages).append("]"); + return builder.toString(); + } } diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/support/TopicFilterConfiguration.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/support/TopicFilterConfiguration.java new file mode 100644 index 0000000000..7170c6b01d --- /dev/null +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/support/TopicFilterConfiguration.java @@ -0,0 +1,46 @@ +/* + * Copyright 2002-2013 the original author or authors. Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. You may obtain a copy of the License at + * http://www.apache.org/licenses/LICENSE-2.0 Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, + * either express or implied. See the License for the specific language governing permissions and limitations under the + * License. + */ +package org.springframework.integration.kafka.support; + +import kafka.consumer.Blacklist; +import kafka.consumer.TopicFilter; +import kafka.consumer.Whitelist; + +/** + * @author Rajasekar Elango + * @since 0.5 + */ +public class TopicFilterConfiguration { + + private final int numberOfStreams; + private TopicFilter topicFilter; + + public TopicFilterConfiguration(final String pattern, final int numberOfStreams, final boolean exclude) { + this.numberOfStreams = numberOfStreams; + if (exclude) { + topicFilter = new Blacklist(pattern); + } + else { + topicFilter = new Whitelist(pattern); + } + } + + public TopicFilter getTopicFilter() { + return topicFilter; + } + + public int getNumberOfStreams() { + return numberOfStreams; + } + + @Override + public String toString() { + return new StringBuilder(topicFilter.toString()).append(" : ").append(numberOfStreams).toString(); + } +} diff --git a/spring-integration-kafka/src/main/resources/org/springframework/integration/config/xml/spring-integration-kafka-1.0.xsd b/spring-integration-kafka/src/main/resources/org/springframework/integration/config/xml/spring-integration-kafka-1.0.xsd index 1c716fe086..b3009d85ce 100644 --- a/spring-integration-kafka/src/main/resources/org/springframework/integration/config/xml/spring-integration-kafka-1.0.xsd +++ b/spring-integration-kafka/src/main/resources/org/springframework/integration/config/xml/spring-integration-kafka-1.0.xsd @@ -232,14 +232,56 @@ ]]> - + - + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + avroBackedKafkaEncoder = new AvroReflectDatumBackedKafkaEncoder(TestObject.class); @@ -45,7 +45,6 @@ public class AvroReflectDatumBackedKafkaSerializerTest { } @Test - @SuppressWarnings("unchecked") public void anotherTest() { final AvroReflectDatumBackedKafkaEncoder avroBackedKafkaEncoder = new AvroReflectDatumBackedKafkaEncoder(java.lang.String.class); final String testString = "Testing Avro"; diff --git a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/serializer/AvroSpecificDatumBackedKafkaSerializerTest.java b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/serializer/AvroSpecificDatumBackedKafkaSerializerTest.java index 8d18befd51..3850a5ab47 100644 --- a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/serializer/AvroSpecificDatumBackedKafkaSerializerTest.java +++ b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/serializer/AvroSpecificDatumBackedKafkaSerializerTest.java @@ -13,7 +13,6 @@ import org.springframework.integration.kafka.test.utils.User; public class AvroSpecificDatumBackedKafkaSerializerTest { @Test - @SuppressWarnings("unchecked") public void testEncodeDecodeFromSpecificDatumSchema() { final AvroSpecificDatumBackedKafkaEncoder avroBackedKafkaEncoder = new AvroSpecificDatumBackedKafkaEncoder(User.class); diff --git a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/support/ConsumerConfigurationTests.java b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/support/ConsumerConfigurationTests.java index 19c267a9aa..9e810769f7 100644 --- a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/support/ConsumerConfigurationTests.java +++ b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/support/ConsumerConfigurationTests.java @@ -16,18 +16,9 @@ package org.springframework.integration.kafka.support; import static org.junit.Assert.assertNull; -import static org.mockito.Mockito.atLeast; -import static org.mockito.Mockito.atMost; -import static org.mockito.Mockito.mock; -import static org.mockito.Mockito.times; -import static org.mockito.Mockito.verify; -import static org.mockito.Mockito.when; +import static org.mockito.Mockito.*; -import java.util.ArrayList; -import java.util.Collection; -import java.util.HashMap; -import java.util.List; -import java.util.Map; +import java.util.*; import kafka.consumer.ConsumerIterator; import kafka.consumer.KafkaStream; @@ -42,7 +33,7 @@ import org.mockito.stubbing.Answer; /** * @author Soby Chacko - * @author Gunnar Hillert + * @author Rajasekar Elango * @since 0.5 */ public class ConsumerConfigurationTests { @@ -55,6 +46,10 @@ public class ConsumerConfigurationTests { final MessageLeftOverTracker messageLeftOverTracker = mock(MessageLeftOverTracker.class); final ConsumerConnector consumerConnector = mock(ConsumerConnector.class); + Map topicStreamMap = new HashMap(); + topicStreamMap.put("topic1", 1); + when(consumerMetadata.getTopicStreamMap()).thenReturn(topicStreamMap); + when(consumerConnectionProvider.getConsumerConnector()).thenReturn(consumerConnector); final ConsumerConfiguration consumerConfiguration = new ConsumerConfiguration(consumerMetadata, @@ -67,7 +62,7 @@ public class ConsumerConfigurationTests { final Map>> messageStreams = new HashMap>>(); messageStreams.put("topic", streams); - when(consumerConfiguration.getConsumerMapWithMessageStreams()).thenReturn(messageStreams); + when(consumerConfiguration.createMessageStreamsForTopic()).thenReturn(messageStreams); final ConsumerIterator iterator = mock(ConsumerIterator.class); when(stream.iterator()).thenReturn(iterator); final MessageAndMetadata messageAndMetadata = mock(MessageAndMetadata.class); @@ -95,6 +90,10 @@ public class ConsumerConfigurationTests { mock(ConsumerConnectionProvider.class); final MessageLeftOverTracker messageLeftOverTracker = mock(MessageLeftOverTracker.class); + Map topicStreamMap = new HashMap(); + topicStreamMap.put("topic1", 1); + when(consumerMetadata.getTopicStreamMap()).thenReturn(topicStreamMap); + final ConsumerConnector consumerConnector = mock(ConsumerConnector.class); when(consumerConnectionProvider.getConsumerConnector()).thenReturn(consumerConnector); @@ -113,7 +112,7 @@ public class ConsumerConfigurationTests { final Map>> messageStreams = new HashMap>>(); messageStreams.put("topic", streams); - when(consumerConfiguration.getConsumerMapWithMessageStreams()).thenReturn(messageStreams); + when(consumerConfiguration.createMessageStreamsForTopic()).thenReturn(messageStreams); final ConsumerIterator iterator1 = mock(ConsumerIterator.class); final ConsumerIterator iterator2 = mock(ConsumerIterator.class); final ConsumerIterator iterator3 = mock(ConsumerIterator.class); @@ -160,7 +159,12 @@ public class ConsumerConfigurationTests { final ConsumerMetadata consumerMetadata = mock(ConsumerMetadata.class); final ConsumerConnectionProvider consumerConnectionProvider = mock(ConsumerConnectionProvider.class); - final MessageLeftOverTracker messageLeftOverTracker = mock(MessageLeftOverTracker.class); + final MessageLeftOverTracker messageLeftOverTracker = mock(MessageLeftOverTracker.class); + + Map topicStreamMap = new HashMap(); + topicStreamMap.put("topic1", 1); + when(consumerMetadata.getTopicStreamMap()).thenReturn(topicStreamMap); + final ConsumerConnector consumerConnector = mock(ConsumerConnector.class); @@ -182,7 +186,7 @@ public class ConsumerConfigurationTests { messageStreams.put("topic2", streams); messageStreams.put("topic3", streams); - when(consumerConfiguration.getConsumerMapWithMessageStreams()).thenReturn(messageStreams); + when(consumerConfiguration.createMessageStreamsForTopic()).thenReturn(messageStreams); final ConsumerIterator iterator1 = mock(ConsumerIterator.class); final ConsumerIterator iterator2 = mock(ConsumerIterator.class); final ConsumerIterator iterator3 = mock(ConsumerIterator.class); @@ -211,6 +215,7 @@ public class ConsumerConfigurationTests { when(messageAndMetadata1.partition()).thenAnswer(getAnswer()); final Map>> messages = consumerConfiguration.receive(); + int sum = 0; final Collection>> values = messages.values(); @@ -224,6 +229,8 @@ public class ConsumerConfigurationTests { Assert.assertEquals(sum, 9); } + + private Answer getAnswer() { return new Answer() { private int count = 0; @@ -250,6 +257,9 @@ public class ConsumerConfigurationTests { final MessageLeftOverTracker messageLeftOverTracker = mock(MessageLeftOverTracker.class); final ConsumerConnector consumerConnector = mock(ConsumerConnector.class); + Map topicStreamMap = new HashMap(); + topicStreamMap.put("topic1", 1); + when(consumerMetadata.getTopicStreamMap()).thenReturn(topicStreamMap); when(messageLeftOverTracker.getCurrentCount()).thenReturn(3); final MessageAndMetadata m1 = new MessageAndMetadata("key1", "value1", "topic1", 1, 1L); final MessageAndMetadata m2 = new MessageAndMetadata("key2", "value2", "topic2", 1, 1L); @@ -273,11 +283,10 @@ public class ConsumerConfigurationTests { streams.add(stream); final Map>> messageStreams = new HashMap>>(); messageStreams.put("topic1", streams); - - when(consumerConfiguration.getConsumerMapWithMessageStreams()).thenReturn(messageStreams); - final ConsumerIterator iterator = mock(ConsumerIterator.class); - when(stream.iterator()).thenReturn((ConsumerIterator) iterator); - final MessageAndMetadata messageAndMetadata = mock(MessageAndMetadata.class); + when(consumerConfiguration.createMessageStreamsForTopic()).thenReturn(messageStreams); + final ConsumerIterator iterator = mock(ConsumerIterator.class); + when(stream.iterator()).thenReturn(iterator); + final MessageAndMetadata messageAndMetadata = mock(MessageAndMetadata.class); when(iterator.next()).thenReturn(messageAndMetadata); when(messageAndMetadata.message()).thenReturn("got message"); when(messageAndMetadata.topic()).thenReturn("topic1"); @@ -329,7 +338,7 @@ public class ConsumerConfigurationTests { final ConsumerConfiguration consumerConfiguration = new ConsumerConfiguration(mockedConsumerMetadata, mockedConsumerConnectionProvider, mockedMessageLeftOverTracker); - consumerConfiguration.getConsumerMapWithMessageStreams(); + consumerConfiguration.createMessageStreamsForTopic(); verify(mockedConsumerMetadata, atLeast(1)).getTopicStreamMap(); verify(mockedConsumerConnector, atLeast(1)).createMessageStreams(topicsStreamMap, null, null); @@ -368,13 +377,122 @@ public class ConsumerConfigurationTests { final ConsumerConfiguration consumerConfiguration = new ConsumerConfiguration(mockedConsumerMetadata, mockedConsumerConnectionProvider, mockedMessageLeftOverTracker); - consumerConfiguration.getConsumerMapWithMessageStreams(); + consumerConfiguration.createMessageStreamsForTopic(); verify(mockedConsumerMetadata, atLeast(1)).getTopicStreamMap(); verify(mockedConsumerConnector, atMost(0)).createMessageStreams(topicsStreamMap); verify(mockedConsumerConnector, atLeast(1)).createMessageStreams(topicsStreamMap, mockedKeyDecoder, mockedValueDecoder); } + + @Test + @SuppressWarnings("unchecked") + public void testReceiveMessageForTopicFilterFromSingleStream() { + final ConsumerMetadata consumerMetadata = mock(ConsumerMetadata.class); + final ConsumerConnectionProvider consumerConnectionProvider = + mock(ConsumerConnectionProvider.class); + final MessageLeftOverTracker messageLeftOverTracker = mock(MessageLeftOverTracker.class); + final ConsumerConnector consumerConnector = mock(ConsumerConnector.class); + + when(consumerMetadata.getTopicFilterConfiguration()).thenReturn(new TopicFilterConfiguration(".*", 1, false)); + + when(consumerConnectionProvider.getConsumerConnector()).thenReturn(consumerConnector); + + final ConsumerConfiguration consumerConfiguration = new ConsumerConfiguration(consumerMetadata, + consumerConnectionProvider, messageLeftOverTracker); + consumerConfiguration.setMaxMessages(1); + + final KafkaStream stream = mock(KafkaStream.class); + final List> streams = new ArrayList>(); + streams.add(stream); + + when(consumerConfiguration.createMessageStreamsForTopicFilter()).thenReturn(streams); + final ConsumerIterator iterator = mock(ConsumerIterator.class); + when(stream.iterator()).thenReturn(iterator); + final MessageAndMetadata messageAndMetadata = mock(MessageAndMetadata.class); + when(iterator.next()).thenReturn(messageAndMetadata); + when(messageAndMetadata.message()).thenReturn("got message"); + when(messageAndMetadata.topic()).thenReturn("topic"); + when(messageAndMetadata.partition()).thenReturn(1); + + final Map>> messages = consumerConfiguration.receive(); + Assert.assertEquals(messages.size(), 1); + Assert.assertEquals(messages.get("topic").size(), 1); + Assert.assertEquals(messages.get("topic").get(1).get(0), "got message"); + + verify(stream, times(1)).iterator(); + verify(iterator, times(1)).next(); + verify(messageAndMetadata, times(1)).message(); + verify(messageAndMetadata, times(1)).topic(); + } + + @Test + @SuppressWarnings("unchecked") + public void testReceiveMessageForTopicFilterFromMultipleStreams() { + final ConsumerMetadata consumerMetadata = mock(ConsumerMetadata.class); + final ConsumerConnectionProvider consumerConnectionProvider = + mock(ConsumerConnectionProvider.class); + final MessageLeftOverTracker messageLeftOverTracker = mock(MessageLeftOverTracker.class); + + when(consumerMetadata.getTopicFilterConfiguration()).thenReturn(new TopicFilterConfiguration(".*", 1, false)); + + final ConsumerConnector consumerConnector = mock(ConsumerConnector.class); + + when(consumerConnectionProvider.getConsumerConnector()).thenReturn(consumerConnector); + + final ConsumerConfiguration consumerConfiguration = new ConsumerConfiguration(consumerMetadata, + consumerConnectionProvider, messageLeftOverTracker); + consumerConfiguration.setMaxMessages(3); + + final KafkaStream stream1 = mock(KafkaStream.class); + final KafkaStream stream2 = mock(KafkaStream.class); + final KafkaStream stream3 = mock(KafkaStream.class); + final List> streams = new ArrayList>(); + streams.add(stream1); + streams.add(stream2); + streams.add(stream3); + + when(consumerConfiguration.createMessageStreamsForTopicFilter()).thenReturn(streams); + final ConsumerIterator iterator1 = mock(ConsumerIterator.class); + final ConsumerIterator iterator2 = mock(ConsumerIterator.class); + final ConsumerIterator iterator3 = mock(ConsumerIterator.class); + + when(stream1.iterator()).thenReturn(iterator1); + when(stream2.iterator()).thenReturn(iterator2); + when(stream3.iterator()).thenReturn(iterator3); + final MessageAndMetadata messageAndMetadata1 = mock(MessageAndMetadata.class); + final MessageAndMetadata messageAndMetadata2 = mock(MessageAndMetadata.class); + final MessageAndMetadata messageAndMetadata3 = mock(MessageAndMetadata.class); + + when(iterator1.next()).thenReturn(messageAndMetadata1); + when(iterator2.next()).thenReturn(messageAndMetadata2); + when(iterator3.next()).thenReturn(messageAndMetadata3); + + when(messageAndMetadata1.message()).thenReturn("got message"); + when(messageAndMetadata1.topic()).thenReturn("topic"); + when(messageAndMetadata1.partition()).thenReturn(1); + + when(messageAndMetadata2.message()).thenReturn("got message"); + when(messageAndMetadata2.topic()).thenReturn("topic"); + when(messageAndMetadata2.partition()).thenReturn(2); + + when(messageAndMetadata3.message()).thenReturn("got message"); + when(messageAndMetadata3.topic()).thenReturn("topic"); + when(messageAndMetadata3.partition()).thenReturn(3); + + final Map>> messages = consumerConfiguration.receive(); + Assert.assertEquals(messages.size(), 1); + int sum = 0; + + final Map> values = messages.get("topic"); + + for (final List l : values.values()) { + sum += l.size(); + } + + Assert.assertEquals(sum, 3); + } + private boolean valueFound(final List l, final String value){ for (final Object o : l){ if (value.equals(o)){ diff --git a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/support/KafkaProducerContextTests.java b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/support/KafkaProducerContextTests.java new file mode 100644 index 0000000000..c1bc1ed32a --- /dev/null +++ b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/support/KafkaProducerContextTests.java @@ -0,0 +1,65 @@ +/* + * Copyright 2002-2013 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.springframework.integration.kafka.support; + +import java.util.HashMap; +import java.util.Map; + +import org.junit.Assert; + +import kafka.javaapi.producer.Producer; + +import org.junit.Test; +import org.mockito.Mockito; +import org.springframework.beans.factory.ListableBeanFactory; + +/** + * @author Rajasekar Elango + * @since 0.5 + */ +public class KafkaProducerContextTests { + + @SuppressWarnings({ "rawtypes", "unchecked" }) + @Test + public void testTopicRegexForProducerConfiguration(){ + + final KafkaProducerContext kafkaProducerContext = new KafkaProducerContext(); + final ListableBeanFactory beanFactory = Mockito.mock(ListableBeanFactory.class); + + final ProducerMetadata producerMetadata = Mockito.mock(ProducerMetadata.class); + + String testRegex = "test.*"; + + Mockito.when(producerMetadata.getTopic()).thenReturn(testRegex); + final Producer producer = Mockito.mock(Producer.class); + + final ProducerConfiguration producerConfiguration = new ProducerConfiguration(producerMetadata, producer); + + + final Map topicConfigurations = new HashMap(); + topicConfigurations.put(testRegex, producerConfiguration); + + Mockito.when(beanFactory.getBeansOfType(ProducerConfiguration.class)).thenReturn(topicConfigurations); + kafkaProducerContext.setBeanFactory(beanFactory); + + Assert.assertNotNull(kafkaProducerContext.getTopicConfiguration("test1")); + Assert.assertNotNull(kafkaProducerContext.getTopicConfiguration("test2")); + Assert.assertNotNull(kafkaProducerContext.getTopicConfiguration("testabc")); + Assert.assertNull(kafkaProducerContext.getTopicConfiguration("dontmatch_testRegex")); + + } + +} diff --git a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/test/utils/User.java b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/test/utils/User.java index c6ad7ddaff..8ceada801e 100644 --- a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/test/utils/User.java +++ b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/test/utils/User.java @@ -46,7 +46,6 @@ public class User extends org.apache.avro.specific.SpecificRecordBase implements } // Used by DatumReader. Applications should not call. - @SuppressWarnings(value = "unchecked") public void put(int field$, java.lang.Object value$) { switch (field$) { case 0: @@ -92,5 +91,3 @@ public class User extends org.apache.avro.specific.SpecificRecordBase implements this.lastName = value; } } - -