Add Pulsar container factory customizers

This commit adds the ability for users to customize the auto-configured
Spring for Apache Pulsar message container factories. Each container
factory holds a set of container properties that is a common target for
users to configure. Allowing the customization of these properties
prevents a rapid increase of configuration properties.

Author: onobc

spring-boot-project/spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/pulsar/ConcurrentPulsarListenerContainerFactoryCustomizer.java

@@ -0,0 +1,36 @@
+/*
+ * Copyright 2012-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.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.boot.autoconfigure.pulsar;
+
+import org.springframework.pulsar.config.ConcurrentPulsarListenerContainerFactory;
+
+/**
+ * Callback interface that can be implemented to customize a
+ * {@link ConcurrentPulsarListenerContainerFactory}.
+ *
+ * @author Chris Bono
+ */
+@FunctionalInterface
+public interface ConcurrentPulsarListenerContainerFactoryCustomizer {
+
+	/**
+	 * Customize a {@link ConcurrentPulsarListenerContainerFactory}.
+	 * @param containerFactory the factory to customize
+	 */
+	void customize(ConcurrentPulsarListenerContainerFactory<?> containerFactory);
+
+}

spring-boot-project/spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/pulsar/DefaultPulsarReaderContainerFactoryCustomizer.java

@@ -0,0 +1,36 @@
+/*
+ * Copyright 2012-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.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.boot.autoconfigure.pulsar;
+
+import org.springframework.pulsar.config.DefaultPulsarReaderContainerFactory;
+
+/**
+ * Callback interface that can be implemented to customize a
+ * {@link DefaultPulsarReaderContainerFactory}.
+ *
+ * @author Chris Bono
+ */
+@FunctionalInterface
+public interface DefaultPulsarReaderContainerFactoryCustomizer {
+
+	/**
+	 * Customize a {@link DefaultPulsarReaderContainerFactory}.
+	 * @param containerFactory the factory to customize
+	 */
+	void customize(DefaultPulsarReaderContainerFactory<?> containerFactory);
+
+}

spring-boot-project/spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/pulsar/DefaultReactivePulsarListenerContainerFactoryCustomizer.java

@@ -0,0 +1,36 @@
+/*
+ * Copyright 2012-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.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.boot.autoconfigure.pulsar;
+
+import org.springframework.pulsar.reactive.config.DefaultReactivePulsarListenerContainerFactory;
+
+/**
+ * Callback interface that can be implemented to customize a
+ * {@link DefaultReactivePulsarListenerContainerFactory}.
+ *
+ * @author Chris Bono
+ */
+@FunctionalInterface
+public interface DefaultReactivePulsarListenerContainerFactoryCustomizer {
+
+	/**
+	 * Customize a {@link DefaultReactivePulsarListenerContainerFactory}.
+	 * @param containerFactory the factory to customize
+	 */
+	void customize(DefaultReactivePulsarListenerContainerFactory<?> containerFactory);
+
+}

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

@@ -178,6 +178,7 @@ private void applyConsumerBuilderCustomizers(List<ConsumerBuilderCustomizer<?>>
 	ConcurrentPulsarListenerContainerFactory<?> pulsarListenerContainerFactory(
 			PulsarConsumerFactory<Object> pulsarConsumerFactory, SchemaResolver schemaResolver,
 			TopicResolver topicResolver, ObjectProvider<PulsarAwareTransactionManager> pulsarTransactionManager,
+			ObjectProvider<ConcurrentPulsarListenerContainerFactoryCustomizer> customizersProvider,
 			Environment environment) {
 		PulsarContainerProperties containerProperties = new PulsarContainerProperties();
 		containerProperties.setSchemaResolver(schemaResolver);
@@ -187,7 +188,10 @@ ConcurrentPulsarListenerContainerFactory<?> pulsarListenerContainerFactory(
 		}
 		pulsarTransactionManager.ifUnique(containerProperties.transactions()::setTransactionManager);
 		this.propertiesMapper.customizeContainerProperties(containerProperties);
-		return new ConcurrentPulsarListenerContainerFactory<>(pulsarConsumerFactory, containerProperties);
+		ConcurrentPulsarListenerContainerFactory<?> containerFactory = new ConcurrentPulsarListenerContainerFactory<>(
+				pulsarConsumerFactory, containerProperties);
+		customizersProvider.orderedStream().forEachOrdered((customizer) -> customizer.customize(containerFactory));
+		return containerFactory;
 	}
 
 	@Bean
@@ -215,14 +219,19 @@ private void applyReaderBuilderCustomizers(List<ReaderBuilderCustomizer<?>> cust
 	@Bean
 	@ConditionalOnMissingBean(name = "pulsarReaderContainerFactory")
 	DefaultPulsarReaderContainerFactory<?> pulsarReaderContainerFactory(PulsarReaderFactory<?> pulsarReaderFactory,
-			SchemaResolver schemaResolver, Environment environment) {
+			SchemaResolver schemaResolver,
+			ObjectProvider<DefaultPulsarReaderContainerFactoryCustomizer> customizersProvider,
+			Environment environment) {
 		PulsarReaderContainerProperties readerContainerProperties = new PulsarReaderContainerProperties();
 		readerContainerProperties.setSchemaResolver(schemaResolver);
 		if (Threading.VIRTUAL.isActive(environment)) {
 			readerContainerProperties.setReaderTaskExecutor(new VirtualThreadTaskExecutor("pulsar-reader-"));
 		}
 		this.propertiesMapper.customizeReaderContainerProperties(readerContainerProperties);
-		return new DefaultPulsarReaderContainerFactory<>(pulsarReaderFactory, readerContainerProperties);
+		DefaultPulsarReaderContainerFactory<?> containerFactory = new DefaultPulsarReaderContainerFactory<>(
+				pulsarReaderFactory, readerContainerProperties);
+		customizersProvider.orderedStream().forEachOrdered((customizer) -> customizer.customize(containerFactory));
+		return containerFactory;
 	}
 
 	@Configuration(proxyBeanMethods = false)

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

@@ -164,12 +164,16 @@ private void applyMessageConsumerBuilderCustomizers(List<ReactiveMessageConsumer
 	@ConditionalOnMissingBean(name = "reactivePulsarListenerContainerFactory")
 	DefaultReactivePulsarListenerContainerFactory<?> reactivePulsarListenerContainerFactory(
 			ReactivePulsarConsumerFactory<Object> reactivePulsarConsumerFactory, SchemaResolver schemaResolver,
-			TopicResolver topicResolver) {
+			TopicResolver topicResolver,
+			ObjectProvider<DefaultReactivePulsarListenerContainerFactoryCustomizer> customizersProvider) {
 		ReactivePulsarContainerProperties<Object> containerProperties = new ReactivePulsarContainerProperties<>();
 		containerProperties.setSchemaResolver(schemaResolver);
 		containerProperties.setTopicResolver(topicResolver);
 		this.propertiesMapper.customizeContainerProperties(containerProperties);
-		return new DefaultReactivePulsarListenerContainerFactory<>(reactivePulsarConsumerFactory, containerProperties);
+		DefaultReactivePulsarListenerContainerFactory<?> containerFactory = new DefaultReactivePulsarListenerContainerFactory<>(
+				reactivePulsarConsumerFactory, containerProperties);
+		customizersProvider.orderedStream().forEachOrdered((customizer) -> customizer.customize(containerFactory));
+		return containerFactory;
 	}
 
 	@Bean

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

@@ -18,6 +18,7 @@
 
 import java.util.ArrayList;
 import java.util.List;
+import java.util.Objects;
 import java.util.concurrent.ThreadFactory;
 import java.util.concurrent.TimeUnit;
 
@@ -585,6 +586,36 @@ void whenTransactionEnabledFalseListenerContainerShouldNotUseTransactions() {
 			});
 		}
 
+		@Test
+		void whenHasUserDefinedCustomizersAppliesInCorrectOrder() {
+			this.contextRunner.withUserConfiguration(ListenerContainerFactoryCustomizersConfig.class)
+				.run((context) -> assertThat(context).getBean(ConcurrentPulsarListenerContainerFactory.class)
+					.hasFieldOrPropertyWithValue("containerProperties.subscriptionName", ":bar:foo"));
+		}
+
+		@TestConfiguration(proxyBeanMethods = false)
+		static class ListenerContainerFactoryCustomizersConfig {
+
+			@Bean
+			@Order(200)
+			ConcurrentPulsarListenerContainerFactoryCustomizer customizerFoo() {
+				return (containerFactory) -> appendToSubscriptionName(containerFactory, ":foo");
+			}
+
+			@Bean
+			@Order(100)
+			ConcurrentPulsarListenerContainerFactoryCustomizer customizerBar() {
+				return (containerFactory) -> appendToSubscriptionName(containerFactory, ":bar");
+			}
+
+			private void appendToSubscriptionName(ConcurrentPulsarListenerContainerFactory<?> containerFactory,
+					String valueToAppend) {
+				String name = Objects.toString(containerFactory.getContainerProperties().getSubscriptionName(), "");
+				containerFactory.getContainerProperties().setSubscriptionName(name.concat(valueToAppend));
+			}
+
+		}
+
 	}
 
 	@Nested
@@ -617,7 +648,7 @@ void hasNoTopicBuilderWhenTopicDefaultsAreDisabled() {
 		}
 
 		@Test
-		<T> void whenHasUserDefinedCustomizersAppliesInCorrectOrder() {
+		<T> void whenHasUserDefinedReaderBuilderCustomizersAppliesInCorrectOrder() {
 			this.contextRunner.withPropertyValues("spring.pulsar.reader.name=fromPropsCustomizer")
 				.withUserConfiguration(ReaderBuilderCustomizersConfig.class)
 				.run((context) -> {
@@ -654,6 +685,13 @@ void whenVirtualThreadsAreEnabledOnJava20AndEarlierReaderShouldNotUseVirtualThre
 			});
 		}
 
+		@Test
+		void whenHasUserDefinedFactoryCustomizersAppliesInCorrectOrder() {
+			this.contextRunner.withUserConfiguration(ReaderContainerFactoryCustomizersConfig.class)
+				.run((context) -> assertThat(context).getBean(DefaultPulsarReaderContainerFactory.class)
+					.hasFieldOrPropertyWithValue("containerProperties.readerListener", ":bar:foo"));
+		}
+
 		@TestConfiguration(proxyBeanMethods = false)
 		static class ReaderBuilderCustomizersConfig {
 
@@ -671,6 +709,29 @@ ReaderBuilderCustomizer<?> customizerBar() {
 
 		}
 
+		@TestConfiguration(proxyBeanMethods = false)
+		static class ReaderContainerFactoryCustomizersConfig {
+
+			@Bean
+			@Order(200)
+			DefaultPulsarReaderContainerFactoryCustomizer customizerFoo() {
+				return (containerFactory) -> appendToReaderListener(containerFactory, ":foo");
+			}
+
+			@Bean
+			@Order(100)
+			DefaultPulsarReaderContainerFactoryCustomizer customizerBar() {
+				return (containerFactory) -> appendToReaderListener(containerFactory, ":bar");
+			}
+
+			private void appendToReaderListener(DefaultPulsarReaderContainerFactory<?> containerFactory,
+					String valueToAppend) {
+				String name = Objects.toString(containerFactory.getContainerProperties().getReaderListener(), "");
+				containerFactory.getContainerProperties().setReaderListener(name.concat(valueToAppend));
+			}
+
+		}
+
 	}
 
 	@Nested

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

@@ -19,6 +19,7 @@
 import java.time.Duration;
 import java.util.ArrayList;
 import java.util.List;
+import java.util.Objects;
 import java.util.function.Supplier;
 
 import com.github.benmanes.caffeine.cache.Caffeine;
@@ -382,6 +383,36 @@ void injectsExpectedBeans() {
 				});
 		}
 
+		@Test
+		void whenHasUserDefinedFactoryCustomizersAppliesInCorrectOrder() {
+			this.contextRunner.withUserConfiguration(ListenerContainerFactoryCustomizersConfig.class)
+				.run((context) -> assertThat(context).getBean(DefaultReactivePulsarListenerContainerFactory.class)
+					.hasFieldOrPropertyWithValue("containerProperties.subscriptionName", ":bar:foo"));
+		}
+
+		@TestConfiguration(proxyBeanMethods = false)
+		static class ListenerContainerFactoryCustomizersConfig {
+
+			@Bean
+			@Order(200)
+			DefaultReactivePulsarListenerContainerFactoryCustomizer customizerFoo() {
+				return (containerFactory) -> appendToSubscriptionName(containerFactory, ":foo");
+			}
+
+			@Bean
+			@Order(100)
+			DefaultReactivePulsarListenerContainerFactoryCustomizer customizerBar() {
+				return (containerFactory) -> appendToSubscriptionName(containerFactory, ":bar");
+			}
+
+			private void appendToSubscriptionName(DefaultReactivePulsarListenerContainerFactory<?> containerFactory,
+					String valueToAppend) {
+				String name = Objects.toString(containerFactory.getContainerProperties().getSubscriptionName(), "");
+				containerFactory.getContainerProperties().setSubscriptionName(name.concat(valueToAppend));
+			}
+
+		}
+
 	}
 
 	@Nested

spring-boot-project/spring-boot-docs/src/docs/antora/modules/reference/pages/messaging/pulsar.adoc

@@ -150,11 +150,11 @@ include-code::MyBean[]
 Spring Boot auto-configuration provides all the components necessary for `PulsarListener`, such as the `PulsarListenerContainerFactory` and the consumer factory it uses to construct the underlying Pulsar consumers.
 You can configure these components by specifying any of the `spring.pulsar.listener.\*` and `spring.pulsar.consumer.*` prefixed application properties.
 
-If you need more control over the consumer factory configuration, consider registering one or more `ConsumerBuilderCustomizer` beans.
+If you need more control over the configuration of the consumer factory used by the container factory to create consumers, consider registering one or more `ConsumerBuilderCustomizer` beans.
 These customizers are applied to all consumers created by the factory, and therefore all `@PulsarListener` instances.
 You can also customize a single listener by setting the `consumerCustomizer` attribute of the `@PulsarListener` annotation.
 
-
+If you need more control over the actual container factory configuration, consider registering one or more `ConcurrentPulsarListenerContainerFactoryCustomizer` beans.
 
 [[messaging.pulsar.receiving-reactive]]
 == Receiving a Message Reactively
@@ -165,13 +165,13 @@ The following component creates a reactive listener endpoint on the `someTopic`
 include-code::MyBean[]
 
 Spring Boot auto-configuration provides all the components necessary for `ReactivePulsarListener`, such as the `ReactivePulsarListenerContainerFactory` and the consumer factory it uses to construct the underlying reactive Pulsar consumers.
-You can configure these components by specifying any of the `spring.pulsar.listener.*` and `spring.pulsar.consumer.*` prefixed application properties.
+You can configure these components by specifying any of the `spring.pulsar.listener.\*` and `spring.pulsar.consumer.*` prefixed application properties.
 
-If you need more control over the consumer factory configuration, consider registering one or more `ReactiveMessageConsumerBuilderCustomizer` beans.
+If you need more control over the configuration of the consumer factory used by the container factory to create consumers, consider registering one or more `ReactiveMessageConsumerBuilderCustomizer` beans.
 These customizers are applied to all consumers created by the factory, and therefore all `@ReactivePulsarListener` instances.
 You can also customize a single listener by setting the `consumerCustomizer` attribute of the `@ReactivePulsarListener` annotation.
 
-
+If you need more control over the actual container factory configuration, consider registering one or more `DefaultReactivePulsarListenerContainerFactoryCustomizer` beans.
 
 [[messaging.pulsar.reading]]
 == Reading a Message
@@ -187,10 +187,11 @@ include-code::MyBean[]
 The `@PulsarReader` relies on a `PulsarReaderFactory` to create the underlying Pulsar reader.
 Spring Boot auto-configuration provides this reader factory which can be customized by setting any of the `spring.pulsar.reader.*` prefixed application properties.
 
-If you need more control over the reader factory configuration, consider registering one or more `ReaderBuilderCustomizer` beans.
+If you need more control over the configuration of the reader factory used by the container factory to create readers, consider registering one or more `ReaderBuilderCustomizer` beans.
 These customizers are applied to all readers created by the factory, and therefore all `@PulsarReader` instances.
 You can also customize a single listener by setting the `readerCustomizer` attribute of the `@PulsarReader` annotation.
 
+If you need more control over the actual container factory configuration, consider registering one or more `DefaultPulsarReaderContainerFactoryCustomizer` beans.
 
 
 [[messaging.pulsar.reading-reactive]]