From de989d65d260ae905cb12df2ec5c5201323cbb60 Mon Sep 17 00:00:00 2001 From: omercelikceng Date: Wed, 29 Nov 2023 19:48:41 +0300 Subject: [PATCH] Multiplex config in ReactorKafkaBinder - The bug for multiplex configuration in ReactorKafkaBinder was resolved and a method was written for common partition-related operations. - Refactor KafkaBinderUnitTests --- .../provisioning/KafkaTopicProvisioner.java | 65 +++++++++++++++++++ .../reactorkafka/ReactorKafkaBinder.java | 58 ++++------------- .../kafka/KafkaMessageChannelBinder.java | 64 +++--------------- .../binder/kafka/KafkaBinderUnitTests.java | 7 +- 4 files changed, 92 insertions(+), 102 deletions(-) diff --git a/binders/kafka-binder/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/provisioning/KafkaTopicProvisioner.java b/binders/kafka-binder/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/provisioning/KafkaTopicProvisioner.java index f9e823e9f..d8e69c539 100644 --- a/binders/kafka-binder/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/provisioning/KafkaTopicProvisioner.java +++ b/binders/kafka-binder/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/provisioning/KafkaTopicProvisioner.java @@ -46,6 +46,8 @@ import org.apache.kafka.clients.admin.ListTopicsResult; import org.apache.kafka.clients.admin.NewPartitions; import org.apache.kafka.clients.admin.NewTopic; import org.apache.kafka.clients.admin.TopicDescription; +import org.apache.kafka.clients.consumer.Consumer; +import org.apache.kafka.clients.producer.Producer; import org.apache.kafka.common.KafkaFuture; import org.apache.kafka.common.PartitionInfo; import org.apache.kafka.common.config.ConfigResource; @@ -57,6 +59,7 @@ import org.springframework.boot.autoconfigure.kafka.KafkaProperties; import org.springframework.cloud.stream.binder.BinderException; import org.springframework.cloud.stream.binder.ExtendedConsumerProperties; import org.springframework.cloud.stream.binder.ExtendedProducerProperties; +import org.springframework.cloud.stream.binder.kafka.common.TopicInformation; import org.springframework.cloud.stream.binder.kafka.properties.KafkaBinderConfigurationProperties; import org.springframework.cloud.stream.binder.kafka.properties.KafkaConsumerProperties; import org.springframework.cloud.stream.binder.kafka.properties.KafkaProducerProperties; @@ -66,6 +69,8 @@ import org.springframework.cloud.stream.provisioning.ConsumerDestination; import org.springframework.cloud.stream.provisioning.ProducerDestination; import org.springframework.cloud.stream.provisioning.ProvisioningException; import org.springframework.cloud.stream.provisioning.ProvisioningProvider; +import org.springframework.kafka.core.ConsumerFactory; +import org.springframework.kafka.core.ProducerFactory; import org.springframework.retry.RetryOperations; import org.springframework.retry.backoff.ExponentialBackOffPolicy; import org.springframework.retry.policy.SimpleRetryPolicy; @@ -542,6 +547,66 @@ public class KafkaTopicProvisioner implements } } + public Collection getListenedPartitions(final String group, + final ExtendedConsumerProperties extendedConsumerProperties, + final ConsumerFactory consumerFactory, int partitionCount, + boolean usingPatterns, boolean groupManagement, final String topic, + Map topicsInUse) { + Collection listenedPartitions; + Collection allPartitions = usingPatterns ? Collections.emptyList() + : getPartitionInfoForConsumer(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); + } + } + } + topicsInUse.put(topic, + new TopicInformation(group, listenedPartitions, usingPatterns)); + return listenedPartitions; + } + + /** + * Check that the topic has the expected number of partitions and return the partition information for consumer. + */ + public Collection getPartitionInfoForConsumer(final String topic, + final ExtendedConsumerProperties extendedConsumerProperties, + final ConsumerFactory consumerFactory, int partitionCount) { + return getPartitionsForTopic(partitionCount, + extendedConsumerProperties.getExtension().isAutoRebalanceEnabled(), + () -> { + try (Consumer consumer = consumerFactory.createConsumer()) { + return consumer.partitionsFor(topic); + } + }, topic); + } + + /** + * Check that the topic has the expected number of partitions and return the partition information for producer. + */ + public Collection getPartitionInfoForProducer(final String topicName, + final ProducerFactory producerFB, + final ExtendedProducerProperties producerProperties) { + return getPartitionsForTopic( + producerProperties.getPartitionCount(), false, () -> { + Producer producer = producerFB.createProducer(); + List partitionsFor = producer + .partitionsFor(topicName); + producer.close(); + return partitionsFor; + }, topicName); + } + /** * Check that the topic has the expected number of partitions and return the partition information. * @param partitionCount the expected count. diff --git a/binders/kafka-binder/spring-cloud-stream-binder-kafka-reactive/src/main/java/org/springframework/cloud/stream/binder/reactorkafka/ReactorKafkaBinder.java b/binders/kafka-binder/spring-cloud-stream-binder-kafka-reactive/src/main/java/org/springframework/cloud/stream/binder/reactorkafka/ReactorKafkaBinder.java index 91ec7fc41..ab08c90b8 100644 --- a/binders/kafka-binder/spring-cloud-stream-binder-kafka-reactive/src/main/java/org/springframework/cloud/stream/binder/reactorkafka/ReactorKafkaBinder.java +++ b/binders/kafka-binder/spring-cloud-stream-binder-kafka-reactive/src/main/java/org/springframework/cloud/stream/binder/reactorkafka/ReactorKafkaBinder.java @@ -19,7 +19,6 @@ package org.springframework.cloud.stream.binder.reactorkafka; import java.util.ArrayList; import java.util.Arrays; import java.util.Collection; -import java.util.Collections; import java.util.List; import java.util.Map; import java.util.UUID; @@ -28,7 +27,6 @@ import java.util.regex.Pattern; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; -import org.apache.kafka.clients.consumer.Consumer; import org.apache.kafka.clients.consumer.ConsumerRecord; import org.apache.kafka.clients.producer.Producer; import org.apache.kafka.clients.producer.ProducerRecord; @@ -67,7 +65,6 @@ import org.springframework.integration.core.MessageProducer; import org.springframework.integration.endpoint.MessageProducerSupport; import org.springframework.integration.handler.AbstractMessageHandler; import org.springframework.integration.support.MessageBuilder; -import org.springframework.kafka.core.ConsumerFactory; import org.springframework.kafka.core.DefaultKafkaConsumerFactory; import org.springframework.kafka.core.DefaultKafkaProducerFactory; import org.springframework.kafka.support.KafkaHeaders; @@ -86,6 +83,7 @@ import org.springframework.util.StringUtils; /** * @author Gary Russell * @author Byungjun You + * @author Omer Celik * @since 4.0 * */ @@ -207,45 +205,6 @@ public class ReactorKafkaBinder return new ReactorMessageHandler(opts, converter, destination.getName(), resultChannel); } - // TODO: Refactor to provide in a common area since KafkaMessageChannelBinder also provides this. - public void 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, usingPatterns)); - } - - private Collection getPartitionInfo(String topic, - final ExtendedConsumerProperties extendedConsumerProperties, - final ConsumerFactory consumerFactory, int partitionCount) { - return provisioningProvider.getPartitionsForTopic(partitionCount, - extendedConsumerProperties.getExtension().isAutoRebalanceEnabled(), - () -> { - try (Consumer consumer = consumerFactory.createConsumer()) { - return consumer.partitionsFor(topic); - } - }, topic); - } - Map getTopicsInUse() { return this.topicsInUse; } @@ -292,8 +251,19 @@ public class ReactorKafkaBinder DefaultKafkaConsumerFactory factory = new DefaultKafkaConsumerFactory<>(props); int partitionCount = properties.getInstanceCount() * properties.getConcurrency(); boolean groupManagement = properties.getExtension().isAutoRebalanceEnabled(); - processTopic(consumerGroup, properties, factory, partitionCount, properties.getExtension().isDestinationIsPattern(), - groupManagement, destination.getName()); + if (!properties.isMultiplex()) { + provisioningProvider.getListenedPartitions(consumerGroup, properties, factory, + partitionCount, properties.getExtension().isDestinationIsPattern(), + groupManagement, destination.getName(), topicsInUse); + } + else { + for (String name : StringUtils + .commaDelimitedListToStringArray(destination.getName())) { + provisioningProvider.getListenedPartitions(consumerGroup, properties, factory, + partitionCount, properties.getExtension().isDestinationIsPattern(), + groupManagement, name.trim(), topicsInUse); + } + } class ReactorMessageProducer extends MessageProducerSupport { diff --git a/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java b/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java index f4fe0f877..663a941c9 100644 --- a/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java +++ b/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java @@ -43,7 +43,6 @@ import org.apache.kafka.clients.consumer.Consumer; import org.apache.kafka.clients.consumer.ConsumerConfig; import org.apache.kafka.clients.consumer.ConsumerRebalanceListener; import org.apache.kafka.clients.consumer.ConsumerRecord; -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.PartitionInfo; @@ -164,6 +163,7 @@ import org.springframework.util.backoff.FixedBackOff; * @author Chris Bono * @author Byungjun You * @author Oliver Führer + * @author Omer Celik */ public class KafkaMessageChannelBinder extends // @checkstyle:off @@ -419,14 +419,8 @@ public class KafkaMessageChannelBinder extends ? transMan.getProducerFactory() : getProducerFactory(null, producerProperties, destination.getName() + ".producer", destination.getName()); - Collection partitions = provisioningProvider.getPartitionsForTopic( - producerProperties.getPartitionCount(), false, () -> { - Producer producer = producerFB.createProducer(); - List partitionsFor = producer - .partitionsFor(destination.getName()); - producer.close(); - return partitionsFor; - }, destination.getName()); + Collection partitions = provisioningProvider.getPartitionInfoForProducer( + destination.getName(), producerFB, producerProperties); this.topicsInUse.put(destination.getName(), new TopicInformation(null, partitions, false)); if (producerProperties.isPartitioned() @@ -609,16 +603,16 @@ public class KafkaMessageChannelBinder extends boolean groupManagement = extendedConsumerProperties.getExtension() .isAutoRebalanceEnabled(); if (!extendedConsumerProperties.isMultiplex()) { - listenedPartitions.addAll(processTopic(consumerGroup, + listenedPartitions.addAll(provisioningProvider.getListenedPartitions(consumerGroup, extendedConsumerProperties, consumerFactory, partitionCount, - usingPatterns, groupManagement, destination.getName())); + usingPatterns, groupManagement, destination.getName(), topicsInUse)); } else { for (String name : StringUtils .commaDelimitedListToStringArray(destination.getName())) { - listenedPartitions.addAll(processTopic(consumerGroup, + listenedPartitions.addAll(provisioningProvider.getListenedPartitions(consumerGroup, extendedConsumerProperties, consumerFactory, partitionCount, - usingPatterns, groupManagement, name.trim())); + usingPatterns, groupManagement, name.trim(), topicsInUse)); } } @@ -883,34 +877,6 @@ public class KafkaMessageChannelBinder extends }); } - 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, usingPatterns)); - return listenedPartitions; - } - /* * Reset the offsets if needed; may update the offsets in in the container's * topicPartitionInitialOffsets. @@ -1040,14 +1006,14 @@ public class KafkaMessageChannelBinder extends // all partitions // not just the ones this binding is listening to; doesn't seem right for a // health check. - Collection partitionInfos = getPartitionInfo( + Collection partitionInfos = provisioningProvider.getPartitionInfoForConsumer( destination.getName(), extendedConsumerProperties, consumerFactory, -1); this.topicsInUse.put(destination.getName(), new TopicInformation(consumerGroup, partitionInfos, false)); } else { for (int i = 0; i < topics.length; i++) { - Collection partitionInfos = getPartitionInfo(topics[i], + Collection partitionInfos = provisioningProvider.getPartitionInfoForConsumer(topics[i], extendedConsumerProperties, consumerFactory, -1); this.topicsInUse.put(topics[i], new TopicInformation(consumerGroup, partitionInfos, false)); @@ -1106,18 +1072,6 @@ public class KafkaMessageChannelBinder extends return mapper; } - private Collection getPartitionInfo(String topic, - final ExtendedConsumerProperties extendedConsumerProperties, - final ConsumerFactory consumerFactory, int partitionCount) { - return provisioningProvider.getPartitionsForTopic(partitionCount, - extendedConsumerProperties.getExtension().isAutoRebalanceEnabled(), - () -> { - try (Consumer consumer = consumerFactory.createConsumer()) { - return consumer.partitionsFor(topic); - } - }, topic); - } - @Override protected ErrorMessageStrategy getErrorMessageStrategy() { return new RawRecordHeaderErrorMessageStrategy(); diff --git a/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderUnitTests.java b/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderUnitTests.java index 4be06e858..1258872ca 100644 --- a/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderUnitTests.java +++ b/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderUnitTests.java @@ -67,6 +67,7 @@ import static org.mockito.Mockito.verify; /** * @author Gary Russell + * @author Omer Celik * @since 1.2.2 * */ @@ -83,7 +84,7 @@ class KafkaBinderUnitTests { KafkaMessageChannelBinder binder = new KafkaMessageChannelBinder( binderConfigurationProperties, provisioningProvider); KafkaConsumerProperties consumerProps = new KafkaConsumerProperties(); - ExtendedConsumerProperties ecp = new ExtendedConsumerProperties( + ExtendedConsumerProperties ecp = new ExtendedConsumerProperties<>( consumerProps); Method method = KafkaMessageChannelBinder.class.getDeclaredMethod( "createKafkaConsumerFactory", boolean.class, String.class, @@ -194,8 +195,8 @@ class KafkaBinderUnitTests { return partitions.stream().map(p -> new PartitionInfo(topic, part.getAndIncrement(), null, null, null)) .collect(Collectors.toList()); - }).given(provisioningProvider).getPartitionsForTopic(anyInt(), anyBoolean(), - any(), any()); + }).given(provisioningProvider).getListenedPartitions(anyString(), any(), any(), + anyInt(), anyBoolean(), anyBoolean(), anyString(), any()); @SuppressWarnings("unchecked") final Consumer consumer = mock(Consumer.class); final CountDownLatch latch = new CountDownLatch(1);