From ca2983c8812ef8054c3b939a555ffd82953d3c08 Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Mon, 14 May 2018 15:08:05 -0400 Subject: [PATCH] GH-373: Support multiplexed consumers 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 --- .../provisioning/KafkaTopicProvisioner.java | 14 +++ .../kafka/KafkaMessageChannelBinder.java | 117 ++++++++++++------ .../stream/binder/kafka/KafkaBinderTests.java | 30 ++++- 3 files changed, 117 insertions(+), 44 deletions(-) diff --git a/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/provisioning/KafkaTopicProvisioner.java b/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/provisioning/KafkaTopicProvisioner.java index d885dd3f6..c7ed28f8d 100644 --- a/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/provisioning/KafkaTopicProvisioner.java +++ b/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/provisioning/KafkaTopicProvisioner.java @@ -151,6 +151,20 @@ public class KafkaTopicProvisioner implements ProvisioningProvider 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 properties) { if (properties.getExtension().isDestinationIsPattern()) { Assert.isTrue(!properties.getExtension().isEnableDlq(), diff --git a/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java b/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java index 8f94bbe84..9ea126b76 100644 --- a/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java +++ b/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java @@ -331,31 +331,29 @@ public class KafkaMessageChannelBinder extends int partitionCount = extendedConsumerProperties.getInstanceCount() * extendedConsumerProperties.getConcurrency(); + Collection listenedPartitions = new ArrayList<>(); + boolean usingPatterns = extendedConsumerProperties.getExtension().isDestinationIsPattern(); - Collection allPartitions = usingPatterns ? Collections.emptyList() - : getPartitionInfo(destination, extendedConsumerProperties, consumerFactory, partitionCount); - - Collection 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 processTopic(final String group, + final ExtendedConsumerProperties extendedConsumerProperties, + final ConsumerFactory consumerFactory, int partitionCount, boolean usingPatterns, + boolean groupManagement, String topic) { + Collection listenedPartitions; + Collection 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 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 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 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 getPartitionInfo(final ConsumerDestination destination, + private Collection getPartitionInfo(String topic, final ExtendedConsumerProperties extendedConsumerProperties, final ConsumerFactory consumerFactory, int partitionCount) { Collection allPartitions = provisioningProvider.getPartitionsForTopic(partitionCount, extendedConsumerProperties.getExtension().isAutoRebalanceEnabled(), () -> { - Consumer consumer = consumerFactory.createConsumer(); - List partitionsFor = consumer.partitionsFor(destination.getName()); - consumer.close(); - return partitionsFor; + try (Consumer consumer = consumerFactory.createConsumer()) { + List 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 { private final KafkaTemplate kafkaTemplate; - private final String dlqName; - DlqSender(KafkaTemplate kafkaTemplate, String dlqName) { + DlqSender(KafkaTemplate 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 producerRecord = new ProducerRecord<>(this.dlqName, consumerRecord.partition(), + ProducerRecord producerRecord = new ProducerRecord<>(dlqName, consumerRecord.partition(), key, value, headers); StringBuilder sb = new StringBuilder().append(" a message with key='") diff --git a/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderTests.java b/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderTests.java index 78fc0be46..3427d99be 100644 --- a/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderTests.java +++ b/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderTests.java @@ -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 producerBinding = binder.bindProducer(producerName, moduleOutputChannel, producerProperties); - Binding consumerBinding = binder.bindConsumer(producerName, + String consumerDest = producerName + ", " + producerName.replaceAll("0", "1"); + Binding consumerBinding = binder.bindConsumer(consumerDest, "testGroup", moduleInputChannel, consumerProperties); + MessageListenerContainer container = TestUtils.getPropertyValue(consumerBinding, + "lifecycle.messageListenerContainer", MessageListenerContainer.class); + assertThat(container.getContainerProperties().getTopicPartitions().length).isEqualTo(2); + ExtendedConsumerProperties 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> boundErrorChannelMessage = new AtomicReference<>(); final AtomicReference> globalErrorChannelMessage = new AtomicReference<>(); @@ -2451,8 +2458,10 @@ public class KafkaBinderTests extends public void testPolledConsumer() throws Exception { KafkaTestBinder binder = getBinder(); PollableSource inboundBindTarget = new DefaultPollableMessageSource(this.messageConverter); - Binding> binding = binder.bindPollableConsumer("pollable", "group", - inboundBindTarget, createConsumerProperties()); + ExtendedConsumerProperties consumerProps = createConsumerProperties(); + consumerProps.setMultiplex(true); + Binding> binding = binder.bindPollableConsumer("pollable,anotherOne", "group", + inboundBindTarget, consumerProps); Map 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(); }