Port auto-config changes from 0.2.x (#402)

* Port "Use builder to autoconfigure PulsarProducerFactory (#397)" from 0.2.x

* Port "Use builder to autoconfigure PulsarConsumerFactory (#399)" from 0.2.x

* Port "Use builder to autoconfigure PulsarReaderFactory (#400)" from 0.2.x

* Port "Use builder to autoconfigure PulsarAdministration (#401)" from 0.2.x

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;
@@ -30,7 +28,6 @@ import java.util.concurrent.LinkedBlockingQueue;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicReference;
 
-import org.apache.pulsar.client.admin.PulsarAdmin;
 import org.apache.pulsar.client.api.DeadLetterPolicy;
 import org.apache.pulsar.client.api.Message;
 import org.apache.pulsar.client.api.MessageId;
@@ -107,7 +104,7 @@ public class ReactivePulsarListenerTests implements PulsarTestContainerSupport {
 
 		@Bean
 		public PulsarProducerFactory<String> pulsarProducerFactory(PulsarClient pulsarClient) {
-			return new DefaultPulsarProducerFactory<>(pulsarClient, new HashMap<>());
+			return new DefaultPulsarProducerFactory<>(pulsarClient);
 		}
 
 		@Bean
@@ -139,8 +136,7 @@ public class ReactivePulsarListenerTests implements PulsarTestContainerSupport {
 
 		@Bean
 		PulsarAdministration pulsarAdministration() {
-			return new PulsarAdministration(
-					PulsarAdmin.builder().serviceHttpUrl(PulsarTestContainerSupport.getHttpServiceUrl()));
+			return new PulsarAdministration(PulsarTestContainerSupport.getHttpServiceUrl());
 		}
 
 		@Bean
@@ -347,8 +343,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 +353,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 +364,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 +376,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 +488,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 +498,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 +509,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 +522,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 +656,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 +667,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/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/DefaultPulsarClientFactory.java

@@ -37,7 +37,7 @@ public class DefaultPulsarClientFactory implements PulsarClientFactory {
 	 * @param serviceUrl the service url
 	 */
 	public DefaultPulsarClientFactory(String serviceUrl) {
-		this((clientBuilder -> clientBuilder.serviceUrl(serviceUrl)));
+		this((clientBuilder) -> clientBuilder.serviceUrl(serviceUrl));
 	}
 
 	/**

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

@@ -18,7 +18,6 @@ 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;
@@ -30,6 +29,7 @@ import org.apache.pulsar.client.api.ConsumerBuilder;
 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.ConsumerBuilderImpl;
 
 import org.springframework.lang.Nullable;
 import org.springframework.util.CollectionUtils;
@@ -41,22 +41,25 @@ import org.springframework.util.CollectionUtils;
  * @author Soby Chacko
  * @author Alexander Preuß
  * @author Christophe Bornet
+ * @author Chris Bono
  */
 public class DefaultPulsarConsumerFactory<T> implements PulsarConsumerFactory<T> {
 
-	private final Map<String, Object> consumerConfig;
-
 	private final PulsarClient pulsarClient;
 
+	@Nullable
+	private final ConsumerBuilderCustomizer<T> defaultConfigCustomizer;
+
 	/**
 	 * Construct a consumer factory instance.
 	 * @param pulsarClient the client used to consume
-	 * @param consumerConfig default configuration to apply to the created consumer or
-	 * empty map to use no default configuration
+	 * @param defaultConfigCustomizer the default configuration to apply to the consumers
+	 * or null to use no default configuration
 	 */
-	public DefaultPulsarConsumerFactory(PulsarClient pulsarClient, Map<String, Object> consumerConfig) {
+	public DefaultPulsarConsumerFactory(PulsarClient pulsarClient,
+			ConsumerBuilderCustomizer<T> defaultConfigCustomizer) {
 		this.pulsarClient = pulsarClient;
-		this.consumerConfig = Collections.unmodifiableMap(consumerConfig);
+		this.defaultConfigCustomizer = defaultConfigCustomizer;
 	}
 
 	@Override
@@ -72,25 +75,35 @@ public class DefaultPulsarConsumerFactory<T> implements PulsarConsumerFactory<T>
 			@Nullable List<ConsumerBuilderCustomizer<T>> customizers) throws PulsarClientException {
 		Objects.requireNonNull(schema, "Schema must be specified");
 		ConsumerBuilder<T> consumerBuilder = this.pulsarClient.newConsumer(schema);
-		Map<String, Object> config = new HashMap<>(this.consumerConfig);
-		if (topics != null) {
-			config.put("topicNames", new HashSet<>(topics));
+
+		// Apply the default config customizer (preserve the topic)
+		if (this.defaultConfigCustomizer != null) {
+			this.defaultConfigCustomizer.customize(consumerBuilder);
 		}
-		if (metadataProperties != null) {
-			config.put("properties", new TreeMap<>(metadataProperties));
+		if (topics != null) {
+			replaceTopicsOnBuilder(consumerBuilder, topics);
 		}
 		if (subscriptionName != null) {
-			config.put("subscriptionName", subscriptionName);
+			consumerBuilder.subscriptionName(subscriptionName);
+		}
+		if (metadataProperties != null) {
+			replaceMetadataPropertiesOnBuilder(consumerBuilder, metadataProperties);
 		}
-		ConsumerBuilderConfigurationUtil.loadConf(consumerBuilder, config);
 		if (!CollectionUtils.isEmpty(customizers)) {
 			customizers.forEach(customizer -> customizer.customize(consumerBuilder));
 		}
 		return consumerBuilder.subscribe();
 	}
 
-	public Map<String, Object> getConsumerConfig() {
-		return this.consumerConfig;
+	private void replaceTopicsOnBuilder(ConsumerBuilder<T> builder, Collection<String> topics) {
+		var builderImpl = (ConsumerBuilderImpl<T>) builder;
+		builderImpl.getConf().setTopicNames(new HashSet<>(topics));
+	}
+
+	private void replaceMetadataPropertiesOnBuilder(ConsumerBuilder<T> builder,
+			Map<String, String> metadataProperties) {
+		var builderImpl = (ConsumerBuilderImpl<T>) builder;
+		builderImpl.getConf().setProperties(new TreeMap<>(metadataProperties));
 	}
 
 }

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/DefaultPulsarReaderFactory.java

@@ -16,9 +16,9 @@
 
 package org.springframework.pulsar.core;
 
-import java.util.Collections;
+import java.util.Collection;
+import java.util.HashSet;
 import java.util.List;
-import java.util.Map;
 import java.util.Objects;
 
 import org.apache.pulsar.client.api.MessageId;
@@ -27,6 +27,7 @@ import org.apache.pulsar.client.api.PulsarClientException;
 import org.apache.pulsar.client.api.Reader;
 import org.apache.pulsar.client.api.ReaderBuilder;
 import org.apache.pulsar.client.api.Schema;
+import org.apache.pulsar.client.impl.ReaderBuilderImpl;
 
 import org.springframework.lang.Nullable;
 import org.springframework.util.CollectionUtils;
@@ -41,15 +42,27 @@ public class DefaultPulsarReaderFactory<T> implements PulsarReaderFactory<T> {
 
 	private final PulsarClient pulsarClient;
 
-	private final Map<String, Object> readerConfig;
+	@Nullable
+	private final ReaderBuilderCustomizer<T> defaultConfigCustomizer;
 
+	/**
+	 * Construct a reader factory instance with no default configuration.
+	 * @param pulsarClient the client used to consume
+	 */
 	public DefaultPulsarReaderFactory(PulsarClient pulsarClient) {
-		this(pulsarClient, Collections.emptyMap());
+		this(pulsarClient, null);
 	}
 
-	public DefaultPulsarReaderFactory(PulsarClient pulsarClient, Map<String, Object> readerConfig) {
+	/**
+	 * Construct a reader factory instance.
+	 * @param pulsarClient the client used to consume
+	 * @param defaultConfigCustomizer the default configuration to apply to the readers or
+	 * null to use no default configuration
+	 */
+	public DefaultPulsarReaderFactory(PulsarClient pulsarClient,
+			@Nullable ReaderBuilderCustomizer<T> defaultConfigCustomizer) {
 		this.pulsarClient = pulsarClient;
-		this.readerConfig = readerConfig;
+		this.defaultConfigCustomizer = defaultConfigCustomizer;
 	}
 
 	@Override
@@ -57,12 +70,19 @@ public class DefaultPulsarReaderFactory<T> implements PulsarReaderFactory<T> {
 			@Nullable List<ReaderBuilderCustomizer<T>> customizers) throws PulsarClientException {
 		Objects.requireNonNull(schema, "Schema must be specified");
 		ReaderBuilder<T> readerBuilder = this.pulsarClient.newReader(schema);
-		if (!CollectionUtils.isEmpty(topics)) {
-			readerBuilder.topics(topics);
-		}
-		readerBuilder.startMessageId(messageId);
 
-		readerBuilder.loadConf(this.readerConfig);
+		// Apply the default config customizer (preserve the topics)
+		if (this.defaultConfigCustomizer != null) {
+			this.defaultConfigCustomizer.customize(readerBuilder);
+		}
+
+		if (!CollectionUtils.isEmpty(topics)) {
+			replaceTopicsOnBuilder(readerBuilder, topics);
+		}
+
+		if (messageId != null) {
+			readerBuilder.startMessageId(messageId);
+		}
 
 		if (!CollectionUtils.isEmpty(customizers)) {
 			customizers.forEach(customizer -> customizer.customize(readerBuilder));
@@ -71,4 +91,9 @@ public class DefaultPulsarReaderFactory<T> implements PulsarReaderFactory<T> {
 		return readerBuilder.create();
 	}
 
+	private void replaceTopicsOnBuilder(ReaderBuilder<T> builder, Collection<String> topics) {
+		var builderImpl = (ReaderBuilderImpl<T>) builder;
+		builderImpl.getConf().setTopicNames(new HashSet<>(topics));
+	}
+
 }

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

@@ -0,0 +1,35 @@
+/*
+ * Copyright 2023-2023 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.pulsar.core;
+
+import org.apache.pulsar.client.admin.PulsarAdminBuilder;
+
+/**
+ * The interface to customize a {@link PulsarAdminBuilder}.
+ *
+ * @author Chris Bono
+ */
+@FunctionalInterface
+public interface PulsarAdminBuilderCustomizer {
+
+	/**
+	 * Customizes a {@link PulsarAdminBuilder}.
+	 * @param adminBuilder the builder to customize
+	 */
+	void customize(PulsarAdminBuilder adminBuilder);
+
+}

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

@@ -18,20 +18,16 @@ package org.springframework.pulsar.core;
 
 import java.util.Arrays;
 import java.util.Collection;
-import java.util.HashMap;
 import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
 import java.util.Set;
-import java.util.concurrent.TimeUnit;
 import java.util.stream.Collectors;
 
 import org.apache.pulsar.client.admin.PulsarAdmin;
-import org.apache.pulsar.client.admin.PulsarAdminBuilder;
 import org.apache.pulsar.client.admin.PulsarAdminException;
 import org.apache.pulsar.client.api.PulsarClientException;
-import org.apache.pulsar.client.api.PulsarClientException.UnsupportedAuthenticationException;
 
 import org.springframework.beans.BeansException;
 import org.springframework.beans.factory.SmartInitializingSingleton;
@@ -40,7 +36,6 @@ import org.springframework.context.ApplicationContextAware;
 import org.springframework.core.log.LogAccessor;
 import org.springframework.lang.Nullable;
 import org.springframework.util.CollectionUtils;
-import org.springframework.util.StringUtils;
 
 /**
  * An administration class that delegates to {@link PulsarAdmin} to create and manage
@@ -55,28 +50,27 @@ public class PulsarAdministration
 
 	private final LogAccessor logger = new LogAccessor(this.getClass());
 
-	private final PulsarAdminBuilder adminBuilder;
-
 	@Nullable
 	private ApplicationContext applicationContext;
 
+	@Nullable
+	private final PulsarAdminBuilderCustomizer adminCustomizer;
+
 	/**
-	 * Construct a {@code PulsarAdministration} instance using the given configuration for
-	 * the underlying {@link PulsarAdmin}.
-	 * @param adminConfig the {@link PulsarAdmin} configuration
+	 * Construct a default instance using the specified service url.
+	 * @param serviceHttpUrl the admin http service url
 	 */
-	public PulsarAdministration(Map<String, Object> adminConfig) {
-		this.adminBuilder = PulsarAdmin.builder();
-		loadConf(this.adminBuilder, adminConfig);
+	public PulsarAdministration(String serviceHttpUrl) {
+		this((adminBuilder) -> adminBuilder.serviceHttpUrl(serviceHttpUrl));
 	}
 
 	/**
-	 * Construct a {@code PulsarAdministration} instance using the given builder for the
-	 * underlying {@link PulsarAdmin}.
-	 * @param adminBuilder the {@link PulsarAdminBuilder}
+	 * Construct an instance with the specified customizations.
+	 * @param adminCustomizer the customizer to apply to the builder or null to use the
+	 * default admin builder without modifications
 	 */
-	public PulsarAdministration(PulsarAdminBuilder adminBuilder) {
-		this.adminBuilder = adminBuilder;
+	public PulsarAdministration(@Nullable PulsarAdminBuilderCustomizer adminCustomizer) {
+		this.adminCustomizer = adminCustomizer;
 	}
 
 	@Override
@@ -89,39 +83,6 @@ public class PulsarAdministration
 		this.applicationContext = applicationContext;
 	}
 
-	private void loadConf(PulsarAdminBuilder builder, Map<String, Object> adminConfig) {
-		var conf = new HashMap<>(adminConfig);
-
-		// Workaround the fact that the PulsarAdminImpl does not attempt to construct the
-		// timeout settings from the config props
-		if (conf.remove("connectionTimeoutMs") instanceof Integer connectTimeout) {
-			builder.connectionTimeout(connectTimeout, TimeUnit.MILLISECONDS);
-		}
-		if (conf.remove("readTimeoutMs") instanceof Integer readTimeout) {
-			builder.readTimeout(readTimeout, TimeUnit.MILLISECONDS);
-		}
-		if (conf.remove("requestTimeoutMs") instanceof Integer requestTimeout) {
-			builder.requestTimeout(requestTimeout, TimeUnit.MILLISECONDS);
-		}
-		if (conf.remove("autoCertRefreshSeconds") instanceof Integer autoCertRefreshTime) {
-			builder.autoCertRefreshTime(autoCertRefreshTime, TimeUnit.SECONDS);
-		}
-		builder.loadConf(conf);
-
-		// Workaround the fact that the PulsarAdminImpl does not attempt to construct the
-		// authentication from the config props
-		var authPluginClassName = (String) conf.get("authPluginClassName");
-		var authParams = (String) conf.get("authParams");
-		if (StringUtils.hasText(authPluginClassName) && StringUtils.hasText(authParams)) {
-			try {
-				builder.authentication(authPluginClassName, authParams);
-			}
-			catch (UnsupportedAuthenticationException ex) {
-				throw new RuntimeException("Unable to create admin auth: " + ex.getMessage(), ex);
-			}
-		}
-	}
-
 	private void initialize() {
 		var topics = Objects.requireNonNull(this.applicationContext, "Application context was not set")
 				.getBeansOfType(PulsarTopic.class, false, false).values();
@@ -129,7 +90,11 @@ public class PulsarAdministration
 	}
 
 	public PulsarAdmin createAdminClient() throws PulsarClientException {
-		return this.adminBuilder.build();
+		var adminBuilder = PulsarAdmin.builder();
+		if (this.adminCustomizer != null) {
+			this.adminCustomizer.customize(adminBuilder);
+		}
+		return adminBuilder.build();
 	}
 
 	@Override

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

@@ -85,10 +85,4 @@ public interface PulsarConsumerFactory<T> {
 			@Nullable Map<String, String> metadataProperties, @Nullable List<ConsumerBuilderCustomizer<T>> customizers)
 			throws PulsarClientException;
 
-	/**
-	 * Return the configuration options to use when creating consumers.
-	 * @return the configuration options
-	 */
-	Map<String, Object> getConsumerConfig();
-
 }

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

@@ -30,9 +30,7 @@ import static org.mockito.Mockito.verify;
 
 import java.time.Duration;
 import java.util.ArrayList;
-import java.util.Collections;
 import java.util.List;
-import java.util.Map;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicInteger;
@@ -61,14 +59,10 @@ class ConsumerAcknowledgmentTests implements PulsarTestContainerSupport {
 
 	@Test
 	void testRecordAck() throws Exception {
-		Map<String, Object> config = Map.of("topicNames", Collections.singleton("cons-ack-tests-011"),
-				"subscriptionName", "cons-ack-tests-sb-011");
-
 		PulsarClient pulsarClient = PulsarClient.builder().serviceUrl(PulsarTestContainerSupport.getPulsarBrokerUrl())
 				.build();
-		DefaultPulsarConsumerFactory<String> pulsarConsumerFactory = spy(
-				new DefaultPulsarConsumerFactory<>(pulsarClient, config));
-
+		DefaultPulsarConsumerFactory<String> pulsarConsumerFactory = spy(new DefaultPulsarConsumerFactory<>(
+				pulsarClient, defaultConfig("cons-ack-tests-011", "cons-ack-tests-sb-011")));
 		PulsarContainerProperties pulsarContainerProperties = new PulsarContainerProperties();
 		pulsarContainerProperties.setMessageListener((PulsarRecordMessageListener<?>) (consumer, msg) -> {
 		});
@@ -85,9 +79,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");
@@ -99,13 +92,10 @@ class ConsumerAcknowledgmentTests implements PulsarTestContainerSupport {
 
 	@Test
 	void testBatchAck() throws Exception {
-		Map<String, Object> config = Map.of("topicNames", Collections.singleton("cons-ack-tests-012"),
-				"subscriptionName", "cons-ack-tests-sb-012");
 		PulsarClient pulsarClient = PulsarClient.builder().serviceUrl(PulsarTestContainerSupport.getPulsarBrokerUrl())
 				.build();
-		DefaultPulsarConsumerFactory<String> pulsarConsumerFactory = spy(
-				new DefaultPulsarConsumerFactory<>(pulsarClient, config));
-
+		DefaultPulsarConsumerFactory<String> pulsarConsumerFactory = spy(new DefaultPulsarConsumerFactory<>(
+				pulsarClient, defaultConfig("cons-ack-tests-012", "cons-ack-tests-sb-012")));
 		PulsarContainerProperties pulsarContainerProperties = new PulsarContainerProperties();
 		CountDownLatch latch = new CountDownLatch(10);
 		pulsarContainerProperties
@@ -115,9 +105,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");
@@ -133,16 +122,12 @@ class ConsumerAcknowledgmentTests implements PulsarTestContainerSupport {
 
 	@Test
 	void testBatchAckButSomeRecordsFail() throws Exception {
-		Map<String, Object> config = Map.of("topicNames", Collections.singleton("cons-ack-tests-013"),
-				"subscriptionName", "cons-ack-tests-sb-013");
 		PulsarClient pulsarClient = PulsarClient.builder().serviceUrl(PulsarTestContainerSupport.getPulsarBrokerUrl())
 				.build();
-		DefaultPulsarConsumerFactory<String> pulsarConsumerFactory = spy(
-				new DefaultPulsarConsumerFactory<>(pulsarClient, config));
-
+		DefaultPulsarConsumerFactory<String> pulsarConsumerFactory = spy(new DefaultPulsarConsumerFactory<>(
+				pulsarClient, defaultConfig("cons-ack-tests-013", "cons-ack-tests-sb-013")));
 		PulsarContainerProperties pulsarContainerProperties = new PulsarContainerProperties();
 		CountDownLatch latch = new CountDownLatch(10);
-
 		pulsarContainerProperties.setMessageListener((PulsarRecordMessageListener<?>) (consumer, msg) -> {
 			latch.countDown();
 			if (latch.getCount() % 2 == 0) {
@@ -160,9 +145,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");
@@ -197,22 +181,17 @@ class ConsumerAcknowledgmentTests implements PulsarTestContainerSupport {
 	@Test
 	@SuppressWarnings("unchecked")
 	void testManualAckForRecordListener() throws Exception {
-		Map<String, Object> config = Map.of("topicNames", Collections.singleton("cons-ack-tests-014"),
-				"subscriptionName", "cons-ack-tests-sb-014");
 		PulsarClient pulsarClient = PulsarClient.builder().serviceUrl(PulsarTestContainerSupport.getPulsarBrokerUrl())
 				.build();
-		DefaultPulsarConsumerFactory<String> pulsarConsumerFactory = spy(
-				new DefaultPulsarConsumerFactory<>(pulsarClient, config));
-
+		DefaultPulsarConsumerFactory<String> pulsarConsumerFactory = spy(new DefaultPulsarConsumerFactory<>(
+				pulsarClient, defaultConfig("cons-ack-tests-014", "cons-ack-tests-sb-014")));
 		PulsarContainerProperties pulsarContainerProperties = new PulsarContainerProperties();
 		List<Acknowledgement> acksObjects = new ArrayList<>();
 		PulsarAcknowledgingMessageListener<?> pulsarAcknowledgingMessageListener = (consumer, msg, acknowledgement) -> {
 			acksObjects.add(acknowledgement);
 			acknowledgement.acknowledge();
 		};
-
 		pulsarContainerProperties.setMessageListener(pulsarAcknowledgingMessageListener);
-
 		pulsarContainerProperties.setSchema(Schema.STRING);
 		pulsarContainerProperties.setAckMode(AckMode.MANUAL);
 		DefaultPulsarMessageListenerContainer<String> container = new DefaultPulsarMessageListenerContainer<>(
@@ -226,9 +205,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");
@@ -247,34 +225,28 @@ class ConsumerAcknowledgmentTests implements PulsarTestContainerSupport {
 	@Test
 	@SuppressWarnings("unchecked")
 	void testBatchAckForBatchListener() throws Exception {
-		Map<String, Object> config = Map.of("topicNames", Collections.singleton("cons-ack-tests-015"),
-				"subscriptionName", "cons-ack-tests-sb-015");
 		PulsarClient pulsarClient = PulsarClient.builder().serviceUrl(PulsarTestContainerSupport.getPulsarBrokerUrl())
 				.build();
-		DefaultPulsarConsumerFactory<String> pulsarConsumerFactory = spy(
-				new DefaultPulsarConsumerFactory<>(pulsarClient, config));
-
+		DefaultPulsarConsumerFactory<String> pulsarConsumerFactory = spy(new DefaultPulsarConsumerFactory<>(
+				pulsarClient, defaultConfig("cons-ack-tests-015", "cons-ack-tests-sb-015")));
 		PulsarContainerProperties pulsarContainerProperties = new PulsarContainerProperties();
 		pulsarContainerProperties.setMaxNumMessages(10);
 		pulsarContainerProperties.setBatchTimeoutMillis(60_000);
 		pulsarContainerProperties.setBatchListener(true);
 		CountDownLatch latch = new CountDownLatch(1);
 		PulsarBatchMessageListener<?> pulsarBatchMessageListener = mock(PulsarBatchMessageListener.class);
-
 		doAnswer(invocation -> {
 			latch.countDown();
 			return null;
 		}).when(pulsarBatchMessageListener).received(any(Consumer.class), any(List.class));
-
 		pulsarContainerProperties.setMessageListener(pulsarBatchMessageListener);
 		pulsarContainerProperties.setSchema(Schema.STRING);
 		DefaultPulsarMessageListenerContainer<String> container = new DefaultPulsarMessageListenerContainer<>(
 				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");
@@ -291,34 +263,28 @@ class ConsumerAcknowledgmentTests implements PulsarTestContainerSupport {
 	@Test
 	@SuppressWarnings("unchecked")
 	void testBatchNackForEntireBatchWhenUsingBatchListener() throws Exception {
-		Map<String, Object> config = Map.of("topicNames", Collections.singleton("cons-ack-tests-016"),
-				"subscriptionName", "cons-ack-tests-sb-016");
 		PulsarClient pulsarClient = PulsarClient.builder().serviceUrl(PulsarTestContainerSupport.getPulsarBrokerUrl())
 				.build();
-		DefaultPulsarConsumerFactory<String> pulsarConsumerFactory = spy(
-				new DefaultPulsarConsumerFactory<>(pulsarClient, config));
-
+		DefaultPulsarConsumerFactory<String> pulsarConsumerFactory = spy(new DefaultPulsarConsumerFactory<>(
+				pulsarClient, defaultConfig("cons-ack-tests-016", "cons-ack-tests-sb-016")));
 		PulsarContainerProperties pulsarContainerProperties = new PulsarContainerProperties();
 		pulsarContainerProperties.setMaxNumMessages(10);
 		pulsarContainerProperties.setBatchTimeoutMillis(60_000);
 		pulsarContainerProperties.setBatchListener(true);
 		PulsarBatchMessageListener<?> pulsarBatchMessageListener = mock(PulsarBatchMessageListener.class);
 		CountDownLatch latch = new CountDownLatch(1);
-
 		doAnswer(invocation -> {
 			latch.countDown();
 			throw new RuntimeException();
 		}).when(pulsarBatchMessageListener).received(any(Consumer.class), any(List.class));
-
 		pulsarContainerProperties.setMessageListener(pulsarBatchMessageListener);
 		pulsarContainerProperties.setSchema(Schema.STRING);
 		DefaultPulsarMessageListenerContainer<String> container = new DefaultPulsarMessageListenerContainer<>(
 				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");
@@ -335,26 +301,21 @@ class ConsumerAcknowledgmentTests implements PulsarTestContainerSupport {
 
 	@Test
 	void messagesAreProperlyAckdOnContainerStopBeforeExitingListenerThread() throws Exception {
-		Map<String, Object> config = Map.of("topicNames", Collections.singleton("duplicate-message-test"),
-				"subscriptionName", "duplicate-sub-1");
 		PulsarClient pulsarClient = PulsarClient.builder().serviceUrl(PulsarTestContainerSupport.getPulsarBrokerUrl())
 				.build();
 		DefaultPulsarConsumerFactory<String> pulsarConsumerFactory = new DefaultPulsarConsumerFactory<>(pulsarClient,
-				config);
-
+				defaultConfig("duplicate-message-test", "duplicate-sub-1"));
 		PulsarContainerProperties pulsarContainerProperties = new PulsarContainerProperties();
 		AtomicInteger counter1 = new AtomicInteger(0);
-		pulsarContainerProperties.setMessageListener((PulsarRecordMessageListener<?>) (consumer, msg) -> {
-			counter1.getAndIncrement();
-		});
+		pulsarContainerProperties
+				.setMessageListener((PulsarRecordMessageListener<?>) (consumer, msg) -> counter1.getAndIncrement());
 		pulsarContainerProperties.setSchema(Schema.STRING);
 		DefaultPulsarMessageListenerContainer<String> container1 = new DefaultPulsarMessageListenerContainer<>(
 				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");
 
@@ -368,9 +329,8 @@ class ConsumerAcknowledgmentTests implements PulsarTestContainerSupport {
 		container1.stop();
 
 		AtomicInteger counter2 = new AtomicInteger(0);
-		pulsarContainerProperties.setMessageListener((PulsarRecordMessageListener<?>) (consumer, msg) -> {
-			counter2.getAndIncrement();
-		});
+		pulsarContainerProperties
+				.setMessageListener((PulsarRecordMessageListener<?>) (consumer, msg) -> counter2.getAndIncrement());
 		pulsarContainerProperties.setSchema(Schema.STRING);
 		DefaultPulsarMessageListenerContainer<String> container2 = new DefaultPulsarMessageListenerContainer<>(
 				pulsarConsumerFactory, pulsarContainerProperties);
@@ -389,4 +349,11 @@ class ConsumerAcknowledgmentTests implements PulsarTestContainerSupport {
 		pulsarClient.close();
 	}
 
+	private <T> ConsumerBuilderCustomizer<T> defaultConfig(String topicName, String subscriptionName) {
+		return (consumerBuilder) -> {
+			consumerBuilder.topic(topicName);
+			consumerBuilder.subscriptionName(subscriptionName);
+		};
+	}
+
 }

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

@@ -23,7 +23,6 @@ import static org.mockito.Mockito.inOrder;
 import static org.mockito.Mockito.mock;
 
 import java.util.Collections;
-import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 
@@ -72,7 +71,7 @@ class DefaultPulsarConsumerFactoryTests implements PulsarTestContainerSupport {
 
 		@BeforeEach
 		void createConsumerFactory() {
-			consumerFactory = new DefaultPulsarConsumerFactory<>(pulsarClient, Collections.emptyMap());
+			consumerFactory = new DefaultPulsarConsumerFactory<>(pulsarClient, null);
 		}
 
 		@Test
@@ -170,11 +169,11 @@ class DefaultPulsarConsumerFactoryTests implements PulsarTestContainerSupport {
 
 		@BeforeEach
 		void createConsumerFactory() {
-			Map<String, Object> defaultConfig = new HashMap<>();
-			defaultConfig.put("topicNames", Collections.singleton(defaultTopic));
-			defaultConfig.put("properties", defaultMetadataProperties);
-			defaultConfig.put("subscriptionName", defaultSubscription);
-			consumerFactory = new DefaultPulsarConsumerFactory<>(pulsarClient, defaultConfig);
+			consumerFactory = new DefaultPulsarConsumerFactory<>(pulsarClient, (consumerBuilder) -> {
+				consumerBuilder.topic(defaultTopic);
+				consumerBuilder.subscriptionName(defaultSubscription);
+				consumerBuilder.properties(defaultMetadataProperties);
+			});
 		}
 
 		@Test

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;
@@ -65,7 +64,7 @@ public class DefaultPulsarReaderFactoryTests implements PulsarTestContainerSuppo
 
 		@BeforeEach
 		void createReaderFactory() {
-			pulsarReaderFactory = new DefaultPulsarReaderFactory<>(pulsarClient, Collections.emptyMap());
+			pulsarReaderFactory = new DefaultPulsarReaderFactory<>(pulsarClient);
 		}
 
 		@Test
@@ -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");
 
@@ -131,6 +127,63 @@ public class DefaultPulsarReaderFactoryTests implements PulsarTestContainerSuppo
 			}
 		}
 
+		@Test
+		void useFactoryDefaults() throws Exception {
+			pulsarReaderFactory = new DefaultPulsarReaderFactory<>(pulsarClient, (readerBuilder) -> {
+				readerBuilder.topic("basic-pulsar-reader-topic");
+				readerBuilder.startMessageId(MessageId.earliest);
+			});
+			// The following code expects the above topic and startMessageId to be used
+			Message<String> message;
+			try (Reader<String> reader = pulsarReaderFactory.createReader(null, null, Schema.STRING,
+					Collections.emptyList())) {
+				PulsarProducerFactory<String> pulsarProducerFactory = new DefaultPulsarProducerFactory<>(pulsarClient,
+						"basic-pulsar-reader-topic");
+				PulsarTemplate<String> pulsarTemplate = new PulsarTemplate<>(pulsarProducerFactory);
+				pulsarTemplate.send("hello john doe");
+				message = reader.readNext();
+			}
+			assertThat(message.getValue()).isEqualTo("hello john doe");
+		}
+
+		@Test
+		void overrideFactoryDefaults() throws Exception {
+			pulsarReaderFactory = new DefaultPulsarReaderFactory<>(pulsarClient, (readerBuilder) -> {
+				readerBuilder.topic("foo-topic");
+				readerBuilder.startMessageId(MessageId.latest);
+			});
+			// The following code expects the above topic and startMessageId to be ignored
+			// (overridden)
+			Message<String> message;
+			try (Reader<String> reader = pulsarReaderFactory.createReader(List.of("basic-pulsar-reader-topic"),
+					MessageId.earliest, Schema.STRING, Collections.emptyList())) {
+
+				PulsarProducerFactory<String> pulsarProducerFactory = new DefaultPulsarProducerFactory<>(pulsarClient,
+						"basic-pulsar-reader-topic");
+				PulsarTemplate<String> pulsarTemplate = new PulsarTemplate<>(pulsarProducerFactory);
+				pulsarTemplate.send("hello john doe");
+
+				message = reader.readNext();
+			}
+			assertThat(message.getValue()).isEqualTo("hello john doe");
+		}
+
+		@Test
+		void customizersAreAppliedLast() throws Exception {
+			ReaderBuilderCustomizer<String> customizer = (readerBuilder) -> readerBuilder
+					.topic("basic-pulsar-reader-topic");
+			// The following code expects the above topic will override the passed in
+			// 'foo-topic'
+			try (var reader = pulsarReaderFactory.createReader(List.of("foo-topic"), MessageId.earliest, Schema.STRING,
+					List.of(customizer))) {
+				var pulsarProducerFactory = new DefaultPulsarProducerFactory<String>(pulsarClient,
+						"basic-pulsar-reader-topic");
+				var pulsarTemplate = new PulsarTemplate<>(pulsarProducerFactory);
+				pulsarTemplate.send("hello john doe");
+				assertThat(reader.readNext().getValue()).isEqualTo("hello john doe");
+			}
+		}
+
 	}
 
 	@Nested
@@ -140,7 +193,7 @@ public class DefaultPulsarReaderFactoryTests implements PulsarTestContainerSuppo
 
 		@BeforeEach
 		void createReaderFactory() {
-			pulsarReaderFactory = new DefaultPulsarReaderFactory<>(pulsarClient, Collections.emptyMap());
+			pulsarReaderFactory = new DefaultPulsarReaderFactory<>(pulsarClient);
 		}
 
 		@Test

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

@@ -19,8 +19,6 @@ package org.springframework.pulsar.core;
 import static org.assertj.core.api.Assertions.assertThat;
 
 import java.io.Serial;
-import java.util.Collections;
-import java.util.Map;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
 
@@ -48,18 +46,16 @@ class FailoverConsumerTests implements PulsarTestContainerSupport {
 	void testFailOverConsumersOnPartitionedTopic() throws Exception {
 		PulsarAdmin admin = PulsarAdmin.builder().serviceHttpUrl(PulsarTestContainerSupport.getHttpServiceUrl())
 				.build();
-
 		String topicName = "persistent://public/default/my-part-topic-1";
-		int numPartitions = 3;
-		admin.topics().createPartitionedTopic(topicName, numPartitions);
-
-		Map<String, Object> config = Map.of("topicNames", Collections.singleton("my-part-topic-1"), "subscriptionName",
-				"my-part-subscription-1");
+		admin.topics().createPartitionedTopic(topicName, 3);
 
 		PulsarClient pulsarClient = PulsarClient.builder().serviceUrl(PulsarTestContainerSupport.getPulsarBrokerUrl())
 				.build();
 		DefaultPulsarConsumerFactory<String> pulsarConsumerFactory = new DefaultPulsarConsumerFactory<>(pulsarClient,
-				config);
+				(consumerBuilder) -> {
+					consumerBuilder.topic("my-part-topic-1");
+					consumerBuilder.subscriptionName("my-part-subscription-1");
+				});
 
 		CountDownLatch latch1 = new CountDownLatch(1);
 		CountDownLatch latch2 = new CountDownLatch(1);
@@ -86,10 +82,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/PulsarAdministrationTests.java

@@ -18,19 +18,13 @@ package org.springframework.pulsar.core;
 
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.assertj.core.api.Assertions.assertThatIllegalStateException;
-import static org.assertj.core.api.InstanceOfAssertFactories.type;
 
 import java.util.Collections;
-import java.util.HashMap;
 import java.util.List;
-import java.util.Map;
 
 import org.apache.pulsar.client.admin.PulsarAdmin;
-import org.apache.pulsar.client.admin.PulsarAdminBuilder;
 import org.apache.pulsar.client.admin.PulsarAdminException;
 import org.apache.pulsar.client.api.PulsarClientException;
-import org.apache.pulsar.client.impl.auth.AuthenticationBasic;
-import org.apache.pulsar.client.impl.conf.ClientConfigurationData;
 import org.junit.jupiter.api.Nested;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.extension.ExtendWith;
@@ -63,19 +57,6 @@ public class PulsarAdministrationTests implements PulsarTestContainerSupport {
 	@Autowired
 	private PulsarAdministration pulsarAdministration;
 
-	@Test
-	void constructorRespectsAuthenticationProps() {
-		Map<String, Object> props = new HashMap<>();
-		props.put("authPluginClassName", "org.apache.pulsar.client.impl.auth.AuthenticationBasic");
-		props.put("authParams", "{\"userId\":\"foo\", \"password\":\"bar\"}");
-		PulsarAdministration admin = new PulsarAdministration(props);
-
-		assertThat(admin).extracting("adminBuilder").asInstanceOf(type(PulsarAdminBuilder.class)).extracting("conf")
-				.asInstanceOf(type(ClientConfigurationData.class))
-				.extracting(ClientConfigurationData::getAuthentication).isInstanceOf(AuthenticationBasic.class)
-				.hasFieldOrPropertyWithValue("userId", "foo").hasFieldOrPropertyWithValue("password", "bar");
-	}
-
 	private void assertThatTopicsExist(List<PulsarTopic> expected) throws PulsarAdminException {
 		assertThatTopicsExistIn(expected, NAMESPACE);
 	}
@@ -106,8 +87,7 @@ public class PulsarAdministrationTests implements PulsarTestContainerSupport {
 
 		@Bean
 		PulsarAdministration pulsarAdministration() {
-			return new PulsarAdministration(
-					PulsarAdmin.builder().serviceHttpUrl(PulsarTestContainerSupport.getHttpServiceUrl()));
+			return new PulsarAdministration(PulsarTestContainerSupport.getHttpServiceUrl());
 		}
 
 	}

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

@@ -18,7 +18,6 @@ package org.springframework.pulsar.core;
 
 import static org.assertj.core.api.Assertions.assertThat;
 
-import java.util.Collections;
 import java.util.HashMap;
 import java.util.Map;
 import java.util.concurrent.CountDownLatch;
@@ -57,9 +56,10 @@ public class SharedSubscriptionConsumerTests implements PulsarTestContainerSuppo
 		try {
 			pulsarClient = PulsarClient.builder().serviceUrl(PulsarTestContainerSupport.getPulsarBrokerUrl()).build();
 			DefaultPulsarConsumerFactory<String> pulsarConsumerFactory = new DefaultPulsarConsumerFactory<>(
-					pulsarClient,
-					Map.of("topicNames", Collections.singleton("shared-subscription-single-msg-test-topic"),
-							"subscriptionName", "shared-subscription-single-msg-test-sub"));
+					pulsarClient, (consumerBuilder) -> {
+						consumerBuilder.topic("shared-subscription-single-msg-test-topic");
+						consumerBuilder.subscriptionName("shared-subscription-single-msg-test-sub");
+					});
 
 			CountDownLatch latch1 = new CountDownLatch(1);
 			CountDownLatch latch2 = new CountDownLatch(1);
@@ -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();
@@ -114,8 +113,10 @@ public class SharedSubscriptionConsumerTests implements PulsarTestContainerSuppo
 		try {
 			pulsarClient = PulsarClient.builder().serviceUrl(PulsarTestContainerSupport.getPulsarBrokerUrl()).build();
 			DefaultPulsarConsumerFactory<String> consumerFactory = new DefaultPulsarConsumerFactory<>(pulsarClient,
-					Map.of("topicNames", Collections.singleton("key-shared-batch-disabled-topic"), "subscriptionName",
-							"key-shared-batch-disabled-sub"));
+					(consumerBuilder) -> {
+						consumerBuilder.topic("key-shared-batch-disabled-topic");
+						consumerBuilder.subscriptionName("key-shared-batch-disabled-sub");
+					});
 
 			CountDownLatch latch = new CountDownLatch(30);
 
@@ -132,7 +133,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

@@ -28,9 +28,7 @@ import static org.mockito.Mockito.when;
 
 import java.time.Duration;
 import java.util.ArrayList;
-import java.util.Collections;
 import java.util.List;
-import java.util.Map;
 import java.util.concurrent.atomic.AtomicInteger;
 
 import org.apache.pulsar.client.api.Consumer;
@@ -56,13 +54,13 @@ public class DefaultPulsarConsumerErrorHandlerTests implements PulsarTestContain
 	@Test
 	@SuppressWarnings("unchecked")
 	void happyPathErrorHandlingForRecordMessageListener() throws Exception {
-		Map<String, Object> config = Map.of("topicNames", Collections.singleton("default-error-handler-tests-1"),
-				"subscriptionName", "default-error-handler-tests-sub-1");
-
 		PulsarClient pulsarClient = PulsarClient.builder().serviceUrl(PulsarTestContainerSupport.getPulsarBrokerUrl())
 				.build();
 		DefaultPulsarConsumerFactory<String> pulsarConsumerFactory = new DefaultPulsarConsumerFactory<>(pulsarClient,
-				config);
+				(consumerBuilder) -> {
+					consumerBuilder.topic("default-error-handler-tests-1");
+					consumerBuilder.subscriptionName("default-error-handler-tests-sub-1");
+				});
 
 		PulsarContainerProperties pulsarContainerProperties = new PulsarContainerProperties();
 		PulsarRecordMessageListener<?> messageListener = mock(PulsarRecordMessageListener.class);
@@ -74,9 +72,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);
 
@@ -105,13 +102,13 @@ public class DefaultPulsarConsumerErrorHandlerTests implements PulsarTestContain
 	@Test
 	@SuppressWarnings("unchecked")
 	void errorHandlingForRecordMessageListenerWithTransientError() throws Exception {
-		Map<String, Object> config = Map.of("topicNames", Collections.singleton("default-error-handler-tests-2"),
-				"subscriptionName", "default-error-handler-tests-sub-2");
-
 		PulsarClient pulsarClient = PulsarClient.builder().serviceUrl(PulsarTestContainerSupport.getPulsarBrokerUrl())
 				.build();
 		DefaultPulsarConsumerFactory<String> pulsarConsumerFactory = new DefaultPulsarConsumerFactory<>(pulsarClient,
-				config);
+				(consumerBuilder) -> {
+					consumerBuilder.topic("default-error-handler-tests-2");
+					consumerBuilder.subscriptionName("default-error-handler-tests-sub-2");
+				});
 
 		PulsarContainerProperties pulsarContainerProperties = new PulsarContainerProperties();
 		PulsarRecordMessageListener<?> messageListener = mock(PulsarRecordMessageListener.class);
@@ -127,9 +124,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);
 
@@ -152,13 +148,13 @@ public class DefaultPulsarConsumerErrorHandlerTests implements PulsarTestContain
 	@Test
 	@SuppressWarnings("unchecked")
 	void everyOtherRecordThrowsNonTransientExceptionsRecordMessageListener() throws Exception {
-		Map<String, Object> config = Map.of("topicNames", Collections.singleton("default-error-handler-tests-3"),
-				"subscriptionName", "default-error-handler-tests-sub-3");
-
 		PulsarClient pulsarClient = PulsarClient.builder().serviceUrl(PulsarTestContainerSupport.getPulsarBrokerUrl())
 				.build();
 		DefaultPulsarConsumerFactory<Integer> pulsarConsumerFactory = new DefaultPulsarConsumerFactory<>(pulsarClient,
-				config);
+				(consumerBuilder) -> {
+					consumerBuilder.topic("default-error-handler-tests-3");
+					consumerBuilder.subscriptionName("default-error-handler-tests-sub-3");
+				});
 
 		PulsarContainerProperties pulsarContainerProperties = new PulsarContainerProperties();
 		PulsarRecordMessageListener<?> messageListener = mock(PulsarRecordMessageListener.class);
@@ -174,9 +170,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);
 
@@ -209,13 +204,13 @@ public class DefaultPulsarConsumerErrorHandlerTests implements PulsarTestContain
 	@Test
 	@SuppressWarnings("unchecked")
 	void batchRecordListenerFirstOneOnlyErrorAndRecover() throws Exception {
-		Map<String, Object> config = Map.of("topicNames", Collections.singleton("default-error-handler-tests-4"),
-				"subscriptionName", "default-error-handler-tests-sub-4");
-
 		PulsarClient pulsarClient = PulsarClient.builder().serviceUrl(PulsarTestContainerSupport.getPulsarBrokerUrl())
 				.build();
 		DefaultPulsarConsumerFactory<Integer> pulsarConsumerFactory = new DefaultPulsarConsumerFactory<>(pulsarClient,
-				config);
+				(consumerBuilder) -> {
+					consumerBuilder.topic("default-error-handler-tests-4");
+					consumerBuilder.subscriptionName("default-error-handler-tests-sub-4");
+				});
 
 		PulsarContainerProperties pulsarContainerProperties = new PulsarContainerProperties();
 		pulsarContainerProperties.setMaxNumMessages(10);
@@ -252,9 +247,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);
@@ -279,13 +273,13 @@ public class DefaultPulsarConsumerErrorHandlerTests implements PulsarTestContain
 	@Test
 	@SuppressWarnings("unchecked")
 	void batchRecordListenerRecordFailsInTheMiddle() throws Exception {
-		Map<String, Object> config = Map.of("topicNames", Collections.singleton("default-error-handler-tests-5"),
-				"subscriptionName", "default-error-handler-tests-sub-5");
-
 		PulsarClient pulsarClient = PulsarClient.builder().serviceUrl(PulsarTestContainerSupport.getPulsarBrokerUrl())
 				.build();
 		DefaultPulsarConsumerFactory<Integer> pulsarConsumerFactory = new DefaultPulsarConsumerFactory<>(pulsarClient,
-				config);
+				(consumerBuilder) -> {
+					consumerBuilder.topic("default-error-handler-tests-5");
+					consumerBuilder.subscriptionName("default-error-handler-tests-sub-5");
+				});
 
 		PulsarContainerProperties pulsarContainerProperties = new PulsarContainerProperties();
 		pulsarContainerProperties.setMaxNumMessages(10);
@@ -321,9 +315,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);
@@ -347,13 +340,13 @@ public class DefaultPulsarConsumerErrorHandlerTests implements PulsarTestContain
 	@Test
 	@SuppressWarnings("unchecked")
 	void batchRecordListenerRecordFailsTwiceInTheMiddle() throws Exception {
-		Map<String, Object> config = Map.of("topicNames", Collections.singleton("default-error-handler-tests-6"),
-				"subscriptionName", "default-error-handler-tests-sub-6");
-
 		PulsarClient pulsarClient = PulsarClient.builder().serviceUrl(PulsarTestContainerSupport.getPulsarBrokerUrl())
 				.build();
 		DefaultPulsarConsumerFactory<Integer> pulsarConsumerFactory = new DefaultPulsarConsumerFactory<>(pulsarClient,
-				config);
+				(consumerBuilder) -> {
+					consumerBuilder.topic("default-error-handler-tests-6");
+					consumerBuilder.subscriptionName("default-error-handler-tests-sub-6");
+				});
 
 		PulsarContainerProperties pulsarContainerProperties = new PulsarContainerProperties();
 		pulsarContainerProperties.setMaxNumMessages(10);
@@ -389,9 +382,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);
@@ -415,13 +407,13 @@ public class DefaultPulsarConsumerErrorHandlerTests implements PulsarTestContain
 	@Test
 	@SuppressWarnings("unchecked")
 	void batchRecordListenerRecordFailsInTheMiddleButTransientError() throws Exception {
-		Map<String, Object> config = Map.of("topicNames", Collections.singleton("default-error-handler-tests-7"),
-				"subscriptionName", "default-error-handler-tests-sub-7");
-
 		PulsarClient pulsarClient = PulsarClient.builder().serviceUrl(PulsarTestContainerSupport.getPulsarBrokerUrl())
 				.build();
 		DefaultPulsarConsumerFactory<Integer> pulsarConsumerFactory = new DefaultPulsarConsumerFactory<>(pulsarClient,
-				config);
+				(consumerBuilder) -> {
+					consumerBuilder.topic("default-error-handler-tests-7");
+					consumerBuilder.subscriptionName("default-error-handler-tests-sub-7");
+				});
 
 		PulsarContainerProperties pulsarContainerProperties = new PulsarContainerProperties();
 		pulsarContainerProperties.setMaxNumMessages(10);
@@ -463,9 +455,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);
@@ -482,13 +473,13 @@ public class DefaultPulsarConsumerErrorHandlerTests implements PulsarTestContain
 	@Test
 	@SuppressWarnings("unchecked")
 	void batchListenerFailsTransientErrorFollowedByNonTransient() throws Exception {
-		Map<String, Object> config = Map.of("topicNames", Collections.singleton("default-error-handler-tests-8"),
-				"subscriptionName", "default-error-handler-tests-sub-8");
-
 		PulsarClient pulsarClient = PulsarClient.builder().serviceUrl(PulsarTestContainerSupport.getPulsarBrokerUrl())
 				.build();
 		DefaultPulsarConsumerFactory<Integer> pulsarConsumerFactory = new DefaultPulsarConsumerFactory<>(pulsarClient,
-				config);
+				(consumerBuilder) -> {
+					consumerBuilder.topic("default-error-handler-tests-8");
+					consumerBuilder.subscriptionName("default-error-handler-tests-sub-8");
+				});
 
 		PulsarContainerProperties pulsarContainerProperties = new PulsarContainerProperties();
 		pulsarContainerProperties.setMaxNumMessages(10);
@@ -533,9 +524,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

@@ -26,9 +26,7 @@ import static org.mockito.Mockito.verify;
 
 import java.time.Duration;
 import java.util.ArrayList;
-import java.util.Collections;
 import java.util.List;
-import java.util.Map;
 import java.util.Set;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
@@ -65,13 +63,13 @@ class DefaultPulsarMessageListenerContainerTests implements PulsarTestContainerS
 
 	@Test
 	void basicDefaultConsumer() throws Exception {
-		Map<String, Object> config = Map.of("topicNames", Collections.singleton("dpmlct-012"), "subscriptionName",
-				"dpmlct-sb-012");
-
 		PulsarClient pulsarClient = PulsarClient.builder().serviceUrl(PulsarTestContainerSupport.getPulsarBrokerUrl())
 				.build();
 		DefaultPulsarConsumerFactory<String> pulsarConsumerFactory = new DefaultPulsarConsumerFactory<>(pulsarClient,
-				config);
+				(consumerBuilder) -> {
+					consumerBuilder.topic("dpmlct-012");
+					consumerBuilder.subscriptionName("dpmlct-sb-012");
+				});
 		CountDownLatch latch = new CountDownLatch(1);
 		PulsarContainerProperties pulsarContainerProperties = new PulsarContainerProperties();
 		pulsarContainerProperties
@@ -81,9 +79,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();
@@ -94,13 +91,13 @@ class DefaultPulsarMessageListenerContainerTests implements PulsarTestContainerS
 	@Disabled
 	@Test
 	void containerPauseAndResumeFeatureUsingWaitAndNotify() throws Exception {
-		Map<String, Object> config = Map.of("topicNames", Collections.singleton("containerPauseResumeWaitNotify-topic"),
-				"subscriptionName", "containerPauseResumeWaitNotify-sub");
-
 		PulsarClient pulsarClient = PulsarClient.builder().serviceUrl(PulsarTestContainerSupport.getPulsarBrokerUrl())
 				.build();
 		DefaultPulsarConsumerFactory<String> pulsarConsumerFactory = new DefaultPulsarConsumerFactory<>(pulsarClient,
-				config);
+				(consumerBuilder) -> {
+					consumerBuilder.topic("containerPauseResumeWaitNotify-topic");
+					consumerBuilder.subscriptionName("containerPauseResumeWaitNotify-sub");
+				});
 		PulsarContainerProperties pulsarContainerProperties = new PulsarContainerProperties();
 		pulsarContainerProperties.setMessageListener((PulsarRecordMessageListener<?>) (consumer, msg) -> {
 		});
@@ -163,13 +160,14 @@ class DefaultPulsarMessageListenerContainerTests implements PulsarTestContainerS
 
 	@Test
 	void subscriptionInitialPositionEarliest() throws Exception {
-		Map<String, Object> config = Map.of("topicNames", Collections.singleton("dpmlct-013"), "subscriptionName",
-				"dpmlct-sb-013", "subscriptionInitialPosition", SubscriptionInitialPosition.Earliest);
-
 		PulsarClient pulsarClient = PulsarClient.builder().serviceUrl(PulsarTestContainerSupport.getPulsarBrokerUrl())
 				.build();
 		DefaultPulsarConsumerFactory<String> pulsarConsumerFactory = new DefaultPulsarConsumerFactory<>(pulsarClient,
-				config);
+				(consumerBuilder) -> {
+					consumerBuilder.topic("dpmlct-013");
+					consumerBuilder.subscriptionName("dpmlct-sb-013");
+					consumerBuilder.subscriptionInitialPosition(SubscriptionInitialPosition.Earliest);
+				});
 		CountDownLatch latch = new CountDownLatch(5);
 		PulsarContainerProperties pulsarContainerProperties = new PulsarContainerProperties();
 		pulsarContainerProperties
@@ -178,9 +176,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);
@@ -194,13 +191,13 @@ class DefaultPulsarMessageListenerContainerTests implements PulsarTestContainerS
 
 	@Test
 	void subscriptionInitialPositionDefaultLatest() throws Exception {
-		Map<String, Object> config = Map.of("topicNames", Collections.singleton("dpmlct-014"), "subscriptionName",
-				"dpmlct-sb-014");
-
 		PulsarClient pulsarClient = PulsarClient.builder().serviceUrl(PulsarTestContainerSupport.getPulsarBrokerUrl())
 				.build();
 		DefaultPulsarConsumerFactory<String> pulsarConsumerFactory = new DefaultPulsarConsumerFactory<>(pulsarClient,
-				config);
+				(consumerBuilder) -> {
+					consumerBuilder.topic("dpmlct-014");
+					consumerBuilder.subscriptionName("dpmlct-sb-014");
+				});
 		PulsarContainerProperties pulsarContainerProperties = new PulsarContainerProperties();
 		List<String> messages = new ArrayList<>();
 		pulsarContainerProperties.setMessageListener(
@@ -209,9 +206,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);
@@ -228,15 +224,16 @@ class DefaultPulsarMessageListenerContainerTests implements PulsarTestContainerS
 
 	@Test
 	void negativeAckRedeliveryBackoff() throws Exception {
-		RedeliveryBackoff redeliveryBackoff = MultiplierRedeliveryBackoff.builder().minDelayMs(1000)
-				.maxDelayMs(5 * 1000).build();
-		Map<String, Object> config = Map.of("topicNames", Collections.singleton("dpmlct-015"), "subscriptionName",
-				"dpmlct-sb-015", "negativeAckRedeliveryBackoff", redeliveryBackoff);
-
 		PulsarClient pulsarClient = PulsarClient.builder().serviceUrl(PulsarTestContainerSupport.getPulsarBrokerUrl())
 				.build();
+		RedeliveryBackoff redeliveryBackoff = MultiplierRedeliveryBackoff.builder().minDelayMs(1000)
+				.maxDelayMs(5 * 1000).build();
 		DefaultPulsarConsumerFactory<String> pulsarConsumerFactory = spy(
-				new DefaultPulsarConsumerFactory<>(pulsarClient, config));
+				new DefaultPulsarConsumerFactory<>(pulsarClient, (consumerBuilder) -> {
+					consumerBuilder.topic("dpmlct-015");
+					consumerBuilder.subscriptionName("dpmlct-sb-015");
+					consumerBuilder.negativeAckRedeliveryBackoff(redeliveryBackoff);
+				}));
 		CountDownLatch latch = new CountDownLatch(10);
 		PulsarContainerProperties pulsarContainerProperties = new PulsarContainerProperties();
 		pulsarContainerProperties.setMessageListener((PulsarRecordMessageListener<?>) (consumer, msg) -> {
@@ -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);
@@ -275,16 +271,17 @@ class DefaultPulsarMessageListenerContainerTests implements PulsarTestContainerS
 
 	@Test
 	void deadLetterPolicyDefault() throws Exception {
-		DeadLetterPolicy deadLetterPolicy = DeadLetterPolicy.builder().maxRedeliverCount(1)
-				.deadLetterTopic("dpmlct-016-dlq-topic").build();
-		Map<String, Object> config = Map.of("topicNames", Collections.singleton("dpmlct-016"), "subscriptionName",
-				"dpmlct-sb-016", "negativeAckRedeliveryDelayMicros", TimeUnit.SECONDS.toMicros(1), "deadLetterPolicy",
-				deadLetterPolicy);
-
 		PulsarClient pulsarClient = PulsarClient.builder().serviceUrl(PulsarTestContainerSupport.getPulsarBrokerUrl())
 				.build();
+		DeadLetterPolicy deadLetterPolicy = DeadLetterPolicy.builder().maxRedeliverCount(1)
+				.deadLetterTopic("dpmlct-016-dlq-topic").build();
 		DefaultPulsarConsumerFactory<Integer> pulsarConsumerFactory = new DefaultPulsarConsumerFactory<>(pulsarClient,
-				config);
+				(consumerBuilder) -> {
+					consumerBuilder.topic("dpmlct-016");
+					consumerBuilder.subscriptionName("dpmlct-sb-016");
+					consumerBuilder.negativeAckRedeliveryDelay(1L, TimeUnit.SECONDS);
+					consumerBuilder.deadLetterPolicy(deadLetterPolicy);
+				});
 
 		CountDownLatch dlqLatch = new CountDownLatch(1);
 		CountDownLatch latch = new CountDownLatch(6);
@@ -312,9 +309,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);
@@ -331,16 +327,17 @@ class DefaultPulsarMessageListenerContainerTests implements PulsarTestContainerS
 
 	@Test
 	void deadLetterPolicyCustom() throws Exception {
-		DeadLetterPolicy deadLetterPolicy = DeadLetterPolicy.builder().maxRedeliverCount(5).deadLetterTopic("dlq-topic")
-				.build();
-		Map<String, Object> config = Map.of("topicNames", Collections.singleton("dpmlct-017"), "subscriptionName",
-				"dpmlct-sb-016", "negativeAckRedeliveryDelayMicros", TimeUnit.SECONDS.toMicros(1), "deadLetterPolicy",
-				deadLetterPolicy);
-
 		PulsarClient pulsarClient = PulsarClient.builder().serviceUrl(PulsarTestContainerSupport.getPulsarBrokerUrl())
 				.build();
+		DeadLetterPolicy deadLetterPolicy = DeadLetterPolicy.builder().maxRedeliverCount(5).deadLetterTopic("dlq-topic")
+				.build();
 		DefaultPulsarConsumerFactory<Integer> pulsarConsumerFactory = new DefaultPulsarConsumerFactory<>(pulsarClient,
-				config);
+				(consumerBuilder) -> {
+					consumerBuilder.topic("dpmlct-017");
+					consumerBuilder.subscriptionName("dpmlct-sb-017");
+					consumerBuilder.negativeAckRedeliveryDelay(1L, TimeUnit.SECONDS);
+					consumerBuilder.deadLetterPolicy(deadLetterPolicy);
+				});
 
 		CountDownLatch dlqLatch = new CountDownLatch(1);
 		CountDownLatch latch = new CountDownLatch(6);
@@ -371,9 +368,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,17 +22,13 @@ 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;
 import java.util.concurrent.TimeUnit;
 import java.util.stream.Collectors;
 
-import org.apache.pulsar.client.admin.PulsarAdmin;
 import org.apache.pulsar.client.api.Consumer;
 import org.apache.pulsar.client.api.DeadLetterPolicy;
 import org.apache.pulsar.client.api.Message;
@@ -105,8 +101,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,22 +116,19 @@ 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, null);
 		}
 
 		@Bean
 		PulsarListenerContainerFactory pulsarListenerContainerFactory(
 				PulsarConsumerFactory<Object> pulsarConsumerFactory) {
-			ConcurrentPulsarListenerContainerFactory<?> pulsarListenerContainerFactory = new ConcurrentPulsarListenerContainerFactory<>(
-					pulsarConsumerFactory, new PulsarContainerProperties());
-			return pulsarListenerContainerFactory;
+			return new ConcurrentPulsarListenerContainerFactory<>(pulsarConsumerFactory,
+					new PulsarContainerProperties());
 		}
 
 		@Bean
 		PulsarAdministration pulsarAdministration() {
-			return new PulsarAdministration(
-					PulsarAdmin.builder().serviceHttpUrl(PulsarTestContainerSupport.getHttpServiceUrl()));
+			return new PulsarAdministration(PulsarTestContainerSupport.getHttpServiceUrl());
 		}
 
 		@Bean
@@ -184,12 +176,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 +193,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 +448,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 +460,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 +472,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 +484,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 +623,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 +634,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 +645,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 +657,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 +727,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 +738,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);
 			}
@@ -1106,9 +1082,7 @@ public class PulsarListenerTests implements PulsarTestContainerSupport {
 
 			@Bean
 			public ConsumerBuilderCustomizer<String> myCustomizer() {
-				return cb -> {
-					cb.subscriptionName("test-changed-subscription-name");
-				};
+				return cb -> cb.subscriptionName("test-changed-subscription-name");
 			}
 
 		}

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

@@ -18,14 +18,12 @@ package org.springframework.pulsar.observation;
 
 import static org.assertj.core.api.Assertions.assertThat;
 
-import java.util.Collections;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
 
-import org.apache.pulsar.client.admin.PulsarAdmin;
 import org.apache.pulsar.client.api.PulsarClient;
 import org.apache.pulsar.client.api.PulsarClientException;
 
@@ -131,7 +129,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
@@ -147,7 +145,7 @@ public class ObservationIntegrationTests extends SampleTestRunner implements Pul
 
 		@Bean
 		public PulsarConsumerFactory<?> pulsarConsumerFactory(PulsarClient pulsarClient) {
-			return new DefaultPulsarConsumerFactory<>(pulsarClient, Collections.emptyMap());
+			return new DefaultPulsarConsumerFactory<>(pulsarClient, null);
 		}
 
 		@Bean
@@ -160,8 +158,7 @@ public class ObservationIntegrationTests extends SampleTestRunner implements Pul
 
 		@Bean
 		PulsarAdministration pulsarAdministration() {
-			return new PulsarAdministration(
-					PulsarAdmin.builder().serviceHttpUrl(PulsarTestContainerSupport.getHttpServiceUrl()));
+			return new PulsarAdministration(PulsarTestContainerSupport.getHttpServiceUrl());
 		}
 
 		@Bean

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

@@ -20,13 +20,11 @@ import static org.assertj.core.api.Assertions.assertThat;
 import static org.awaitility.Awaitility.await;
 
 import java.util.Arrays;
-import java.util.Collections;
 import java.util.Deque;
 import java.util.List;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
 
-import org.apache.pulsar.client.admin.PulsarAdmin;
 import org.apache.pulsar.client.api.Message;
 import org.apache.pulsar.client.api.PulsarClient;
 import org.apache.pulsar.client.api.PulsarClientException;
@@ -170,7 +168,7 @@ public class ObservationTests implements PulsarTestContainerSupport {
 
 		@Bean
 		PulsarProducerFactory<String> pulsarProducerFactory(PulsarClient pulsarClient) {
-			return new DefaultPulsarProducerFactory<>(pulsarClient, Collections.emptyMap());
+			return new DefaultPulsarProducerFactory<>(pulsarClient);
 		}
 
 		@Bean
@@ -197,7 +195,7 @@ public class ObservationTests implements PulsarTestContainerSupport {
 
 		@Bean
 		PulsarConsumerFactory<?> pulsarConsumerFactory(PulsarClient pulsarClient) {
-			return new DefaultPulsarConsumerFactory<>(pulsarClient, Collections.emptyMap());
+			return new DefaultPulsarConsumerFactory<>(pulsarClient, null);
 		}
 
 		@Bean
@@ -224,8 +222,7 @@ public class ObservationTests implements PulsarTestContainerSupport {
 
 		@Bean
 		PulsarAdministration pulsarAdministration() {
-			return new PulsarAdministration(
-					PulsarAdmin.builder().serviceHttpUrl(PulsarTestContainerSupport.getHttpServiceUrl()));
+			return new PulsarAdministration(PulsarTestContainerSupport.getHttpServiceUrl());
 		}
 
 		@Bean

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

@@ -18,7 +18,6 @@ package org.springframework.pulsar.reader;
 
 import static org.assertj.core.api.Assertions.assertThat;
 
-import java.util.Collections;
 import java.util.List;
 import java.util.Map;
 import java.util.concurrent.CountDownLatch;
@@ -44,6 +43,7 @@ import org.springframework.pulsar.test.support.PulsarTestContainerSupport;
  * Basic tests for {@link DefaultPulsarMessageReaderContainer}.
  *
  * @author Soby Chacko
+ * @author Chris Bono
  */
 public class DefaultPulsarMessageReaderContainerTests implements PulsarTestContainerSupport {
 
@@ -67,9 +67,12 @@ public class DefaultPulsarMessageReaderContainerTests implements PulsarTestConta
 	@Test
 	void basicDefaultReader() throws Exception {
 		var latch = new CountDownLatch(1);
-		var config = Map.of("topicNames", Collections.singleton("dprlct-001"), "subscriptionName", "dprlct-sub-001");
 
-		DefaultPulsarReaderFactory<String> pulsarReaderFactory = new DefaultPulsarReaderFactory<>(pulsarClient, config);
+		DefaultPulsarReaderFactory<String> pulsarReaderFactory = new DefaultPulsarReaderFactory<>(pulsarClient,
+				(readerBuilder -> {
+					readerBuilder.topic("dprlct-001");
+					readerBuilder.subscriptionName("dprlct-sub-001");
+				}));
 		var readerContainerProperties = new PulsarReaderContainerProperties();
 		readerContainerProperties.setReaderListener((ReaderListener<?>) (reader, msg) -> {
 			assertThat(msg.getValue()).isEqualTo("hello john doe");
@@ -83,9 +86,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();
@@ -99,9 +101,8 @@ public class DefaultPulsarMessageReaderContainerTests implements PulsarTestConta
 	void topicProvidedThroughContainerProperties() throws Exception {
 		var latch = new CountDownLatch(1);
 		var containerProps = new PulsarReaderContainerProperties();
-		var config = Collections.<String, Object>emptyMap();
 
-		DefaultPulsarReaderFactory<String> pulsarReaderFactory = new DefaultPulsarReaderFactory<>(pulsarClient, config);
+		DefaultPulsarReaderFactory<String> pulsarReaderFactory = new DefaultPulsarReaderFactory<>(pulsarClient);
 		containerProps.setReaderListener((ReaderListener<?>) (reader, msg) -> {
 			assertThat(msg.getValue()).isEqualTo("hello buzz doe");
 			latch.countDown();
@@ -113,10 +114,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();
@@ -138,14 +137,14 @@ public class DefaultPulsarMessageReaderContainerTests implements PulsarTestConta
 		containerProps.setTopics(List.of("dprlct-003"));
 		containerProps.setSchema(Schema.STRING);
 
-		var readerConfig = Collections.<String, Object>emptyMap();
-		var readerFactory = new DefaultPulsarReaderFactory<String>(pulsarClient, readerConfig);
+		var readerFactory = new DefaultPulsarReaderFactory<String>(pulsarClient);
 		DefaultPulsarMessageReaderContainer<String> container = null;
 		try {
 			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,6 @@ 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 +53,7 @@ import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
  * {@link PulsarReader} integration tests.
  *
  * @author Soby Chacko
+ * @author Chris Bono
  */
 @SpringJUnitConfig
 @DirtiesContext
@@ -73,8 +71,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 +86,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);
 		}
 
 		@Bean
 		PulsarReaderContainerFactory pulsarReaderContainerFactory(PulsarReaderFactory<Object> pulsarReaderFactory) {
-			DefaultPulsarReaderContainerFactory<?> pulsarReaderContainerFactory = new DefaultPulsarReaderContainerFactory<>(
-					pulsarReaderFactory, new PulsarReaderContainerProperties());
-			return pulsarReaderContainerFactory;
+			return new DefaultPulsarReaderContainerFactory<>(pulsarReaderFactory,
+					new PulsarReaderContainerProperties());
 		}
 
 	}