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

Add support for Pulsar default tenant/namespace

Browse Source 
This commit allows Pulsar users to configure a default tenant
and/or namespace to be used when producing or consuming messages
to topic URLs that are not fully-qualified.

See gh-41851
This commit is contained in:
Chris Bono
2024-08-13 19:53:21 -05:00
committed by Andy Wilkinson
parent 5c76189583
commit 98da3aa27c
9 changed files with 191 additions and 20 deletions
Download Patch File Download Diff File Expand all files Collapse all files

34
spring-boot-project/spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/pulsar/PulsarAutoConfiguration.java
View File

@@ -52,6 +52,7 @@ import org.springframework.pulsar.core.PulsarConsumerFactory;
import org.springframework.pulsar.core.PulsarProducerFactory;
import org.springframework.pulsar.core.PulsarReaderFactory;
import org.springframework.pulsar.core.PulsarTemplate;
import org.springframework.pulsar.core.PulsarTopicBuilder;
import org.springframework.pulsar.core.ReaderBuilderCustomizer;
import org.springframework.pulsar.core.SchemaResolver;
import org.springframework.pulsar.core.TopicResolver;
@@ -88,24 +89,29 @@ public class PulsarAutoConfiguration {
@ConditionalOnMissingBean(PulsarProducerFactory.class)
@ConditionalOnProperty(name = "spring.pulsar.producer.cache.enabled", havingValue = "false")
DefaultPulsarProducerFactory<?> pulsarProducerFactory(PulsarClient pulsarClient, TopicResolver topicResolver,
ObjectProvider<ProducerBuilderCustomizer<?>> customizersProvider) {
ObjectProvider<ProducerBuilderCustomizer<?>> customizersProvider, PulsarTopicBuilder topicBuilder) {
List<ProducerBuilderCustomizer<Object>> lambdaSafeCustomizers = lambdaSafeProducerBuilderCustomizers(
customizersProvider);
return new DefaultPulsarProducerFactory<>(pulsarClient, this.properties.getProducer().getTopicName(),
lambdaSafeCustomizers, topicResolver);
DefaultPulsarProducerFactory<?> producerFactory = new DefaultPulsarProducerFactory<>(pulsarClient,
this.properties.getProducer().getTopicName(), lambdaSafeCustomizers, topicResolver);
producerFactory.setTopicBuilder(topicBuilder);
return producerFactory;
}
@Bean
@ConditionalOnMissingBean(PulsarProducerFactory.class)
@ConditionalOnProperty(name = "spring.pulsar.producer.cache.enabled", havingValue = "true", matchIfMissing = true)
CachingPulsarProducerFactory<?> cachingPulsarProducerFactory(PulsarClient pulsarClient, TopicResolver topicResolver,
ObjectProvider<ProducerBuilderCustomizer<?>> customizersProvider) {
ObjectProvider<ProducerBuilderCustomizer<?>> customizersProvider, PulsarTopicBuilder topicBuilder) {
PulsarProperties.Producer.Cache cacheProperties = this.properties.getProducer().getCache();
List<ProducerBuilderCustomizer<Object>> lambdaSafeCustomizers = lambdaSafeProducerBuilderCustomizers(
customizersProvider);
return new CachingPulsarProducerFactory<>(pulsarClient, this.properties.getProducer().getTopicName(),
lambdaSafeCustomizers, topicResolver, cacheProperties.getExpireAfterAccess(),
cacheProperties.getMaximumSize(), cacheProperties.getInitialCapacity());
CachingPulsarProducerFactory<?> producerFactory = new CachingPulsarProducerFactory<>(pulsarClient,
this.properties.getProducer().getTopicName(), lambdaSafeCustomizers, topicResolver,
cacheProperties.getExpireAfterAccess(), cacheProperties.getMaximumSize(),
cacheProperties.getInitialCapacity());
producerFactory.setTopicBuilder(topicBuilder);
return producerFactory;
}
private List<ProducerBuilderCustomizer<Object>> lambdaSafeProducerBuilderCustomizers(
@@ -138,13 +144,16 @@ public class PulsarAutoConfiguration {
@Bean
@ConditionalOnMissingBean(PulsarConsumerFactory.class)
DefaultPulsarConsumerFactory<?> pulsarConsumerFactory(PulsarClient pulsarClient,
ObjectProvider<ConsumerBuilderCustomizer<?>> customizersProvider) {
ObjectProvider<ConsumerBuilderCustomizer<?>> customizersProvider, PulsarTopicBuilder topicBuilder) {
List<ConsumerBuilderCustomizer<?>> customizers = new ArrayList<>();
customizers.add(this.propertiesMapper::customizeConsumerBuilder);
customizers.addAll(customizersProvider.orderedStream().toList());
List<ConsumerBuilderCustomizer<Object>> lambdaSafeCustomizers = List
.of((builder) -> applyConsumerBuilderCustomizers(customizers, builder));
return new DefaultPulsarConsumerFactory<>(pulsarClient, lambdaSafeCustomizers);
DefaultPulsarConsumerFactory<?> consumerFactory = new DefaultPulsarConsumerFactory<>(pulsarClient,
lambdaSafeCustomizers);
consumerFactory.setTopicBuilder(topicBuilder);
return consumerFactory;
}
@Bean
@@ -181,13 +190,16 @@ public class PulsarAutoConfiguration {
@Bean
@ConditionalOnMissingBean(PulsarReaderFactory.class)
DefaultPulsarReaderFactory<?> pulsarReaderFactory(PulsarClient pulsarClient,
ObjectProvider<ReaderBuilderCustomizer<?>> customizersProvider) {
ObjectProvider<ReaderBuilderCustomizer<?>> customizersProvider, PulsarTopicBuilder topicBuilder) {
List<ReaderBuilderCustomizer<?>> customizers = new ArrayList<>();
customizers.add(this.propertiesMapper::customizeReaderBuilder);
customizers.addAll(customizersProvider.orderedStream().toList());
List<ReaderBuilderCustomizer<Object>> lambdaSafeCustomizers = List
.of((builder) -> applyReaderBuilderCustomizers(customizers, builder));
return new DefaultPulsarReaderFactory<>(pulsarClient, lambdaSafeCustomizers);
DefaultPulsarReaderFactory<?> readerFactory = new DefaultPulsarReaderFactory<>(pulsarClient,
lambdaSafeCustomizers);
readerFactory.setTopicBuilder(topicBuilder);
return readerFactory;
}
@SuppressWarnings("unchecked")

12
spring-boot-project/spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/pulsar/PulsarConfiguration.java
View File

@@ -23,6 +23,7 @@ import org.apache.pulsar.client.admin.PulsarAdminBuilder;
import org.apache.pulsar.client.api.ClientBuilder;
import org.apache.pulsar.client.api.PulsarClient;
import org.apache.pulsar.client.api.Schema;
import org.apache.pulsar.common.naming.TopicDomain;
import org.apache.pulsar.common.schema.SchemaType;
import org.springframework.beans.factory.ObjectProvider;
@@ -34,6 +35,7 @@ import org.springframework.boot.context.properties.EnableConfigurationProperties
import org.springframework.boot.util.LambdaSafe;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Scope;
import org.springframework.pulsar.core.DefaultPulsarClientFactory;
import org.springframework.pulsar.core.DefaultSchemaResolver;
import org.springframework.pulsar.core.DefaultTopicResolver;
@@ -41,6 +43,7 @@ import org.springframework.pulsar.core.PulsarAdminBuilderCustomizer;
import org.springframework.pulsar.core.PulsarAdministration;
import org.springframework.pulsar.core.PulsarClientBuilderCustomizer;
import org.springframework.pulsar.core.PulsarClientFactory;
import org.springframework.pulsar.core.PulsarTopicBuilder;
import org.springframework.pulsar.core.SchemaResolver;
import org.springframework.pulsar.core.SchemaResolver.SchemaResolverCustomizer;
import org.springframework.pulsar.core.TopicResolver;
@@ -176,4 +179,13 @@ class PulsarConfiguration {
properties.isFailFast(), properties.isPropagateFailures(), properties.isPropagateStopFailures());
}
@Bean
@Scope("prototype")
@ConditionalOnMissingBean
@ConditionalOnProperty(name = "spring.pulsar.defaults.topic.enabled", havingValue = "true", matchIfMissing = true)
PulsarTopicBuilder pulsarTopicBuilder() {
return new PulsarTopicBuilder(TopicDomain.persistent, this.properties.getDefaults().getTopic().getTenant(),
this.properties.getDefaults().getTopic().getNamespace());
}
}

44
spring-boot-project/spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/pulsar/PulsarProperties.java
View File

@@ -257,6 +257,16 @@ public class PulsarProperties {
*/
private List<TypeMapping> typeMappings = new ArrayList<>();
private Topic topic = new Topic();
public Topic getTopic() {
return this.topic;
}
public void setTopic(Topic topic) {
this.topic = topic;
}
public List<TypeMapping> getTypeMappings() {
return this.typeMappings;
}
@@ -301,6 +311,40 @@ public class PulsarProperties {
}
public static class Topic {
/**
* Default tenant to use when producing or consuming messages against a
* non-fully-qualified topic URL. When not specified Pulsar uses a default
* tenant of 'public'.
*/
private String tenant;
/**
* Default namespace to use when producing or consuming messages against a
* non-fully-qualified topic URL. When not specified Pulsar uses a default
* namespace of 'default'.
*/
private String namespace;
public String getTenant() {
return this.tenant;
}
public void setTenant(String tenant) {
this.tenant = tenant;
}
public String getNamespace() {
return this.namespace;
}
public void setNamespace(String namespace) {
this.namespace = namespace;
}
}
}
public static class Function {

21
spring-boot-project/spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/pulsar/PulsarReactiveAutoConfiguration.java
View File

@@ -41,6 +41,7 @@ import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Import;
import org.springframework.pulsar.config.PulsarAnnotationSupportBeanNames;
import org.springframework.pulsar.core.PulsarTopicBuilder;
import org.springframework.pulsar.core.SchemaResolver;
import org.springframework.pulsar.core.TopicResolver;
import org.springframework.pulsar.reactive.config.DefaultReactivePulsarListenerContainerFactory;
@@ -112,7 +113,8 @@ public class PulsarReactiveAutoConfiguration {
@ConditionalOnMissingBean(ReactivePulsarSenderFactory.class)
DefaultReactivePulsarSenderFactory<?> reactivePulsarSenderFactory(ReactivePulsarClient reactivePulsarClient,
ObjectProvider<ReactiveMessageSenderCache> reactiveMessageSenderCache, TopicResolver topicResolver,
ObjectProvider<ReactiveMessageSenderBuilderCustomizer<?>> customizersProvider) {
ObjectProvider<ReactiveMessageSenderBuilderCustomizer<?>> customizersProvider,
PulsarTopicBuilder topicBuilder) {
List<ReactiveMessageSenderBuilderCustomizer<?>> customizers = new ArrayList<>();
customizers.add(this.propertiesMapper::customizeMessageSenderBuilder);
customizers.addAll(customizersProvider.orderedStream().toList());
@@ -122,6 +124,7 @@ public class PulsarReactiveAutoConfiguration {
.withDefaultConfigCustomizers(lambdaSafeCustomizers)
.withMessageSenderCache(reactiveMessageSenderCache.getIfAvailable())
.withTopicResolver(topicResolver)
.withTopicBuilder(topicBuilder)
.build();
}
@@ -136,13 +139,17 @@ public class PulsarReactiveAutoConfiguration {
@ConditionalOnMissingBean(ReactivePulsarConsumerFactory.class)
DefaultReactivePulsarConsumerFactory<?> reactivePulsarConsumerFactory(
ReactivePulsarClient pulsarReactivePulsarClient,
ObjectProvider<ReactiveMessageConsumerBuilderCustomizer<?>> customizersProvider) {
ObjectProvider<ReactiveMessageConsumerBuilderCustomizer<?>> customizersProvider,
PulsarTopicBuilder topicBuilder) {
List<ReactiveMessageConsumerBuilderCustomizer<?>> customizers = new ArrayList<>();
customizers.add(this.propertiesMapper::customizeMessageConsumerBuilder);
customizers.addAll(customizersProvider.orderedStream().toList());
List<ReactiveMessageConsumerBuilderCustomizer<Object>> lambdaSafeCustomizers = List
.of((builder) -> applyMessageConsumerBuilderCustomizers(customizers, builder));
return new DefaultReactivePulsarConsumerFactory<>(pulsarReactivePulsarClient, lambdaSafeCustomizers);
DefaultReactivePulsarConsumerFactory<?> consumerFactory = new DefaultReactivePulsarConsumerFactory<>(
pulsarReactivePulsarClient, lambdaSafeCustomizers);
consumerFactory.setTopicBuilder(topicBuilder);
return consumerFactory;
}
@SuppressWarnings("unchecked")
@@ -167,13 +174,17 @@ public class PulsarReactiveAutoConfiguration {
@Bean
@ConditionalOnMissingBean(ReactivePulsarReaderFactory.class)
DefaultReactivePulsarReaderFactory<?> reactivePulsarReaderFactory(ReactivePulsarClient reactivePulsarClient,
ObjectProvider<ReactiveMessageReaderBuilderCustomizer<?>> customizersProvider) {
ObjectProvider<ReactiveMessageReaderBuilderCustomizer<?>> customizersProvider,
PulsarTopicBuilder topicBuilder) {
List<ReactiveMessageReaderBuilderCustomizer<?>> customizers = new ArrayList<>();
customizers.add(this.propertiesMapper::customizeMessageReaderBuilder);
customizers.addAll(customizersProvider.orderedStream().toList());
List<ReactiveMessageReaderBuilderCustomizer<Object>> lambdaSafeCustomizers = List
.of((builder) -> applyMessageReaderBuilderCustomizers(customizers, builder));
return new DefaultReactivePulsarReaderFactory<>(reactivePulsarClient, lambdaSafeCustomizers);
DefaultReactivePulsarReaderFactory<?> readerFactory = new DefaultReactivePulsarReaderFactory<>(
reactivePulsarClient, lambdaSafeCustomizers);
readerFactory.setTopicBuilder(topicBuilder);
return readerFactory;
}
@SuppressWarnings("unchecked")

6
spring-boot-project/spring-boot-autoconfigure/src/main/resources/META-INF/additional-spring-configuration-metadata.json
View File

@@ -2068,6 +2068,12 @@
"name": "spring.neo4j.uri",
"defaultValue": "bolt://localhost:7687"
},
{
"name": "spring.pulsar.defaults.topic.enabled",
"type": "java.lang.Boolean",
"description": "Whether to enable default tenant and namespace support for topics.",
"defaultValue": true
},
{
"name": "spring.pulsar.function.enabled",
"type": "java.lang.Boolean",

12
spring-boot-project/spring-boot-autoconfigure/src/test/java/org/springframework/boot/autoconfigure/pulsar/PulsarAutoConfigurationTests.java
View File

@@ -219,7 +219,9 @@ class PulsarAutoConfigurationTests {
"spring.pulsar.producer.cache.enabled=false")
.run((context) -> assertThat(context).getBean(DefaultPulsarProducerFactory.class)
.hasFieldOrPropertyWithValue("pulsarClient", context.getBean(PulsarClient.class))
.hasFieldOrPropertyWithValue("topicResolver", context.getBean(TopicResolver.class)));
.hasFieldOrPropertyWithValue("topicResolver", context.getBean(TopicResolver.class))
.extracting("topicBuilder")
.isNotNull()); // prototype so only check not-null
}
@ParameterizedTest
@@ -375,7 +377,9 @@ class PulsarAutoConfigurationTests {
@Test
void injectsExpectedBeans() {
this.contextRunner.run((context) -> assertThat(context).getBean(DefaultPulsarConsumerFactory.class)
.hasFieldOrPropertyWithValue("pulsarClient", context.getBean(PulsarClient.class)));
.hasFieldOrPropertyWithValue("pulsarClient", context.getBean(PulsarClient.class))
.extracting("topicBuilder")
.isNotNull()); // prototype so only check not-null
}
@Test
@@ -574,7 +578,9 @@ class PulsarAutoConfigurationTests {
@Test
void injectsExpectedBeans() {
this.contextRunner.run((context) -> assertThat(context).getBean(DefaultPulsarReaderFactory.class)
.hasFieldOrPropertyWithValue("pulsarClient", context.getBean(PulsarClient.class)));
.hasFieldOrPropertyWithValue("pulsarClient", context.getBean(PulsarClient.class))
.extracting("topicBuilder")
.isNotNull()); // prototype so only check not-null
}
@Test

41
spring-boot-project/spring-boot-autoconfigure/src/test/java/org/springframework/boot/autoconfigure/pulsar/PulsarConfigurationTests.java
View File

@@ -49,6 +49,7 @@ import org.springframework.pulsar.core.PulsarAdminBuilderCustomizer;
import org.springframework.pulsar.core.PulsarAdministration;
import org.springframework.pulsar.core.PulsarClientBuilderCustomizer;
import org.springframework.pulsar.core.PulsarClientFactory;
import org.springframework.pulsar.core.PulsarTopicBuilder;
import org.springframework.pulsar.core.SchemaResolver;
import org.springframework.pulsar.core.SchemaResolver.SchemaResolverCustomizer;
import org.springframework.pulsar.core.TopicResolver;
@@ -320,6 +321,46 @@ class PulsarConfigurationTests {
}
@Nested
class TopicBuilderTests {
private final ApplicationContextRunner contextRunner = PulsarConfigurationTests.this.contextRunner;
@Test
void whenHasUserDefinedBeanDoesNotAutoConfigureBean() {
PulsarTopicBuilder topicBuilder = mock(PulsarTopicBuilder.class);
this.contextRunner.withBean("customPulsarTopicBuilder", PulsarTopicBuilder.class, () -> topicBuilder)
.run((context) -> assertThat(context).getBean(PulsarTopicBuilder.class).isSameAs(topicBuilder));
}
@Test
void whenHasDefaultsTopicDisabledPropertyDoesNotCreateBean() {
this.contextRunner.withPropertyValues("spring.pulsar.defaults.topic.enabled=false")
.run((context) -> assertThat(context).doesNotHaveBean(PulsarTopicBuilder.class));
}
@Test
void whenHasDefaultsTenantAndNamespaceAppliedToTopicBuilder() {
List<String> properties = new ArrayList<>();
properties.add("spring.pulsar.defaults.topic.tenant=my-tenant");
properties.add("spring.pulsar.defaults.topic.namespace=my-namespace");
this.contextRunner.withPropertyValues(properties.toArray(String[]::new))
.run((context) -> assertThat(context).getBean(PulsarTopicBuilder.class)
.asInstanceOf(InstanceOfAssertFactories.type(PulsarTopicBuilder.class))
.satisfies((topicBuilder) -> {
assertThat(topicBuilder).hasFieldOrPropertyWithValue("defaultTenant", "my-tenant");
assertThat(topicBuilder).hasFieldOrPropertyWithValue("defaultNamespace", "my-namespace");
}));
}
@Test
void beanHasScopePrototype() {
this.contextRunner.run((context) -> assertThat(context.getBean(PulsarTopicBuilder.class))
.isNotSameAs(context.getBean(PulsarTopicBuilder.class)));
}
}
@Nested
class FunctionAdministrationTests {

25
spring-boot-project/spring-boot-autoconfigure/src/test/java/org/springframework/boot/autoconfigure/pulsar/PulsarPropertiesTests.java
View File

@@ -152,7 +152,7 @@ class PulsarPropertiesTests {
}
@Nested
class DefaultsProperties {
class DefaultsTypeMappingProperties {
@Test
void bindWhenNoTypeMappings() {
@@ -242,6 +242,29 @@ class PulsarPropertiesTests {
}
@Nested
class DefaultsTenantNamespaceProperties {
@Test
void bindWhenValuesNotSpecified() {
assertThat(new PulsarProperties().getDefaults().getTopic()).satisfies((defaults) -> {
assertThat(defaults.getTenant()).isNull();
assertThat(defaults.getNamespace()).isNull();
});
}
@Test
void bindWhenValuesSpecified() {
Map<String, String> map = new HashMap<>();
map.put("spring.pulsar.defaults.topic.tenant", "my-tenant");
map.put("spring.pulsar.defaults.topic.namespace", "my-namespace");
PulsarProperties.Defaults.Topic properties = bindProperties(map).getDefaults().getTopic();
assertThat(properties.getTenant()).isEqualTo("my-tenant");
assertThat(properties.getNamespace()).isEqualTo("my-namespace");
}
}
@Nested
class FunctionProperties {

16
spring-boot-project/spring-boot-autoconfigure/src/test/java/org/springframework/boot/autoconfigure/pulsar/PulsarReactiveAutoConfigurationTests.java
View File

@@ -48,6 +48,7 @@ import org.springframework.core.annotation.Order;
import org.springframework.pulsar.core.DefaultSchemaResolver;
import org.springframework.pulsar.core.DefaultTopicResolver;
import org.springframework.pulsar.core.PulsarAdministration;
import org.springframework.pulsar.core.PulsarTopicBuilder;
import org.springframework.pulsar.core.SchemaResolver;
import org.springframework.pulsar.core.TopicResolver;
import org.springframework.pulsar.reactive.config.DefaultReactivePulsarListenerContainerFactory;
@@ -177,6 +178,11 @@ class PulsarReactiveAutoConfigurationTests {
assertThat(senderFactory)
.extracting("topicResolver", InstanceOfAssertFactories.type(TopicResolver.class))
.isSameAs(context.getBean(TopicResolver.class));
assertThat(senderFactory).extracting("topicBuilder").isNotNull(); // prototype
// so
// only
// check
// not-null
});
}
@@ -252,13 +258,18 @@ class PulsarReactiveAutoConfigurationTests {
@Test
void injectsExpectedBeans() {
ReactivePulsarClient client = mock(ReactivePulsarClient.class);
PulsarTopicBuilder topicBuilder = mock(PulsarTopicBuilder.class);
this.contextRunner.withBean("customReactivePulsarClient", ReactivePulsarClient.class, () -> client)
.withBean("customTopicBuilder", PulsarTopicBuilder.class, () -> topicBuilder)
.run((context) -> {
ReactivePulsarConsumerFactory<?> consumerFactory = context
.getBean(DefaultReactivePulsarConsumerFactory.class);
assertThat(consumerFactory)
.extracting("reactivePulsarClient", InstanceOfAssertFactories.type(ReactivePulsarClient.class))
.isSameAs(client);
assertThat(consumerFactory)
.extracting("topicBuilder", InstanceOfAssertFactories.type(PulsarTopicBuilder.class))
.isSameAs(topicBuilder);
});
}
@@ -362,14 +373,19 @@ class PulsarReactiveAutoConfigurationTests {
@Test
void injectsExpectedBeans() {
ReactivePulsarClient client = mock(ReactivePulsarClient.class);
PulsarTopicBuilder topicBuilder = mock(PulsarTopicBuilder.class);
this.contextRunner.withPropertyValues("spring.pulsar.reader.name=test-reader")
.withBean("customReactivePulsarClient", ReactivePulsarClient.class, () -> client)
.withBean("customPulsarTopicBuilder", PulsarTopicBuilder.class, () -> topicBuilder)
.run((context) -> {
DefaultReactivePulsarReaderFactory<?> readerFactory = context
.getBean(DefaultReactivePulsarReaderFactory.class);
assertThat(readerFactory)
.extracting("reactivePulsarClient", InstanceOfAssertFactories.type(ReactivePulsarClient.class))
.isSameAs(client);
assertThat(readerFactory)
.extracting("topicBuilder", InstanceOfAssertFactories.type(PulsarTopicBuilder.class))
.isSameAs(topicBuilder);
});
}