Add specialized builder customizers for listener annotations (#487)

Introduces specializations of the consumer builders that can be used by
 the annotation listeners. This differentiates between the existing
 consumer builders that are picked up by Spring Boot
 autoconfiguration and used to customizer the consumer factories
 globally.

* Add PulsarListenerConsumerBuilderCustomizer
* Add PulsarReaderReaderBuilderCustomizer
* Add ReactivePulsarListenerMessageConsumerBuilderCustomizer

Resolves #486

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

@@ -1029,12 +1029,12 @@ At that point, the message is acknowledged by the container, and the listener is
 === Consumer Customization on PulsarListener
 
 Spring for Apache Pulsar provides a convenient way to customize the consumer created by the container used by the `PulsarListener`.
-Applications can provide a bean for `ConsumerBuilderCustomizer`.
+Applications can provide a bean for `PulsarListenerConsumerBuilderCustomizer`.
 Here is an example.
 [source, java]
 ----
 @Bean
-public ConsumerBuilderCustomizer<String> myCustomizer() {
+public PulsarListenerConsumerBuilderCustomizer<String> myCustomizer() {
     return cb -> {
         cb.subscriptionName("modified-subscription-name");
     };
@@ -1107,8 +1107,8 @@ 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 `ReaderBuilderCustomizer` in Spring for Apache Pulsar.
-You can provide a `@Bean` from `ReaderBuilderCustomizer` and then make it available to the `PulsarReader` as below.
+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.
 
 [source, java]
 ----
@@ -1119,7 +1119,7 @@ void read(String message) {
 }
 
 @Bean
-public ReaderBuilderCustomizer<String> myCustomizer() {
+public PulsarReaderReaderBuilderCustomizer<String> myCustomizer() {
     return readerBuilder -> {
         readerBuilder.startMessageId(messageId); // the first message read is after this message id.
         // Any other customizations on the readerBuilder

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

@@ -245,7 +245,7 @@ Spring Boot provides this consumer factory which can be configured with any of t
 [[reactive-consumer-customizer]]
 ==== Consumer Customization
 
-You can specify a `ReactiveMessageConsumerBuilderCustomizer` to configure the underlying Pulsar consumer builder that ultimately constructs the consumer used by the listener to receive the messages.
+You can specify a `ReactivePulsarListenerMessageConsumerBuilderCustomizer` to configure the underlying Pulsar consumer builder that ultimately constructs the consumer used by the listener to receive the messages.
 
 WARNING: Use with caution as this gives full access to the consumer builder and invoking some of its methods (such as `create`) may have unintended side effects.
 
@@ -260,7 +260,7 @@ Mono<Void> listen(String message) {
 }
 
 @Bean
-ReactiveMessageConsumerBuilderCustomizer<String> myConsumerCustomizer() {
+ReactivePulsarListenerMessageConsumerBuilderCustomizer<String> myConsumerCustomizer() {
     return b -> b.subscriptionInitialPosition(SubscriptionInitialPosition.Earliest);
 }
 ----
@@ -273,7 +273,7 @@ The following customizer example uses direct Pulsar consumer properties:
 [source, java]
 ----
 @Bean
-ReactiveMessageConsumerBuilderCustomizer<String> directConsumerPropsCustomizer() {
+ReactivePulsarListenerMessageConsumerBuilderCustomizer<String> directConsumerPropsCustomizer() {
     return b -> b.property("subscriptionName", "subscription-1").property("topicNames", "foo-1");
 }
 ----

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

@@ -53,7 +53,6 @@ import org.springframework.pulsar.reactive.config.MethodReactivePulsarListenerEn
 import org.springframework.pulsar.reactive.config.ReactivePulsarListenerContainerFactory;
 import org.springframework.pulsar.reactive.config.ReactivePulsarListenerEndpoint;
 import org.springframework.pulsar.reactive.config.ReactivePulsarListenerEndpointRegistry;
-import org.springframework.pulsar.reactive.core.ReactiveMessageConsumerBuilderCustomizer;
 import org.springframework.util.Assert;
 import org.springframework.util.StringUtils;
 
@@ -277,19 +276,20 @@ public class ReactivePulsarListenerAnnotationBeanPostProcessor<V> extends Abstra
 		}
 	}
 
-	@SuppressWarnings("unchecked")
+	@SuppressWarnings({ "rawtypes", "unchecked" })
 	private void resolveConsumerCustomizer(MethodReactivePulsarListenerEndpoint<?> endpoint,
 			ReactivePulsarListener reactivePulsarListener) {
-		Object customizer = resolveExpression(reactivePulsarListener.consumerCustomizer());
-		if (customizer instanceof ReactiveMessageConsumerBuilderCustomizer<?>) {
-			endpoint.setConsumerCustomizer((ReactiveMessageConsumerBuilderCustomizer) customizer);
+		Object consumerCustomizer = resolveExpression(reactivePulsarListener.consumerCustomizer());
+		if (consumerCustomizer instanceof ReactivePulsarListenerMessageConsumerBuilderCustomizer customizer) {
+			endpoint.setConsumerCustomizer(customizer::customize);
 		}
 		else {
-			String consumerCustomizerBeanName = resolveExpressionAsString(reactivePulsarListener.consumerCustomizer(),
+			String customizerBeanName = resolveExpressionAsString(reactivePulsarListener.consumerCustomizer(),
 					"consumerCustomizer");
-			if (StringUtils.hasText(consumerCustomizerBeanName)) {
-				endpoint.setConsumerCustomizer(this.beanFactory.getBean(consumerCustomizerBeanName,
-						ReactiveMessageConsumerBuilderCustomizer.class));
+			if (StringUtils.hasText(customizerBeanName)) {
+				var customizer = this.beanFactory.getBean(customizerBeanName,
+						ReactivePulsarListenerMessageConsumerBuilderCustomizer.class);
+				endpoint.setConsumerCustomizer(customizer::customize);
 			}
 		}
 	}

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

@@ -0,0 +1,43 @@
+/*
+ * 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.config.annotation;
+
+import org.apache.pulsar.reactive.client.api.ReactiveMessageConsumerBuilder;
+
+/**
+ * Callback interface that can be implemented by a bean to customize the
+ * {@link ReactiveMessageConsumerBuilder builder} that is used to create the underlying
+ * Pulsar reactive message consumer used by a {@link ReactivePulsarListener} to receive
+ * messages.
+ * <p>
+ * Unlike the {@link ReactiveMessageConsumerBuilder} which is applied to all created
+ * reactive message consumer builders, this customizer is only applied to the individual
+ * consumer builder(s) of the {@code @ReactivePulsarListener(s)} it is associated with.
+ *
+ * @param <T> The message payload type
+ * @author Chris Bono
+ */
+@FunctionalInterface
+public interface ReactivePulsarListenerMessageConsumerBuilderCustomizer<T> {
+
+	/**
+	 * Customize the {@link ReactiveMessageConsumerBuilder}.
+	 * @param reactiveMessageConsumerBuilder the builder to customize
+	 */
+	void customize(ReactiveMessageConsumerBuilder<T> reactiveMessageConsumerBuilder);
+
+}

spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/core/ReactiveMessageConsumerBuilderCustomizer.java

@@ -19,7 +19,17 @@ package org.springframework.pulsar.reactive.core;
 import org.apache.pulsar.reactive.client.api.ReactiveMessageConsumerBuilder;
 
 /**
- * The interface to customize a {@link ReactiveMessageConsumerBuilder}.
+ * Callback interface that can be implemented to customize the
+ * {@link ReactiveMessageConsumerBuilder builder} that is used by the
+ * {@link ReactivePulsarConsumerFactory} to create consumers.
+ * <p>
+ * When using Spring Boot autoconfiguration, any beans implementing this interface will be
+ * used as default configuration by the {@link DefaultReactivePulsarConsumerFactory} and
+ * as such will apply to all created consumers.
+ * <p>
+ * The consumer factory also supports passing in a specific instance of this callback when
+ * {@link ReactivePulsarConsumerFactory#createConsumer creating a consumer} and as such
+ * the passed in customizer only applies to the single created consumer.
  *
  * @param <T> The message payload type
  * @author Christophe Bornet
@@ -28,7 +38,7 @@ import org.apache.pulsar.reactive.client.api.ReactiveMessageConsumerBuilder;
 public interface ReactiveMessageConsumerBuilderCustomizer<T> {
 
 	/**
-	 * Customizes a {@link ReactiveMessageConsumerBuilder}.
+	 * Customize the {@link ReactiveMessageConsumerBuilder}.
 	 * @param reactiveMessageConsumerBuilder the builder to customize
 	 */
 	void customize(ReactiveMessageConsumerBuilder<T> reactiveMessageConsumerBuilder);

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

@@ -69,8 +69,8 @@ import org.springframework.pulsar.reactive.config.ReactivePulsarListenerContaine
 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.ReactiveMessageConsumerBuilderCustomizer;
 import org.springframework.pulsar.reactive.core.ReactivePulsarConsumerFactory;
 import org.springframework.pulsar.reactive.listener.ReactivePulsarListenerTests.SchemaCustomMappingsTestCases.SchemaCustomMappingsTestConfig.User2;
 import org.springframework.pulsar.support.PulsarHeaders;
@@ -145,7 +145,7 @@ public class ReactivePulsarListenerTests implements PulsarTestContainerSupport {
 		}
 
 		@Bean
-		ReactiveMessageConsumerBuilderCustomizer<?> subscriptionInitialPositionEarliest() {
+		ReactivePulsarListenerMessageConsumerBuilderCustomizer<?> subscriptionInitialPositionEarliest() {
 			return b -> b.subscriptionInitialPosition(SubscriptionInitialPosition.Earliest);
 		}
 
@@ -207,7 +207,7 @@ public class ReactivePulsarListenerTests implements PulsarTestContainerSupport {
 			}
 
 			@Bean
-			ReactiveMessageConsumerBuilderCustomizer<String> listen1Customizer() {
+			ReactivePulsarListenerMessageConsumerBuilderCustomizer<String> listen1Customizer() {
 				return b -> b.subscriptionInitialPosition(SubscriptionInitialPosition.Earliest);
 			}
 
@@ -218,7 +218,7 @@ public class ReactivePulsarListenerTests implements PulsarTestContainerSupport {
 			}
 
 			@Bean
-			ReactiveMessageConsumerBuilderCustomizer<String> listen2Customizer() {
+			ReactivePulsarListenerMessageConsumerBuilderCustomizer<String> listen2Customizer() {
 				return b -> b.topics(List.of("topic-2"))
 					.subscriptionInitialPosition(SubscriptionInitialPosition.Earliest);
 			}
@@ -231,7 +231,7 @@ public class ReactivePulsarListenerTests implements PulsarTestContainerSupport {
 			}
 
 			@Bean
-			ReactiveMessageConsumerBuilderCustomizer<String> listen3Customizer() {
+			ReactivePulsarListenerMessageConsumerBuilderCustomizer<String> listen3Customizer() {
 				return b -> b.topicsPatternAutoDiscoveryPeriod(Duration.ofSeconds(5))
 					.subscriptionInitialPosition(SubscriptionInitialPosition.Earliest);
 			}
@@ -331,7 +331,7 @@ public class ReactivePulsarListenerTests implements PulsarTestContainerSupport {
 			}
 
 			@Bean
-			ReactiveMessageConsumerBuilderCustomizer<String> consumerCustomizer() {
+			ReactivePulsarListenerMessageConsumerBuilderCustomizer<String> consumerCustomizer() {
 				return b -> b.negativeAckRedeliveryDelay(Duration.ofSeconds(1))
 					.subscriptionInitialPosition(SubscriptionInitialPosition.Earliest);
 			}

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

@@ -28,7 +28,6 @@ import org.apache.pulsar.common.schema.SchemaType;
 import org.springframework.messaging.handler.annotation.MessageMapping;
 import org.springframework.pulsar.config.PulsarListenerContainerFactory;
 import org.springframework.pulsar.config.PulsarListenerEndpointRegistry;
-import org.springframework.pulsar.core.ConsumerBuilderCustomizer;
 import org.springframework.pulsar.listener.AckMode;
 
 /**
@@ -210,8 +209,10 @@ public @interface PulsarListener {
 
 	/**
 	 * The bean name or a 'SpEL' expression that resolves to a
-	 * {@link ConsumerBuilderCustomizer} to use to configure the consumer.
-	 * @return the bean name or empty string to not configure the consumer.
+	 * {@link PulsarListenerConsumerBuilderCustomizer} to use to configure the underlying
+	 * consumer.
+	 * @return the bean name or SpEL expression to the customizer or an empty string to
+	 * not customize the consumer
 	 */
 	String consumerCustomizer() default "";
 

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

@@ -51,7 +51,6 @@ import org.springframework.pulsar.config.PulsarListenerContainerFactory;
 import org.springframework.pulsar.config.PulsarListenerEndpoint;
 import org.springframework.pulsar.config.PulsarListenerEndpointRegistrar;
 import org.springframework.pulsar.config.PulsarListenerEndpointRegistry;
-import org.springframework.pulsar.core.ConsumerBuilderCustomizer;
 import org.springframework.pulsar.listener.PulsarConsumerErrorHandler;
 import org.springframework.util.Assert;
 import org.springframework.util.StringUtils;
@@ -268,18 +267,19 @@ public class PulsarListenerAnnotationBeanPostProcessor<V> extends AbstractPulsar
 		}
 	}
 
-	@SuppressWarnings({ "rawtypes" })
+	@SuppressWarnings({ "rawtypes", "unchecked" })
 	private void resolveConsumerCustomizer(MethodPulsarListenerEndpoint<?> endpoint, PulsarListener pulsarListener) {
 		Object consumerCustomizer = resolveExpression(pulsarListener.consumerCustomizer());
-		if (consumerCustomizer instanceof ConsumerBuilderCustomizer<?>) {
-			endpoint.setConsumerBuilderCustomizer((ConsumerBuilderCustomizer<?>) consumerCustomizer);
+		if (consumerCustomizer instanceof PulsarListenerConsumerBuilderCustomizer customizer) {
+			endpoint.setConsumerBuilderCustomizer(customizer::customize);
 		}
 		else {
-			String consumerCustomizerBeanName = resolveExpressionAsString(pulsarListener.consumerCustomizer(),
+			String customizerBeanName = resolveExpressionAsString(pulsarListener.consumerCustomizer(),
 					"consumerCustomizer");
-			if (StringUtils.hasText(consumerCustomizerBeanName)) {
-				endpoint.setConsumerBuilderCustomizer(
-						this.beanFactory.getBean(consumerCustomizerBeanName, ConsumerBuilderCustomizer.class));
+			if (StringUtils.hasText(customizerBeanName)) {
+				var customizer = this.beanFactory.getBean(customizerBeanName,
+						PulsarListenerConsumerBuilderCustomizer.class);
+				endpoint.setConsumerBuilderCustomizer(customizer::customize);
 			}
 		}
 	}

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

@@ -0,0 +1,45 @@
+/*
+ * 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.annotation;
+
+import org.apache.pulsar.client.api.ConsumerBuilder;
+
+import org.springframework.pulsar.core.ConsumerBuilderCustomizer;
+
+/**
+ * Callback interface that can be implemented by a bean to customize the
+ * {@link ConsumerBuilder} that is used to create the underlying Pulsar consumer used by a
+ * {@link PulsarListener} to receive messages.
+ *
+ * <p>
+ * Unlike the {@link ConsumerBuilderCustomizer} which is applied to all created consumer
+ * builders, this customizer is only applied to the individual consumer builder(s) of the
+ * {@code @PulsarListener(s)} it is associated with.
+ *
+ * @param <T> The message payload type
+ * @author Chris Bono
+ */
+@FunctionalInterface
+public interface PulsarListenerConsumerBuilderCustomizer<T> {
+
+	/**
+	 * Customize the {@link ConsumerBuilder}.
+	 * @param consumerBuilder the builder to customize
+	 */
+	void customize(ConsumerBuilder<T> consumerBuilder);
+
+}

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

@@ -44,8 +44,6 @@ import org.springframework.pulsar.config.PulsarReaderContainerFactory;
 import org.springframework.pulsar.config.PulsarReaderEndpoint;
 import org.springframework.pulsar.config.PulsarReaderEndpointRegistrar;
 import org.springframework.pulsar.config.PulsarReaderEndpointRegistry;
-import org.springframework.pulsar.core.ConsumerBuilderCustomizer;
-import org.springframework.pulsar.core.ReaderBuilderCustomizer;
 import org.springframework.util.Assert;
 import org.springframework.util.StringUtils;
 
@@ -232,17 +230,18 @@ public class PulsarReaderAnnotationBeanPostProcessor<V> extends AbstractPulsarAn
 		resolveReaderCustomizer(endpoint, pulsarReader);
 	}
 
+	@SuppressWarnings({ "rawtypes", "unchecked" })
 	private void resolveReaderCustomizer(MethodPulsarReaderEndpoint<?> endpoint, PulsarReader pulsarReader) {
 		Object readerCustomizer = resolveExpression(pulsarReader.readerCustomizer());
-		if (readerCustomizer instanceof ConsumerBuilderCustomizer<?>) {
-			endpoint.setReaderBuilderCustomizer((ReaderBuilderCustomizer<?>) readerCustomizer);
+		if (readerCustomizer instanceof PulsarReaderReaderBuilderCustomizer customizer) {
+			endpoint.setReaderBuilderCustomizer(customizer::customize);
 		}
 		else {
-			String readerCustomizerBeanName = resolveExpressionAsString(pulsarReader.readerCustomizer(),
-					"readerCustomizer");
-			if (StringUtils.hasText(readerCustomizerBeanName)) {
-				endpoint.setReaderBuilderCustomizer(
-						this.beanFactory.getBean(readerCustomizerBeanName, ReaderBuilderCustomizer.class));
+			String customizerBeanName = resolveExpressionAsString(pulsarReader.readerCustomizer(), "readerCustomizer");
+			if (StringUtils.hasText(customizerBeanName)) {
+				var customizer = this.beanFactory.getBean(customizerBeanName,
+						PulsarReaderReaderBuilderCustomizer.class);
+				endpoint.setReaderBuilderCustomizer(customizer::customize);
 			}
 		}
 	}

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

@@ -0,0 +1,45 @@
+/*
+ * 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.annotation;
+
+import org.apache.pulsar.client.api.ReaderBuilder;
+
+import org.springframework.pulsar.core.ReaderBuilderCustomizer;
+
+/**
+ * Callback interface that can be implemented by a bean to customize the
+ * {@link ReaderBuilder} that is used to create the underlying Pulsar reader used by a
+ * {@link PulsarReader @PulsarReader} to receive messages.
+ *
+ * <p>
+ * Unlike the {@link ReaderBuilderCustomizer} which is applied to all created reader
+ * builders, this customizer is only applied to the individual reader builder(s) of the
+ * {@code @PulsarReader(s)} it is associated with.
+ *
+ * @param <T> The message payload type
+ * @author Chris Bono
+ */
+@FunctionalInterface
+public interface PulsarReaderReaderBuilderCustomizer<T> {
+
+	/**
+	 * Customize the {@link ReaderBuilder}.
+	 * @param readerBuilder the builder to customize
+	 */
+	void customize(ReaderBuilder<T> readerBuilder);
+
+}

spring-pulsar/src/main/java/org/springframework/pulsar/core/ConsumerBuilderCustomizer.java

@@ -19,7 +19,16 @@ package org.springframework.pulsar.core;
 import org.apache.pulsar.client.api.ConsumerBuilder;
 
 /**
- * The interface to customize a {@link ConsumerBuilder}.
+ * Callback interface that can be implemented to customize the {@link ConsumerBuilder}
+ * that is used by the {@link PulsarConsumerFactory} to create consumers.
+ * <p>
+ * When using Spring Boot autoconfiguration, any beans implementing this interface will be
+ * used as default configuration by the {@link DefaultPulsarConsumerFactory} and as such
+ * will apply to all created consumers.
+ * <p>
+ * The consumer factory also supports passing in a specific instance of this callback when
+ * {@link PulsarConsumerFactory#createConsumer creating a consumer} and as such the passed
+ * in customizer only applies to the single created consumer.
  *
  * @param <T> The message payload type
  * @author Christophe Bornet
@@ -28,7 +37,7 @@ import org.apache.pulsar.client.api.ConsumerBuilder;
 public interface ConsumerBuilderCustomizer<T> {
 
 	/**
-	 * Customizes a {@link ConsumerBuilder}.
+	 * Customize the {@link ConsumerBuilder}.
 	 * @param consumerBuilder the builder to customize
 	 */
 	void customize(ConsumerBuilder<T> consumerBuilder);

spring-pulsar/src/main/java/org/springframework/pulsar/core/ReaderBuilderCustomizer.java

@@ -19,7 +19,16 @@ package org.springframework.pulsar.core;
 import org.apache.pulsar.client.api.ReaderBuilder;
 
 /**
- * The interface to customize a {@link ReaderBuilder}.
+ * Callback interface that can be implemented to customize the {@link ReaderBuilder} that
+ * is used by the {@link PulsarReaderFactory} to create readers.
+ * <p>
+ * When using Spring Boot autoconfiguration, any beans implementing this interface will be
+ * used as default configuration by the {@link DefaultPulsarReaderFactory} and as such
+ * will apply to all created readers.
+ * <p>
+ * The reader factory also supports passing in a specific instance of this callback when
+ * {@link PulsarReaderFactory#createReader creating a reader} and as such the passed in
+ * customizer only applies to the single created reader.
  *
  * @param <T> The message payload type
  * @author Soby Chacko

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

@@ -57,10 +57,10 @@ import org.springframework.context.annotation.Configuration;
 import org.springframework.messaging.handler.annotation.Header;
 import org.springframework.pulsar.annotation.EnablePulsar;
 import org.springframework.pulsar.annotation.PulsarListener;
+import org.springframework.pulsar.annotation.PulsarListenerConsumerBuilderCustomizer;
 import org.springframework.pulsar.config.ConcurrentPulsarListenerContainerFactory;
 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;
@@ -1095,7 +1095,7 @@ public class PulsarListenerTests implements PulsarTestContainerSupport {
 			}
 
 			@Bean
-			public ConsumerBuilderCustomizer<String> myCustomizer() {
+			public PulsarListenerConsumerBuilderCustomizer<String> myCustomizer() {
 				return cb -> cb.subscriptionName("test-changed-subscription-name");
 			}
 

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

@@ -35,6 +35,7 @@ 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;
@@ -43,7 +44,6 @@ 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.core.ReaderBuilderCustomizer;
 import org.springframework.pulsar.test.support.PulsarTestContainerSupport;
 import org.springframework.test.annotation.DirtiesContext;
 import org.springframework.test.context.ContextConfiguration;
@@ -204,7 +204,7 @@ public class PulsarReaderTests implements PulsarTestContainerSupport {
 			}
 
 			@Bean
-			public ReaderBuilderCustomizer<String> myCustomizer(PulsarTemplate<String> pulsarTemplate) {
+			public PulsarReaderReaderBuilderCustomizer<String> myCustomizer(PulsarTemplate<String> pulsarTemplate) {
 				return cb -> {
 					for (int i = 0; i < 10; i++) {
 						try {