Automatically set customizer on listeners (#495)

When there is only a single customizer and a single listener defined in
the application then the customizer will automatically be associated
with the listener. This works for @PulsarListener, @PulsarReader, and
@ReactivePulsarListener.

See #480

spring-pulsar-docs/src/main/antora/modules/ROOT/pages/reference/pulsar.adoc

@@ -299,6 +299,26 @@ void listen(String message) {
 
 TIP: The properties used are direct Pulsar consumer properties, not the `spring.pulsar.consumer` application configuration properties
 
+==== Customizing the ConsumerBuilder
+
+You can customize any fields available through `ConsumerBuilder` using a `PulsarListenerConsumerBuilderCustomizer` by providing a `@Bean` of type `PulsarListenerConsumerBuilderCustomizer` and then making it available to the `PulsarListener` as shown below.
+
+[source, java]
+----
+@PulsarListener(topics = "hello-topic", consumerCustomizer = "myCustomizer")
+public void listen(String message) {
+    System.out.println("Message Received: " + message);
+}
+
+@Bean
+PulsarListenerConsumerBuilderCustomizer<String> myCustomizer() {
+    return (builder) -> builder.consumerName("myConsumer");
+}
+----
+
+TIP: If your application only has a single `@PulsarListener` and a single `PulsarListenerConsumerBuilderCustomizer` bean registered then the customizer will be automatically applied.
+
+
 [[schema-info-listener-imperative]]
 :listener-class: PulsarListener
 include::schema-info/schema-info-listener.adoc[leveloffset=+1]
@@ -1112,7 +1132,7 @@ Suppose you want the reader to start reading messages arbitrarily from a topic o
 ==== Customizing the ReaderBuilder
 
 You can customize any fields available through `ReaderBuilder` using a `PulsarReaderReaderBuilderCustomizer` in Spring for Apache Pulsar.
-You can provide a `@Bean` of type `PulsarReaderBuilderCustomizer` and then make it available to the `PulsarReader` as below.
+You can provide a `@Bean` of type `PulsarReaderReaderBuilderCustomizer` and then make it available to the `PulsarReader` as below.
 
 [source, java]
 ----
@@ -1131,6 +1151,8 @@ public PulsarReaderReaderBuilderCustomizer<String> myCustomizer() {
 }
 ----
 
+TIP: If your application only has a single `@PulsarReader` and a single `PulsarReaderReaderBuilderCustomizer` bean registered then the customizer will be automatically applied.
+
 [[topic-resolution-process-imperative]]
 == Topic Resolution
 include::topic-resolution.adoc[leveloffset=+1]

spring-pulsar-docs/src/main/antora/modules/ROOT/pages/reference/reactive-pulsar.adoc

@@ -270,6 +270,8 @@ ReactivePulsarListenerMessageConsumerBuilderCustomizer<String> myConsumerCustomi
 }
 ----
 
+TIP: If your application only has a single `@ReactivePulsarListener` and a single `ReactivePulsarListenerMessageConsumerBuilderCustomizer` bean registered then the customizer will be automatically applied.
+
 You can also use the customizer to provide direct Pulsar consumer properties to the consumer builder.
 This is convenient if you do not want to use the Boot configuration properties mentioned earlier or have multiple `ReactivePulsarListener` methods whose configuration varies.
 

spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/config/MethodReactivePulsarListenerEndpoint.java

@@ -226,6 +226,10 @@ public class MethodReactivePulsarListenerEndpoint<V> extends AbstractReactivePul
 		this.deadLetterPolicy = deadLetterPolicy;
 	}
 
+	public ReactiveMessageConsumerBuilderCustomizer<V> getConsumerCustomizer() {
+		return this.consumerCustomizer;
+	}
+
 	public void setConsumerCustomizer(ReactiveMessageConsumerBuilderCustomizer<V> consumerCustomizer) {
 		this.consumerCustomizer = consumerCustomizer;
 	}

spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/config/annotation/ReactivePulsarListenerAnnotationBeanPostProcessor.java

@@ -110,13 +110,13 @@ public class ReactivePulsarListenerAnnotationBeanPostProcessor<V> extends Abstra
 
 	private final AtomicInteger counter = new AtomicInteger();
 
+	private final List<MethodReactivePulsarListenerEndpoint<?>> processedEndpoints = new ArrayList<>();
+
 	@Override
 	public void afterSingletonsInstantiated() {
 		this.registrar.setBeanFactory(this.beanFactory);
-
 		this.beanFactory.getBeanProvider(PulsarListenerConfigurer.class)
 			.forEach(c -> c.configurePulsarListeners(this.registrar));
-
 		if (this.registrar.getEndpointRegistry() == null) {
 			if (this.endpointRegistry == null) {
 				Assert.state(this.beanFactory != null,
@@ -127,12 +127,11 @@ public class ReactivePulsarListenerAnnotationBeanPostProcessor<V> extends Abstra
 			}
 			this.registrar.setEndpointRegistry(this.endpointRegistry);
 		}
-
 		if (this.defaultContainerFactoryBeanName != null) {
 			this.registrar.setContainerFactoryBeanName(this.defaultContainerFactoryBeanName);
 		}
 		addFormatters(this.messageHandlerMethodFactory.getDefaultFormattingConversionService());
-
+		postProcessEndpointsBeforeRegistration();
 		// Actually register all listeners
 		this.registrar.afterPropertiesSet();
 	}
@@ -201,14 +200,11 @@ public class ReactivePulsarListenerAnnotationBeanPostProcessor<V> extends Abstra
 	@Nullable
 	private ReactivePulsarListenerContainerFactory<?> resolveContainerFactory(
 			ReactivePulsarListener ReactivePulsarListener, Object factoryTarget, String beanName) {
-
 		String containerFactory = ReactivePulsarListener.containerFactory();
 		if (!StringUtils.hasText(containerFactory)) {
 			return null;
 		}
-
 		ReactivePulsarListenerContainerFactory<?> factory = null;
-
 		Object resolved = resolveExpression(containerFactory);
 		if (resolved instanceof ReactivePulsarListenerContainerFactory) {
 			return (ReactivePulsarListenerContainerFactory<?>) resolved;
@@ -230,7 +226,6 @@ public class ReactivePulsarListenerAnnotationBeanPostProcessor<V> extends Abstra
 
 	private void processReactivePulsarListenerAnnotation(MethodReactivePulsarListenerEndpoint<?> endpoint,
 			ReactivePulsarListener reactivePulsarListener, Object bean, String[] topics, String topicPattern) {
-
 		endpoint.setBean(bean);
 		endpoint.setMessageHandlerMethodFactory(this.messageHandlerMethodFactory);
 		endpoint.setSubscriptionName(getEndpointSubscriptionName(reactivePulsarListener));
@@ -239,7 +234,6 @@ public class ReactivePulsarListenerAnnotationBeanPostProcessor<V> extends Abstra
 		endpoint.setTopicPattern(topicPattern);
 		resolveSubscriptionType(endpoint, reactivePulsarListener);
 		endpoint.setSchemaType(reactivePulsarListener.schemaType());
-
 		String concurrency = reactivePulsarListener.concurrency();
 		if (StringUtils.hasText(concurrency)) {
 			endpoint.setConcurrency(resolveExpressionAsInteger(concurrency, "concurrency"));
@@ -249,16 +243,15 @@ public class ReactivePulsarListenerAnnotationBeanPostProcessor<V> extends Abstra
 			endpoint.setUseKeyOrderedProcessing(
 					resolveExpressionAsBoolean(useKeyOrderedProcessing, "useKeyOrderedProcessing"));
 		}
-
 		String autoStartup = reactivePulsarListener.autoStartup();
 		if (StringUtils.hasText(autoStartup)) {
 			endpoint.setAutoStartup(resolveExpressionAsBoolean(autoStartup, "autoStartup"));
 		}
 		endpoint.setFluxListener(reactivePulsarListener.stream());
 		endpoint.setBeanFactory(this.beanFactory);
-
 		resolveDeadLetterPolicy(endpoint, reactivePulsarListener);
 		resolveConsumerCustomizer(endpoint, reactivePulsarListener);
+		this.processedEndpoints.add(endpoint);
 	}
 
 	private void resolveSubscriptionType(MethodReactivePulsarListenerEndpoint<?> endpoint,
@@ -286,9 +279,29 @@ public class ReactivePulsarListenerAnnotationBeanPostProcessor<V> extends Abstra
 		}
 	}
 
+	@SuppressWarnings("unchecked")
+	protected void postProcessEndpointsBeforeRegistration() {
+		if (this.processedEndpoints.size() == 1) {
+			MethodReactivePulsarListenerEndpoint<?> endpoint = this.processedEndpoints.get(0);
+			if (endpoint.getConsumerCustomizer() != null) {
+				return;
+			}
+			this.beanFactory.getBeanProvider(ReactivePulsarListenerMessageConsumerBuilderCustomizer.class)
+				.ifUnique((customizer) -> {
+					this.logger.info(() -> String
+						.format("Setting the only registered ReactivePulsarListenerMessageConsumerBuilderCustomizer "
+								+ "on the only registered @ReactivePulsarListener (%s)", endpoint.getId()));
+					endpoint.setConsumerCustomizer(customizer::customize);
+				});
+		}
+	}
+
 	@SuppressWarnings({ "rawtypes", "unchecked" })
 	private void resolveConsumerCustomizer(MethodReactivePulsarListenerEndpoint<?> endpoint,
 			ReactivePulsarListener reactivePulsarListener) {
+		if (!StringUtils.hasText(reactivePulsarListener.consumerCustomizer())) {
+			return;
+		}
 		Object consumerCustomizer = resolveExpression(reactivePulsarListener.consumerCustomizer());
 		if (consumerCustomizer instanceof ReactivePulsarListenerMessageConsumerBuilderCustomizer customizer) {
 			endpoint.setConsumerCustomizer(customizer::customize);

spring-pulsar-reactive/src/test/java/org/springframework/pulsar/reactive/listener/ReactivePulsarListenerCustomizerTests.java

@@ -0,0 +1,238 @@
+/*
+ * Copyright 2022-2023 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.pulsar.reactive.listener;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.verify;
+
+import java.util.Collections;
+
+import org.apache.pulsar.client.api.PulsarClient;
+import org.apache.pulsar.client.api.PulsarClientException;
+import org.apache.pulsar.reactive.client.adapter.AdaptedReactivePulsarClientFactory;
+import org.apache.pulsar.reactive.client.api.ReactiveMessageConsumerBuilder;
+import org.apache.pulsar.reactive.client.api.ReactivePulsarClient;
+import org.assertj.core.api.InstanceOfAssertFactories;
+import org.assertj.core.api.ObjectAssert;
+import org.junit.jupiter.api.Nested;
+import org.junit.jupiter.api.Test;
+
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.pulsar.core.DefaultPulsarClientFactory;
+import org.springframework.pulsar.core.DefaultPulsarProducerFactory;
+import org.springframework.pulsar.core.PulsarAdministration;
+import org.springframework.pulsar.core.PulsarProducerFactory;
+import org.springframework.pulsar.core.PulsarTemplate;
+import org.springframework.pulsar.reactive.config.DefaultReactivePulsarListenerContainerFactory;
+import org.springframework.pulsar.reactive.config.ReactivePulsarListenerContainerFactory;
+import org.springframework.pulsar.reactive.config.ReactivePulsarListenerEndpointRegistry;
+import org.springframework.pulsar.reactive.config.annotation.EnableReactivePulsar;
+import org.springframework.pulsar.reactive.config.annotation.ReactivePulsarListener;
+import org.springframework.pulsar.reactive.config.annotation.ReactivePulsarListenerMessageConsumerBuilderCustomizer;
+import org.springframework.pulsar.reactive.core.DefaultReactivePulsarConsumerFactory;
+import org.springframework.pulsar.reactive.core.ReactivePulsarConsumerFactory;
+import org.springframework.pulsar.reactive.listener.ReactivePulsarListenerCustomizerTests.WithMultipleListenersAndSingleCustomizer.WithMultipleListenersAndSingleCustomizerConfig;
+import org.springframework.pulsar.reactive.listener.ReactivePulsarListenerCustomizerTests.WithSingleListenerAndMultipleCustomizers.WithSingleListenerAndMultipleCustomizersConfig;
+import org.springframework.pulsar.reactive.listener.ReactivePulsarListenerCustomizerTests.WithSingleListenerAndSingleCustomizer.WithSingleListenerAndSingleCustomizerConfig;
+import org.springframework.pulsar.test.support.PulsarTestContainerSupport;
+import org.springframework.test.annotation.DirtiesContext;
+import org.springframework.test.context.ContextConfiguration;
+import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
+
+/**
+ * Tests for the customizers on the {@link ReactivePulsarListener} annotation.
+ *
+ * @author Chris Bono
+ */
+@SpringJUnitConfig
+@DirtiesContext
+@SuppressWarnings({ "unchecked", "rawtypes" })
+class ReactivePulsarListenerCustomizerTests implements PulsarTestContainerSupport {
+
+	private ObjectAssert<DefaultReactivePulsarMessageListenerContainer> assertContainer(
+			ReactivePulsarListenerEndpointRegistry registry, String containerId) {
+		return assertThat(registry.getListenerContainer(containerId)).isNotNull()
+			.isInstanceOf(DefaultReactivePulsarMessageListenerContainer.class)
+			.asInstanceOf(InstanceOfAssertFactories.type(DefaultReactivePulsarMessageListenerContainer.class));
+	}
+
+	@Configuration(proxyBeanMethods = false)
+	@EnableReactivePulsar
+	static class TopLevelConfig {
+
+		@Bean
+		PulsarProducerFactory<String> pulsarProducerFactory(PulsarClient pulsarClient) {
+			return new DefaultPulsarProducerFactory<>(pulsarClient);
+		}
+
+		@Bean
+		PulsarClient pulsarClient() throws PulsarClientException {
+			return new DefaultPulsarClientFactory(PulsarTestContainerSupport.getPulsarBrokerUrl()).createClient();
+		}
+
+		@Bean
+		ReactivePulsarClient pulsarReactivePulsarClient(PulsarClient pulsarClient) {
+			return AdaptedReactivePulsarClientFactory.create(pulsarClient);
+		}
+
+		@Bean
+		PulsarTemplate<String> pulsarTemplate(PulsarProducerFactory<String> pulsarProducerFactory) {
+			return new PulsarTemplate<>(pulsarProducerFactory);
+		}
+
+		@Bean
+		ReactivePulsarConsumerFactory<String> pulsarConsumerFactory(ReactivePulsarClient pulsarClient) {
+			return new DefaultReactivePulsarConsumerFactory<>(pulsarClient, Collections.emptyList());
+		}
+
+		@Bean
+		ReactivePulsarListenerContainerFactory<String> reactivePulsarListenerContainerFactory(
+				ReactivePulsarConsumerFactory<String> pulsarConsumerFactory) {
+			return new DefaultReactivePulsarListenerContainerFactory<>(pulsarConsumerFactory,
+					new ReactivePulsarContainerProperties<>());
+		}
+
+		@Bean
+		PulsarAdministration pulsarAdministration() {
+			return new PulsarAdministration(PulsarTestContainerSupport.getHttpServiceUrl());
+		}
+
+	}
+
+	@Nested
+	@ContextConfiguration(classes = WithSingleListenerAndSingleCustomizerConfig.class)
+	class WithSingleListenerAndSingleCustomizer {
+
+		private static ReactivePulsarListenerMessageConsumerBuilderCustomizer<Object> MY_CUSTOMIZER = mock(
+				ReactivePulsarListenerMessageConsumerBuilderCustomizer.class);
+
+		@Test
+		void customizerIsAutoAssociated(@Autowired ReactivePulsarListenerEndpointRegistry registry) {
+			assertContainer(registry, "singleListenerSingleCustomizer-id").satisfies((container) -> {
+				var builder = mock(ReactiveMessageConsumerBuilder.class);
+				container.getConsumerCustomizer().customize(builder);
+				verify(MY_CUSTOMIZER).customize(builder);
+			});
+		}
+
+		@Configuration(proxyBeanMethods = false)
+		static class WithSingleListenerAndSingleCustomizerConfig {
+
+			@ReactivePulsarListener(id = "singleListenerSingleCustomizer-id",
+					topics = "singleListenerSingleCustomizer-topic")
+			void listen(String ignored) {
+			}
+
+			@Bean
+			ReactivePulsarListenerMessageConsumerBuilderCustomizer<?> myCustomizer() {
+				return MY_CUSTOMIZER;
+			}
+
+		}
+
+	}
+
+	@Nested
+	@ContextConfiguration(classes = WithMultipleListenersAndSingleCustomizerConfig.class)
+	class WithMultipleListenersAndSingleCustomizer {
+
+		private static ReactivePulsarListenerMessageConsumerBuilderCustomizer<Object> MY_CUSTOMIZER = mock(
+				ReactivePulsarListenerMessageConsumerBuilderCustomizer.class);
+
+		@Test
+		void customizerIsNotAutoAssociated(@Autowired ReactivePulsarListenerEndpointRegistry registry) {
+			assertContainer(registry, "multiListenerSingleCustomizer1-id").satisfies((container) -> {
+				var builder = mock(ReactiveMessageConsumerBuilder.class);
+				container.getConsumerCustomizer().customize(builder);
+				verify(MY_CUSTOMIZER, never()).customize(builder);
+			});
+			assertContainer(registry, "multiListenerSingleCustomizer2-id").satisfies((container) -> {
+				var builder = mock(ReactiveMessageConsumerBuilder.class);
+				container.getConsumerCustomizer().customize(builder);
+				verify(MY_CUSTOMIZER, never()).customize(builder);
+			});
+		}
+
+		@Configuration(proxyBeanMethods = false)
+		static class WithMultipleListenersAndSingleCustomizerConfig {
+
+			@ReactivePulsarListener(id = "multiListenerSingleCustomizer1-id",
+					topics = "multiListenerSingleCustomizer1-topic")
+			void listen1(String ignored) {
+			}
+
+			@ReactivePulsarListener(id = "multiListenerSingleCustomizer2-id",
+					topics = "multiListenerSingleCustomizer2-topic")
+			void listen2(String ignored) {
+			}
+
+			@Bean
+			ReactivePulsarListenerMessageConsumerBuilderCustomizer<?> myCustomizer() {
+				return MY_CUSTOMIZER;
+			}
+
+		}
+
+	}
+
+	@Nested
+	@ContextConfiguration(classes = WithSingleListenerAndMultipleCustomizersConfig.class)
+	class WithSingleListenerAndMultipleCustomizers {
+
+		private static ReactivePulsarListenerMessageConsumerBuilderCustomizer<Object> MY_CUSTOMIZER = mock(
+				ReactivePulsarListenerMessageConsumerBuilderCustomizer.class);
+
+		private static ReactivePulsarListenerMessageConsumerBuilderCustomizer<Object> MY_CUSTOMIZER2 = mock(
+				ReactivePulsarListenerMessageConsumerBuilderCustomizer.class);
+
+		@Test
+		void customizerIsNotAutoAssociated(@Autowired ReactivePulsarListenerEndpointRegistry registry) {
+			assertContainer(registry, "singleListenerMultiCustomizers-id").satisfies((container) -> {
+				var builder = mock(ReactiveMessageConsumerBuilder.class);
+				container.getConsumerCustomizer().customize(builder);
+				verify(MY_CUSTOMIZER, never()).customize(builder);
+				verify(MY_CUSTOMIZER2, never()).customize(builder);
+			});
+		}
+
+		@Configuration(proxyBeanMethods = false)
+		static class WithSingleListenerAndMultipleCustomizersConfig {
+
+			@ReactivePulsarListener(id = "singleListenerMultiCustomizers-id",
+					topics = "singleListenerMultiCustomizers-topic")
+			void listen(String ignored) {
+			}
+
+			@Bean
+			ReactivePulsarListenerMessageConsumerBuilderCustomizer<?> myCustomizer1() {
+				return MY_CUSTOMIZER;
+			}
+
+			@Bean
+			ReactivePulsarListenerMessageConsumerBuilderCustomizer<?> myCustomizer2() {
+				return MY_CUSTOMIZER2;
+			}
+
+		}
+
+	}
+
+}

spring-pulsar/src/main/java/org/springframework/pulsar/annotation/PulsarListenerAnnotationBeanPostProcessor.java

@@ -108,13 +108,13 @@ public class PulsarListenerAnnotationBeanPostProcessor<V> extends AbstractPulsar
 
 	private final AtomicInteger counter = new AtomicInteger();
 
+	private final List<MethodPulsarListenerEndpoint<?>> processedEndpoints = new ArrayList<>();
+
 	@Override
 	public void afterSingletonsInstantiated() {
 		this.registrar.setBeanFactory(this.beanFactory);
-
 		this.beanFactory.getBeanProvider(PulsarListenerConfigurer.class)
 			.forEach(c -> c.configurePulsarListeners(this.registrar));
-
 		if (this.registrar.getEndpointRegistry() == null) {
 			if (this.endpointRegistry == null) {
 				Assert.state(this.beanFactory != null,
@@ -125,12 +125,11 @@ public class PulsarListenerAnnotationBeanPostProcessor<V> extends AbstractPulsar
 			}
 			this.registrar.setEndpointRegistry(this.endpointRegistry);
 		}
-
 		if (this.defaultContainerFactoryBeanName != null) {
 			this.registrar.setContainerFactoryBeanName(this.defaultContainerFactoryBeanName);
 		}
-
 		addFormatters(this.messageHandlerMethodFactory.defaultFormattingConversionService);
+		postProcessEndpointsBeforeRegistration();
 		// Actually register all listeners
 		this.registrar.afterPropertiesSet();
 	}
@@ -167,7 +166,6 @@ public class PulsarListenerAnnotationBeanPostProcessor<V> extends AbstractPulsar
 		Method methodToUse = checkProxy(method, bean);
 		MethodPulsarListenerEndpoint<V> endpoint = new MethodPulsarListenerEndpoint<>();
 		endpoint.setMethod(methodToUse);
-
 		String beanRef = pulsarListener.beanRef();
 		this.listenerScope.addListener(beanRef, bean);
 		String[] topics = resolveTopics(pulsarListener);
@@ -178,27 +176,21 @@ public class PulsarListenerAnnotationBeanPostProcessor<V> extends AbstractPulsar
 
 	protected void processListener(MethodPulsarListenerEndpoint<?> endpoint, PulsarListener PulsarListener, Object bean,
 			String beanName, String[] topics, String topicPattern) {
-
 		processPulsarListenerAnnotation(endpoint, PulsarListener, bean, topics, topicPattern);
-
 		String containerFactory = resolve(PulsarListener.containerFactory());
 		PulsarListenerContainerFactory listenerContainerFactory = resolveContainerFactory(PulsarListener,
 				containerFactory, beanName);
-
 		this.registrar.registerEndpoint(endpoint, listenerContainerFactory);
 	}
 
 	@Nullable
 	private PulsarListenerContainerFactory resolveContainerFactory(PulsarListener PulsarListener, Object factoryTarget,
 			String beanName) {
-
 		String containerFactory = PulsarListener.containerFactory();
 		if (!StringUtils.hasText(containerFactory)) {
 			return null;
 		}
-
 		PulsarListenerContainerFactory factory = null;
-
 		Object resolved = resolveExpression(containerFactory);
 		if (resolved instanceof PulsarListenerContainerFactory) {
 			return (PulsarListenerContainerFactory) resolved;
@@ -219,7 +211,6 @@ public class PulsarListenerAnnotationBeanPostProcessor<V> extends AbstractPulsar
 
 	private void processPulsarListenerAnnotation(MethodPulsarListenerEndpoint<?> endpoint,
 			PulsarListener pulsarListener, Object bean, String[] topics, String topicPattern) {
-
 		endpoint.setBean(bean);
 		endpoint.setMessageHandlerMethodFactory(this.messageHandlerMethodFactory);
 		endpoint.setSubscriptionName(getEndpointSubscriptionName(pulsarListener));
@@ -229,12 +220,10 @@ public class PulsarListenerAnnotationBeanPostProcessor<V> extends AbstractPulsar
 		resolveSubscriptionType(endpoint, pulsarListener);
 		endpoint.setSchemaType(pulsarListener.schemaType());
 		endpoint.setAckMode(pulsarListener.ackMode());
-
 		String concurrency = pulsarListener.concurrency();
 		if (StringUtils.hasText(concurrency)) {
 			endpoint.setConcurrency(resolveExpressionAsInteger(concurrency, "concurrency"));
 		}
-
 		String autoStartup = pulsarListener.autoStartup();
 		if (StringUtils.hasText(autoStartup)) {
 			endpoint.setAutoStartup(resolveExpressionAsBoolean(autoStartup, "autoStartup"));
@@ -242,12 +231,12 @@ public class PulsarListenerAnnotationBeanPostProcessor<V> extends AbstractPulsar
 		resolvePulsarProperties(endpoint, pulsarListener.properties());
 		endpoint.setBatchListener(pulsarListener.batch());
 		endpoint.setBeanFactory(this.beanFactory);
-
 		resolveNegativeAckRedeliveryBackoff(endpoint, pulsarListener);
 		resolveAckTimeoutRedeliveryBackoff(endpoint, pulsarListener);
 		resolveDeadLetterPolicy(endpoint, pulsarListener);
 		resolvePulsarConsumerErrorHandler(endpoint, pulsarListener);
 		resolveConsumerCustomizer(endpoint, pulsarListener);
+		this.processedEndpoints.add(endpoint);
 	}
 
 	private void resolveSubscriptionType(MethodPulsarListenerEndpoint<?> endpoint, PulsarListener pulsarListener) {
@@ -275,8 +264,27 @@ public class PulsarListenerAnnotationBeanPostProcessor<V> extends AbstractPulsar
 		}
 	}
 
+	@SuppressWarnings("unchecked")
+	protected void postProcessEndpointsBeforeRegistration() {
+		if (this.processedEndpoints.size() == 1) {
+			MethodPulsarListenerEndpoint<?> endpoint = this.processedEndpoints.get(0);
+			if (endpoint.getConsumerBuilderCustomizer() != null) {
+				return;
+			}
+			this.beanFactory.getBeanProvider(PulsarListenerConsumerBuilderCustomizer.class).ifUnique((customizer) -> {
+				this.logger
+					.info(() -> String.format("Setting the only registered PulsarListenerConsumerBuilderCustomizer "
+							+ "on the only registered @PulsarListener (%s)", endpoint.getId()));
+				endpoint.setConsumerBuilderCustomizer(customizer::customize);
+			});
+		}
+	}
+
 	@SuppressWarnings({ "rawtypes", "unchecked" })
 	private void resolveConsumerCustomizer(MethodPulsarListenerEndpoint<?> endpoint, PulsarListener pulsarListener) {
+		if (!StringUtils.hasText(pulsarListener.consumerCustomizer())) {
+			return;
+		}
 		Object consumerCustomizer = resolveExpression(pulsarListener.consumerCustomizer());
 		if (consumerCustomizer instanceof PulsarListenerConsumerBuilderCustomizer customizer) {
 			endpoint.setConsumerBuilderCustomizer(customizer::customize);

spring-pulsar/src/main/java/org/springframework/pulsar/annotation/PulsarReaderAnnotationBeanPostProcessor.java

@@ -100,13 +100,13 @@ public class PulsarReaderAnnotationBeanPostProcessor<V> extends AbstractPulsarAn
 
 	private final AtomicInteger counter = new AtomicInteger();
 
+	private final List<MethodPulsarReaderEndpoint<?>> processedEndpoints = new ArrayList<>();
+
 	@Override
 	public void afterSingletonsInstantiated() {
 		this.registrar.setBeanFactory(this.beanFactory);
-
 		this.beanFactory.getBeanProvider(PulsarReaderConfigurer.class)
 			.forEach(c -> c.configurePulsarReaders(this.registrar));
-
 		if (this.registrar.getEndpointRegistry() == null) {
 			if (this.endpointRegistry == null) {
 				Assert.state(this.beanFactory != null,
@@ -117,11 +117,10 @@ public class PulsarReaderAnnotationBeanPostProcessor<V> extends AbstractPulsarAn
 			}
 			this.registrar.setEndpointRegistry(this.endpointRegistry);
 		}
-
 		if (this.defaultContainerFactoryBeanName != null) {
 			this.registrar.setContainerFactoryBeanName(this.defaultContainerFactoryBeanName);
 		}
-
+		postProcessEndpointsBeforeRegistration();
 		// Register all readers
 		this.registrar.afterPropertiesSet();
 	}
@@ -228,10 +227,29 @@ public class PulsarReaderAnnotationBeanPostProcessor<V> extends AbstractPulsarAn
 		endpoint.setBeanFactory(this.beanFactory);
 
 		resolveReaderCustomizer(endpoint, pulsarReader);
+		this.processedEndpoints.add(endpoint);
+	}
+
+	@SuppressWarnings("unchecked")
+	protected void postProcessEndpointsBeforeRegistration() {
+		if (this.processedEndpoints.size() == 1) {
+			MethodPulsarReaderEndpoint<?> endpoint = this.processedEndpoints.get(0);
+			if (endpoint.getReaderBuilderCustomizer() != null) {
+				return;
+			}
+			this.beanFactory.getBeanProvider(PulsarReaderReaderBuilderCustomizer.class).ifUnique((customizer) -> {
+				this.logger.info(() -> String.format("Setting the only registered PulsarReaderReaderBuilderCustomizer "
+						+ "on the only registered @PulsarReader (%s)", endpoint.getId()));
+				endpoint.setReaderBuilderCustomizer(customizer::customize);
+			});
+		}
 	}
 
 	@SuppressWarnings({ "rawtypes", "unchecked" })
 	private void resolveReaderCustomizer(MethodPulsarReaderEndpoint<?> endpoint, PulsarReader pulsarReader) {
+		if (!StringUtils.hasText(pulsarReader.readerCustomizer())) {
+			return;
+		}
 		Object readerCustomizer = resolveExpression(pulsarReader.readerCustomizer());
 		if (readerCustomizer instanceof PulsarReaderReaderBuilderCustomizer customizer) {
 			endpoint.setReaderBuilderCustomizer(customizer::customize);

spring-pulsar/src/main/java/org/springframework/pulsar/config/MethodPulsarListenerEndpoint.java

@@ -247,6 +247,10 @@ public class MethodPulsarListenerEndpoint<V> extends AbstractPulsarListenerEndpo
 		this.ackTimeoutRedeliveryBackoff = ackTimeoutRedeliveryBackoff;
 	}
 
+	public ConsumerBuilderCustomizer<?> getConsumerBuilderCustomizer() {
+		return this.consumerBuilderCustomizer;
+	}
+
 	public void setConsumerBuilderCustomizer(ConsumerBuilderCustomizer<?> consumerBuilderCustomizer) {
 		this.consumerBuilderCustomizer = consumerBuilderCustomizer;
 	}

spring-pulsar/src/main/java/org/springframework/pulsar/config/MethodPulsarReaderEndpoint.java

@@ -199,6 +199,10 @@ public class MethodPulsarReaderEndpoint<V> extends AbstractPulsarReaderEndpoint<
 		this.messageHandlerMethodFactory = messageHandlerMethodFactory;
 	}
 
+	public ReaderBuilderCustomizer<?> getReaderBuilderCustomizer() {
+		return this.readerBuilderCustomizer;
+	}
+
 	public void setReaderBuilderCustomizer(ReaderBuilderCustomizer<?> readerBuilderCustomizer) {
 		this.readerBuilderCustomizer = readerBuilderCustomizer;
 	}

spring-pulsar/src/test/java/org/springframework/pulsar/listener/PulsarListenerCustomizerTests.java

@@ -0,0 +1,194 @@
+/*
+ * Copyright 2022-2023 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.pulsar.listener;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.pulsar.client.api.Consumer;
+import org.apache.pulsar.client.api.ConsumerBuilder;
+import org.assertj.core.api.InstanceOfAssertFactories;
+import org.assertj.core.api.ObjectAssert;
+import org.junit.jupiter.api.Nested;
+import org.junit.jupiter.api.Test;
+
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.pulsar.annotation.PulsarListener;
+import org.springframework.pulsar.annotation.PulsarListenerConsumerBuilderCustomizer;
+import org.springframework.pulsar.config.PulsarListenerEndpointRegistry;
+import org.springframework.pulsar.core.ConsumerBuilderCustomizer;
+import org.springframework.pulsar.listener.PulsarListenerCustomizerTests.WithCustomizerOnListener.WithCustomizerOnListenerConfig;
+import org.springframework.pulsar.listener.PulsarListenerCustomizerTests.WithMultipleListenersAndSingleCustomizer.WithMultipleListenersAndSingleCustomizerConfig;
+import org.springframework.pulsar.listener.PulsarListenerCustomizerTests.WithSingleListenerAndMultipleCustomizers.WithSingleListenerAndMultipleCustomizersConfig;
+import org.springframework.pulsar.listener.PulsarListenerCustomizerTests.WithSingleListenerAndSingleCustomizer.WithSingleListenerAndSingleCustomizerConfig;
+import org.springframework.test.context.ContextConfiguration;
+
+/**
+ * Tests setting the consumer customizer on the {@link PulsarListener @PulsarListener}.
+ *
+ * @author Chris Bono
+ */
+@SuppressWarnings({ "unchecked", "rawtypes" })
+class PulsarListenerCustomizerTests extends PulsarListenerTestsBase {
+
+	private ObjectAssert<AbstractPulsarMessageListenerContainer> assertContainer(
+			PulsarListenerEndpointRegistry registry, String containerId) {
+		return assertThat(registry.getListenerContainer(containerId)).isNotNull()
+			.isInstanceOf(AbstractPulsarMessageListenerContainer.class)
+			.asInstanceOf(InstanceOfAssertFactories.type(AbstractPulsarMessageListenerContainer.class));
+	}
+
+	@Nested
+	@ContextConfiguration(classes = WithCustomizerOnListenerConfig.class)
+	class WithCustomizerOnListener {
+
+		private static final CountDownLatch latch = new CountDownLatch(1);
+
+		@Test
+		void overridesDefaultCustomizer() throws Exception {
+			pulsarTemplate.send("overrides-default-topic", "hello");
+			assertThat(latch.await(5, TimeUnit.SECONDS)).isTrue();
+		}
+
+		@Configuration(proxyBeanMethods = false)
+		static class WithCustomizerOnListenerConfig {
+
+			@PulsarListener(topics = "overrides-default-topic", consumerCustomizer = "myCustomizer")
+			void listen(String ignored, Consumer<String> consumer) {
+				assertThat(consumer.getConsumerName()).isEqualTo("fromMyCustomizer");
+				latch.countDown();
+			}
+
+			@Bean
+			ConsumerBuilderCustomizer<String> defaultCustomizer() {
+				return (cb) -> cb.consumerName("fromDefaultCustomizer");
+			}
+
+			@Bean
+			PulsarListenerConsumerBuilderCustomizer<String> myCustomizer() {
+				return (cb) -> cb.consumerName("fromMyCustomizer");
+			}
+
+		}
+
+	}
+
+	@Nested
+	@ContextConfiguration(classes = WithSingleListenerAndSingleCustomizerConfig.class)
+	class WithSingleListenerAndSingleCustomizer {
+
+		private static PulsarListenerConsumerBuilderCustomizer<Object> MY_CUSTOMIZER = mock(
+				PulsarListenerConsumerBuilderCustomizer.class);
+
+		@Test
+		void customizerIsAutoAssociated(@Autowired PulsarListenerEndpointRegistry registry) {
+			assertContainer(registry, "singleListenerSingleCustomizer-id").satisfies((container) -> {
+				var builder = mock(ConsumerBuilder.class);
+				container.getConsumerBuilderCustomizer().customize(builder);
+				verify(MY_CUSTOMIZER).customize(builder);
+			});
+		}
+
+		@Configuration(proxyBeanMethods = false)
+		static class WithSingleListenerAndSingleCustomizerConfig {
+
+			@PulsarListener(id = "singleListenerSingleCustomizer-id", topics = "singleListenerSingleCustomizer-topic")
+			void listen(String ignored) {
+			}
+
+			@Bean
+			PulsarListenerConsumerBuilderCustomizer<?> myCustomizer() {
+				return MY_CUSTOMIZER;
+			}
+
+		}
+
+	}
+
+	@Nested
+	@ContextConfiguration(classes = WithMultipleListenersAndSingleCustomizerConfig.class)
+	class WithMultipleListenersAndSingleCustomizer {
+
+		@Test
+		void customizerIsNotAutoAssociated(@Autowired PulsarListenerEndpointRegistry registry) {
+			assertContainer(registry, "multiListenerSingleCustomizer1-id")
+				.extracting(AbstractPulsarMessageListenerContainer::getConsumerBuilderCustomizer)
+				.isNull();
+			assertContainer(registry, "multiListenerSingleCustomizer2-id")
+				.extracting(AbstractPulsarMessageListenerContainer::getConsumerBuilderCustomizer)
+				.isNull();
+		}
+
+		@Configuration(proxyBeanMethods = false)
+		static class WithMultipleListenersAndSingleCustomizerConfig {
+
+			@PulsarListener(id = "multiListenerSingleCustomizer1-id", topics = "multiListenerSingleCustomizer1-topic")
+			void listen1(String ignored) {
+			}
+
+			@PulsarListener(id = "multiListenerSingleCustomizer2-id", topics = "multiListenerSingleCustomizer2-topic")
+			void listen2(String ignored) {
+			}
+
+			@Bean
+			PulsarListenerConsumerBuilderCustomizer<?> myCustomizer() {
+				return mock(PulsarListenerConsumerBuilderCustomizer.class);
+			}
+
+		}
+
+	}
+
+	@Nested
+	@ContextConfiguration(classes = WithSingleListenerAndMultipleCustomizersConfig.class)
+	class WithSingleListenerAndMultipleCustomizers {
+
+		@Test
+		void customizerIsNotAutoAssociated(@Autowired PulsarListenerEndpointRegistry registry) {
+			assertContainer(registry, "singleListenerMultiCustomizers-id")
+				.extracting(AbstractPulsarMessageListenerContainer::getConsumerBuilderCustomizer)
+				.isNull();
+		}
+
+		@Configuration(proxyBeanMethods = false)
+		static class WithSingleListenerAndMultipleCustomizersConfig {
+
+			@PulsarListener(id = "singleListenerMultiCustomizers-id", topics = "singleListenerMultiCustomizers-topic")
+			void listen(String ignored) {
+			}
+
+			@Bean
+			PulsarListenerConsumerBuilderCustomizer<?> myCustomizer1() {
+				return mock(PulsarListenerConsumerBuilderCustomizer.class);
+			}
+
+			@Bean
+			PulsarListenerConsumerBuilderCustomizer<?> myCustomizer2() {
+				return mock(PulsarListenerConsumerBuilderCustomizer.class);
+			}
+
+		}
+
+	}
+
+}

spring-pulsar/src/test/java/org/springframework/pulsar/listener/PulsarListenerTests.java

@@ -34,8 +34,6 @@ import org.apache.pulsar.client.api.DeadLetterPolicy;
 import org.apache.pulsar.client.api.Message;
 import org.apache.pulsar.client.api.MessageId;
 import org.apache.pulsar.client.api.Messages;
-import org.apache.pulsar.client.api.PulsarClient;
-import org.apache.pulsar.client.api.PulsarClientException;
 import org.apache.pulsar.client.api.RedeliveryBackoff;
 import org.apache.pulsar.client.api.Schema;
 import org.apache.pulsar.client.api.SubscriptionType;
@@ -53,7 +51,6 @@ import org.awaitility.Awaitility;
 import org.junit.jupiter.api.Nested;
 import org.junit.jupiter.api.Test;
 
-import org.springframework.beans.factory.ObjectProvider;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.context.annotation.AnnotationConfigApplicationContext;
 import org.springframework.context.annotation.Bean;
@@ -66,25 +63,17 @@ import org.springframework.pulsar.config.ConcurrentPulsarListenerContainerFactor
 import org.springframework.pulsar.config.PulsarListenerContainerFactory;
 import org.springframework.pulsar.config.PulsarListenerEndpointRegistry;
 import org.springframework.pulsar.core.ConsumerBuilderCustomizer;
-import org.springframework.pulsar.core.DefaultPulsarClientFactory;
-import org.springframework.pulsar.core.DefaultPulsarConsumerFactory;
 import org.springframework.pulsar.core.DefaultPulsarProducerFactory;
 import org.springframework.pulsar.core.DefaultSchemaResolver;
 import org.springframework.pulsar.core.DefaultTopicResolver;
-import org.springframework.pulsar.core.PulsarAdministration;
 import org.springframework.pulsar.core.PulsarConsumerFactory;
-import org.springframework.pulsar.core.PulsarProducerFactory;
 import org.springframework.pulsar.core.PulsarTemplate;
-import org.springframework.pulsar.core.PulsarTopic;
 import org.springframework.pulsar.core.SchemaResolver;
 import org.springframework.pulsar.core.TopicResolver;
 import org.springframework.pulsar.listener.PulsarListenerTests.SubscriptionTypeTests.WithDefaultType.WithDefaultTypeConfig;
 import org.springframework.pulsar.listener.PulsarListenerTests.SubscriptionTypeTests.WithSpecificTypes.WithSpecificTypesConfig;
 import org.springframework.pulsar.support.PulsarHeaders;
-import org.springframework.pulsar.test.support.PulsarTestContainerSupport;
-import org.springframework.test.annotation.DirtiesContext;
 import org.springframework.test.context.ContextConfiguration;
-import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
 import org.springframework.util.backoff.FixedBackOff;
 
 /**
@@ -92,60 +81,7 @@ import org.springframework.util.backoff.FixedBackOff;
  * @author Alexander Preuß
  * @author Chris Bono
  */
-@SpringJUnitConfig
-@DirtiesContext
-public class PulsarListenerTests implements PulsarTestContainerSupport {
-
-	@Autowired
-	PulsarTemplate<String> pulsarTemplate;
-
-	@Autowired
-	private PulsarClient pulsarClient;
-
-	@Configuration(proxyBeanMethods = false)
-	@EnablePulsar
-	public static class TopLevelConfig {
-
-		@Bean
-		public PulsarProducerFactory<String> pulsarProducerFactory(PulsarClient pulsarClient) {
-			return new DefaultPulsarProducerFactory<>(pulsarClient, "foo-1");
-		}
-
-		@Bean
-		public PulsarClient pulsarClient() throws PulsarClientException {
-			return new DefaultPulsarClientFactory(PulsarTestContainerSupport.getPulsarBrokerUrl()).createClient();
-		}
-
-		@Bean
-		public PulsarTemplate<String> pulsarTemplate(PulsarProducerFactory<String> pulsarProducerFactory) {
-			return new PulsarTemplate<>(pulsarProducerFactory);
-		}
-
-		@Bean
-		public PulsarConsumerFactory<?> pulsarConsumerFactory(PulsarClient pulsarClient,
-				ObjectProvider<ConsumerBuilderCustomizer<String>> defaultConsumerCustomizersProvider) {
-			return new DefaultPulsarConsumerFactory<>(pulsarClient,
-					defaultConsumerCustomizersProvider.orderedStream().toList());
-		}
-
-		@Bean
-		PulsarListenerContainerFactory pulsarListenerContainerFactory(
-				PulsarConsumerFactory<Object> pulsarConsumerFactory) {
-			return new ConcurrentPulsarListenerContainerFactory<>(pulsarConsumerFactory,
-					new PulsarContainerProperties());
-		}
-
-		@Bean
-		PulsarAdministration pulsarAdministration() {
-			return new PulsarAdministration(PulsarTestContainerSupport.getHttpServiceUrl());
-		}
-
-		@Bean
-		PulsarTopic partitionedTopic() {
-			return PulsarTopic.builder("persistent://public/default/concurrency-on-pl").numberOfPartitions(3).build();
-		}
-
-	}
+class PulsarListenerTests extends PulsarListenerTestsBase {
 
 	@Nested
 	@ContextConfiguration(classes = PulsarListenerBasicTestCases.TestPulsarListenersForBasicScenario.class)
@@ -1080,38 +1016,6 @@ public class PulsarListenerTests implements PulsarTestContainerSupport {
 
 	}
 
-	@Nested
-	@ContextConfiguration(classes = PulsarListenerCustomizerTests.WithCustomizerConfig.class)
-	class PulsarListenerCustomizerTests {
-
-		private static final CountDownLatch latch = new CountDownLatch(1);
-
-		@Test
-		void withCustomizerOverridingSubscriptionName() throws Exception {
-			pulsarTemplate.send("with-customizer-listener-topic", "hello");
-			assertThat(latch.await(10, TimeUnit.SECONDS)).isTrue();
-		}
-
-		@EnablePulsar
-		@Configuration
-		static class WithCustomizerConfig {
-
-			@PulsarListener(id = "with-customizer-listener", subscriptionName = "with-customizer-listener-subscription",
-					topics = "with-customizer-listener-topic", consumerCustomizer = "myCustomizer")
-			void listen(String ignored, Consumer<String> consumer) {
-				assertThat(consumer.getSubscription()).isEqualTo("test-changed-subscription-name");
-				latch.countDown();
-			}
-
-			@Bean
-			public PulsarListenerConsumerBuilderCustomizer<String> myCustomizer() {
-				return cb -> cb.subscriptionName("test-changed-subscription-name");
-			}
-
-		}
-
-	}
-
 	@Nested
 	class SubscriptionTypeTests {
 

spring-pulsar/src/test/java/org/springframework/pulsar/listener/PulsarListenerTestsBase.java

@@ -0,0 +1,102 @@
+/*
+ * Copyright 2022-2023 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.pulsar.listener;
+
+import org.apache.pulsar.client.api.PulsarClient;
+import org.apache.pulsar.client.api.PulsarClientException;
+
+import org.springframework.beans.factory.ObjectProvider;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.pulsar.annotation.EnablePulsar;
+import org.springframework.pulsar.config.ConcurrentPulsarListenerContainerFactory;
+import org.springframework.pulsar.config.PulsarListenerContainerFactory;
+import org.springframework.pulsar.core.ConsumerBuilderCustomizer;
+import org.springframework.pulsar.core.DefaultPulsarClientFactory;
+import org.springframework.pulsar.core.DefaultPulsarConsumerFactory;
+import org.springframework.pulsar.core.DefaultPulsarProducerFactory;
+import org.springframework.pulsar.core.PulsarAdministration;
+import org.springframework.pulsar.core.PulsarConsumerFactory;
+import org.springframework.pulsar.core.PulsarProducerFactory;
+import org.springframework.pulsar.core.PulsarTemplate;
+import org.springframework.pulsar.core.PulsarTopic;
+import org.springframework.pulsar.test.support.PulsarTestContainerSupport;
+import org.springframework.test.annotation.DirtiesContext;
+import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
+
+/**
+ * @author Soby Chacko
+ * @author Alexander Preuß
+ * @author Chris Bono
+ */
+@SpringJUnitConfig
+@DirtiesContext
+abstract class PulsarListenerTestsBase implements PulsarTestContainerSupport {
+
+	@Autowired
+	protected PulsarTemplate<String> pulsarTemplate;
+
+	@Autowired
+	protected PulsarClient pulsarClient;
+
+	@Configuration(proxyBeanMethods = false)
+	@EnablePulsar
+	static class TopLevelConfig {
+
+		@Bean
+		PulsarProducerFactory<String> pulsarProducerFactory(PulsarClient pulsarClient) {
+			return new DefaultPulsarProducerFactory<>(pulsarClient, "foo-1");
+		}
+
+		@Bean
+		PulsarClient pulsarClient() throws PulsarClientException {
+			return new DefaultPulsarClientFactory(PulsarTestContainerSupport.getPulsarBrokerUrl()).createClient();
+		}
+
+		@Bean
+		PulsarTemplate<String> pulsarTemplate(PulsarProducerFactory<String> pulsarProducerFactory) {
+			return new PulsarTemplate<>(pulsarProducerFactory);
+		}
+
+		@Bean
+		public PulsarConsumerFactory<?> pulsarConsumerFactory(PulsarClient pulsarClient,
+				ObjectProvider<ConsumerBuilderCustomizer<String>> defaultConsumerCustomizersProvider) {
+			return new DefaultPulsarConsumerFactory<>(pulsarClient,
+					defaultConsumerCustomizersProvider.orderedStream().toList());
+		}
+
+		@Bean
+		PulsarListenerContainerFactory pulsarListenerContainerFactory(
+				PulsarConsumerFactory<Object> pulsarConsumerFactory) {
+			return new ConcurrentPulsarListenerContainerFactory<>(pulsarConsumerFactory,
+					new PulsarContainerProperties());
+		}
+
+		@Bean
+		PulsarAdministration pulsarAdministration() {
+			return new PulsarAdministration(PulsarTestContainerSupport.getHttpServiceUrl());
+		}
+
+		@Bean
+		PulsarTopic partitionedTopic() {
+			return PulsarTopic.builder("persistent://public/default/concurrency-on-pl").numberOfPartitions(3).build();
+		}
+
+	}
+
+}

spring-pulsar/src/test/java/org/springframework/pulsar/reader/PulsarReaderCustomizerTests.java

@@ -0,0 +1,196 @@
+/*
+ * Copyright 2022-2023 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.pulsar.reader;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.pulsar.client.api.ReaderBuilder;
+import org.assertj.core.api.InstanceOfAssertFactories;
+import org.assertj.core.api.ObjectAssert;
+import org.junit.jupiter.api.Nested;
+import org.junit.jupiter.api.Test;
+
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.pulsar.annotation.PulsarReader;
+import org.springframework.pulsar.annotation.PulsarReaderReaderBuilderCustomizer;
+import org.springframework.pulsar.config.PulsarReaderEndpointRegistry;
+import org.springframework.pulsar.core.ReaderBuilderCustomizer;
+import org.springframework.pulsar.reader.PulsarReaderCustomizerTests.WithCustomizerOnReader.WithCustomizerOnReaderConfig;
+import org.springframework.pulsar.reader.PulsarReaderCustomizerTests.WithMultipleReadersAndSingleCustomizer.WithMultipleReadersAndSingleCustomizerConfig;
+import org.springframework.pulsar.reader.PulsarReaderCustomizerTests.WithSingleReaderAndMultipleCustomizers.WithSingleReaderAndMultipleCustomizersConfig;
+import org.springframework.pulsar.reader.PulsarReaderCustomizerTests.WithSingleReaderAndSingleCustomizer.WithSingleReaderAndSingleCustomizerConfig;
+import org.springframework.test.context.ContextConfiguration;
+
+/**
+ * Tests setting the consumer customizer on the {@link PulsarReader @PulsarReader}.
+ *
+ * @author Chris Bono
+ */
+@SuppressWarnings({ "unchecked", "rawtypes" })
+class PulsarReaderCustomizerTests extends PulsarReaderTestsBase {
+
+	private ObjectAssert<AbstractPulsarMessageReaderContainer> assertContainer(PulsarReaderEndpointRegistry registry,
+			String containerId) {
+		return assertThat(registry.getReaderContainer(containerId)).isNotNull()
+			.isInstanceOf(AbstractPulsarMessageReaderContainer.class)
+			.asInstanceOf(InstanceOfAssertFactories.type(AbstractPulsarMessageReaderContainer.class));
+	}
+
+	@Nested
+	@ContextConfiguration(classes = WithCustomizerOnReaderConfig.class)
+	class WithCustomizerOnReader {
+
+		private static final CountDownLatch latch = new CountDownLatch(1);
+
+		@Test
+		void overridesDefaultCustomizer() throws Exception {
+			pulsarTemplate.send("myCustomizerTopic", "hello");
+			assertThat(latch.await(5, TimeUnit.SECONDS)).isTrue();
+		}
+
+		@Configuration(proxyBeanMethods = false)
+		static class WithCustomizerOnReaderConfig {
+
+			@PulsarReader(id = "overrides-default-id", readerCustomizer = "myCustomizer", startMessageId = "earliest")
+			void listen(String ignored) {
+				latch.countDown();
+			}
+
+			@Bean
+			ReaderBuilderCustomizer<String> defaultCustomizer() {
+				return (rb) -> rb.topic("defaultCustomizerTopic");
+			}
+
+			@Bean
+			PulsarReaderReaderBuilderCustomizer<String> myCustomizer() {
+				return (rb) -> rb.topic("myCustomizerTopic");
+			}
+
+		}
+
+	}
+
+	@Nested
+	@ContextConfiguration(classes = WithSingleReaderAndSingleCustomizerConfig.class)
+	class WithSingleReaderAndSingleCustomizer {
+
+		private static PulsarReaderReaderBuilderCustomizer<Object> MY_CUSTOMIZER = mock(
+				PulsarReaderReaderBuilderCustomizer.class);
+
+		@Test
+		void customizerIsAutoAssociated(@Autowired PulsarReaderEndpointRegistry registry) {
+			assertContainer(registry, "singleListenerSingleCustomizer-id").satisfies((container) -> {
+				var builder = mock(ReaderBuilder.class);
+				container.getReaderBuilderCustomizer().customize(builder);
+				verify(MY_CUSTOMIZER).customize(builder);
+			});
+		}
+
+		@Configuration(proxyBeanMethods = false)
+		static class WithSingleReaderAndSingleCustomizerConfig {
+
+			@PulsarReader(id = "singleListenerSingleCustomizer-id", topics = "singleListenerSingleCustomizer-topic",
+					startMessageId = "earliest")
+			void listen(String ignored) {
+			}
+
+			@Bean
+			PulsarReaderReaderBuilderCustomizer<?> myCustomizer() {
+				return MY_CUSTOMIZER;
+			}
+
+		}
+
+	}
+
+	@Nested
+	@ContextConfiguration(classes = WithMultipleReadersAndSingleCustomizerConfig.class)
+	class WithMultipleReadersAndSingleCustomizer {
+
+		@Test
+		void customizerIsNotAutoAssociated(@Autowired PulsarReaderEndpointRegistry registry) {
+			assertContainer(registry, "multiReaderSingleCustomizer1-id")
+				.extracting(AbstractPulsarMessageReaderContainer::getReaderBuilderCustomizer)
+				.isNull();
+			assertContainer(registry, "multiReaderSingleCustomizer2-id")
+				.extracting(AbstractPulsarMessageReaderContainer::getReaderBuilderCustomizer)
+				.isNull();
+		}
+
+		@Configuration(proxyBeanMethods = false)
+		static class WithMultipleReadersAndSingleCustomizerConfig {
+
+			@PulsarReader(id = "multiReaderSingleCustomizer1-id", topics = "multiReaderSingleCustomizer1-topic",
+					startMessageId = "earliest")
+			void listen1(String ignored) {
+			}
+
+			@PulsarReader(id = "multiReaderSingleCustomizer2-id", topics = "multiReaderSingleCustomizer2-topic",
+					startMessageId = "earliest")
+			void listen2(String ignored) {
+			}
+
+			@Bean
+			PulsarReaderReaderBuilderCustomizer<?> myCustomizer() {
+				return mock(PulsarReaderReaderBuilderCustomizer.class);
+			}
+
+		}
+
+	}
+
+	@Nested
+	@ContextConfiguration(classes = WithSingleReaderAndMultipleCustomizersConfig.class)
+	class WithSingleReaderAndMultipleCustomizers {
+
+		@Test
+		void customizerIsNotAutoAssociated(@Autowired PulsarReaderEndpointRegistry registry) {
+			assertContainer(registry, "singleReaderMultiCustomizers-id")
+				.extracting(AbstractPulsarMessageReaderContainer::getReaderBuilderCustomizer)
+				.isNull();
+		}
+
+		@Configuration(proxyBeanMethods = false)
+		static class WithSingleReaderAndMultipleCustomizersConfig {
+
+			@PulsarReader(id = "singleReaderMultiCustomizers-id", topics = "singleReaderMultiCustomizers-topic",
+					startMessageId = "earliest")
+			void listen(String ignored) {
+			}
+
+			@Bean
+			PulsarReaderReaderBuilderCustomizer<?> myCustomizer1() {
+				return mock(PulsarReaderReaderBuilderCustomizer.class);
+			}
+
+			@Bean
+			PulsarReaderReaderBuilderCustomizer<?> myCustomizer2() {
+				return mock(PulsarReaderReaderBuilderCustomizer.class);
+			}
+
+		}
+
+	}
+
+}

spring-pulsar/src/test/java/org/springframework/pulsar/reader/PulsarReaderTests.java

@@ -24,30 +24,18 @@ import java.util.concurrent.TimeUnit;
 
 import org.apache.pulsar.client.api.Message;
 import org.apache.pulsar.client.api.MessageId;
-import org.apache.pulsar.client.api.PulsarClient;
 import org.apache.pulsar.client.api.PulsarClientException;
 import org.junit.jupiter.api.Nested;
 import org.junit.jupiter.api.Test;
 
-import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.context.annotation.AnnotationConfigApplicationContext;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 import org.springframework.pulsar.annotation.EnablePulsar;
 import org.springframework.pulsar.annotation.PulsarReader;
 import org.springframework.pulsar.annotation.PulsarReaderReaderBuilderCustomizer;
-import org.springframework.pulsar.config.DefaultPulsarReaderContainerFactory;
-import org.springframework.pulsar.config.PulsarReaderContainerFactory;
-import org.springframework.pulsar.core.DefaultPulsarClientFactory;
-import org.springframework.pulsar.core.DefaultPulsarProducerFactory;
-import org.springframework.pulsar.core.DefaultPulsarReaderFactory;
-import org.springframework.pulsar.core.PulsarProducerFactory;
-import org.springframework.pulsar.core.PulsarReaderFactory;
 import org.springframework.pulsar.core.PulsarTemplate;
-import org.springframework.pulsar.test.support.PulsarTestContainerSupport;
-import org.springframework.test.annotation.DirtiesContext;
 import org.springframework.test.context.ContextConfiguration;
-import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
 
 /**
  * Tests for {@link PulsarReader}.
@@ -55,47 +43,7 @@ import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
  * @author Soby Chacko
  * @author Chris Bono
  */
-@SpringJUnitConfig
-@DirtiesContext
-public class PulsarReaderTests implements PulsarTestContainerSupport {
-
-	@Autowired
-	PulsarTemplate<String> pulsarTemplate;
-
-	@Autowired
-	private PulsarClient pulsarClient;
-
-	@Configuration(proxyBeanMethods = false)
-	@EnablePulsar
-	public static class TopLevelConfig {
-
-		@Bean
-		public PulsarProducerFactory<String> pulsarProducerFactory(PulsarClient pulsarClient) {
-			return new DefaultPulsarProducerFactory<>(pulsarClient);
-		}
-
-		@Bean
-		public PulsarClient pulsarClient() throws PulsarClientException {
-			return new DefaultPulsarClientFactory(PulsarTestContainerSupport.getPulsarBrokerUrl()).createClient();
-		}
-
-		@Bean
-		public PulsarTemplate<String> pulsarTemplate(PulsarProducerFactory<String> pulsarProducerFactory) {
-			return new PulsarTemplate<>(pulsarProducerFactory);
-		}
-
-		@Bean
-		public PulsarReaderFactory<?> pulsarReaderFactory(PulsarClient pulsarClient) {
-			return new DefaultPulsarReaderFactory<>(pulsarClient);
-		}
-
-		@Bean
-		PulsarReaderContainerFactory pulsarReaderContainerFactory(PulsarReaderFactory<Object> pulsarReaderFactory) {
-			return new DefaultPulsarReaderContainerFactory<>(pulsarReaderFactory,
-					new PulsarReaderContainerProperties());
-		}
-
-	}
+public class PulsarReaderTests extends PulsarReaderTestsBase {
 
 	@Nested
 	@ContextConfiguration(classes = StartMessageIdEarliest.PulsarReaderStartMessageIdEarliest.class)
@@ -109,8 +57,7 @@ public class PulsarReaderTests implements PulsarTestContainerSupport {
 			assertThat(latch.await(5, TimeUnit.SECONDS)).isTrue();
 		}
 
-		@EnablePulsar
-		@Configuration
+		@Configuration(proxyBeanMethods = false)
 		static class PulsarReaderStartMessageIdEarliest {
 
 			@PulsarReader(id = "pulsarReaderBasicScenario-id-1", topics = "pulsarReaderBasicScenario-topic-1",
@@ -162,8 +109,7 @@ public class PulsarReaderTests implements PulsarTestContainerSupport {
 			assertThat(latch.await(5, TimeUnit.SECONDS)).isTrue();
 		}
 
-		@EnablePulsar
-		@Configuration
+		@Configuration(proxyBeanMethods = false)
 		static class PulsarReaderStartMessageIdLatest {
 
 			@PulsarReader(id = "pulsarReaderBasicScenario-id-3", topics = "pulsarReaderBasicScenario-topic-3",
@@ -185,11 +131,10 @@ public class PulsarReaderTests implements PulsarTestContainerSupport {
 
 		@Test
 		void startMessageIdProvidedThroughReaderCustomizer() throws Exception {
-			assertThat(latch.await(10, TimeUnit.SECONDS)).isTrue();
+			assertThat(latch.await(5, TimeUnit.SECONDS)).isTrue();
 		}
 
-		@EnablePulsar
-		@Configuration
+		@Configuration(proxyBeanMethods = false)
 		static class WithCustomizerConfig {
 
 			int currentIndex = 5;

spring-pulsar/src/test/java/org/springframework/pulsar/reader/PulsarReaderTestsBase.java

@@ -0,0 +1,87 @@
+/*
+ * Copyright 2023 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.pulsar.reader;
+
+import org.apache.pulsar.client.api.PulsarClient;
+import org.apache.pulsar.client.api.PulsarClientException;
+
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.pulsar.annotation.EnablePulsar;
+import org.springframework.pulsar.annotation.PulsarReader;
+import org.springframework.pulsar.config.DefaultPulsarReaderContainerFactory;
+import org.springframework.pulsar.config.PulsarReaderContainerFactory;
+import org.springframework.pulsar.core.DefaultPulsarClientFactory;
+import org.springframework.pulsar.core.DefaultPulsarProducerFactory;
+import org.springframework.pulsar.core.DefaultPulsarReaderFactory;
+import org.springframework.pulsar.core.PulsarProducerFactory;
+import org.springframework.pulsar.core.PulsarReaderFactory;
+import org.springframework.pulsar.core.PulsarTemplate;
+import org.springframework.pulsar.test.support.PulsarTestContainerSupport;
+import org.springframework.test.annotation.DirtiesContext;
+import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
+
+/**
+ * Tests for {@link PulsarReader}.
+ *
+ * @author Soby Chacko
+ * @author Chris Bono
+ */
+@SpringJUnitConfig
+@DirtiesContext
+class PulsarReaderTestsBase implements PulsarTestContainerSupport {
+
+	@Autowired
+	protected PulsarTemplate<String> pulsarTemplate;
+
+	@Autowired
+	protected PulsarClient pulsarClient;
+
+	@Configuration(proxyBeanMethods = false)
+	@EnablePulsar
+	static class TopLevelConfig {
+
+		@Bean
+		PulsarProducerFactory<String> pulsarProducerFactory(PulsarClient pulsarClient) {
+			return new DefaultPulsarProducerFactory<>(pulsarClient);
+		}
+
+		@Bean
+		PulsarClient pulsarClient() throws PulsarClientException {
+			return new DefaultPulsarClientFactory(PulsarTestContainerSupport.getPulsarBrokerUrl()).createClient();
+		}
+
+		@Bean
+		PulsarTemplate<String> pulsarTemplate(PulsarProducerFactory<String> pulsarProducerFactory) {
+			return new PulsarTemplate<>(pulsarProducerFactory);
+		}
+
+		@Bean
+		PulsarReaderFactory<?> pulsarReaderFactory(PulsarClient pulsarClient) {
+			return new DefaultPulsarReaderFactory<>(pulsarClient);
+		}
+
+		@Bean
+		PulsarReaderContainerFactory pulsarReaderContainerFactory(PulsarReaderFactory<Object> pulsarReaderFactory) {
+			return new DefaultPulsarReaderContainerFactory<>(pulsarReaderFactory,
+					new PulsarReaderContainerProperties());
+		}
+
+	}
+
+}