Add Pulsar container factory customizers

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

@@ -178,7 +178,7 @@ private void applyConsumerBuilderCustomizers(List<ConsumerBuilderCustomizer<?>>
 	ConcurrentPulsarListenerContainerFactory<?> pulsarListenerContainerFactory(
 			PulsarConsumerFactory<Object> pulsarConsumerFactory, SchemaResolver schemaResolver,
 			TopicResolver topicResolver, ObjectProvider<PulsarAwareTransactionManager> pulsarTransactionManager,
-			Environment environment) {
+			PulsarContainerFactoryCustomizers containerFactoryCustomizers, Environment environment) {
 		PulsarContainerProperties containerProperties = new PulsarContainerProperties();
 		containerProperties.setSchemaResolver(schemaResolver);
 		containerProperties.setTopicResolver(topicResolver);
@@ -187,7 +187,10 @@ ConcurrentPulsarListenerContainerFactory<?> pulsarListenerContainerFactory(
 		}
 		pulsarTransactionManager.ifUnique(containerProperties.transactions()::setTransactionManager);
 		this.propertiesMapper.customizeContainerProperties(containerProperties);
-		return new ConcurrentPulsarListenerContainerFactory<>(pulsarConsumerFactory, containerProperties);
+		ConcurrentPulsarListenerContainerFactory<?> containerFactory = new ConcurrentPulsarListenerContainerFactory<>(
+				pulsarConsumerFactory, containerProperties);
+		containerFactoryCustomizers.customize(containerFactory);
+		return containerFactory;
 	}
 
 	@Bean
@@ -215,14 +218,18 @@ private void applyReaderBuilderCustomizers(List<ReaderBuilderCustomizer<?>> cust
 	@Bean
 	@ConditionalOnMissingBean(name = "pulsarReaderContainerFactory")
 	DefaultPulsarReaderContainerFactory<?> pulsarReaderContainerFactory(PulsarReaderFactory<?> pulsarReaderFactory,
-			SchemaResolver schemaResolver, Environment environment) {
+			SchemaResolver schemaResolver, PulsarContainerFactoryCustomizers containerFactoryCustomizers,
+			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);
+		containerFactoryCustomizers.customize(containerFactory);
+		return containerFactory;
 	}
 
 	@Configuration(proxyBeanMethods = false)

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

@@ -188,4 +188,11 @@ PulsarTopicBuilder pulsarTopicBuilder() {
 				this.properties.getDefaults().getTopic().getNamespace());
 	}
 
+	@Bean
+	@ConditionalOnMissingBean
+	PulsarContainerFactoryCustomizers pulsarContainerFactoryCustomizers(
+			ObjectProvider<PulsarContainerFactoryCustomizer<?>> customizers) {
+		return new PulsarContainerFactoryCustomizers(customizers.orderedStream().toList());
+	}
+
 }

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

@@ -0,0 +1,39 @@
+/*
+ * 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.PulsarContainerFactory;
+
+/**
+ * Callback interface that can be implemented by beans wishing to customize a
+ * {@link PulsarContainerFactory} before it is fully initialized, in particular to tune
+ * its configuration.
+ *
+ * @param <T> the type of the {@link PulsarContainerFactory}
+ * @author Chris Bono
+ * @since 3.4.0
+ */
+@FunctionalInterface
+public interface PulsarContainerFactoryCustomizer<T extends PulsarContainerFactory<?, ?>> {
+
+	/**
+	 * Customize the container factory.
+	 * @param containerFactory the {@code PulsarContainerFactory} to customize
+	 */
+	void customize(T containerFactory);
+
+}

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

@@ -0,0 +1,58 @@
+/*
+ * 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 java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+
+import org.springframework.boot.util.LambdaSafe;
+import org.springframework.pulsar.config.PulsarContainerFactory;
+import org.springframework.pulsar.core.PulsarConsumerFactory;
+
+/**
+ * Invokes the available {@link PulsarContainerFactoryCustomizer} instances in the context
+ * for a given {@link PulsarConsumerFactory}.
+ *
+ * @author Chris Bono
+ * @since 3.4.0
+ */
+public class PulsarContainerFactoryCustomizers {
+
+	private final List<PulsarContainerFactoryCustomizer<?>> customizers;
+
+	public PulsarContainerFactoryCustomizers(List<? extends PulsarContainerFactoryCustomizer<?>> customizers) {
+		this.customizers = (customizers != null) ? new ArrayList<>(customizers) : Collections.emptyList();
+	}
+
+	/**
+	 * Customize the specified {@link PulsarContainerFactory}. Locates all
+	 * {@link PulsarContainerFactoryCustomizer} beans able to handle the specified
+	 * instance and invoke {@link PulsarContainerFactoryCustomizer#customize} on them.
+	 * @param <T> the type of container factory
+	 * @param containerFactory the container factory to customize
+	 * @return the customized container factory
+	 */
+	@SuppressWarnings("unchecked")
+	public <T extends PulsarContainerFactory<?, ?>> T customize(T containerFactory) {
+		LambdaSafe.callbacks(PulsarContainerFactoryCustomizer.class, this.customizers, containerFactory)
+			.withLogger(PulsarContainerFactoryCustomizers.class)
+			.invoke((customizer) -> customizer.customize(containerFactory));
+		return containerFactory;
+	}
+
+}

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

@@ -164,12 +164,15 @@ private void applyMessageConsumerBuilderCustomizers(List<ReactiveMessageConsumer
 	@ConditionalOnMissingBean(name = "reactivePulsarListenerContainerFactory")
 	DefaultReactivePulsarListenerContainerFactory<?> reactivePulsarListenerContainerFactory(
 			ReactivePulsarConsumerFactory<Object> reactivePulsarConsumerFactory, SchemaResolver schemaResolver,
-			TopicResolver topicResolver) {
+			TopicResolver topicResolver, PulsarContainerFactoryCustomizers containerFactoryCustomizers) {
 		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);
+		containerFactoryCustomizers.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;
 
@@ -72,6 +73,7 @@
 import org.springframework.pulsar.core.SchemaResolver;
 import org.springframework.pulsar.core.TopicResolver;
 import org.springframework.pulsar.listener.PulsarContainerProperties.TransactionSettings;
+import org.springframework.pulsar.reactive.config.DefaultReactivePulsarListenerContainerFactory;
 import org.springframework.pulsar.transaction.PulsarAwareTransactionManager;
 import org.springframework.test.util.ReflectionTestUtils;
 
@@ -585,6 +587,44 @@ 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(50)
+			PulsarContainerFactoryCustomizer<DefaultReactivePulsarListenerContainerFactory<?>> customizerIgnored() {
+				return (__) -> {
+					throw new RuntimeException("should-not-have-matched");
+				};
+			}
+
+			@Bean
+			@Order(200)
+			PulsarContainerFactoryCustomizer<ConcurrentPulsarListenerContainerFactory<?>> customizerFoo() {
+				return (containerFactory) -> appendToSubscriptionName(containerFactory, ":foo");
+			}
+
+			@Bean
+			@Order(100)
+			PulsarContainerFactoryCustomizer<ConcurrentPulsarListenerContainerFactory<?>> 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 +657,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 +694,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 +718,37 @@ ReaderBuilderCustomizer<?> customizerBar() {
 
 		}
 
+		@TestConfiguration(proxyBeanMethods = false)
+		static class ReaderContainerFactoryCustomizersConfig {
+
+			@Bean
+			@Order(50)
+			PulsarContainerFactoryCustomizer<DefaultReactivePulsarListenerContainerFactory<?>> customizerIgnored() {
+				return (__) -> {
+					throw new RuntimeException("should-not-have-matched");
+				};
+			}
+
+			@Bean
+			@Order(200)
+			PulsarContainerFactoryCustomizer<DefaultPulsarReaderContainerFactory<?>> customizerFoo() {
+				return (containerFactory) -> appendToReaderListener(containerFactory, ":foo");
+			}
+
+			@Bean
+			@Order(100)
+			PulsarContainerFactoryCustomizer<DefaultPulsarReaderContainerFactory<?>> 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/PulsarConfigurationTests.java

@@ -86,6 +86,15 @@ void whenHasUserDefinedConnectionDetailsBeanDoesNotAutoConfigureBean() {
 				.isSameAs(customConnectionDetails));
 	}
 
+	@Test
+	void whenHasUserDefinedContainerFactoryCustomizersBeanDoesNotAutoConfigureBean() {
+		PulsarContainerFactoryCustomizers customizers = mock(PulsarContainerFactoryCustomizers.class);
+		this.contextRunner
+			.withBean("customContainerFactoryCustomizers", PulsarContainerFactoryCustomizers.class, () -> customizers)
+			.run((context) -> assertThat(context).getBean(PulsarContainerFactoryCustomizers.class)
+				.isSameAs(customizers));
+	}
+
 	@Nested
 	class ClientTests {