SpringCloud/spring-cloud-stream
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

GH-373: Support multiplexed consumers

Browse Source 
Resolves https://github.com/spring-cloud/spring-cloud-stream-binder-kafka/issues/373

When a consumer is multiplexed, configure the container to listen to multiple topics.
Also for the polled consumer.

When using a DLQ, determine the queue name from the topic in the failed record (unless
an explicit DLQ name has been provisioned - in which case, the same DLQ will be used
for all topics.

Resolves #383
This commit is contained in:
Gary Russell
2018-05-14 15:08:05 -04:00
committed by Oleg Zhurakousky
parent 1b00179f25
commit ca2983c881
3 changed files with 117 additions and 44 deletions
Download Patch File Download Diff File Expand all files Collapse all files

14
spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/provisioning/KafkaTopicProvisioner.java
View File

@@ -151,6 +151,20 @@ public class KafkaTopicProvisioner implements ProvisioningProvider<ExtendedConsu
@Override
public ConsumerDestination provisionConsumerDestination(final String name, final String group,
ExtendedConsumerProperties<KafkaConsumerProperties> properties) {
if (!properties.isMultiplex()) {
return doProvisionConsumerDestination(name, group, properties);
}
else {
String[] destinations = StringUtils.commaDelimitedListToStringArray(name);
for (String destination : destinations) {
doProvisionConsumerDestination(destination.trim(), group, properties);
}
return new KafkaConsumerDestination(name);
}
}
private ConsumerDestination doProvisionConsumerDestination(final String name, final String group,
ExtendedConsumerProperties<KafkaConsumerProperties> properties) {
if (properties.getExtension().isDestinationIsPattern()) {
Assert.isTrue(!properties.getExtension().isEnableDlq(),

117
spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java
View File

@@ -331,31 +331,29 @@ public class KafkaMessageChannelBinder extends
int partitionCount = extendedConsumerProperties.getInstanceCount()
* extendedConsumerProperties.getConcurrency();
Collection<PartitionInfo> listenedPartitions = new ArrayList<>();
boolean usingPatterns = extendedConsumerProperties.getExtension().isDestinationIsPattern();
Collection<PartitionInfo> allPartitions = usingPatterns ? Collections.emptyList()
: getPartitionInfo(destination, extendedConsumerProperties, consumerFactory, partitionCount);
Collection<PartitionInfo> listenedPartitions;
Assert.isTrue(!usingPatterns || !extendedConsumerProperties.isMultiplex(),
"Cannot use a pattern with multiplexed destinations; "
+ "use the regex pattern to specify multiple topics instead");
boolean groupManagement = extendedConsumerProperties.getExtension().isAutoRebalanceEnabled();
if (groupManagement ||
extendedConsumerProperties.getInstanceCount() == 1) {
listenedPartitions = allPartitions;
if (!extendedConsumerProperties.isMultiplex()) {
listenedPartitions.addAll(processTopic(group, extendedConsumerProperties, consumerFactory,
partitionCount, usingPatterns, groupManagement, destination.getName()));
}
else {
listenedPartitions = new ArrayList<>();
for (PartitionInfo partition : allPartitions) {
// divide partitions across modules
if ((partition.partition()
% extendedConsumerProperties.getInstanceCount()) == extendedConsumerProperties
.getInstanceIndex()) {
listenedPartitions.add(partition);
}
for (String name : StringUtils.commaDelimitedListToStringArray(destination.getName())) {
listenedPartitions.addAll(processTopic(group, extendedConsumerProperties, consumerFactory,
partitionCount, usingPatterns, groupManagement, name.trim()));
}
}
String topics = destination.getName();
this.topicsInUse.put(topics, new TopicInformation(group, listenedPartitions));
String[] topics = extendedConsumerProperties.isMultiplex() ? StringUtils.commaDelimitedListToStringArray(destination.getName())
: new String[] { destination.getName() };
for (int i = 0; i < topics.length; i++) {
topics[i] = topics[i].trim();
}
Assert.isTrue(usingPatterns
|| !CollectionUtils.isEmpty(listenedPartitions), "A list of partitions must be provided");
final TopicPartitionInitialOffset[] topicPartitionInitialOffsets = getTopicPartitionInitialOffsets(
@@ -363,7 +361,7 @@ public class KafkaMessageChannelBinder extends
final ContainerProperties containerProperties = anonymous
|| extendedConsumerProperties.getExtension().isAutoRebalanceEnabled()
? usingPatterns
? new ContainerProperties(Pattern.compile(topics))
? new ContainerProperties(Pattern.compile(topics[0]))
: new ContainerProperties(topics)
: new ContainerProperties(topicPartitionInitialOffsets);
if (this.transactionManager != null) {
@@ -425,6 +423,33 @@ public class KafkaMessageChannelBinder extends
return kafkaMessageDrivenChannelAdapter;
}
public Collection<PartitionInfo> processTopic(final String group,
final ExtendedConsumerProperties<KafkaConsumerProperties> extendedConsumerProperties,
final ConsumerFactory<?, ?> consumerFactory, int partitionCount, boolean usingPatterns,
boolean groupManagement, String topic) {
Collection<PartitionInfo> listenedPartitions;
Collection<PartitionInfo> allPartitions = usingPatterns ? Collections.emptyList()
: getPartitionInfo(topic, extendedConsumerProperties, consumerFactory, partitionCount);
if (groupManagement ||
extendedConsumerProperties.getInstanceCount() == 1) {
listenedPartitions = allPartitions;
}
else {
listenedPartitions = new ArrayList<>();
for (PartitionInfo partition : allPartitions) {
// divide partitions across modules
if ((partition.partition()
% extendedConsumerProperties.getInstanceCount()) == extendedConsumerProperties
.getInstanceIndex()) {
listenedPartitions.add(partition);
}
}
}
this.topicsInUse.put(topic, new TopicInformation(group, listenedPartitions));
return listenedPartitions;
}
/*
* Reset the offsets if needed; may update the offsets in in the container's
* topicPartitionInitialOffsets.
@@ -486,15 +511,29 @@ public class KafkaMessageChannelBinder extends
String consumerGroup = anonymous ? "anonymous." + UUID.randomUUID().toString() : group;
final ConsumerFactory<?, ?> consumerFactory = createKafkaConsumerFactory(anonymous, consumerGroup,
consumerProperties);
KafkaMessageSource<?, ?> source = new KafkaMessageSource<>(consumerFactory, destination.getName());
String[] topics = consumerProperties.isMultiplex() ? StringUtils.commaDelimitedListToStringArray(destination.getName())
: new String[] { destination.getName() };
for (int i = 0; i < topics.length; i++) {
topics[i] = topics[i].trim();
}
KafkaMessageSource<?, ?> source = new KafkaMessageSource<>(consumerFactory, topics);
source.setMessageConverter(getMessageConverter(consumerProperties));
source.setRawMessageHeader(consumerProperties.getExtension().isEnableDlq());
// I copied this from the regular consumer - it looks bogus to me - includes all partitions
// not just the ones this binding is listening to; doesn't seem right for a health check.
Collection<PartitionInfo> partitionInfos = getPartitionInfo(destination, consumerProperties, consumerFactory,
-1);
this.topicsInUse.put(destination.getName(), new TopicInformation(group, partitionInfos));
if (!consumerProperties.isMultiplex()) {
// I copied this from the regular consumer - it looks bogus to me - includes all partitions
// not just the ones this binding is listening to; doesn't seem right for a health check.
Collection<PartitionInfo> partitionInfos = getPartitionInfo(destination.getName(), consumerProperties,
consumerFactory, -1);
this.topicsInUse.put(destination.getName(), new TopicInformation(group, partitionInfos));
}
else {
for (int i = 0; i < topics.length; i++) {
Collection<PartitionInfo> partitionInfos = getPartitionInfo(topics[i], consumerProperties,
consumerFactory, -1);
this.topicsInUse.put(topics[i], new TopicInformation(group, partitionInfos));
}
}
source.setRebalanceListener(new ConsumerRebalanceListener() {
@@ -576,16 +615,16 @@ public class KafkaMessageChannelBinder extends
return mapper;
}
private Collection<PartitionInfo> getPartitionInfo(final ConsumerDestination destination,
private Collection<PartitionInfo> getPartitionInfo(String topic,
final ExtendedConsumerProperties<KafkaConsumerProperties> extendedConsumerProperties,
final ConsumerFactory<?, ?> consumerFactory, int partitionCount) {
Collection<PartitionInfo> allPartitions = provisioningProvider.getPartitionsForTopic(partitionCount,
extendedConsumerProperties.getExtension().isAutoRebalanceEnabled(),
() -> {
Consumer<?, ?> consumer = consumerFactory.createConsumer();
List<PartitionInfo> partitionsFor = consumer.partitionsFor(destination.getName());
consumer.close();
return partitionsFor;
try (Consumer<?, ?> consumer = consumerFactory.createConsumer()) {
List<PartitionInfo> partitionsFor = consumer.partitionsFor(topic);
return partitionsFor;
}
});
return allPartitions;
}
@@ -606,12 +645,9 @@ public class KafkaMessageChannelBinder extends
: getProducerFactory(null,
new ExtendedProducerProperties<>(dlqProducerProperties));
final KafkaTemplate<?,?> kafkaTemplate = new KafkaTemplate<>(producerFactory);
String dlqName = StringUtils.hasText(kafkaConsumerProperties.getDlqName())
? kafkaConsumerProperties.getDlqName()
: "error." + destination.getName() + "." + group;
@SuppressWarnings({"unchecked", "rawtypes"})
DlqSender<?,?> dlqSender = new DlqSender(kafkaTemplate, dlqName);
DlqSender<?,?> dlqSender = new DlqSender(kafkaTemplate);
return message -> {
@SuppressWarnings("unchecked")
@@ -672,7 +708,10 @@ public class KafkaMessageChannelBinder extends
}
}
}
dlqSender.sendToDlq(recordToSend.get(), kafkaHeaders);
String dlqName = StringUtils.hasText(kafkaConsumerProperties.getDlqName())
? kafkaConsumerProperties.getDlqName()
: "error." + record.topic() + "." + group;
dlqSender.sendToDlq(recordToSend.get(), kafkaHeaders, dlqName);
};
}
return null;
@@ -861,18 +900,16 @@ public class KafkaMessageChannelBinder extends
private final class DlqSender<K,V> {
private final KafkaTemplate<K,V> kafkaTemplate;
private final String dlqName;
DlqSender(KafkaTemplate<K, V> kafkaTemplate, String dlqName) {
DlqSender(KafkaTemplate<K, V> kafkaTemplate) {
this.kafkaTemplate = kafkaTemplate;
this.dlqName = dlqName;
}
@SuppressWarnings("unchecked")
void sendToDlq(ConsumerRecord<?, ?> consumerRecord, Headers headers) {
void sendToDlq(ConsumerRecord<?, ?> consumerRecord, Headers headers, String dlqName) {
K key = (K)consumerRecord.key();
V value = (V)consumerRecord.value();
ProducerRecord<K,V> producerRecord = new ProducerRecord<>(this.dlqName, consumerRecord.partition(),
ProducerRecord<K,V> producerRecord = new ProducerRecord<>(dlqName, consumerRecord.partition(),
key, value, headers);
StringBuilder sb = new StringBuilder().append(" a message with key='")

30
spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderTests.java
View File

@@ -102,6 +102,7 @@ import org.springframework.kafka.core.ProducerFactory;
import org.springframework.kafka.listener.AbstractMessageListenerContainer;
import org.springframework.kafka.listener.AbstractMessageListenerContainer.AckMode;
import org.springframework.kafka.listener.ConcurrentMessageListenerContainer;
import org.springframework.kafka.listener.MessageListenerContainer;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.kafka.support.KafkaHeaders;
import org.springframework.kafka.support.SendResult;
@@ -607,6 +608,7 @@ public class KafkaBinderTests extends
consumerProperties.getExtension().setEnableDlq(true);
consumerProperties.getExtension().setAutoRebalanceEnabled(false);
consumerProperties.setHeaderMode(headerMode);
consumerProperties.setMultiplex(true);
DirectChannel moduleInputChannel = createBindableChannel("input", createConsumerBindingProperties(consumerProperties));
@@ -619,9 +621,14 @@ public class KafkaBinderTests extends
String producerName = "dlqTest." + uniqueBindingId + ".0";
Binding<MessageChannel> producerBinding = binder.bindProducer(producerName,
moduleOutputChannel, producerProperties);
Binding<MessageChannel> consumerBinding = binder.bindConsumer(producerName,
String consumerDest = producerName + ", " + producerName.replaceAll("0", "1");
Binding<MessageChannel> consumerBinding = binder.bindConsumer(consumerDest,
"testGroup", moduleInputChannel, consumerProperties);
MessageListenerContainer container = TestUtils.getPropertyValue(consumerBinding,
"lifecycle.messageListenerContainer", MessageListenerContainer.class);
assertThat(container.getContainerProperties().getTopicPartitions().length).isEqualTo(2);
ExtendedConsumerProperties<KafkaConsumerProperties> dlqConsumerProperties = createConsumerProperties();
dlqConsumerProperties.setMaxAttempts(1);
dlqConsumerProperties.setHeaderMode(headerMode);
@@ -629,7 +636,7 @@ public class KafkaBinderTests extends
ApplicationContext context = TestUtils.getPropertyValue(binder.getBinder(), "applicationContext",
ApplicationContext.class);
SubscribableChannel boundErrorChannel = context
.getBean(producerName + ".testGroup.errors-0", SubscribableChannel.class);
.getBean(consumerDest + ".testGroup.errors-0", SubscribableChannel.class);
SubscribableChannel globalErrorChannel = context.getBean("errorChannel", SubscribableChannel.class);
final AtomicReference<Message<?>> boundErrorChannelMessage = new AtomicReference<>();
final AtomicReference<Message<?>> globalErrorChannelMessage = new AtomicReference<>();
@@ -2451,8 +2458,10 @@ public class KafkaBinderTests extends
public void testPolledConsumer() throws Exception {
KafkaTestBinder binder = getBinder();
PollableSource<MessageHandler> inboundBindTarget = new DefaultPollableMessageSource(this.messageConverter);
Binding<PollableSource<MessageHandler>> binding = binder.bindPollableConsumer("pollable", "group",
inboundBindTarget, createConsumerProperties());
ExtendedConsumerProperties<KafkaConsumerProperties> consumerProps = createConsumerProperties();
consumerProps.setMultiplex(true);
Binding<PollableSource<MessageHandler>> binding = binder.bindPollableConsumer("pollable,anotherOne", "group",
inboundBindTarget, consumerProps);
Map<String, Object> producerProps = KafkaTestUtils.producerProps(embeddedKafka);
KafkaTemplate template = new KafkaTemplate(new DefaultKafkaProducerFactory<>(producerProps));
template.send("pollable", "testPollable");
@@ -2467,6 +2476,19 @@ public class KafkaBinderTests extends
Thread.sleep(100);
}
assertThat(polled).isTrue();
template.send("anotherOne", "testPollable2");
polled = inboundBindTarget.poll(m -> {
assertThat(m.getPayload()).isEqualTo("testPollable2");
});
n = 0;
while (n++ < 100 && !polled) {
polled = inboundBindTarget.poll(m -> {
assertThat(m.getPayload()).isEqualTo("testPollable2".getBytes());
});
Thread.sleep(100);
}
assertThat(polled).isTrue();
binding.unbind();
}