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

Kafka: KCCtx: return null if consumedData.isEmpty

Browse Source 
Fixes: https://github.com/spring-projects/spring-integration-extensions/issues/68
This commit is contained in:
Artem Bilan
2014-07-01 15:01:22 +03:00
committed by Artem Bilan
parent fb40e521c7
commit 58ff34d83b
1 changed files with 8 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -15,6 +15,11 @@
*/
package org.springframework.integration.kafka.support;
import java.util.Collection;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.beans.factory.BeanFactoryAware;
@@ -22,11 +27,7 @@ import org.springframework.beans.factory.ListableBeanFactory;
import org.springframework.integration.kafka.core.KafkaConsumerDefaults;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.messaging.Message;
import java.util.Collection;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import org.springframework.util.CollectionUtils;
/**
* @author Soby Chacko
@@ -54,11 +55,11 @@ public class KafkaConsumerContext<K,V> implements BeanFactoryAware {
for (final ConsumerConfiguration<K,V> consumerConfiguration : getConsumerConfigurations()) {
final Map<String, Map<Integer, List<Object>>> messages = consumerConfiguration.receive();
if (messages != null){
if (!CollectionUtils.isEmpty(messages)){
consumedData.putAll(messages);
}
}
return MessageBuilder.withPayload(consumedData).build();
return consumedData.isEmpty() ? null : MessageBuilder.withPayload(consumedData).build();
}
public String getConsumerTimeout() {