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

Ensure default topics are fully qualified (#849)

Browse Source 
* Ensure default topics are fully qualified

Prior to this commit, only the topics specified on the `@PulsarListener`
were being fully qualified in the `DefaultPulsarConsumerFactory`. With
this change all topics (including the default topics driven by the
Spring Boot `spring.pulsar.consumer.topics` config prop) are
fully-qualified.
This commit is contained in:
Chris Bono
2024-09-13 10:32:03 -05:00
committed by GitHub
parent b16bfae4c0
commit bce10bdccf
2 changed files with 17 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
integration-tests/src/intTest/java/org/springframework/pulsar/inttest/listener/PulsarListenerIntegrationTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2022-2023 the original author or authors.
* Copyright 2022-2024 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -160,7 +160,8 @@ class PulsarListenerIntegrationTests implements PulsarTestContainerSupport {
public void listen(String ignored, Consumer<String> consumer) {
assertThat(consumer).extracting("conf", InstanceOfAssertFactories.type(ConsumerConfigurationData.class))
.satisfies((conf) -> {
assertThat(conf.getSingleTopic()).isEqualTo("plit-config-props-topic-dev");
assertThat(conf.getSingleTopic())
.isEqualTo("persistent://public/default/plit-config-props-topic-dev");
assertThat(conf.getSubscriptionType()).isEqualTo(SubscriptionType.Shared);
assertThat(conf.getSubscriptionName()).isEqualTo("plit-config-props-subs-dev");
});

18
spring-pulsar/src/main/java/org/springframework/pulsar/core/DefaultPulsarConsumerFactory.java
View File

@@ -101,10 +101,11 @@ public class DefaultPulsarConsumerFactory<T> implements PulsarConsumerFactory<T>
Objects.requireNonNull(schema, "Schema must be specified");
ConsumerBuilder<T> consumerBuilder = this.pulsarClient.newConsumer(schema);
// Apply the default config customizer (preserve the topic)
// Apply the default config customizer
if (!CollectionUtils.isEmpty(this.defaultConfigCustomizers)) {
this.defaultConfigCustomizers.forEach((customizer -> customizer.customize(consumerBuilder)));
}
// Preserve the passed in topics (don't let default config customizer win)
if (topics != null) {
replaceTopicsOnBuilder(consumerBuilder, topics);
}
@@ -117,6 +118,9 @@ public class DefaultPulsarConsumerFactory<T> implements PulsarConsumerFactory<T>
if (!CollectionUtils.isEmpty(customizers)) {
customizers.forEach(customizer -> customizer.customize(consumerBuilder));
}
if (this.topicBuilder != null) {
this.ensureTopicNamesFullyQualified(consumerBuilder);
}
try {
return consumerBuilder.subscribe();
}
@@ -126,9 +130,6 @@ public class DefaultPulsarConsumerFactory<T> implements PulsarConsumerFactory<T>
}
private void replaceTopicsOnBuilder(ConsumerBuilder<T> builder, Collection<String> topics) {
if (this.topicBuilder != null) {
topics = topics.stream().map(this.topicBuilder::getFullyQualifiedNameForTopic).toList();
}
var builderImpl = (ConsumerBuilderImpl<T>) builder;
builderImpl.getConf().setTopicNames(new HashSet<>(topics));
}
@@ -139,4 +140,13 @@ public class DefaultPulsarConsumerFactory<T> implements PulsarConsumerFactory<T>
builderImpl.getConf().setProperties(new TreeMap<>(metadataProperties));
}
protected void ensureTopicNamesFullyQualified(ConsumerBuilder<T> builder) {
var builderImpl = (ConsumerBuilderImpl<T>) builder;
var topics = builderImpl.getConf().getTopicNames();
if (!CollectionUtils.isEmpty(topics)) {
var fullyQualifiedTopics = topics.stream().map(this.topicBuilder::getFullyQualifiedNameForTopic).toList();
builderImpl.getConf().setTopicNames(new HashSet<>(fullyQualifiedTopics));
}
}
}