Use builder to autoconfigure PulsarProducerFactory (#397)

- Apply PulsarProperties to producer builder rather than raw
map of Pulsar properties

- Move producer props 'toMap' into PulsarBinderUtils

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

@@ -20,8 +20,6 @@ import static org.assertj.core.api.Assertions.assertThat;
 
 import java.nio.charset.StandardCharsets;
 import java.time.Duration;
-import java.util.Collections;
-import java.util.HashMap;
 import java.util.List;
 import java.util.Objects;
 import java.util.concurrent.BlockingQueue;
@@ -107,7 +105,7 @@ public class ReactivePulsarListenerTests implements PulsarTestContainerSupport {
 
 		@Bean
 		public PulsarProducerFactory<String> pulsarProducerFactory(PulsarClient pulsarClient) {
-			return new DefaultPulsarProducerFactory<>(pulsarClient, new HashMap<>());
+			return new DefaultPulsarProducerFactory<>(pulsarClient);
 		}
 
 		@Bean
@@ -347,8 +345,7 @@ public class ReactivePulsarListenerTests implements PulsarTestContainerSupport {
 
 		@Test
 		void jsonSchema() throws Exception {
-			PulsarProducerFactory<User> pulsarProducerFactory = new DefaultPulsarProducerFactory<>(pulsarClient,
-					Collections.emptyMap());
+			PulsarProducerFactory<User> pulsarProducerFactory = new DefaultPulsarProducerFactory<>(pulsarClient);
 			PulsarTemplate<User> template = new PulsarTemplate<>(pulsarProducerFactory);
 			for (int i = 0; i < 3; i++) {
 				template.send("json-topic", new User("Jason", i), JSONSchema.of(User.class));
@@ -358,8 +355,7 @@ public class ReactivePulsarListenerTests implements PulsarTestContainerSupport {
 
 		@Test
 		void avroSchema() throws Exception {
-			PulsarProducerFactory<User> pulsarProducerFactory = new DefaultPulsarProducerFactory<>(pulsarClient,
-					Collections.emptyMap());
+			PulsarProducerFactory<User> pulsarProducerFactory = new DefaultPulsarProducerFactory<>(pulsarClient);
 			PulsarTemplate<User> template = new PulsarTemplate<>(pulsarProducerFactory);
 			for (int i = 0; i < 3; i++) {
 				template.send("avro-topic", new User("Avi", i), AvroSchema.of(User.class));
@@ -370,7 +366,7 @@ public class ReactivePulsarListenerTests implements PulsarTestContainerSupport {
 		@Test
 		void keyvalueSchema() throws Exception {
 			PulsarProducerFactory<KeyValue<String, Integer>> pulsarProducerFactory = new DefaultPulsarProducerFactory<>(
-					pulsarClient, Collections.emptyMap());
+					pulsarClient);
 			PulsarTemplate<KeyValue<String, Integer>> template = new PulsarTemplate<>(pulsarProducerFactory);
 			Schema<KeyValue<String, Integer>> kvSchema = Schema.KeyValue(Schema.STRING, Schema.INT32,
 					KeyValueEncodingType.INLINE);
@@ -382,8 +378,8 @@ public class ReactivePulsarListenerTests implements PulsarTestContainerSupport {
 
 		@Test
 		void protobufSchema() throws Exception {
-			PulsarProducerFactory<Proto.Person> pulsarProducerFactory = new DefaultPulsarProducerFactory<>(pulsarClient,
-					Collections.emptyMap());
+			PulsarProducerFactory<Proto.Person> pulsarProducerFactory = new DefaultPulsarProducerFactory<>(
+					pulsarClient);
 			PulsarTemplate<Proto.Person> template = new PulsarTemplate<>(pulsarProducerFactory);
 			for (int i = 0; i < 3; i++) {
 				template.send("protobuf-topic", Proto.Person.newBuilder().setId(i).setName("Paul").build(),
@@ -494,8 +490,7 @@ public class ReactivePulsarListenerTests implements PulsarTestContainerSupport {
 
 		@Test
 		void jsonSchema() throws Exception {
-			PulsarProducerFactory<User2> pulsarProducerFactory = new DefaultPulsarProducerFactory<>(pulsarClient,
-					Collections.emptyMap());
+			PulsarProducerFactory<User2> pulsarProducerFactory = new DefaultPulsarProducerFactory<>(pulsarClient);
 			PulsarTemplate<User2> template = new PulsarTemplate<>(pulsarProducerFactory);
 			for (int i = 0; i < 3; i++) {
 				template.send("json-custom-schema-topic", new User2("Jason", i), JSONSchema.of(User2.class));
@@ -505,8 +500,7 @@ public class ReactivePulsarListenerTests implements PulsarTestContainerSupport {
 
 		@Test
 		void avroSchema() throws Exception {
-			PulsarProducerFactory<User> pulsarProducerFactory = new DefaultPulsarProducerFactory<>(pulsarClient,
-					Collections.emptyMap());
+			PulsarProducerFactory<User> pulsarProducerFactory = new DefaultPulsarProducerFactory<>(pulsarClient);
 			PulsarTemplate<User> template = new PulsarTemplate<>(pulsarProducerFactory);
 			for (int i = 0; i < 3; i++) {
 				template.send("avro-custom-schema-topic", new User("Avi", i), AvroSchema.of(User.class));
@@ -517,7 +511,7 @@ public class ReactivePulsarListenerTests implements PulsarTestContainerSupport {
 		@Test
 		void keyvalueSchema() throws Exception {
 			PulsarProducerFactory<KeyValue<String, User2>> pulsarProducerFactory = new DefaultPulsarProducerFactory<>(
-					pulsarClient, Collections.emptyMap());
+					pulsarClient);
 			PulsarTemplate<KeyValue<String, User2>> template = new PulsarTemplate<>(pulsarProducerFactory);
 			Schema<KeyValue<String, User2>> kvSchema = Schema.KeyValue(Schema.STRING, Schema.JSON(User2.class),
 					KeyValueEncodingType.INLINE);
@@ -530,8 +524,8 @@ public class ReactivePulsarListenerTests implements PulsarTestContainerSupport {
 
 		@Test
 		void protobufSchema() throws Exception {
-			PulsarProducerFactory<Proto.Person> pulsarProducerFactory = new DefaultPulsarProducerFactory<>(pulsarClient,
-					Collections.emptyMap());
+			PulsarProducerFactory<Proto.Person> pulsarProducerFactory = new DefaultPulsarProducerFactory<>(
+					pulsarClient);
 			PulsarTemplate<Proto.Person> template = new PulsarTemplate<>(pulsarProducerFactory);
 			for (int i = 0; i < 3; i++) {
 				template.send("protobuf-custom-schema-topic",
@@ -664,8 +658,7 @@ public class ReactivePulsarListenerTests implements PulsarTestContainerSupport {
 
 		@Test
 		void complexMessageTypeTopicMapping() throws Exception {
-			PulsarProducerFactory<User2> pulsarProducerFactory = new DefaultPulsarProducerFactory<>(pulsarClient,
-					Collections.emptyMap());
+			PulsarProducerFactory<User2> pulsarProducerFactory = new DefaultPulsarProducerFactory<>(pulsarClient);
 			PulsarTemplate<User2> template = new PulsarTemplate<>(pulsarProducerFactory);
 			Schema<User2> schema = Schema.JSON(User2.class);
 			for (int i = 0; i < 3; i++) {
@@ -676,8 +669,7 @@ public class ReactivePulsarListenerTests implements PulsarTestContainerSupport {
 
 		@Test
 		void primitiveMessageTypeTopicMapping() throws Exception {
-			PulsarProducerFactory<String> pulsarProducerFactory = new DefaultPulsarProducerFactory<>(pulsarClient,
-					Collections.emptyMap());
+			PulsarProducerFactory<String> pulsarProducerFactory = new DefaultPulsarProducerFactory<>(pulsarClient);
 			PulsarTemplate<String> template = new PulsarTemplate<>(pulsarProducerFactory);
 			for (int i = 0; i < 3; i++) {
 				template.send("rplt-topicMapping-string-topic", "Susan " + i, Schema.STRING);

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

@@ -21,6 +21,7 @@ import java.util.HashMap;
 import java.util.HashSet;
 import java.util.Map;
 import java.util.Set;
+import java.util.concurrent.TimeUnit;
 
 import org.apache.pulsar.client.api.CompressionType;
 import org.apache.pulsar.client.api.HashingScheme;
@@ -31,7 +32,7 @@ import org.apache.pulsar.client.api.ProducerCryptoFailureAction;
 import org.springframework.boot.context.properties.PropertyMapper;
 import org.springframework.lang.Nullable;
 import org.springframework.pulsar.autoconfigure.PulsarProperties.Cache;
-import org.springframework.pulsar.autoconfigure.PulsarProperties.Properties;
+import org.springframework.pulsar.core.ProducerBuilderCustomizer;
 import org.springframework.util.unit.DataSize;
 
 /**
@@ -365,40 +366,41 @@ public class ProducerConfigProperties {
 		return this.cache;
 	}
 
-	public Map<String, Object> buildProperties() {
-		PulsarProperties.Properties properties = new Properties();
-
-		PropertyMapper map = PropertyMapper.get().alwaysApplyingWhenNonNull();
-
-		map.from(this::getTopicName).to(properties.in("topicName"));
-		map.from(this::getProducerName).to(properties.in("producerName"));
-		map.from(this::getSendTimeout).asInt(Duration::toMillis).to(properties.in("sendTimeoutMs"));
-		map.from(this::getBlockIfQueueFull).to(properties.in("blockIfQueueFull"));
-		map.from(this::getMaxPendingMessages).to(properties.in("maxPendingMessages"));
-		map.from(this::getMaxPendingMessagesAcrossPartitions).to(properties.in("maxPendingMessagesAcrossPartitions"));
-		map.from(this::getMessageRoutingMode).to(properties.in("messageRoutingMode"));
-		map.from(this::getHashingScheme).to(properties.in("hashingScheme"));
-		map.from(this::getCryptoFailureAction).to(properties.in("cryptoFailureAction"));
-		map.from(this::getBatchingMaxPublishDelay).as(it -> it.toNanos() / 1000)
-				.to(properties.in("batchingMaxPublishDelayMicros"));
-		map.from(this::getBatchingPartitionSwitchFrequencyByPublishDelay)
-				.to(properties.in("batchingPartitionSwitchFrequencyByPublishDelay"));
-		map.from(this::getBatchingMaxMessages).to(properties.in("batchingMaxMessages"));
-		map.from(this::getBatchingMaxBytes).asInt(DataSize::toBytes).to(properties.in("batchingMaxBytes"));
-		map.from(this::getBatchingEnabled).to(properties.in("batchingEnabled"));
-		map.from(this::getChunkingEnabled).to(properties.in("chunkingEnabled"));
-		map.from(this::getEncryptionKeys).to(properties.in("encryptionKeys"));
-		map.from(this::getCompressionType).to(properties.in("compressionType"));
-		map.from(this::getInitialSequenceId).to(properties.in("initialSequenceId"));
-		map.from(this::getAutoUpdatePartitions).to(properties.in("autoUpdatePartitions"));
-		map.from(this::getAutoUpdatePartitionsInterval).as(Duration::toSeconds)
-				.to(properties.in("autoUpdatePartitionsIntervalSeconds"));
-		map.from(this::getMultiSchema).to(properties.in("multiSchema"));
-		map.from(this::getProducerAccessMode).to(properties.in("accessMode"));
-		map.from(this::getLazyStartPartitionedProducers).to(properties.in("lazyStartPartitionedProducers"));
-		map.from(this::getProperties).to(properties.in("properties"));
-
-		return properties;
+	@SuppressWarnings("deprecation")
+	public ProducerBuilderCustomizer<?> toProducerBuilderCustomizer() {
+		return (producerBuilder) -> {
+			PropertyMapper map = PropertyMapper.get().alwaysApplyingWhenNonNull();
+			map.from(this::getTopicName).to(producerBuilder::topic);
+			map.from(this::getProducerName).to(producerBuilder::producerName);
+			map.from(this::getSendTimeout).asInt(Duration::toMillis).to(producerBuilder,
+					(pb, val) -> pb.sendTimeout(val, TimeUnit.MILLISECONDS));
+			map.from(this::getBlockIfQueueFull).to(producerBuilder::blockIfQueueFull);
+			map.from(this::getMaxPendingMessages).to(producerBuilder::maxPendingMessages);
+			map.from(this::getMaxPendingMessagesAcrossPartitions)
+					.to(producerBuilder::maxPendingMessagesAcrossPartitions);
+			map.from(this::getMessageRoutingMode).to(producerBuilder::messageRoutingMode);
+			map.from(this::getHashingScheme).to(producerBuilder::hashingScheme);
+			map.from(this::getCryptoFailureAction).to(producerBuilder::cryptoFailureAction);
+			map.from(this::getBatchingMaxPublishDelay).as(Duration::toMillis).to(producerBuilder,
+					(pb, val) -> pb.batchingMaxPublishDelay(val, TimeUnit.MILLISECONDS));
+			map.from(this::getBatchingPartitionSwitchFrequencyByPublishDelay)
+					.to(producerBuilder::roundRobinRouterBatchingPartitionSwitchFrequency);
+			map.from(this::getBatchingMaxMessages).to(producerBuilder::batchingMaxMessages);
+			map.from(this::getBatchingMaxBytes).asInt(DataSize::toBytes).to(producerBuilder::batchingMaxBytes);
+			map.from(this::getBatchingEnabled).to(producerBuilder::enableBatching);
+			map.from(this::getChunkingEnabled).to(producerBuilder::enableChunking);
+			map.from(this::getEncryptionKeys)
+					.to((encryptionKeys) -> encryptionKeys.forEach(producerBuilder::addEncryptionKey));
+			map.from(this::getCompressionType).to(producerBuilder::compressionType);
+			map.from(this::getInitialSequenceId).to(producerBuilder::initialSequenceId);
+			map.from(this::getAutoUpdatePartitions).to(producerBuilder::autoUpdatePartitions);
+			map.from(this::getAutoUpdatePartitionsInterval).asInt(Duration::toMillis).to(producerBuilder,
+					(pb, val) -> pb.autoUpdatePartitionsInterval(val, TimeUnit.MILLISECONDS));
+			map.from(this::getMultiSchema).to(producerBuilder::enableMultiSchema);
+			map.from(this::getProducerAccessMode).to(producerBuilder::accessMode);
+			map.from(this::getLazyStartPartitionedProducers).to(producerBuilder::enableLazyStartPartitionedProducers);
+			map.from(this::getProperties).to(producerBuilder::properties);
+		};
 	}
 
 }

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

@@ -94,8 +94,8 @@ public class PulsarAutoConfiguration {
 	@ConditionalOnMissingBean
 	@ConditionalOnProperty(name = "spring.pulsar.producer.cache.enabled", havingValue = "false")
 	public PulsarProducerFactory<?> pulsarProducerFactory(PulsarClient pulsarClient, TopicResolver topicResolver) {
-		return new DefaultPulsarProducerFactory<>(pulsarClient, this.properties.buildProducerProperties(),
-				topicResolver);
+		return new DefaultPulsarProducerFactory<>(pulsarClient, this.properties.getProducer().getTopicName(),
+				this.properties.getProducer().toProducerBuilderCustomizer(), topicResolver);
 	}
 
 	@Bean
@@ -103,8 +103,9 @@ public class PulsarAutoConfiguration {
 	@ConditionalOnProperty(name = "spring.pulsar.producer.cache.enabled", havingValue = "true", matchIfMissing = true)
 	public PulsarProducerFactory<?> cachingPulsarProducerFactory(PulsarClient pulsarClient,
 			TopicResolver topicResolver) {
-		return new CachingPulsarProducerFactory<>(pulsarClient, this.properties.buildProducerProperties(),
-				topicResolver, this.properties.getProducer().getCache().getExpireAfterAccess(),
+		return new CachingPulsarProducerFactory<>(pulsarClient, this.properties.getProducer().getTopicName(),
+				this.properties.getProducer().toProducerBuilderCustomizer(), topicResolver,
+				this.properties.getProducer().getCache().getExpireAfterAccess(),
 				this.properties.getProducer().getCache().getMaximumSize(),
 				this.properties.getProducer().getCache().getInitialCapacity());
 	}

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

@@ -110,10 +110,6 @@ public class PulsarProperties {
 		return new HashMap<>(this.consumer.buildProperties());
 	}
 
-	public Map<String, Object> buildProducerProperties() {
-		return new HashMap<>(this.producer.buildProperties());
-	}
-
 	public Map<String, Object> buildAdminProperties() {
 		return new HashMap<>(this.admin.buildProperties());
 	}

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

@@ -22,6 +22,7 @@ import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.spy;
 import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
 
 import java.util.Collections;
 import java.util.List;
@@ -42,6 +43,7 @@ import org.junit.jupiter.api.Test;
 
 import org.springframework.boot.autoconfigure.AutoConfigurations;
 import org.springframework.boot.test.context.FilteredClassLoader;
+import org.springframework.boot.test.context.TestConfiguration;
 import org.springframework.boot.test.context.assertj.AssertableApplicationContext;
 import org.springframework.boot.test.context.runner.ApplicationContextRunner;
 import org.springframework.context.annotation.Bean;
@@ -58,6 +60,7 @@ import org.springframework.pulsar.core.DefaultPulsarProducerFactory;
 import org.springframework.pulsar.core.DefaultPulsarReaderFactory;
 import org.springframework.pulsar.core.DefaultSchemaResolver;
 import org.springframework.pulsar.core.DefaultTopicResolver;
+import org.springframework.pulsar.core.ProducerBuilderCustomizer;
 import org.springframework.pulsar.core.PulsarAdministration;
 import org.springframework.pulsar.core.PulsarClientBuilderCustomizer;
 import org.springframework.pulsar.core.PulsarConsumerFactory;
@@ -369,10 +372,8 @@ class PulsarAutoConfigurationTests {
 
 		@Test
 		void clientConfigurerWithNoUserDefinedCustomizers() {
-			contextRunner.run((context) -> {
-				assertThat(context).getBean(PulsarClientBuilderConfigurer.class)
-						.hasFieldOrPropertyWithValue("customizers", Collections.emptyList());
-			});
+			contextRunner.run((context) -> assertThat(context).getBean(PulsarClientBuilderConfigurer.class)
+					.hasFieldOrPropertyWithValue("customizers", Collections.emptyList()));
 		}
 
 		@Test
@@ -520,18 +521,28 @@ class PulsarAutoConfigurationTests {
 
 		@Test
 		void beansAreInjectedInNonCachingProducerFactory() {
-			contextRunner.withPropertyValues("spring.pulsar.producer.cache.enabled=false")
-					.run((context -> assertThat(context).hasNotFailed().getBean(DefaultPulsarProducerFactory.class)
+			contextRunner.withUserConfiguration(ProducerCustomizerConfig.class)
+					.withPropertyValues("spring.pulsar.producer.topic-name=foo-topic",
+							"spring.pulsar.producer.cache.enabled=false")
+					.run((context) -> assertThat(context).getBean(DefaultPulsarProducerFactory.class)
+							.hasFieldOrPropertyWithValue("defaultTopic", "foo-topic")
+							.hasFieldOrPropertyWithValue("defaultConfigCustomizer",
+									ProducerCustomizerConfig.testCustomizer)
 							.hasFieldOrPropertyWithValue("pulsarClient", context.getBean(PulsarClient.class))
-							.hasFieldOrPropertyWithValue("topicResolver", context.getBean(TopicResolver.class))));
+							.hasFieldOrPropertyWithValue("topicResolver", context.getBean(TopicResolver.class)));
 		}
 
 		@Test
 		void beansAreInjectedInCachingProducerFactory() {
-			contextRunner.withPropertyValues("spring.pulsar.producer.cache.enabled=true")
-					.run((context -> assertThat(context).hasNotFailed().getBean(CachingPulsarProducerFactory.class)
+			contextRunner.withUserConfiguration(ProducerCustomizerConfig.class)
+					.withPropertyValues("spring.pulsar.producer.topic-name=foo-topic",
+							"spring.pulsar.producer.cache.enabled=true")
+					.run((context) -> assertThat(context).getBean(CachingPulsarProducerFactory.class)
+							.hasFieldOrPropertyWithValue("defaultTopic", "foo-topic")
+							.hasFieldOrPropertyWithValue("defaultConfigCustomizer",
+									ProducerCustomizerConfig.testCustomizer)
 							.hasFieldOrPropertyWithValue("pulsarClient", context.getBean(PulsarClient.class))
-							.hasFieldOrPropertyWithValue("topicResolver", context.getBean(TopicResolver.class))));
+							.hasFieldOrPropertyWithValue("topicResolver", context.getBean(TopicResolver.class)));
 		}
 
 		private void assertHasProducerFactoryOfType(Class<?> producerFactoryType,
@@ -610,4 +621,29 @@ class PulsarAutoConfigurationTests {
 
 	}
 
+	/*
+	 * Use '@TestConfiguration' and exact name of the PulsarProperties bean that is
+	 * created via the '@EnableConfigurationProperties' on the actual auto-config in order
+	 * to 'replace' the PulsarProperties bean - all so we can make sure the returned
+	 * producer builder customizer is the one we expect.
+	 */
+	@TestConfiguration(proxyBeanMethods = false)
+	static class ProducerCustomizerConfig {
+
+		@SuppressWarnings("rawtypes")
+		static ProducerBuilderCustomizer testCustomizer = (producerBuilder) -> {
+		};
+
+		@Bean(name = "spring.pulsar-org.springframework.pulsar.autoconfigure.PulsarProperties")
+		PulsarProperties pulsarProperties() {
+			var pulsarProps = new PulsarProperties();
+			var producerProps = spy(pulsarProps.getProducer());
+			when(producerProps.toProducerBuilderCustomizer()).thenReturn(testCustomizer);
+			var spyPulsarProps = spy(pulsarProps);
+			when(spyPulsarProps.getProducer()).thenReturn(producerProps);
+			return spyPulsarProps;
+		}
+
+	}
+
 }

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

@@ -21,6 +21,7 @@ import static org.assertj.core.api.Assertions.assertThatExceptionOfType;
 import static org.assertj.core.api.Assertions.assertThatIllegalArgumentException;
 import static org.assertj.core.api.Assertions.assertThatNoException;
 import static org.assertj.core.api.Assertions.assertThatRuntimeException;
+import static org.assertj.core.api.Assertions.entry;
 
 import java.time.Duration;
 import java.util.HashMap;
@@ -41,7 +42,6 @@ import org.apache.pulsar.client.api.SubscriptionMode;
 import org.apache.pulsar.client.api.SubscriptionType;
 import org.apache.pulsar.client.impl.conf.ConfigurationDataUtils;
 import org.apache.pulsar.client.impl.conf.ConsumerConfigurationData;
-import org.apache.pulsar.client.impl.conf.ProducerConfigurationData;
 import org.apache.pulsar.client.impl.conf.ReaderConfigurationData;
 import org.apache.pulsar.common.schema.SchemaType;
 import org.assertj.core.api.InstanceOfAssertFactories;
@@ -58,7 +58,7 @@ import org.springframework.pulsar.autoconfigure.PulsarProperties.TypeMapping;
 import org.springframework.util.unit.DataSize;
 
 /**
- * Unit tests for {@link PulsarProperties}.
+ * Tests for {@link PulsarProperties}.
  *
  * @author Chris Bono
  * @author Christophe Bornet
@@ -389,7 +389,7 @@ public class PulsarPropertiesTests {
 
 		@Test
 		void producerProperties() {
-			Map<String, String> props = new HashMap<>();
+			var props = new HashMap<String, String>();
 			props.put("spring.pulsar.producer.topic-name", "my-topic");
 			props.put("spring.pulsar.producer.producer-name", "my-producer");
 			props.put("spring.pulsar.producer.send-timeout", "2s");
@@ -413,32 +413,29 @@ public class PulsarPropertiesTests {
 			props.put("spring.pulsar.producer.properties[my-prop]", "my-prop-value");
 
 			bind(props);
-			Map<String, Object> producerProps = properties.buildProducerProperties();
 
-			// Verify that the props can be loaded in a ProducerBuilder
-			assertThatNoException().isThrownBy(() -> ConfigurationDataUtils.loadData(producerProps,
-					new ProducerConfigurationData(), ProducerConfigurationData.class));
-
-			assertThat(producerProps).containsEntry("topicName", "my-topic")
-					.containsEntry("producerName", "my-producer").containsEntry("sendTimeoutMs", 2_000)
-					.containsEntry("blockIfQueueFull", true).containsEntry("maxPendingMessages", 3)
-					.containsEntry("maxPendingMessagesAcrossPartitions", 4)
-					.containsEntry("messageRoutingMode", MessageRoutingMode.CustomPartition)
-					.containsEntry("hashingScheme", HashingScheme.Murmur3_32Hash)
-					.containsEntry("cryptoFailureAction", ProducerCryptoFailureAction.SEND)
-					.containsEntry("batchingMaxPublishDelayMicros", 5_000_000L)
-					.containsEntry("batchingPartitionSwitchFrequencyByPublishDelay", 6)
-					.containsEntry("batchingMaxMessages", 7).containsEntry("batchingMaxBytes", 8)
-					.containsEntry("batchingEnabled", false).containsEntry("chunkingEnabled", true)
-					.hasEntrySatisfying("encryptionKeys",
-							keys -> assertThat(keys).asInstanceOf(InstanceOfAssertFactories.collection(String.class))
-									.containsExactly("my-key"))
-					.containsEntry("compressionType", CompressionType.LZ4).containsEntry("initialSequenceId", 9L)
-					.containsEntry("accessMode", ProducerAccessMode.Exclusive)
-					.containsEntry("lazyStartPartitionedProducers", true).hasEntrySatisfying("properties",
-							properties -> assertThat(properties)
-									.asInstanceOf(InstanceOfAssertFactories.map(String.class, String.class))
-									.containsEntry("my-prop", "my-prop-value"));
+			var producerProps = properties.getProducer();
+			assertThat(producerProps.getTopicName()).isEqualTo("my-topic");
+			assertThat(producerProps.getProducerName()).isEqualTo("my-producer");
+			assertThat(producerProps.getSendTimeout()).isEqualTo(Duration.ofMillis(2000));
+			assertThat(producerProps.getBlockIfQueueFull()).isTrue();
+			assertThat(producerProps.getMaxPendingMessages()).isEqualTo(3);
+			assertThat(producerProps.getMaxPendingMessagesAcrossPartitions()).isEqualTo(4);
+			assertThat(producerProps.getMessageRoutingMode()).isEqualTo(MessageRoutingMode.CustomPartition);
+			assertThat(producerProps.getHashingScheme()).isEqualTo(HashingScheme.Murmur3_32Hash);
+			assertThat(producerProps.getCryptoFailureAction()).isEqualTo(ProducerCryptoFailureAction.SEND);
+			assertThat(producerProps.getBatchingMaxPublishDelay()).isEqualTo(Duration.ofMillis(5000));
+			assertThat(producerProps.getBatchingPartitionSwitchFrequencyByPublishDelay()).isEqualTo(6);
+			assertThat(producerProps.getBatchingMaxMessages()).isEqualTo(7);
+			assertThat(producerProps.getBatchingMaxBytes()).isEqualTo(DataSize.ofBytes(8));
+			assertThat(producerProps.getBatchingEnabled()).isFalse();
+			assertThat(producerProps.getChunkingEnabled()).isTrue();
+			assertThat(producerProps.getEncryptionKeys()).containsExactly("my-key");
+			assertThat(producerProps.getCompressionType()).isEqualTo(CompressionType.LZ4);
+			assertThat(producerProps.getInitialSequenceId()).isEqualTo(9);
+			assertThat(producerProps.getProducerAccessMode()).isEqualTo(ProducerAccessMode.Exclusive);
+			assertThat(producerProps.getLazyStartPartitionedProducers()).isTrue();
+			assertThat(producerProps.getProperties()).containsExactly(entry("my-prop", "my-prop-value"));
 		}
 
 	}

spring-pulsar-spring-cloud-stream-binder/src/main/java/org/springframework/pulsar/spring/cloud/stream/binder/PulsarBinderUtils.java

@@ -16,15 +16,19 @@
 
 package org.springframework.pulsar.spring.cloud.stream.binder;
 
+import java.time.Duration;
 import java.util.HashMap;
 import java.util.Map;
 import java.util.Objects;
 import java.util.UUID;
 
+import org.springframework.boot.context.properties.PropertyMapper;
 import org.springframework.cloud.stream.provisioning.ConsumerDestination;
 import org.springframework.core.log.LogAccessor;
+import org.springframework.pulsar.autoconfigure.ProducerConfigProperties;
 import org.springframework.pulsar.spring.cloud.stream.binder.properties.PulsarConsumerProperties;
 import org.springframework.util.StringUtils;
+import org.springframework.util.unit.DataSize;
 
 /**
  * Binder utility methods.
@@ -100,4 +104,52 @@ final class PulsarBinderUtils {
 		return newOrModifiedProps;
 	}
 
+	/**
+	 * Gets a map representation of a {@link ProducerConfigProperties}.
+	 * @param producerProps the producer props
+	 * @return map representation of producer props where each entry is a field and its
+	 * associated value
+	 */
+	static Map<String, Object> convertProducerPropertiesToMap(ProducerConfigProperties producerProps) {
+		var properties = new PulsarBinderUtils.Properties();
+		var map = PropertyMapper.get().alwaysApplyingWhenNonNull();
+		map.from(producerProps::getTopicName).to(properties.in("topicName"));
+		map.from(producerProps::getProducerName).to(properties.in("producerName"));
+		map.from(producerProps::getSendTimeout).asInt(Duration::toMillis).to(properties.in("sendTimeoutMs"));
+		map.from(producerProps::getBlockIfQueueFull).to(properties.in("blockIfQueueFull"));
+		map.from(producerProps::getMaxPendingMessages).to(properties.in("maxPendingMessages"));
+		map.from(producerProps::getMaxPendingMessagesAcrossPartitions)
+				.to(properties.in("maxPendingMessagesAcrossPartitions"));
+		map.from(producerProps::getMessageRoutingMode).to(properties.in("messageRoutingMode"));
+		map.from(producerProps::getHashingScheme).to(properties.in("hashingScheme"));
+		map.from(producerProps::getCryptoFailureAction).to(properties.in("cryptoFailureAction"));
+		map.from(producerProps::getBatchingMaxPublishDelay).as(it -> it.toNanos() / 1000)
+				.to(properties.in("batchingMaxPublishDelayMicros"));
+		map.from(producerProps::getBatchingPartitionSwitchFrequencyByPublishDelay)
+				.to(properties.in("batchingPartitionSwitchFrequencyByPublishDelay"));
+		map.from(producerProps::getBatchingMaxMessages).to(properties.in("batchingMaxMessages"));
+		map.from(producerProps::getBatchingMaxBytes).asInt(DataSize::toBytes).to(properties.in("batchingMaxBytes"));
+		map.from(producerProps::getBatchingEnabled).to(properties.in("batchingEnabled"));
+		map.from(producerProps::getChunkingEnabled).to(properties.in("chunkingEnabled"));
+		map.from(producerProps::getEncryptionKeys).to(properties.in("encryptionKeys"));
+		map.from(producerProps::getCompressionType).to(properties.in("compressionType"));
+		map.from(producerProps::getInitialSequenceId).to(properties.in("initialSequenceId"));
+		map.from(producerProps::getAutoUpdatePartitions).to(properties.in("autoUpdatePartitions"));
+		map.from(producerProps::getAutoUpdatePartitionsInterval).as(Duration::toSeconds)
+				.to(properties.in("autoUpdatePartitionsIntervalSeconds"));
+		map.from(producerProps::getMultiSchema).to(properties.in("multiSchema"));
+		map.from(producerProps::getProducerAccessMode).to(properties.in("accessMode"));
+		map.from(producerProps::getLazyStartPartitionedProducers).to(properties.in("lazyStartPartitionedProducers"));
+		map.from(producerProps::getProperties).to(properties.in("properties"));
+		return properties;
+	}
+
+	static class Properties extends HashMap<String, Object> {
+
+		<V> java.util.function.Consumer<V> in(String key) {
+			return (value) -> put(key, value);
+		}
+
+	}
+
 }

spring-pulsar-spring-cloud-stream-binder/src/main/java/org/springframework/pulsar/spring/cloud/stream/binder/PulsarMessageChannelBinder.java

@@ -110,9 +110,11 @@ public class PulsarMessageChannelBinder extends
 		else {
 			schema = null;
 		}
-		var baseProducerProps = new ProducerConfigProperties().buildProperties();
-		var binderProducerProps = this.binderConfigProps.getProducer().buildProperties();
-		var bindingProducerProps = producerProperties.getExtension().buildProperties();
+
+		var baseProducerProps = PulsarBinderUtils.convertProducerPropertiesToMap(new ProducerConfigProperties());
+		var binderProducerProps = PulsarBinderUtils
+				.convertProducerPropertiesToMap(this.binderConfigProps.getProducer());
+		var bindingProducerProps = PulsarBinderUtils.convertProducerPropertiesToMap(producerProperties.getExtension());
 		var mergedProducerProps = PulsarBinderUtils.mergePropertiesWithPrecedence(baseProducerProps,
 				binderProducerProps, bindingProducerProps);
 

spring-pulsar-spring-cloud-stream-binder/src/test/java/org/springframework/pulsar/spring/cloud/stream/binder/PulsarBinderConfigurationPropertiesTests.java

@@ -68,7 +68,7 @@ public class PulsarBinderConfigurationPropertiesTests {
 		props.put("spring.cloud.stream.pulsar.binder.producer.properties[my-prop]", "my-prop-value");
 
 		bind(props);
-		Map<String, Object> producerProps = properties.getProducer().buildProperties();
+		Map<String, Object> producerProps = PulsarBinderUtils.convertProducerPropertiesToMap(properties.getProducer());
 
 		// Verify that the props can be loaded in a ProducerBuilder
 		assertThatNoException().isThrownBy(() -> ConfigurationDataUtils.loadData(producerProps,

spring-pulsar-spring-cloud-stream-binder/src/test/java/org/springframework/pulsar/spring/cloud/stream/binder/PulsarBinderIntegrationTests.java

@@ -639,10 +639,14 @@ class PulsarBinderIntegrationTests implements PulsarTestContainerSupport {
 	@Import(PrimitiveTextConfig.class)
 	static class BinderAndBindingPropsTestConfig {
 
+		@SuppressWarnings("unchecked")
 		@Bean
 		public PulsarProducerFactory<?> pulsarProducerFactory(PulsarClient pulsarClient,
 				PulsarProperties pulsarProperties, TopicResolver topicResolver) {
-			return new TrackingProducerFactory(pulsarClient, pulsarProperties.buildProducerProperties(), topicResolver);
+			var customizer = (ProducerBuilderCustomizer<String>) pulsarProperties.getProducer()
+					.toProducerBuilderCustomizer();
+			return new TrackingProducerFactory(pulsarClient, pulsarProperties.getProducer().getTopicName(), customizer,
+					topicResolver);
 		}
 
 		@Bean
@@ -657,8 +661,9 @@ class PulsarBinderIntegrationTests implements PulsarTestContainerSupport {
 
 		List<Producer<String>> producersCreated = new ArrayList<>();
 
-		TrackingProducerFactory(PulsarClient pulsarClient, Map<String, Object> config, TopicResolver topicResolver) {
-			super(pulsarClient, config, topicResolver);
+		TrackingProducerFactory(PulsarClient pulsarClient, @Nullable String defaultTopic,
+				ProducerBuilderCustomizer<String> defaultConfigCustomizer, TopicResolver topicResolver) {
+			super(pulsarClient, defaultTopic, defaultConfigCustomizer, topicResolver);
 		}
 
 		@Override

spring-pulsar-spring-cloud-stream-binder/src/test/java/org/springframework/pulsar/spring/cloud/stream/binder/PulsarBinderTests.java

@@ -19,7 +19,6 @@ package org.springframework.pulsar.spring.cloud.stream.binder;
 import static org.assertj.core.api.Assertions.assertThat;
 
 import java.nio.charset.StandardCharsets;
-import java.util.Collections;
 import java.util.Map;
 import java.util.UUID;
 import java.util.concurrent.CountDownLatch;
@@ -105,7 +104,7 @@ public class PulsarBinderTests extends
 				Map.of("serviceUrl", PulsarTestContainerSupport.getHttpServiceUrl()));
 		var configProps = new PulsarBinderConfigurationProperties();
 		var provisioner = new PulsarTopicProvisioner(pulsarAdministration, configProps);
-		var producerFactory = new DefaultPulsarProducerFactory<>(pulsarClient, Collections.emptyMap());
+		var producerFactory = new DefaultPulsarProducerFactory<>(pulsarClient);
 		var pulsarTemplate = new PulsarTemplate<>(producerFactory);
 		var config = Map.<String, Object>of("subscriptionInitialPosition", SubscriptionInitialPosition.Earliest);
 		var consumerFactory = new DefaultPulsarConsumerFactory<>(pulsarClient, config);

spring-pulsar-spring-cloud-stream-binder/src/test/java/org/springframework/pulsar/spring/cloud/stream/binder/PulsarBinderUtilsTests.java

@@ -17,21 +17,36 @@
 package org.springframework.pulsar.spring.cloud.stream.binder;
 
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatNoException;
 import static org.junit.jupiter.params.provider.Arguments.arguments;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.when;
 
 import java.util.Collections;
+import java.util.HashMap;
 import java.util.Map;
 import java.util.stream.Stream;
 
+import org.apache.pulsar.client.api.CompressionType;
+import org.apache.pulsar.client.api.HashingScheme;
+import org.apache.pulsar.client.api.MessageRoutingMode;
+import org.apache.pulsar.client.api.ProducerAccessMode;
+import org.apache.pulsar.client.api.ProducerCryptoFailureAction;
+import org.apache.pulsar.client.impl.conf.ConfigurationDataUtils;
+import org.apache.pulsar.client.impl.conf.ProducerConfigurationData;
+import org.assertj.core.api.InstanceOfAssertFactories;
 import org.junit.jupiter.api.Nested;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.params.ParameterizedTest;
 import org.junit.jupiter.params.provider.Arguments;
 import org.junit.jupiter.params.provider.MethodSource;
 
+import org.springframework.boot.context.properties.bind.Bindable;
+import org.springframework.boot.context.properties.bind.Binder;
+import org.springframework.boot.context.properties.source.ConfigurationPropertySource;
+import org.springframework.boot.context.properties.source.MapConfigurationPropertySource;
 import org.springframework.cloud.stream.provisioning.ConsumerDestination;
+import org.springframework.pulsar.autoconfigure.ProducerConfigProperties;
 import org.springframework.pulsar.spring.cloud.stream.binder.properties.PulsarConsumerProperties;
 
 /**
@@ -130,4 +145,70 @@ public class PulsarBinderUtilsTests {
 
 	}
 
+	@Nested
+	class ConvertedPropertiesTests {
+
+		private final ProducerConfigProperties properties = new ProducerConfigProperties();
+
+		private void bind(Map<String, String> map) {
+			ConfigurationPropertySource source = new MapConfigurationPropertySource(map);
+			new Binder(source).bind("spring.pulsar.producer", Bindable.ofInstance(this.properties));
+		}
+
+		@Test
+		void producerPropertiesToMap() {
+			Map<String, String> props = new HashMap<>();
+			props.put("spring.pulsar.producer.topic-name", "my-topic");
+			props.put("spring.pulsar.producer.producer-name", "my-producer");
+			props.put("spring.pulsar.producer.send-timeout", "2s");
+			props.put("spring.pulsar.producer.block-if-queue-full", "true");
+			props.put("spring.pulsar.producer.max-pending-messages", "3");
+			props.put("spring.pulsar.producer.max-pending-messages-across-partitions", "4");
+			props.put("spring.pulsar.producer.message-routing-mode", "custompartition");
+			props.put("spring.pulsar.producer.hashing-scheme", "murmur3_32hash");
+			props.put("spring.pulsar.producer.crypto-failure-action", "send");
+			props.put("spring.pulsar.producer.batching-max-publish-delay", "5s");
+			props.put("spring.pulsar.producer.batching-partition-switch-frequency-by-publish-delay", "6");
+			props.put("spring.pulsar.producer.batching-max-messages", "7");
+			props.put("spring.pulsar.producer.batching-max-bytes", "8");
+			props.put("spring.pulsar.producer.batching-enabled", "false");
+			props.put("spring.pulsar.producer.chunking-enabled", "true");
+			props.put("spring.pulsar.producer.encryption-keys[0]", "my-key");
+			props.put("spring.pulsar.producer.compression-type", "lz4");
+			props.put("spring.pulsar.producer.initial-sequence-id", "9");
+			props.put("spring.pulsar.producer.producer-access-mode", "exclusive");
+			props.put("spring.pulsar.producer.lazy-start=partitioned-producers", "true");
+			props.put("spring.pulsar.producer.properties[my-prop]", "my-prop-value");
+
+			bind(props);
+			Map<String, Object> producerProps = PulsarBinderUtils.convertProducerPropertiesToMap(properties);
+
+			// Verify that the props can be loaded in a ProducerBuilder
+			assertThatNoException().isThrownBy(() -> ConfigurationDataUtils.loadData(producerProps,
+					new ProducerConfigurationData(), ProducerConfigurationData.class));
+
+			assertThat(producerProps).containsEntry("topicName", "my-topic")
+					.containsEntry("producerName", "my-producer").containsEntry("sendTimeoutMs", 2_000)
+					.containsEntry("blockIfQueueFull", true).containsEntry("maxPendingMessages", 3)
+					.containsEntry("maxPendingMessagesAcrossPartitions", 4)
+					.containsEntry("messageRoutingMode", MessageRoutingMode.CustomPartition)
+					.containsEntry("hashingScheme", HashingScheme.Murmur3_32Hash)
+					.containsEntry("cryptoFailureAction", ProducerCryptoFailureAction.SEND)
+					.containsEntry("batchingMaxPublishDelayMicros", 5_000_000L)
+					.containsEntry("batchingPartitionSwitchFrequencyByPublishDelay", 6)
+					.containsEntry("batchingMaxMessages", 7).containsEntry("batchingMaxBytes", 8)
+					.containsEntry("batchingEnabled", false).containsEntry("chunkingEnabled", true)
+					.hasEntrySatisfying("encryptionKeys",
+							keys -> assertThat(keys).asInstanceOf(InstanceOfAssertFactories.collection(String.class))
+									.containsExactly("my-key"))
+					.containsEntry("compressionType", CompressionType.LZ4).containsEntry("initialSequenceId", 9L)
+					.containsEntry("accessMode", ProducerAccessMode.Exclusive)
+					.containsEntry("lazyStartPartitionedProducers", true).hasEntrySatisfying("properties",
+							properties -> assertThat(properties)
+									.asInstanceOf(InstanceOfAssertFactories.map(String.class, String.class))
+									.containsEntry("my-prop", "my-prop-value"));
+		}
+
+	}
+
 }

spring-pulsar-spring-cloud-stream-binder/src/test/java/org/springframework/pulsar/spring/cloud/stream/binder/PulsarExtendedBindingPropertiesTests.java

@@ -65,7 +65,8 @@ public class PulsarExtendedBindingPropertiesTests {
 		bind(props);
 
 		assertThat(properties.getBindings()).containsOnlyKeys("my-foo");
-		Map<String, Object> producerProps = properties.getExtendedProducerProperties("my-foo").buildProperties();
+		Map<String, Object> producerProps = PulsarBinderUtils
+				.convertProducerPropertiesToMap(properties.getExtendedProducerProperties("my-foo"));
 		// Verify that the props can be loaded in a ProducerBuilder
 		assertThatNoException().isThrownBy(() -> ConfigurationDataUtils.loadData(producerProps,
 				new ProducerConfigurationData(), ProducerConfigurationData.class));

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

@@ -20,7 +20,6 @@ import java.time.Duration;
 import java.util.Collection;
 import java.util.HashSet;
 import java.util.List;
-import java.util.Map;
 import java.util.Objects;
 import java.util.Set;
 import java.util.concurrent.CompletableFuture;
@@ -67,16 +66,17 @@ public class CachingPulsarProducerFactory<T> extends DefaultPulsarProducerFactor
 	 * Construct a caching producer factory with the specified values for the cache
 	 * configuration.
 	 * @param pulsarClient the client used to create the producers
-	 * @param producerConfig the configuration to use when creating a producer
+	 * @param defaultTopic the default topic to use for the producers
+	 * @param defaultConfigCustomizer the default configuration to apply to the 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, Map<String, Object> producerConfig,
-			TopicResolver topicResolver, Duration cacheExpireAfterAccess, Long cacheMaximumSize,
-			Integer cacheInitialCapacity) {
-		super(pulsarClient, producerConfig, topicResolver);
+	public CachingPulsarProducerFactory(PulsarClient pulsarClient, @Nullable String defaultTopic,
+			ProducerBuilderCustomizer<T> defaultConfigCustomizer, TopicResolver topicResolver,
+			Duration cacheExpireAfterAccess, Long cacheMaximumSize, Integer cacheInitialCapacity) {
+		super(pulsarClient, defaultTopic, defaultConfigCustomizer, topicResolver);
 		var cacheFactory = CacheProviderFactory.<ProducerCacheKey<T>, Producer<T>>load();
 		this.producerCache = cacheFactory.create(cacheExpireAfterAccess, cacheMaximumSize, cacheInitialCapacity,
 				(key, producer, cause) -> {
@@ -90,8 +90,8 @@ public class CachingPulsarProducerFactory<T> extends DefaultPulsarProducerFactor
 	protected Producer<T> doCreateProducer(Schema<T> schema, @Nullable String topic,
 			@Nullable Collection<String> encryptionKeys, @Nullable List<ProducerBuilderCustomizer<T>> customizers) {
 		Objects.requireNonNull(schema, "Schema must be specified");
-		String resolveTopicName = resolveTopicName(topic);
-		ProducerCacheKey<T> producerCacheKey = new ProducerCacheKey<>(schema, resolveTopicName,
+		var resolveTopicName = resolveTopicName(topic);
+		var producerCacheKey = new ProducerCacheKey<>(schema, resolveTopicName,
 				encryptionKeys == null ? null : new HashSet<>(encryptionKeys), customizers);
 		return this.producerCache.getOrCreateIfAbsent(producerCacheKey,
 				(st) -> createCacheableProducer(st.schema, st.topic, st.encryptionKeys, customizers));
@@ -100,7 +100,7 @@ public class CachingPulsarProducerFactory<T> extends DefaultPulsarProducerFactor
 	private Producer<T> createCacheableProducer(Schema<T> schema, String topic,
 			@Nullable Collection<String> encryptionKeys, @Nullable List<ProducerBuilderCustomizer<T>> customizers) {
 		try {
-			Producer<T> producer = super.doCreateProducer(schema, topic, encryptionKeys, customizers);
+			var producer = super.doCreateProducer(schema, topic, encryptionKeys, customizers);
 			return new ProducerWithCloseCallback<>(producer,
 					(p) -> this.logger.trace(() -> "Client closed producer %s but will skip actual closing"
 							.formatted(ProducerUtils.formatProducer(producer))));
@@ -174,7 +174,7 @@ public class CachingPulsarProducerFactory<T> extends DefaultPulsarProducerFactor
 			if (o == null || getClass() != o.getClass()) {
 				return false;
 			}
-			ProducerCacheKey<?> that = (ProducerCacheKey<?>) o;
+			var that = (ProducerCacheKey<?>) o;
 			return this.topic.equals(that.topic) && this.schemaHash.equals(that.schemaHash)
 					&& Objects.equals(this.encryptionKeys, that.encryptionKeys)
 					&& Objects.equals(this.customizers, that.customizers);

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

@@ -18,9 +18,8 @@ package org.springframework.pulsar.core;
 
 import java.util.Collection;
 import java.util.Collections;
-import java.util.HashMap;
+import java.util.HashSet;
 import java.util.List;
-import java.util.Map;
 import java.util.Objects;
 
 import org.apache.pulsar.client.api.Producer;
@@ -28,6 +27,7 @@ import org.apache.pulsar.client.api.ProducerBuilder;
 import org.apache.pulsar.client.api.PulsarClient;
 import org.apache.pulsar.client.api.PulsarClientException;
 import org.apache.pulsar.client.api.Schema;
+import org.apache.pulsar.client.impl.ProducerBuilderImpl;
 
 import org.springframework.core.log.LogAccessor;
 import org.springframework.lang.Nullable;
@@ -46,20 +46,58 @@ public class DefaultPulsarProducerFactory<T> implements PulsarProducerFactory<T>
 
 	private final LogAccessor logger = new LogAccessor(this.getClass());
 
-	private final Map<String, Object> producerConfig;
-
 	private final PulsarClient pulsarClient;
 
+	@Nullable
+	private final String defaultTopic;
+
+	@Nullable
+	private final ProducerBuilderCustomizer<T> defaultConfigCustomizer;
+
 	private final TopicResolver topicResolver;
 
-	public DefaultPulsarProducerFactory(PulsarClient pulsarClient, Map<String, Object> config) {
-		this(pulsarClient, config, new DefaultTopicResolver());
+	/**
+	 * Construct a producer factory that uses a default topic resolver.
+	 * @param pulsarClient the client used to create the producers
+	 */
+	public DefaultPulsarProducerFactory(PulsarClient pulsarClient) {
+		this(pulsarClient, null, (pb) -> {
+		}, new DefaultTopicResolver());
 	}
 
-	public DefaultPulsarProducerFactory(PulsarClient pulsarClient, Map<String, Object> config,
-			TopicResolver topicResolver) {
+	/**
+	 * 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
+	 */
+	public DefaultPulsarProducerFactory(PulsarClient pulsarClient, @Nullable String defaultTopic) {
+		this(pulsarClient, defaultTopic, (pb) -> {
+		}, 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
+	 */
+	public DefaultPulsarProducerFactory(PulsarClient pulsarClient, @Nullable String defaultTopic,
+			@Nullable ProducerBuilderCustomizer<T> defaultConfigCustomizer) {
+		this(pulsarClient, defaultTopic, defaultConfigCustomizer, 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 topicResolver the topic resolver to use
+	 */
+	public DefaultPulsarProducerFactory(PulsarClient pulsarClient, @Nullable String defaultTopic,
+			@Nullable ProducerBuilderCustomizer<T> defaultConfigCustomizer, TopicResolver topicResolver) {
 		this.pulsarClient = pulsarClient;
-		this.producerConfig = Collections.unmodifiableMap(config);
+		this.defaultTopic = defaultTopic;
+		this.defaultConfigCustomizer = defaultConfigCustomizer;
 		this.topicResolver = topicResolver;
 	}
 
@@ -99,36 +137,42 @@ public class DefaultPulsarProducerFactory<T> implements PulsarProducerFactory<T>
 			@Nullable Collection<String> encryptionKeys, @Nullable List<ProducerBuilderCustomizer<T>> customizers)
 			throws PulsarClientException {
 		Objects.requireNonNull(schema, "Schema must be specified");
-		String resolvedTopic = resolveTopicName(topic);
+		var resolvedTopic = resolveTopicName(topic);
 		this.logger.trace(() -> "Creating producer for '%s' topic".formatted(resolvedTopic));
-		ProducerBuilder<T> producerBuilder = this.pulsarClient.newProducer(schema);
+		var producerBuilder = this.pulsarClient.newProducer(schema);
 
-		Map<String, Object> config = new HashMap<>(this.producerConfig);
-
-		// Replace default keys - workaround as they can't be replaced through the builder
-		if (encryptionKeys != null) {
-			config.put("encryptionKeys", encryptionKeys);
+		// Apply the default config customizer (preserve the topic)
+		if (this.defaultConfigCustomizer != null) {
+			this.defaultConfigCustomizer.customize(producerBuilder);
 		}
-		ProducerBuilderConfigurationUtil.loadConf(producerBuilder, config);
 		producerBuilder.topic(resolvedTopic);
 
+		// Replace default keys - workaround as they can't be replaced through the builder
+		maybeSetEncryptionKeys(producerBuilder, encryptionKeys);
+
+		// Apply any user-specified customizers (preserve the topic)
 		if (!CollectionUtils.isEmpty(customizers)) {
 			customizers.forEach((c) -> c.customize(producerBuilder));
 		}
-		// make sure the customizer do not override the topic
 		producerBuilder.topic(resolvedTopic);
 
 		return producerBuilder.create();
 	}
 
 	protected String resolveTopicName(String userSpecifiedTopic) {
-		String defaultTopic = Objects.toString(getProducerConfig().get("topicName"), null);
-		return this.topicResolver.resolveTopic(userSpecifiedTopic, () -> defaultTopic).orElseThrow();
+		return this.topicResolver.resolveTopic(userSpecifiedTopic, this::getDefaultTopic).orElseThrow();
 	}
 
 	@Override
-	public Map<String, Object> getProducerConfig() {
-		return this.producerConfig;
+	public String getDefaultTopic() {
+		return this.defaultTopic;
+	}
+
+	private void maybeSetEncryptionKeys(ProducerBuilder<T> builder, @Nullable Collection<String> encryptionKeys) {
+		if (encryptionKeys != null) {
+			var builderImpl = (ProducerBuilderImpl<T>) builder;
+			builderImpl.getConf().setEncryptionKeys(new HashSet<>(encryptionKeys));
+		}
 	}
 
 }

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

@@ -18,7 +18,6 @@ package org.springframework.pulsar.core;
 
 import java.util.Collection;
 import java.util.List;
-import java.util.Map;
 
 import org.apache.pulsar.client.api.Producer;
 import org.apache.pulsar.client.api.ProducerBuilder;
@@ -78,9 +77,11 @@ public interface PulsarProducerFactory<T> {
 			@Nullable List<ProducerBuilderCustomizer<T>> customizers) throws PulsarClientException;
 
 	/**
-	 * Return a map of configuration options to use when creating producers.
-	 * @return the map of configuration options
+	 * Get the default topic to use for all created producers.
+	 * @return the default topic to use for all created producers or null if no default
+	 * set
 	 */
-	Map<String, Object> getProducerConfig();
+	@Nullable
+	String getDefaultTopic();
 
 }

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

@@ -221,7 +221,7 @@ public class PulsarTemplate<T>
 			@Nullable Schema<T> schema, @Nullable Collection<String> encryptionKeys,
 			@Nullable TypedMessageBuilderCustomizer<T> typedMessageBuilderCustomizer,
 			@Nullable ProducerBuilderCustomizer<T> producerCustomizer) throws PulsarClientException {
-		String defaultTopic = Objects.toString(this.producerFactory.getProducerConfig().get("topicName"), null);
+		String defaultTopic = Objects.toString(this.producerFactory.getDefaultTopic(), null);
 		String topicName = this.topicResolver.resolveTopic(topic, message, () -> defaultTopic).orElseThrow();
 		this.logger.trace(() -> "Sending msg to '%s' topic".formatted(topicName));
 

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

@@ -27,7 +27,6 @@ import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collections;
 import java.util.List;
-import java.util.Map;
 import java.util.Set;
 import java.util.stream.Stream;
 
@@ -46,6 +45,7 @@ import org.junit.jupiter.params.ParameterizedTest;
 import org.junit.jupiter.params.provider.Arguments;
 import org.junit.jupiter.params.provider.MethodSource;
 
+import org.springframework.lang.Nullable;
 import org.springframework.pulsar.core.CachingPulsarProducerFactory.ProducerCacheKey;
 import org.springframework.pulsar.core.CachingPulsarProducerFactory.ProducerWithCloseCallback;
 import org.springframework.test.util.ReflectionTestUtils;
@@ -167,7 +167,7 @@ class CachingPulsarProducerFactoryTests extends PulsarProducerFactoryTests {
 
 	@Test
 	void factoryDestroyCleansUpCacheAndClosesProducers() throws PulsarClientException {
-		CachingPulsarProducerFactory<String> producerFactory = producerFactory(pulsarClient, Collections.emptyMap());
+		CachingPulsarProducerFactory<String> producerFactory = producerFactory(pulsarClient, null, null);
 		var actualProducer1 = actualProducer(producerFactory.createProducer(schema, "topic1"));
 		var actualProducer2 = actualProducer(producerFactory.createProducer(schema, "topic2"));
 		var cacheKey1 = new ProducerCacheKey<>(schema, "topic1", null, null);
@@ -183,8 +183,8 @@ class CachingPulsarProducerFactoryTests extends PulsarProducerFactoryTests {
 
 	@Test
 	void producerEvictedFromCache() throws PulsarClientException {
-		CachingPulsarProducerFactory<String> producerFactory = new CachingPulsarProducerFactory<>(pulsarClient,
-				Collections.emptyMap(), new DefaultTopicResolver(), Duration.ofSeconds(3L), 10L, 2);
+		CachingPulsarProducerFactory<String> producerFactory = new CachingPulsarProducerFactory<>(pulsarClient, null,
+				null, new DefaultTopicResolver(), Duration.ofSeconds(3L), 10L, 2);
 		var actualProducer = actualProducer(producerFactory.createProducer(schema, "topic1"));
 		var cacheKey = new ProducerCacheKey<>(schema, "topic1", null, null);
 		var producerCache = getAssertedProducerCache(producerFactory, Collections.singletonList(cacheKey));
@@ -198,7 +198,7 @@ class CachingPulsarProducerFactoryTests extends PulsarProducerFactoryTests {
 	void createProducerEncountersException() {
 		pulsarClient = spy(pulsarClient);
 		when(this.pulsarClient.newProducer(schema)).thenThrow(new RuntimeException("5150"));
-		var producerFactory = producerFactory(pulsarClient, Collections.emptyMap());
+		var producerFactory = producerFactory(pulsarClient, null, null);
 		assertThatThrownBy(() -> producerFactory.createProducer(schema, "topic1")).isInstanceOf(RuntimeException.class)
 				.hasMessage("5150");
 		getAssertedProducerCache(producerFactory, Collections.emptyList());
@@ -228,9 +228,9 @@ class CachingPulsarProducerFactoryTests extends PulsarProducerFactoryTests {
 
 	@Override
 	protected CachingPulsarProducerFactory<String> producerFactory(PulsarClient pulsarClient,
-			Map<String, Object> producerConfig) {
-		var producerFactory = new CachingPulsarProducerFactory<String>(pulsarClient, producerConfig,
-				new DefaultTopicResolver(), Duration.ofMinutes(5L), 30L, 2);
+			@Nullable String defaultTopic, @Nullable ProducerBuilderCustomizer<String> defaultConfigCustomizer) {
+		var producerFactory = new CachingPulsarProducerFactory<String>(pulsarClient, defaultTopic,
+				defaultConfigCustomizer, new DefaultTopicResolver(), Duration.ofMinutes(5L), 30L, 2);
 		producerFactories.add(producerFactory);
 		return producerFactory;
 	}

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

@@ -85,9 +85,8 @@ class ConsumerAcknowledgmentTests implements PulsarTestContainerSupport {
 			return invocation.callRealMethod();
 		}).when(containerConsumer).acknowledge(any(MessageId.class));
 
-		Map<String, Object> prodConfig = Map.of("topicName", "cons-ack-tests-011");
 		DefaultPulsarProducerFactory<String> pulsarProducerFactory = new DefaultPulsarProducerFactory<>(pulsarClient,
-				prodConfig);
+				"cons-ack-tests-011");
 		PulsarTemplate<String> pulsarTemplate = new PulsarTemplate<>(pulsarProducerFactory);
 		for (int i = 0; i < 10; i++) {
 			pulsarTemplate.sendAsync("hello john doe");
@@ -115,9 +114,8 @@ class ConsumerAcknowledgmentTests implements PulsarTestContainerSupport {
 				pulsarConsumerFactory, pulsarContainerProperties);
 		Consumer<String> containerConsumer = ConsumerTestUtils.startContainerAndSpyOnConsumer(container);
 
-		Map<String, Object> prodConfig = Map.of("topicName", "cons-ack-tests-012");
 		DefaultPulsarProducerFactory<String> pulsarProducerFactory = new DefaultPulsarProducerFactory<>(pulsarClient,
-				prodConfig);
+				"cons-ack-tests-012");
 		PulsarTemplate<String> pulsarTemplate = new PulsarTemplate<>(pulsarProducerFactory);
 		for (int i = 0; i < 10; i++) {
 			pulsarTemplate.sendAsync("hello john doe");
@@ -160,9 +158,8 @@ class ConsumerAcknowledgmentTests implements PulsarTestContainerSupport {
 			return invocation.callRealMethod();
 		}).when(containerConsumer).acknowledge(any(MessageId.class));
 
-		Map<String, Object> prodConfig = Map.of("topicName", "cons-ack-tests-013");
 		DefaultPulsarProducerFactory<String> pulsarProducerFactory = new DefaultPulsarProducerFactory<>(pulsarClient,
-				prodConfig);
+				"cons-ack-tests-013");
 		PulsarTemplate<String> pulsarTemplate = new PulsarTemplate<>(pulsarProducerFactory);
 		for (int i = 0; i < 10; i++) {
 			pulsarTemplate.sendAsync("hello john doe");
@@ -226,9 +223,8 @@ class ConsumerAcknowledgmentTests implements PulsarTestContainerSupport {
 			return invocation.callRealMethod();
 		}).when(containerConsumer).acknowledge(any(MessageId.class));
 
-		Map<String, Object> prodConfig = Map.of("topicName", "cons-ack-tests-014");
 		DefaultPulsarProducerFactory<String> pulsarProducerFactory = new DefaultPulsarProducerFactory<>(pulsarClient,
-				prodConfig);
+				"cons-ack-tests-014");
 		PulsarTemplate<String> pulsarTemplate = new PulsarTemplate<>(pulsarProducerFactory);
 		for (int i = 0; i < 10; i++) {
 			pulsarTemplate.sendAsync("hello john doe");
@@ -272,9 +268,8 @@ class ConsumerAcknowledgmentTests implements PulsarTestContainerSupport {
 				pulsarConsumerFactory, pulsarContainerProperties);
 		Consumer<String> containerConsumer = ConsumerTestUtils.startContainerAndSpyOnConsumer(container);
 
-		Map<String, Object> prodConfig = Map.of("topicName", "cons-ack-tests-015");
 		DefaultPulsarProducerFactory<String> pulsarProducerFactory = new DefaultPulsarProducerFactory<>(pulsarClient,
-				prodConfig);
+				"cons-ack-tests-015");
 		PulsarTemplate<String> pulsarTemplate = new PulsarTemplate<>(pulsarProducerFactory);
 		for (int i = 0; i < 10; i++) {
 			pulsarTemplate.sendAsync("hello john doe");
@@ -316,9 +311,8 @@ class ConsumerAcknowledgmentTests implements PulsarTestContainerSupport {
 				pulsarConsumerFactory, pulsarContainerProperties);
 		Consumer<String> containerConsumer = ConsumerTestUtils.startContainerAndSpyOnConsumer(container);
 
-		Map<String, Object> prodConfig = Map.of("topicName", "cons-ack-tests-016");
 		DefaultPulsarProducerFactory<String> pulsarProducerFactory = new DefaultPulsarProducerFactory<>(pulsarClient,
-				prodConfig);
+				"cons-ack-tests-016");
 		PulsarTemplate<String> pulsarTemplate = new PulsarTemplate<>(pulsarProducerFactory);
 		for (int i = 0; i < 10; i++) {
 			pulsarTemplate.sendAsync("hello john doe");
@@ -352,9 +346,8 @@ class ConsumerAcknowledgmentTests implements PulsarTestContainerSupport {
 				pulsarConsumerFactory, pulsarContainerProperties);
 		container1.start();
 
-		Map<String, Object> prodConfig = Collections.singletonMap("topicName", "duplicate-message-test");
 		DefaultPulsarProducerFactory<String> pulsarProducerFactory = new DefaultPulsarProducerFactory<>(pulsarClient,
-				prodConfig);
+				"duplicate-message-test");
 		PulsarTemplate<String> pulsarTemplate = new PulsarTemplate<>(pulsarProducerFactory);
 		pulsarTemplate.send("hello john doe");
 

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

@@ -18,14 +18,13 @@ package org.springframework.pulsar.core;
 
 import static org.assertj.core.api.Assertions.assertThat;
 
-import java.util.Collections;
-import java.util.Map;
-
 import org.apache.pulsar.client.api.Producer;
 import org.apache.pulsar.client.api.PulsarClient;
 import org.apache.pulsar.client.api.PulsarClientException;
 import org.junit.jupiter.api.Test;
 
+import org.springframework.lang.Nullable;
+
 /**
  * Tests for {@link DefaultPulsarProducerFactory}.
  *
@@ -35,8 +34,7 @@ class DefaultPulsarProducerFactoryTests extends PulsarProducerFactoryTests {
 
 	@Test
 	void createProducerMultipleTimeDoesNotCacheProducer() throws PulsarClientException {
-		Map<String, Object> producerConfig = Collections.emptyMap();
-		PulsarProducerFactory<String> producerFactory = producerFactory(pulsarClient, producerConfig);
+		PulsarProducerFactory<String> producerFactory = newProducerFactory();
 		try (Producer<String> producer1 = producerFactory.createProducer(schema, "topic1")) {
 			try (Producer<String> producer2 = producerFactory.createProducer(schema, "topic1")) {
 				try (Producer<String> producer3 = producerFactory.createProducer(schema, "topic1")) {
@@ -47,9 +45,9 @@ class DefaultPulsarProducerFactoryTests extends PulsarProducerFactoryTests {
 	}
 
 	@Override
-	protected PulsarProducerFactory<String> producerFactory(PulsarClient pulsarClient,
-			Map<String, Object> producerConfig) {
-		return new DefaultPulsarProducerFactory<>(pulsarClient, producerConfig);
+	protected PulsarProducerFactory<String> producerFactory(PulsarClient pulsarClient, @Nullable String defaultTopic,
+			@Nullable ProducerBuilderCustomizer<String> defaultConfigCustomizer) {
+		return new DefaultPulsarProducerFactory<>(pulsarClient, defaultTopic, defaultConfigCustomizer);
 	}
 
 }

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

@@ -21,7 +21,6 @@ import static org.assertj.core.api.Assertions.assertThatThrownBy;
 
 import java.util.Collections;
 import java.util.List;
-import java.util.Map;
 import java.util.concurrent.TimeUnit;
 
 import org.apache.pulsar.client.api.Message;
@@ -74,9 +73,8 @@ public class DefaultPulsarReaderFactoryTests implements PulsarTestContainerSuppo
 			try (Reader<String> reader = pulsarReaderFactory.createReader(List.of("basic-pulsar-reader-topic"),
 					MessageId.earliest, Schema.STRING, Collections.emptyList())) {
 
-				Map<String, Object> prodConfig = Map.of("topicName", "basic-pulsar-reader-topic");
 				PulsarProducerFactory<String> pulsarProducerFactory = new DefaultPulsarProducerFactory<>(pulsarClient,
-						prodConfig);
+						"basic-pulsar-reader-topic");
 				PulsarTemplate<String> pulsarTemplate = new PulsarTemplate<>(pulsarProducerFactory);
 				pulsarTemplate.send("hello john doe");
 
@@ -87,9 +85,8 @@ public class DefaultPulsarReaderFactoryTests implements PulsarTestContainerSuppo
 
 		@Test
 		void readingFromTheMiddleOfTheTopic() throws Exception {
-			Map<String, Object> prodConfig = Map.of("topicName", "reading-from-the-middle-of-topic");
 			PulsarProducerFactory<String> pulsarProducerFactory = new DefaultPulsarProducerFactory<>(pulsarClient,
-					prodConfig);
+					"reading-from-the-middle-of-topic");
 			PulsarTemplate<String> pulsarTemplate = new PulsarTemplate<>(pulsarProducerFactory);
 			MessageId[] messageIds = new MessageId[10];
 
@@ -113,9 +110,8 @@ public class DefaultPulsarReaderFactoryTests implements PulsarTestContainerSuppo
 		void readingFromTheEndOfTheTopic() throws Exception {
 			Message<String> message;
 
-			Map<String, Object> prodConfig = Map.of("topicName", "basic-pulsar-reader-topic");
 			PulsarProducerFactory<String> pulsarProducerFactory = new DefaultPulsarProducerFactory<>(pulsarClient,
-					prodConfig);
+					"basic-pulsar-reader-topic");
 			PulsarTemplate<String> pulsarTemplate = new PulsarTemplate<>(pulsarProducerFactory);
 			pulsarTemplate.send("hello john doe");
 

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

@@ -86,10 +86,8 @@ class FailoverConsumerTests implements PulsarTestContainerSupport {
 				pulsarConsumerFactory, pulsarContainerProperties);
 		container3.start();
 
-		Map<String, Object> prodConfig = Map.of("topicName", "my-part-topic-1", "messageRoutingMode",
-				MessageRoutingMode.CustomPartition);
 		DefaultPulsarProducerFactory<String> pulsarProducerFactory = new DefaultPulsarProducerFactory<>(pulsarClient,
-				prodConfig);
+				"my-part-topic-1", (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,7 +26,6 @@ import static org.mockito.Mockito.verify;
 
 import java.util.Arrays;
 import java.util.Collections;
-import java.util.Map;
 import java.util.Set;
 
 import org.apache.pulsar.client.api.Producer;
@@ -42,6 +41,7 @@ import org.junit.jupiter.api.Nested;
 import org.junit.jupiter.api.Test;
 import org.mockito.InOrder;
 
+import org.springframework.lang.Nullable;
 import org.springframework.pulsar.test.support.PulsarTestContainerSupport;
 
 /**
@@ -91,15 +91,15 @@ abstract class PulsarProducerFactoryTests implements PulsarTestContainerSupport
 	}
 
 	protected PulsarProducerFactory<String> newProducerFactory() {
-		return producerFactory(pulsarClient, Collections.emptyMap());
+		return producerFactory(pulsarClient, null, null);
 	}
 
 	protected PulsarProducerFactory<String> newProducerFactoryWithDefaultTopic(String defaultTopic) {
-		return producerFactory(pulsarClient, Collections.singletonMap("topicName", defaultTopic));
+		return producerFactory(pulsarClient, defaultTopic, null);
 	}
 
 	private PulsarProducerFactory<String> newProducerFactoryWithDefaultKeys(Set<String> defaultKeys) {
-		return producerFactory(pulsarClient, Collections.singletonMap("encryptionKeys", defaultKeys));
+		return producerFactory(pulsarClient, null, (pb) -> defaultKeys.forEach(pb::addEncryptionKey));
 	}
 
 	/**
@@ -114,11 +114,12 @@ abstract class PulsarProducerFactoryTests implements PulsarTestContainerSupport
 	/**
 	 * Subclasses override to provide concrete {@link PulsarProducerFactory} instance.
 	 * @param pulsarClient the Pulsar client
-	 * @param producerConfig the Pulsar producers config
+	 * @param defaultTopic the default topic to use for the producers
+	 * @param defaultConfigCustomizer the default configuration to apply to the producers
 	 * @return a Pulsar producer factory instance to use for the tests
 	 */
 	protected abstract PulsarProducerFactory<String> producerFactory(PulsarClient pulsarClient,
-			Map<String, Object> producerConfig);
+			@Nullable String defaultTopic, @Nullable ProducerBuilderCustomizer<String> defaultConfigCustomizer);
 
 	@Test
 	@SuppressWarnings("unchecked")
@@ -223,19 +224,20 @@ abstract class PulsarProducerFactoryTests implements PulsarTestContainerSupport
 
 		@Test
 		void withDefaultEncryptionKeys() throws PulsarClientException {
-			var keys = Set.of("key");
-			var producerFactory = newProducerFactoryWithDefaultKeys(keys);
+			var defaultKeys = Set.of("default-key");
+			var producerFactory = newProducerFactoryWithDefaultKeys(defaultKeys);
 			try (var producer = producerFactory.createProducer(schema, "topic0")) {
-				assertThatProducerHasEncryptionKeys(producer, keys);
+				assertThatProducerHasEncryptionKeys(producer, defaultKeys);
 			}
 		}
 
 		@Test
 		void specificEncryptionKeys() throws PulsarClientException {
-			var keys = Set.of("key");
-			var producerFactory = newProducerFactory();
-			try (var producer = producerFactory.createProducer(schema, "topic0", keys, null)) {
-				assertThatProducerHasEncryptionKeys(producer, keys);
+			var defaultKeys = Set.of("default-key");
+			var userSpecifiedKeys = Set.of("user-key");
+			var producerFactory = newProducerFactoryWithDefaultKeys(defaultKeys);
+			try (var producer = producerFactory.createProducer(schema, "topic0", userSpecifiedKeys, null)) {
+				assertThatProducerHasEncryptionKeys(producer, userSpecifiedKeys);
 			}
 		}
 

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

@@ -29,9 +29,7 @@ import static org.mockito.Mockito.when;
 import java.time.Duration;
 import java.util.ArrayList;
 import java.util.Collections;
-import java.util.HashMap;
 import java.util.List;
-import java.util.Map;
 import java.util.Objects;
 import java.util.Set;
 import java.util.UUID;
@@ -192,8 +190,7 @@ class PulsarTemplateTests implements PulsarTestContainerSupport {
 	@ParameterizedTest(name = "{0}")
 	@MethodSource("interceptorInvocationTestProvider")
 	void interceptorInvocationTest(String topic, List<ProducerInterceptor> interceptors) throws Exception {
-		PulsarProducerFactory<String> producerFactory = new DefaultPulsarProducerFactory<>(client,
-				Collections.singletonMap("topicName", topic));
+		PulsarProducerFactory<String> producerFactory = new DefaultPulsarProducerFactory<>(client, topic);
 		PulsarTemplate<String> pulsarTemplate = new PulsarTemplate<>(producerFactory, interceptors);
 		pulsarTemplate.send("test-interceptor");
 		for (ProducerInterceptor interceptor : interceptors) {
@@ -214,8 +211,7 @@ class PulsarTemplateTests implements PulsarTestContainerSupport {
 	void sendMessageWithTopicInferredByTypeMappings(boolean producerFactoryHasDefaultTopic) throws Exception {
 		String topic = "ptt-topicInferred-" + producerFactoryHasDefaultTopic + "-topic";
 		PulsarProducerFactory<Foo> producerFactory = new DefaultPulsarProducerFactory<>(client,
-				producerFactoryHasDefaultTopic ? Collections.singletonMap("topicName", "fake-topic")
-						: Collections.emptyMap());
+				producerFactoryHasDefaultTopic ? "fake-topic" : null);
 		// Topic mappings allows not specifying the topic when sending (nor having
 		// default on producer)
 		DefaultTopicResolver topicResolver = new DefaultTopicResolver();
@@ -229,8 +225,7 @@ class PulsarTemplateTests implements PulsarTestContainerSupport {
 
 	@Test
 	void sendMessageWithoutTopicFails() {
-		PulsarProducerFactory<String> senderFactory = new DefaultPulsarProducerFactory<>(client,
-				Collections.emptyMap());
+		PulsarProducerFactory<String> senderFactory = new DefaultPulsarProducerFactory<>(client);
 		PulsarTemplate<String> pulsarTemplate = new PulsarTemplate<>(senderFactory);
 		assertThatIllegalArgumentException().isThrownBy(() -> pulsarTemplate.send("test-message"))
 				.withMessage("Topic must be specified when no default topic is configured");
@@ -238,11 +233,8 @@ class PulsarTemplateTests implements PulsarTestContainerSupport {
 
 	private <T> Message<T> sendAndConsume(ThrowingConsumer<PulsarTemplate<T>> sendFunction, String topic,
 			Schema<T> schema, T expectedValue, Boolean withDefaultTopic) throws Exception {
-		Map<String, Object> config = new HashMap<>();
-		if (withDefaultTopic) {
-			config.put("topicName", topic);
-		}
-		PulsarProducerFactory<T> senderFactory = new DefaultPulsarProducerFactory<>(client, config);
+		PulsarProducerFactory<T> senderFactory = new DefaultPulsarProducerFactory<>(client,
+				withDefaultTopic ? topic : null);
 		PulsarTemplate<T> pulsarTemplate = new PulsarTemplate<>(senderFactory);
 		return sendAndConsume(pulsarTemplate, sendFunction, topic, schema, expectedValue);
 	}
@@ -282,8 +274,7 @@ class PulsarTemplateTests implements PulsarTestContainerSupport {
 		@Test
 		void withSchemaInferredByTypeMappings() throws Exception {
 			String topic = "ptt-schemaInferred-topic";
-			PulsarProducerFactory<Foo> producerFactory = new DefaultPulsarProducerFactory<>(client,
-					Collections.singletonMap("topicName", topic));
+			PulsarProducerFactory<Foo> producerFactory = new DefaultPulsarProducerFactory<>(client, topic);
 			// Custom schema resolver allows not specifying the schema when sending
 			DefaultSchemaResolver schemaResolver = new DefaultSchemaResolver();
 			schemaResolver.addCustomSchemaMapping(Foo.class, Schema.JSON(Foo.class));
@@ -301,9 +292,8 @@ class PulsarTemplateTests implements PulsarTestContainerSupport {
 
 		@Test
 		void sendNullWithDefaultTopicFails() {
-			HashMap<String, Object> config = new HashMap<>();
-			config.put("topicName", "sendNullWithDefaultTopicFails");
-			PulsarProducerFactory<String> senderFactory = new DefaultPulsarProducerFactory<>(client, config);
+			PulsarProducerFactory<String> senderFactory = new DefaultPulsarProducerFactory<>(client,
+					"sendNullWithDefaultTopicFails");
 			PulsarTemplate<String> pulsarTemplate = new PulsarTemplate<>(senderFactory);
 			assertThatIllegalArgumentException().isThrownBy(() -> pulsarTemplate.send(null, Schema.STRING))
 					.withMessage("Topic must be specified when the message is null");
@@ -311,8 +301,7 @@ class PulsarTemplateTests implements PulsarTestContainerSupport {
 
 		@Test
 		void sendNullWithoutSchemaFails() {
-			PulsarProducerFactory<Object> senderFactory = new DefaultPulsarProducerFactory<>(client,
-					Collections.emptyMap());
+			PulsarProducerFactory<Object> senderFactory = new DefaultPulsarProducerFactory<>(client);
 			PulsarTemplate<Object> pulsarTemplate = new PulsarTemplate<>(senderFactory);
 			assertThatIllegalArgumentException()
 					.isThrownBy(() -> pulsarTemplate.send("sendNullWithoutSchemaFails", null, null))

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

@@ -76,9 +76,8 @@ public class SharedSubscriptionConsumerTests implements PulsarTestContainerSuppo
 			container3 = createAndStartContainer(pulsarConsumerFactory, latch3, "three", messageCountByKey3,
 					SubscriptionType.Shared);
 
-			Map<String, Object> prodConfig = Map.of("topicName", "shared-subscription-single-msg-test-topic");
 			DefaultPulsarProducerFactory<String> pulsarProducerFactory = new DefaultPulsarProducerFactory<>(
-					pulsarClient, prodConfig);
+					pulsarClient, "shared-subscription-single-msg-test-topic");
 			PulsarTemplate<String> pulsarTemplate = new PulsarTemplate<>(pulsarProducerFactory);
 
 			pulsarTemplate.newMessage("hello john doe").sendAsync();
@@ -132,7 +131,7 @@ public class SharedSubscriptionConsumerTests implements PulsarTestContainerSuppo
 			Thread.sleep(5_000);
 
 			DefaultPulsarProducerFactory<String> producerFactory = new DefaultPulsarProducerFactory<>(pulsarClient,
-					Map.of("topicName", "key-shared-batch-disabled-topic", "batchingEnabled", "false"));
+					"key-shared-batch-disabled-topic", (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/DefaultPulsarConsumerErrorHandlerTests.java

@@ -74,9 +74,8 @@ public class DefaultPulsarConsumerErrorHandlerTests implements PulsarTestContain
 		pulsarContainerProperties.setMessageListener(messageListener);
 		pulsarContainerProperties.setSchema(Schema.STRING);
 
-		Map<String, Object> prodConfig = Map.of("topicName", "default-error-handler-tests-1");
 		DefaultPulsarProducerFactory<String> pulsarProducerFactory = new DefaultPulsarProducerFactory<>(pulsarClient,
-				prodConfig);
+				"default-error-handler-tests-1");
 		PulsarTemplate<String> pulsarTemplate = new PulsarTemplate<>(pulsarProducerFactory);
 		PulsarTemplate<String> mockPulsarTemplate = mock(PulsarTemplate.class, RETURNS_DEEP_STUBS);
 
@@ -127,9 +126,8 @@ public class DefaultPulsarConsumerErrorHandlerTests implements PulsarTestContain
 		pulsarContainerProperties.setMessageListener(messageListener);
 		pulsarContainerProperties.setSchema(Schema.STRING);
 
-		Map<String, Object> prodConfig = Map.of("topicName", "default-error-handler-tests-2");
 		DefaultPulsarProducerFactory<String> pulsarProducerFactory = new DefaultPulsarProducerFactory<>(pulsarClient,
-				prodConfig);
+				"default-error-handler-tests-2");
 		PulsarTemplate<String> pulsarTemplate = new PulsarTemplate<>(pulsarProducerFactory);
 		PulsarTemplate<String> mockPulsarTemplate = mock(PulsarTemplate.class);
 
@@ -174,9 +172,8 @@ public class DefaultPulsarConsumerErrorHandlerTests implements PulsarTestContain
 		pulsarContainerProperties.setMessageListener(messageListener);
 		pulsarContainerProperties.setSchema(Schema.INT32);
 
-		Map<String, Object> prodConfig = Map.of("topicName", "default-error-handler-tests-3");
 		DefaultPulsarProducerFactory<Integer> pulsarProducerFactory = new DefaultPulsarProducerFactory<>(pulsarClient,
-				prodConfig);
+				"default-error-handler-tests-3");
 		PulsarTemplate<Integer> pulsarTemplate = new PulsarTemplate<>(pulsarProducerFactory);
 		PulsarTemplate<Integer> mockPulsarTemplate = mock(PulsarTemplate.class, RETURNS_DEEP_STUBS);
 
@@ -252,9 +249,8 @@ public class DefaultPulsarConsumerErrorHandlerTests implements PulsarTestContain
 
 		container.start();
 
-		Map<String, Object> prodConfig = Map.of("topicName", "default-error-handler-tests-4");
 		DefaultPulsarProducerFactory<Integer> pulsarProducerFactory = new DefaultPulsarProducerFactory<>(pulsarClient,
-				prodConfig);
+				"default-error-handler-tests-4");
 		PulsarTemplate<Integer> pulsarTemplate = new PulsarTemplate<>(pulsarProducerFactory);
 		for (int i = 0; i < 10; i++) {
 			pulsarTemplate.sendAsync(i);
@@ -321,9 +317,8 @@ public class DefaultPulsarConsumerErrorHandlerTests implements PulsarTestContain
 
 		container.start();
 
-		Map<String, Object> prodConfig = Map.of("topicName", "default-error-handler-tests-5");
 		DefaultPulsarProducerFactory<Integer> pulsarProducerFactory = new DefaultPulsarProducerFactory<>(pulsarClient,
-				prodConfig);
+				"default-error-handler-tests-5");
 		PulsarTemplate<Integer> pulsarTemplate = new PulsarTemplate<>(pulsarProducerFactory);
 		for (int i = 0; i < 10; i++) {
 			pulsarTemplate.sendAsync(i);
@@ -389,9 +384,8 @@ public class DefaultPulsarConsumerErrorHandlerTests implements PulsarTestContain
 
 		container.start();
 
-		Map<String, Object> prodConfig = Map.of("topicName", "default-error-handler-tests-6");
 		DefaultPulsarProducerFactory<Integer> pulsarProducerFactory = new DefaultPulsarProducerFactory<>(pulsarClient,
-				prodConfig);
+				"default-error-handler-tests-6");
 		PulsarTemplate<Integer> pulsarTemplate = new PulsarTemplate<>(pulsarProducerFactory);
 		for (int i = 0; i < 10; i++) {
 			pulsarTemplate.sendAsync(i);
@@ -463,9 +457,8 @@ public class DefaultPulsarConsumerErrorHandlerTests implements PulsarTestContain
 
 		container.start();
 
-		Map<String, Object> prodConfig = Map.of("topicName", "default-error-handler-tests-7");
 		DefaultPulsarProducerFactory<Integer> pulsarProducerFactory = new DefaultPulsarProducerFactory<>(pulsarClient,
-				prodConfig);
+				"default-error-handler-tests-7");
 		PulsarTemplate<Integer> pulsarTemplate = new PulsarTemplate<>(pulsarProducerFactory);
 		for (int i = 0; i < 10; i++) {
 			pulsarTemplate.sendAsync(i);
@@ -533,9 +526,8 @@ public class DefaultPulsarConsumerErrorHandlerTests implements PulsarTestContain
 
 		container.start();
 
-		Map<String, Object> prodConfig = Map.of("topicName", "default-error-handler-tests-8");
 		DefaultPulsarProducerFactory<Integer> pulsarProducerFactory = new DefaultPulsarProducerFactory<>(pulsarClient,
-				prodConfig);
+				"default-error-handler-tests-8");
 		PulsarTemplate<Integer> pulsarTemplate = new PulsarTemplate<>(pulsarProducerFactory);
 		for (int i = 0; i < 10; i++) {
 			pulsarTemplate.sendAsync(i);

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

@@ -81,9 +81,8 @@ class DefaultPulsarMessageListenerContainerTests implements PulsarTestContainerS
 				pulsarConsumerFactory, pulsarContainerProperties);
 		container.start();
 
-		Map<String, Object> prodConfig = Map.of("topicName", "dpmlct-012");
 		DefaultPulsarProducerFactory<String> pulsarProducerFactory = new DefaultPulsarProducerFactory<>(pulsarClient,
-				prodConfig);
+				"dpmlct-012");
 		PulsarTemplate<String> pulsarTemplate = new PulsarTemplate<>(pulsarProducerFactory);
 		pulsarTemplate.sendAsync("hello john doe");
 		assertThat(latch.await(10, TimeUnit.SECONDS)).isTrue();
@@ -178,9 +177,8 @@ class DefaultPulsarMessageListenerContainerTests implements PulsarTestContainerS
 		DefaultPulsarMessageListenerContainer<String> container = new DefaultPulsarMessageListenerContainer<>(
 				pulsarConsumerFactory, pulsarContainerProperties);
 
-		Map<String, Object> prodConfig = Map.of("topicName", "dpmlct-013");
 		DefaultPulsarProducerFactory<String> pulsarProducerFactory = new DefaultPulsarProducerFactory<>(pulsarClient,
-				prodConfig);
+				"dpmlct-013");
 		PulsarTemplate<String> pulsarTemplate = new PulsarTemplate<>(pulsarProducerFactory);
 		for (int i = 0; i < 5; i++) {
 			pulsarTemplate.send("hello john doe" + i);
@@ -209,9 +207,8 @@ class DefaultPulsarMessageListenerContainerTests implements PulsarTestContainerS
 		DefaultPulsarMessageListenerContainer<String> container = new DefaultPulsarMessageListenerContainer<>(
 				pulsarConsumerFactory, pulsarContainerProperties);
 
-		Map<String, Object> prodConfig = Map.of("topicName", "dpmlct-014");
 		DefaultPulsarProducerFactory<String> pulsarProducerFactory = new DefaultPulsarProducerFactory<>(pulsarClient,
-				prodConfig);
+				"dpmlct-014");
 		PulsarTemplate<String> pulsarTemplate = new PulsarTemplate<>(pulsarProducerFactory);
 		for (int i = 0; i < 5; i++) {
 			pulsarTemplate.send("hello john doe" + i);
@@ -252,9 +249,8 @@ class DefaultPulsarMessageListenerContainerTests implements PulsarTestContainerS
 
 		Consumer<String> containerConsumer = ConsumerTestUtils.startContainerAndSpyOnConsumer(container);
 
-		Map<String, Object> prodConfig = Collections.singletonMap("topicName", "dpmlct-015");
 		DefaultPulsarProducerFactory<String> pulsarProducerFactory = new DefaultPulsarProducerFactory<>(pulsarClient,
-				prodConfig);
+				"dpmlct-015");
 		PulsarTemplate<String> pulsarTemplate = new PulsarTemplate<>(pulsarProducerFactory);
 		for (int i = 0; i < 5; i++) {
 			pulsarTemplate.send("hello john doe" + i);
@@ -312,9 +308,8 @@ class DefaultPulsarMessageListenerContainerTests implements PulsarTestContainerS
 				pulsarConsumerFactory, pulsarContainerProperties);
 		container.start();
 
-		Map<String, Object> prodConfig = Collections.singletonMap("topicName", "dpmlct-016");
 		DefaultPulsarProducerFactory<Integer> pulsarProducerFactory = new DefaultPulsarProducerFactory<>(pulsarClient,
-				prodConfig);
+				"dpmlct-016");
 		PulsarTemplate<Integer> pulsarTemplate = new PulsarTemplate<>(pulsarProducerFactory);
 		for (int i = 1; i < 6; i++) {
 			pulsarTemplate.send(i);
@@ -371,9 +366,8 @@ class DefaultPulsarMessageListenerContainerTests implements PulsarTestContainerS
 				pulsarConsumerFactory, pulsarContainerProperties);
 		container.start();
 
-		Map<String, Object> prodConfig = Collections.singletonMap("topicName", "dpmlct-017");
 		DefaultPulsarProducerFactory<Integer> pulsarProducerFactory = new DefaultPulsarProducerFactory<>(pulsarClient,
-				prodConfig);
+				"dpmlct-017");
 		PulsarTemplate<Integer> pulsarTemplate = new PulsarTemplate<>(pulsarProducerFactory);
 		for (int i = 1; i < 6; i++) {
 			pulsarTemplate.send(i);

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

@@ -22,10 +22,8 @@ import static org.assertj.core.api.Assertions.assertThatThrownBy;
 import java.nio.charset.StandardCharsets;
 import java.time.Duration;
 import java.util.ArrayList;
-import java.util.Collections;
 import java.util.HashMap;
 import java.util.List;
-import java.util.Map;
 import java.util.Objects;
 import java.util.Properties;
 import java.util.concurrent.CountDownLatch;
@@ -105,8 +103,7 @@ public class PulsarListenerTests implements PulsarTestContainerSupport {
 
 		@Bean
 		public PulsarProducerFactory<String> pulsarProducerFactory(PulsarClient pulsarClient) {
-			Map<String, Object> config = Map.of("topicName", "foo-1");
-			return new DefaultPulsarProducerFactory<>(pulsarClient, config);
+			return new DefaultPulsarProducerFactory<>(pulsarClient, "foo-1");
 		}
 
 		@Bean
@@ -121,16 +118,14 @@ public class PulsarListenerTests implements PulsarTestContainerSupport {
 
 		@Bean
 		public PulsarConsumerFactory<?> pulsarConsumerFactory(PulsarClient pulsarClient) {
-			Map<String, Object> config = new HashMap<>();
-			return new DefaultPulsarConsumerFactory<>(pulsarClient, config);
+			return new DefaultPulsarConsumerFactory<>(pulsarClient, new HashMap<>());
 		}
 
 		@Bean
 		PulsarListenerContainerFactory pulsarListenerContainerFactory(
 				PulsarConsumerFactory<Object> pulsarConsumerFactory) {
-			ConcurrentPulsarListenerContainerFactory<?> pulsarListenerContainerFactory = new ConcurrentPulsarListenerContainerFactory<>(
-					pulsarConsumerFactory, new PulsarContainerProperties());
-			return pulsarListenerContainerFactory;
+			return new ConcurrentPulsarListenerContainerFactory<>(pulsarConsumerFactory,
+					new PulsarContainerProperties());
 		}
 
 		@Bean
@@ -184,12 +179,10 @@ public class PulsarListenerTests implements PulsarTestContainerSupport {
 		@Test
 		void concurrencyOnPulsarListenerWithFailoverSubscription(@Autowired PulsarListenerEndpointRegistry registry)
 				throws Exception {
-			PulsarProducerFactory<String> pulsarProducerFactory = new DefaultPulsarProducerFactory<>(pulsarClient,
-					Map.of("batchingEnabled", false));
-			PulsarTemplate<String> customTemplate = new PulsarTemplate<>(pulsarProducerFactory);
-
-			ConcurrentPulsarMessageListenerContainer<?> bar = (ConcurrentPulsarMessageListenerContainer<?>) registry
-					.getListenerContainer("bar");
+			var pulsarProducerFactory = new DefaultPulsarProducerFactory<String>(pulsarClient, null,
+					(pb) -> pb.enableBatching(false));
+			var customTemplate = new PulsarTemplate<>(pulsarProducerFactory);
+			var bar = (ConcurrentPulsarMessageListenerContainer<?>) registry.getListenerContainer("bar");
 
 			assertThat(bar.getConcurrency()).isEqualTo(3);
 
@@ -203,12 +196,10 @@ public class PulsarListenerTests implements PulsarTestContainerSupport {
 		@Test
 		void nonDefaultConcurrencySettingNotAllowedOnExclusiveSubscriptions(
 				@Autowired PulsarListenerEndpointRegistry registry) throws Exception {
-			PulsarProducerFactory<String> pulsarProducerFactory = new DefaultPulsarProducerFactory<>(pulsarClient,
-					Map.of("batchingEnabled", false));
-			PulsarTemplate<String> customTemplate = new PulsarTemplate<>(pulsarProducerFactory);
-
-			ConcurrentPulsarMessageListenerContainer<?> bar = (ConcurrentPulsarMessageListenerContainer<?>) registry
-					.getListenerContainer("bar");
+			var pulsarProducerFactory = new DefaultPulsarProducerFactory<String>(pulsarClient, null,
+					(pb) -> pb.enableBatching(false));
+			var customTemplate = new PulsarTemplate<>(pulsarProducerFactory);
+			var bar = (ConcurrentPulsarMessageListenerContainer<?>) registry.getListenerContainer("bar");
 
 			assertThat(bar.getConcurrency()).isEqualTo(3);
 
@@ -460,10 +451,9 @@ public class PulsarListenerTests implements PulsarTestContainerSupport {
 
 		@Test
 		void jsonSchema() throws Exception {
-			PulsarProducerFactory<User> pulsarProducerFactory = new DefaultPulsarProducerFactory<>(pulsarClient,
-					Collections.emptyMap());
-			PulsarTemplate<User> template = new PulsarTemplate<>(pulsarProducerFactory);
-			Schema<User> schema = JSONSchema.of(User.class);
+			var pulsarProducerFactory = new DefaultPulsarProducerFactory<User>(pulsarClient);
+			var template = new PulsarTemplate<>(pulsarProducerFactory);
+			var schema = JSONSchema.of(User.class);
 			for (int i = 0; i < 3; i++) {
 				template.send("json-topic", new User("Jason", i), schema);
 			}
@@ -473,10 +463,9 @@ public class PulsarListenerTests implements PulsarTestContainerSupport {
 
 		@Test
 		void avroSchema() throws Exception {
-			PulsarProducerFactory<User> pulsarProducerFactory = new DefaultPulsarProducerFactory<>(pulsarClient,
-					Collections.emptyMap());
-			PulsarTemplate<User> template = new PulsarTemplate<>(pulsarProducerFactory);
-			Schema<User> schema = AvroSchema.of(User.class);
+			var pulsarProducerFactory = new DefaultPulsarProducerFactory<User>(pulsarClient);
+			var template = new PulsarTemplate<>(pulsarProducerFactory);
+			var schema = AvroSchema.of(User.class);
 			for (int i = 0; i < 3; i++) {
 				template.send("avro-topic", new User("Avi", i), schema);
 			}
@@ -486,11 +475,9 @@ public class PulsarListenerTests implements PulsarTestContainerSupport {
 
 		@Test
 		void keyvalueSchema() throws Exception {
-			PulsarProducerFactory<KeyValue<String, Integer>> pulsarProducerFactory = new DefaultPulsarProducerFactory<>(
-					pulsarClient, Collections.emptyMap());
-			PulsarTemplate<KeyValue<String, Integer>> template = new PulsarTemplate<>(pulsarProducerFactory);
-			Schema<KeyValue<String, Integer>> kvSchema = Schema.KeyValue(Schema.STRING, Schema.INT32,
-					KeyValueEncodingType.INLINE);
+			var pulsarProducerFactory = new DefaultPulsarProducerFactory<KeyValue<String, Integer>>(pulsarClient);
+			var template = new PulsarTemplate<>(pulsarProducerFactory);
+			var kvSchema = Schema.KeyValue(Schema.STRING, Schema.INT32, KeyValueEncodingType.INLINE);
 			for (int i = 0; i < 3; i++) {
 				template.send("keyvalue-topic", new KeyValue<>("Kevin", i), kvSchema);
 			}
@@ -500,10 +487,9 @@ public class PulsarListenerTests implements PulsarTestContainerSupport {
 
 		@Test
 		void protobufSchema() throws Exception {
-			PulsarProducerFactory<Proto.Person> pulsarProducerFactory = new DefaultPulsarProducerFactory<>(pulsarClient,
-					Collections.emptyMap());
-			PulsarTemplate<Proto.Person> template = new PulsarTemplate<>(pulsarProducerFactory);
-			Schema<Proto.Person> schema = ProtobufSchema.of(Proto.Person.class);
+			var pulsarProducerFactory = new DefaultPulsarProducerFactory<Proto.Person>(pulsarClient);
+			var template = new PulsarTemplate<>(pulsarProducerFactory);
+			var schema = ProtobufSchema.of(Proto.Person.class);
 			for (int i = 0; i < 3; i++) {
 				template.send("protobuf-topic", Proto.Person.newBuilder().setId(i).setName("Paul").build(), schema);
 			}
@@ -640,10 +626,9 @@ public class PulsarListenerTests implements PulsarTestContainerSupport {
 
 		@Test
 		void jsonSchema() throws Exception {
-			PulsarProducerFactory<User2> pulsarProducerFactory = new DefaultPulsarProducerFactory<>(pulsarClient,
-					Collections.emptyMap());
-			PulsarTemplate<User2> template = new PulsarTemplate<>(pulsarProducerFactory);
-			Schema<User2> schema = Schema.JSON(User2.class);
+			var pulsarProducerFactory = new DefaultPulsarProducerFactory<User2>(pulsarClient);
+			var template = new PulsarTemplate<>(pulsarProducerFactory);
+			var schema = Schema.JSON(User2.class);
 			for (int i = 0; i < 3; i++) {
 				template.send("json-custom-mappings-topic", new User2("Jason", i), schema);
 			}
@@ -652,10 +637,9 @@ public class PulsarListenerTests implements PulsarTestContainerSupport {
 
 		@Test
 		void avroSchema() throws Exception {
-			PulsarProducerFactory<User> pulsarProducerFactory = new DefaultPulsarProducerFactory<>(pulsarClient,
-					Collections.emptyMap());
-			PulsarTemplate<User> template = new PulsarTemplate<>(pulsarProducerFactory);
-			Schema<User> schema = AvroSchema.of(User.class);
+			var pulsarProducerFactory = new DefaultPulsarProducerFactory<User>(pulsarClient);
+			var template = new PulsarTemplate<>(pulsarProducerFactory);
+			var schema = AvroSchema.of(User.class);
 			for (int i = 0; i < 3; i++) {
 				template.send("avro-custom-mappings-topic", new User("Avi", i), schema);
 			}
@@ -664,11 +648,9 @@ public class PulsarListenerTests implements PulsarTestContainerSupport {
 
 		@Test
 		void keyvalueSchema() throws Exception {
-			PulsarProducerFactory<KeyValue<String, User2>> pulsarProducerFactory = new DefaultPulsarProducerFactory<>(
-					pulsarClient, Collections.emptyMap());
-			PulsarTemplate<KeyValue<String, User2>> template = new PulsarTemplate<>(pulsarProducerFactory);
-			Schema<KeyValue<String, User2>> kvSchema = Schema.KeyValue(Schema.STRING, Schema.JSON(User2.class),
-					KeyValueEncodingType.INLINE);
+			var pulsarProducerFactory = new DefaultPulsarProducerFactory<KeyValue<String, User2>>(pulsarClient);
+			var template = new PulsarTemplate<>(pulsarProducerFactory);
+			var kvSchema = Schema.KeyValue(Schema.STRING, Schema.JSON(User2.class), KeyValueEncodingType.INLINE);
 			for (int i = 0; i < 3; i++) {
 				template.send("keyvalue-custom-mappings-topic", new KeyValue<>("Kevin", new User2("Kevin", 5150)),
 						kvSchema);
@@ -678,10 +660,9 @@ public class PulsarListenerTests implements PulsarTestContainerSupport {
 
 		@Test
 		void protobufSchema() throws Exception {
-			PulsarProducerFactory<Proto.Person> pulsarProducerFactory = new DefaultPulsarProducerFactory<>(pulsarClient,
-					Collections.emptyMap());
-			PulsarTemplate<Proto.Person> template = new PulsarTemplate<>(pulsarProducerFactory);
-			Schema<Proto.Person> schema = ProtobufSchema.of(Proto.Person.class);
+			var pulsarProducerFactory = new DefaultPulsarProducerFactory<Proto.Person>(pulsarClient);
+			var template = new PulsarTemplate<>(pulsarProducerFactory);
+			var schema = ProtobufSchema.of(Proto.Person.class);
 			for (int i = 0; i < 3; i++) {
 				template.send("protobuf-custom-mappings-topic",
 						Proto.Person.newBuilder().setId(i).setName("Paul").build(), schema);
@@ -749,10 +730,9 @@ public class PulsarListenerTests implements PulsarTestContainerSupport {
 
 		@Test
 		void complexMessageTypeTopicMapping() throws Exception {
-			PulsarProducerFactory<User2> pulsarProducerFactory = new DefaultPulsarProducerFactory<>(pulsarClient,
-					Collections.emptyMap());
-			PulsarTemplate<User2> template = new PulsarTemplate<>(pulsarProducerFactory);
-			Schema<User2> schema = Schema.JSON(User2.class);
+			var pulsarProducerFactory = new DefaultPulsarProducerFactory<User2>(pulsarClient);
+			var template = new PulsarTemplate<>(pulsarProducerFactory);
+			var schema = Schema.JSON(User2.class);
 			for (int i = 0; i < 3; i++) {
 				template.send("plt-topicMapping-user-topic", new User2("Jason", i), schema);
 			}
@@ -761,9 +741,8 @@ public class PulsarListenerTests implements PulsarTestContainerSupport {
 
 		@Test
 		void primitiveMessageTypeTopicMapping() throws Exception {
-			PulsarProducerFactory<String> pulsarProducerFactory = new DefaultPulsarProducerFactory<>(pulsarClient,
-					Collections.emptyMap());
-			PulsarTemplate<String> template = new PulsarTemplate<>(pulsarProducerFactory);
+			var pulsarProducerFactory = new DefaultPulsarProducerFactory<String>(pulsarClient);
+			var template = new PulsarTemplate<>(pulsarProducerFactory);
 			for (int i = 0; i < 3; i++) {
 				template.send("plt-topicMapping-string-topic", "Susan " + i, Schema.STRING);
 			}

spring-pulsar/src/test/java/org/springframework/pulsar/observation/ObservationIntegrationTests.java

@@ -131,7 +131,7 @@ public class ObservationIntegrationTests extends SampleTestRunner implements Pul
 
 		@Bean
 		public PulsarProducerFactory<String> pulsarProducerFactory(PulsarClient pulsarClient) {
-			return new DefaultPulsarProducerFactory<>(pulsarClient, Collections.emptyMap());
+			return new DefaultPulsarProducerFactory<>(pulsarClient);
 		}
 
 		@Bean

spring-pulsar/src/test/java/org/springframework/pulsar/observation/ObservationTests.java

@@ -170,7 +170,7 @@ public class ObservationTests implements PulsarTestContainerSupport {
 
 		@Bean
 		PulsarProducerFactory<String> pulsarProducerFactory(PulsarClient pulsarClient) {
-			return new DefaultPulsarProducerFactory<>(pulsarClient, Collections.emptyMap());
+			return new DefaultPulsarProducerFactory<>(pulsarClient);
 		}
 
 		@Bean

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

@@ -44,6 +44,7 @@ import org.springframework.pulsar.test.support.PulsarTestContainerSupport;
  * Basic tests for {@link DefaultPulsarMessageReaderContainer}.
  *
  * @author Soby Chacko
+ * @author Chris Bono
  */
 public class DefaultPulsarMessageReaderContainerTests implements PulsarTestContainerSupport {
 
@@ -83,9 +84,8 @@ public class DefaultPulsarMessageReaderContainerTests implements PulsarTestConta
 			container = new DefaultPulsarMessageReaderContainer<>(pulsarReaderFactory, readerContainerProperties);
 			container.start();
 
-			Map<String, Object> prodConfig = Map.of("topicName", "dprlct-001");
 			DefaultPulsarProducerFactory<String> pulsarProducerFactory = new DefaultPulsarProducerFactory<>(
-					pulsarClient, prodConfig);
+					pulsarClient, "dprlct-001", (pb) -> pb.topic("dprlct-001"));
 			PulsarTemplate<String> pulsarTemplate = new PulsarTemplate<>(pulsarProducerFactory);
 			pulsarTemplate.sendAsync("hello john doe");
 			assertThat(latch.await(10, TimeUnit.SECONDS)).isTrue();
@@ -113,10 +113,8 @@ public class DefaultPulsarMessageReaderContainerTests implements PulsarTestConta
 		try {
 			container = new DefaultPulsarMessageReaderContainer<>(pulsarReaderFactory, containerProps);
 			container.start();
-
-			Map<String, Object> prodConfig = Map.of("topicName", "dprlct-002");
 			DefaultPulsarProducerFactory<String> pulsarProducerFactory = new DefaultPulsarProducerFactory<>(
-					pulsarClient, prodConfig);
+					pulsarClient, "dprlct-002", (pb) -> pb.topic("dprlct-002"));
 			PulsarTemplate<String> pulsarTemplate = new PulsarTemplate<>(pulsarProducerFactory);
 			pulsarTemplate.sendAsync("hello buzz doe");
 			assertThat(latch.await(10, TimeUnit.SECONDS)).isTrue();
@@ -145,7 +143,8 @@ public class DefaultPulsarMessageReaderContainerTests implements PulsarTestConta
 			container = new DefaultPulsarMessageReaderContainer<>(readerFactory, containerProps);
 
 			var prodConfig = Map.<String, Object>of("topicName", "dprlct-003");
-			var producerFactory = new DefaultPulsarProducerFactory<>(pulsarClient, prodConfig);
+			var producerFactory = new DefaultPulsarProducerFactory<>(pulsarClient, "dprlct-003",
+					(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

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

@@ -19,9 +19,7 @@ package org.springframework.pulsar.reader;
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.assertj.core.api.Assertions.assertThatThrownBy;
 
-import java.util.Collections;
 import java.util.HashMap;
-import java.util.Map;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
 
@@ -56,6 +54,7 @@ import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
  * {@link PulsarReader} integration tests.
  *
  * @author Soby Chacko
+ * @author Chris Bono
  */
 @SpringJUnitConfig
 @DirtiesContext
@@ -73,8 +72,7 @@ public class PulsarReaderTests implements PulsarTestContainerSupport {
 
 		@Bean
 		public PulsarProducerFactory<String> pulsarProducerFactory(PulsarClient pulsarClient) {
-			Map<String, Object> config = Collections.emptyMap();
-			return new DefaultPulsarProducerFactory<>(pulsarClient, config);
+			return new DefaultPulsarProducerFactory<>(pulsarClient);
 		}
 
 		@Bean
@@ -89,15 +87,13 @@ public class PulsarReaderTests implements PulsarTestContainerSupport {
 
 		@Bean
 		public PulsarReaderFactory<?> pulsarReaderFactory(PulsarClient pulsarClient) {
-			Map<String, Object> config = new HashMap<>();
-			return new DefaultPulsarReaderFactory<>(pulsarClient, config);
+			return new DefaultPulsarReaderFactory<>(pulsarClient, new HashMap<>());
 		}
 
 		@Bean
 		PulsarReaderContainerFactory pulsarReaderContainerFactory(PulsarReaderFactory<Object> pulsarReaderFactory) {
-			DefaultPulsarReaderContainerFactory<?> pulsarReaderContainerFactory = new DefaultPulsarReaderContainerFactory<>(
-					pulsarReaderFactory, new PulsarReaderContainerProperties());
-			return pulsarReaderContainerFactory;
+			return new DefaultPulsarReaderContainerFactory<>(pulsarReaderFactory,
+					new PulsarReaderContainerProperties());
 		}
 
 	}