DefaultPulsarProducerFactory accepts multiple customizers (#434)

See #432

Author: onobc

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

@@ -67,16 +67,17 @@ public class CachingPulsarProducerFactory<T> extends DefaultPulsarProducerFactor
 	 * configuration.
 	 * @param pulsarClient the client used to create the producers
 	 * @param defaultTopic the default topic to use for the producers
-	 * @param defaultConfigCustomizer the default configuration to apply to the producers
+	 * @param defaultConfigCustomizers the optional list of customizers to apply to the
+	 * created producers
 	 * @param topicResolver the topic resolver to use
 	 * @param cacheExpireAfterAccess time period to expire unused entries in the cache
 	 * @param cacheMaximumSize maximum size of cache (entries)
 	 * @param cacheInitialCapacity the initial size of cache
 	 */
 	public CachingPulsarProducerFactory(PulsarClient pulsarClient, @Nullable String defaultTopic,
-			ProducerBuilderCustomizer<T> defaultConfigCustomizer, TopicResolver topicResolver,
+			List<ProducerBuilderCustomizer<T>> defaultConfigCustomizers, TopicResolver topicResolver,
 			Duration cacheExpireAfterAccess, Long cacheMaximumSize, Integer cacheInitialCapacity) {
-		super(pulsarClient, defaultTopic, defaultConfigCustomizer, topicResolver);
+		super(pulsarClient, defaultTopic, defaultConfigCustomizers, topicResolver);
 		var cacheFactory = CacheProviderFactory.<ProducerCacheKey<T>, Producer<T>>load();
 		this.producerCache = cacheFactory.create(cacheExpireAfterAccess, cacheMaximumSize, cacheInitialCapacity,
 				(key, producer, cause) -> {

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

@@ -52,7 +52,7 @@ public class DefaultPulsarProducerFactory<T> implements PulsarProducerFactory<T>
 	private final String defaultTopic;
 
 	@Nullable
-	private final ProducerBuilderCustomizer<T> defaultConfigCustomizer;
+	private final List<ProducerBuilderCustomizer<T>> defaultConfigCustomizers;
 
 	private final TopicResolver topicResolver;
 
@@ -61,8 +61,7 @@ public class DefaultPulsarProducerFactory<T> implements PulsarProducerFactory<T>
 	 * @param pulsarClient the client used to create the producers
 	 */
 	public DefaultPulsarProducerFactory(PulsarClient pulsarClient) {
-		this(pulsarClient, null, (pb) -> {
-		}, new DefaultTopicResolver());
+		this(pulsarClient, null, null, new DefaultTopicResolver());
 	}
 
 	/**
@@ -71,33 +70,34 @@ public DefaultPulsarProducerFactory(PulsarClient pulsarClient) {
 	 * @param defaultTopic the default topic to use for the producers
 	 */
 	public DefaultPulsarProducerFactory(PulsarClient pulsarClient, @Nullable String defaultTopic) {
-		this(pulsarClient, defaultTopic, (pb) -> {
-		}, new DefaultTopicResolver());
+		this(pulsarClient, defaultTopic, null, new DefaultTopicResolver());
 	}
 
 	/**
 	 * Construct a producer factory that uses a default topic resolver.
 	 * @param pulsarClient the client used to create the producers
 	 * @param defaultTopic the default topic to use for the producers
-	 * @param defaultConfigCustomizer the default configuration to apply to the producers
+	 * @param defaultConfigCustomizers the optional list of customizers to apply to the
+	 * created producers
 	 */
 	public DefaultPulsarProducerFactory(PulsarClient pulsarClient, @Nullable String defaultTopic,
-			@Nullable ProducerBuilderCustomizer<T> defaultConfigCustomizer) {
-		this(pulsarClient, defaultTopic, defaultConfigCustomizer, new DefaultTopicResolver());
+			@Nullable List<ProducerBuilderCustomizer<T>> defaultConfigCustomizers) {
+		this(pulsarClient, defaultTopic, defaultConfigCustomizers, new DefaultTopicResolver());
 	}
 
 	/**
 	 * Construct a producer factory that uses the specified parameters.
 	 * @param pulsarClient the client used to create the producers
 	 * @param defaultTopic the default topic to use for the producers
-	 * @param defaultConfigCustomizer the default configuration to apply to the producers
+	 * @param defaultConfigCustomizers the optional list of customizers to apply to the
+	 * created producers
 	 * @param topicResolver the topic resolver to use
 	 */
 	public DefaultPulsarProducerFactory(PulsarClient pulsarClient, @Nullable String defaultTopic,
-			@Nullable ProducerBuilderCustomizer<T> defaultConfigCustomizer, TopicResolver topicResolver) {
+			@Nullable List<ProducerBuilderCustomizer<T>> defaultConfigCustomizers, TopicResolver topicResolver) {
 		this.pulsarClient = pulsarClient;
 		this.defaultTopic = defaultTopic;
-		this.defaultConfigCustomizer = defaultConfigCustomizer;
+		this.defaultConfigCustomizers = defaultConfigCustomizers;
 		this.topicResolver = topicResolver;
 	}
 
@@ -142,8 +142,8 @@ protected Producer<T> doCreateProducer(Schema<T> schema, @Nullable String topic,
 		var producerBuilder = this.pulsarClient.newProducer(schema);
 
 		// Apply the default config customizer (preserve the topic)
-		if (this.defaultConfigCustomizer != null) {
-			this.defaultConfigCustomizer.customize(producerBuilder);
+		if (this.defaultConfigCustomizers != null) {
+			this.defaultConfigCustomizers.forEach((customizer) -> customizer.customize(producerBuilder));
 		}
 		producerBuilder.topic(resolvedTopic);
 

spring-pulsar/src/test/java/org/springframework/pulsar/core/CachingPulsarProducerFactoryTests.java

@@ -228,9 +228,9 @@ protected Producer<String> actualProducer(Producer<String> wrappedProducer) {
 
 	@Override
 	protected CachingPulsarProducerFactory<String> producerFactory(PulsarClient pulsarClient,
-			@Nullable String defaultTopic, @Nullable ProducerBuilderCustomizer<String> defaultConfigCustomizer) {
+			@Nullable String defaultTopic, @Nullable List<ProducerBuilderCustomizer<String>> defaultConfigCustomizers) {
 		var producerFactory = new CachingPulsarProducerFactory<String>(pulsarClient, defaultTopic,
-				defaultConfigCustomizer, new DefaultTopicResolver(), Duration.ofMinutes(5L), 30L, 2);
+				defaultConfigCustomizers, new DefaultTopicResolver(), Duration.ofMinutes(5L), 30L, 2);
 		producerFactories.add(producerFactory);
 		return producerFactory;
 	}

spring-pulsar/src/test/java/org/springframework/pulsar/core/DefaultPulsarProducerFactoryTests.java

@@ -17,11 +17,19 @@
 package org.springframework.pulsar.core;
 
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.inOrder;
+import static org.mockito.Mockito.mock;
+
+import java.util.List;
 
 import org.apache.pulsar.client.api.Producer;
+import org.apache.pulsar.client.api.ProducerBuilder;
 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.mockito.InOrder;
 
 import org.springframework.lang.Nullable;
 
@@ -46,8 +54,42 @@ void createProducerMultipleTimeDoesNotCacheProducer() throws PulsarClientExcepti
 
 	@Override
 	protected PulsarProducerFactory<String> producerFactory(PulsarClient pulsarClient, @Nullable String defaultTopic,
-			@Nullable ProducerBuilderCustomizer<String> defaultConfigCustomizer) {
-		return new DefaultPulsarProducerFactory<>(pulsarClient, defaultTopic, defaultConfigCustomizer);
+			@Nullable List<ProducerBuilderCustomizer<String>> defaultConfigCustomizers) {
+		return new DefaultPulsarProducerFactory<>(pulsarClient, defaultTopic, defaultConfigCustomizers);
+	}
+
+	@Nested
+	@SuppressWarnings("unchecked")
+	class DefaultConfigCustomizerApi {
+
+		private ProducerBuilderCustomizer<String> configCustomizer1 = mock(ProducerBuilderCustomizer.class);
+
+		private ProducerBuilderCustomizer<String> configCustomizer2 = mock(ProducerBuilderCustomizer.class);
+
+		private ProducerBuilderCustomizer<String> createProducerCustomizer = mock(ProducerBuilderCustomizer.class);
+
+		@Test
+		void singleConfigCustomizer() throws PulsarClientException {
+			try (var ignored = newProducerFactoryWithDefaultConfigCustomizers(List.of(configCustomizer1))
+				.createProducer(schema, "topic0", createProducerCustomizer)) {
+				InOrder inOrder = inOrder(configCustomizer1, createProducerCustomizer);
+				inOrder.verify(configCustomizer1).customize(any(ProducerBuilder.class));
+				inOrder.verify(createProducerCustomizer).customize(any(ProducerBuilder.class));
+			}
+		}
+
+		@Test
+		void multipleConfigCustomizers() throws PulsarClientException {
+			try (var ignored = newProducerFactoryWithDefaultConfigCustomizers(
+					List.of(configCustomizer2, configCustomizer1))
+				.createProducer(schema, "topic0", createProducerCustomizer)) {
+				InOrder inOrder = inOrder(configCustomizer1, configCustomizer2, createProducerCustomizer);
+				inOrder.verify(configCustomizer2).customize(any(ProducerBuilder.class));
+				inOrder.verify(configCustomizer1).customize(any(ProducerBuilder.class));
+				inOrder.verify(createProducerCustomizer).customize(any(ProducerBuilder.class));
+			}
+		}
+
 	}
 
 }

spring-pulsar/src/test/java/org/springframework/pulsar/core/FailoverConsumerTests.java

@@ -19,6 +19,7 @@
 import static org.assertj.core.api.Assertions.assertThat;
 
 import java.io.Serial;
+import java.util.List;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
 
@@ -85,7 +86,7 @@ void testFailOverConsumersOnPartitionedTopic() throws Exception {
 		container3.start();
 
 		DefaultPulsarProducerFactory<String> pulsarProducerFactory = new DefaultPulsarProducerFactory<>(pulsarClient,
-				"my-part-topic-1", (pb) -> pb.messageRoutingMode(MessageRoutingMode.CustomPartition));
+				"my-part-topic-1", List.of((pb) -> pb.messageRoutingMode(MessageRoutingMode.CustomPartition)));
 		PulsarTemplate<String> pulsarTemplate = new PulsarTemplate<>(pulsarProducerFactory);
 
 		pulsarTemplate.newMessage("hello john doe")

spring-pulsar/src/test/java/org/springframework/pulsar/core/PulsarProducerFactoryTests.java

@@ -26,6 +26,7 @@
 
 import java.util.Arrays;
 import java.util.Collections;
+import java.util.List;
 import java.util.Set;
 
 import org.apache.pulsar.client.api.Producer;
@@ -100,7 +101,12 @@ protected PulsarProducerFactory<String> newProducerFactoryWithDefaultTopic(Strin
 	}
 
 	private PulsarProducerFactory<String> newProducerFactoryWithDefaultKeys(Set<String> defaultKeys) {
-		return producerFactory(pulsarClient, null, (pb) -> defaultKeys.forEach(pb::addEncryptionKey));
+		return producerFactory(pulsarClient, null, List.of((pb) -> defaultKeys.forEach(pb::addEncryptionKey)));
+	}
+
+	protected PulsarProducerFactory<String> newProducerFactoryWithDefaultConfigCustomizers(
+			List<ProducerBuilderCustomizer<String>> customizers) {
+		return producerFactory(pulsarClient, null, customizers);
 	}
 
 	/**
@@ -116,11 +122,12 @@ protected Producer<String> actualProducer(Producer<String> producer) {
 	 * Subclasses override to provide concrete {@link PulsarProducerFactory} instance.
 	 * @param pulsarClient the Pulsar client
 	 * @param defaultTopic the default topic to use for the producers
-	 * @param defaultConfigCustomizer the default configuration to apply to the producers
+	 * @param defaultConfigCustomizers the optional list of customizers to apply to the
+	 * created producers
 	 * @return a Pulsar producer factory instance to use for the tests
 	 */
 	protected abstract PulsarProducerFactory<String> producerFactory(PulsarClient pulsarClient,
-			@Nullable String defaultTopic, @Nullable ProducerBuilderCustomizer<String> defaultConfigCustomizer);
+			@Nullable String defaultTopic, @Nullable List<ProducerBuilderCustomizer<String>> defaultConfigCustomizers);
 
 	@Test
 	@SuppressWarnings("unchecked")

spring-pulsar/src/test/java/org/springframework/pulsar/core/SharedSubscriptionConsumerTests.java

@@ -19,6 +19,7 @@
 import static org.assertj.core.api.Assertions.assertThat;
 
 import java.util.HashMap;
+import java.util.List;
 import java.util.Map;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
@@ -133,7 +134,7 @@ void keySharedSubscriptionWithDefaultAutoSplitHashingRange() throws Exception {
 			Thread.sleep(5_000);
 
 			DefaultPulsarProducerFactory<String> producerFactory = new DefaultPulsarProducerFactory<>(pulsarClient,
-					"key-shared-batch-disabled-topic", (pb) -> pb.enableBatching(false));
+					"key-shared-batch-disabled-topic", List.of((pb) -> pb.enableBatching(false)));
 			PulsarTemplate<String> pulsarTemplate = new PulsarTemplate<>(producerFactory);
 			for (int i = 0; i < 10; i++) {
 				pulsarTemplate.newMessage("alice-" + i)

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

@@ -177,7 +177,7 @@ void pulsarListenerProvidedConsumerProperties(@Autowired PulsarListenerEndpointR
 		void concurrencyOnPulsarListenerWithFailoverSubscription(@Autowired PulsarListenerEndpointRegistry registry)
 				throws Exception {
 			var pulsarProducerFactory = new DefaultPulsarProducerFactory<String>(pulsarClient, null,
-					(pb) -> pb.enableBatching(false));
+					List.of((pb) -> pb.enableBatching(false)));
 			var customTemplate = new PulsarTemplate<>(pulsarProducerFactory);
 			var bar = (ConcurrentPulsarMessageListenerContainer<?>) registry.getListenerContainer("bar");
 
@@ -194,7 +194,7 @@ void concurrencyOnPulsarListenerWithFailoverSubscription(@Autowired PulsarListen
 		void nonDefaultConcurrencySettingNotAllowedOnExclusiveSubscriptions(
 				@Autowired PulsarListenerEndpointRegistry registry) throws Exception {
 			var pulsarProducerFactory = new DefaultPulsarProducerFactory<String>(pulsarClient, null,
-					(pb) -> pb.enableBatching(false));
+					List.of((pb) -> pb.enableBatching(false)));
 			var customTemplate = new PulsarTemplate<>(pulsarProducerFactory);
 			var bar = (ConcurrentPulsarMessageListenerContainer<?>) registry.getListenerContainer("bar");
 

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

@@ -87,7 +87,7 @@ void basicDefaultReader() throws Exception {
 			container.start();
 
 			DefaultPulsarProducerFactory<String> pulsarProducerFactory = new DefaultPulsarProducerFactory<>(
-					pulsarClient, "dprlct-001", (pb) -> pb.topic("dprlct-001"));
+					pulsarClient, "dprlct-001", List.of((pb) -> pb.topic("dprlct-001")));
 			PulsarTemplate<String> pulsarTemplate = new PulsarTemplate<>(pulsarProducerFactory);
 			pulsarTemplate.sendAsync("hello john doe");
 			assertThat(latch.await(10, TimeUnit.SECONDS)).isTrue();
@@ -115,7 +115,7 @@ void topicProvidedThroughContainerProperties() throws Exception {
 			container = new DefaultPulsarMessageReaderContainer<>(pulsarReaderFactory, containerProps);
 			container.start();
 			DefaultPulsarProducerFactory<String> pulsarProducerFactory = new DefaultPulsarProducerFactory<>(
-					pulsarClient, "dprlct-002", (pb) -> pb.topic("dprlct-002"));
+					pulsarClient, "dprlct-002", List.of((pb) -> pb.topic("dprlct-002")));
 			PulsarTemplate<String> pulsarTemplate = new PulsarTemplate<>(pulsarProducerFactory);
 			pulsarTemplate.sendAsync("hello buzz doe");
 			assertThat(latch.await(10, TimeUnit.SECONDS)).isTrue();
@@ -144,7 +144,7 @@ void latestMessageId() throws Exception {
 
 			var prodConfig = Map.<String, Object>of("topicName", "dprlct-003");
 			var producerFactory = new DefaultPulsarProducerFactory<>(pulsarClient, "dprlct-003",
-					(pb) -> pb.topic("dprlct-003"));
+					List.of((pb) -> pb.topic("dprlct-003")));
 			var pulsarTemplate = new PulsarTemplate<>(producerFactory);
 
 			// The following sends will not be received by the reader as we are using the