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

Multiplex config in ReactorKafkaBinder

Browse Source 
- The bug for multiplex configuration in ReactorKafkaBinder was resolved
   and a method was written for common partition-related operations.
 - Refactor KafkaBinderUnitTests
This commit is contained in:
omercelikceng
2023-11-29 19:48:41 +03:00
committed by Soby Chacko
parent 116813d1bb
commit de989d65d2
4 changed files with 92 additions and 102 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -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<PartitionInfo> getListenedPartitions(final String group,
final ExtendedConsumerProperties<KafkaConsumerProperties> extendedConsumerProperties,
final ConsumerFactory<?, ?> consumerFactory, int partitionCount,
boolean usingPatterns, boolean groupManagement, final String topic,
Map<String, TopicInformation> topicsInUse) {
Collection<PartitionInfo> listenedPartitions;
Collection<PartitionInfo> 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<PartitionInfo> getPartitionInfoForConsumer(final String topic,
final ExtendedConsumerProperties<KafkaConsumerProperties> 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<PartitionInfo> getPartitionInfoForProducer(final String topicName,
final ProducerFactory<byte[], byte[]> producerFB,
final ExtendedProducerProperties<KafkaProducerProperties> producerProperties) {
return getPartitionsForTopic(
producerProperties.getPartitionCount(), false, () -> {
Producer<byte[], byte[]> producer = producerFB.createProducer();
List<PartitionInfo> 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.

58
binders/kafka-binder/spring-cloud-stream-binder-kafka-reactive/src/main/java/org/springframework/cloud/stream/binder/reactorkafka/ReactorKafkaBinder.java
View File

@@ -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<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, usingPatterns));
}
private Collection<PartitionInfo> getPartitionInfo(String topic,
final ExtendedConsumerProperties<KafkaConsumerProperties> extendedConsumerProperties,
final ConsumerFactory<?, ?> consumerFactory, int partitionCount) {
return provisioningProvider.getPartitionsForTopic(partitionCount,
extendedConsumerProperties.getExtension().isAutoRebalanceEnabled(),
() -> {
try (Consumer<?, ?> consumer = consumerFactory.createConsumer()) {
return consumer.partitionsFor(topic);
}
}, topic);
}
Map<String, TopicInformation> getTopicsInUse() {
return this.topicsInUse;
}
@@ -292,8 +251,19 @@ public class ReactorKafkaBinder
DefaultKafkaConsumerFactory<Object, Object> 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 {

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

@@ -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<PartitionInfo> partitions = provisioningProvider.getPartitionsForTopic(
producerProperties.getPartitionCount(), false, () -> {
Producer<byte[], byte[]> producer = producerFB.createProducer();
List<PartitionInfo> partitionsFor = producer
.partitionsFor(destination.getName());
producer.close();
return partitionsFor;
}, destination.getName());
Collection<PartitionInfo> 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<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, 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<PartitionInfo> partitionInfos = getPartitionInfo(
Collection<PartitionInfo> 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<PartitionInfo> partitionInfos = getPartitionInfo(topics[i],
Collection<PartitionInfo> 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<PartitionInfo> getPartitionInfo(String topic,
final ExtendedConsumerProperties<KafkaConsumerProperties> 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();

7
binders/kafka-binder/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderUnitTests.java
View File

@@ -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<KafkaConsumerProperties> ecp = new ExtendedConsumerProperties<KafkaConsumerProperties>(
ExtendedConsumerProperties<KafkaConsumerProperties> 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<byte[], byte[]> consumer = mock(Consumer.class);
final CountDownLatch latch = new CountDownLatch(1);