Add multi-customizers to reactive reader and consumer (#436)

* Also add builder to ReactivePulsarSenderFactory

See #432

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

@@ -20,13 +20,11 @@ import java.util.Collections;
 import java.util.List;
 
 import org.apache.pulsar.client.api.Schema;
-import org.apache.pulsar.reactive.client.api.ImmutableReactiveMessageConsumerSpec;
-import org.apache.pulsar.reactive.client.api.MutableReactiveMessageConsumerSpec;
 import org.apache.pulsar.reactive.client.api.ReactiveMessageConsumer;
 import org.apache.pulsar.reactive.client.api.ReactiveMessageConsumerBuilder;
-import org.apache.pulsar.reactive.client.api.ReactiveMessageConsumerSpec;
 import org.apache.pulsar.reactive.client.api.ReactivePulsarClient;
 
+import org.springframework.lang.Nullable;
 import org.springframework.util.CollectionUtils;
 
 /**
@@ -34,18 +32,25 @@ import org.springframework.util.CollectionUtils;
  *
  * @param <T> underlying payload type for the reactive consumer.
  * @author Christophe Bornet
+ * @author Chris Bono
  */
 public class DefaultReactivePulsarConsumerFactory<T> implements ReactivePulsarConsumerFactory<T> {
 
-	private final ReactiveMessageConsumerSpec consumerSpec;
-
 	private final ReactivePulsarClient reactivePulsarClient;
 
+	@Nullable
+	private final List<ReactiveMessageConsumerBuilderCustomizer<T>> defaultConfigCustomizers;
+
+	/**
+	 * Construct an instance.
+	 * @param reactivePulsarClient the reactive client
+	 * @param defaultConfigCustomizers the optional list of customizers that defines the
+	 * default configuration for each created consumer.
+	 */
 	public DefaultReactivePulsarConsumerFactory(ReactivePulsarClient reactivePulsarClient,
-			ReactiveMessageConsumerSpec consumerSpec) {
-		this.consumerSpec = new ImmutableReactiveMessageConsumerSpec(
-				consumerSpec != null ? consumerSpec : new MutableReactiveMessageConsumerSpec());
+			List<ReactiveMessageConsumerBuilderCustomizer<T>> defaultConfigCustomizers) {
 		this.reactivePulsarClient = reactivePulsarClient;
+		this.defaultConfigCustomizers = defaultConfigCustomizers;
 	}
 
 	@Override
@@ -57,14 +62,19 @@ public class DefaultReactivePulsarConsumerFactory<T> implements ReactivePulsarCo
 	public ReactiveMessageConsumer<T> createConsumer(Schema<T> schema,
 			List<ReactiveMessageConsumerBuilderCustomizer<T>> customizers) {
 
-		ReactiveMessageConsumerBuilder<T> consumer = this.reactivePulsarClient.messageConsumer(schema);
+		ReactiveMessageConsumerBuilder<T> consumerBuilder = this.reactivePulsarClient.messageConsumer(schema);
 
-		consumer.applySpec(this.consumerSpec);
-		if (!CollectionUtils.isEmpty(customizers)) {
-			customizers.forEach((c) -> c.customize(consumer));
+		// Apply the default customizers
+		if (!CollectionUtils.isEmpty(this.defaultConfigCustomizers)) {
+			this.defaultConfigCustomizers.forEach((customizer -> customizer.customize(consumerBuilder)));
 		}
 
-		return consumer.build();
+		// Apply the user specified customizers
+		if (!CollectionUtils.isEmpty(customizers)) {
+			customizers.forEach((c) -> c.customize(consumerBuilder));
+		}
+
+		return consumerBuilder.build();
 	}
 
 }

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

@@ -22,9 +22,9 @@ import java.util.List;
 import org.apache.pulsar.client.api.Schema;
 import org.apache.pulsar.reactive.client.api.ReactiveMessageReader;
 import org.apache.pulsar.reactive.client.api.ReactiveMessageReaderBuilder;
-import org.apache.pulsar.reactive.client.api.ReactiveMessageReaderSpec;
 import org.apache.pulsar.reactive.client.api.ReactivePulsarClient;
 
+import org.springframework.lang.Nullable;
 import org.springframework.util.CollectionUtils;
 
 /**
@@ -32,17 +32,25 @@ import org.springframework.util.CollectionUtils;
  *
  * @param <T> underlying payload type for the reactive reader.
  * @author Christophe Bornet
+ * @author Chris Bono
  */
 public class DefaultReactivePulsarReaderFactory<T> implements ReactivePulsarReaderFactory<T> {
 
-	private final ReactiveMessageReaderSpec readerSpec;
-
 	private final ReactivePulsarClient reactivePulsarClient;
 
+	@Nullable
+	private final List<ReactiveMessageReaderBuilderCustomizer<T>> defaultConfigCustomizers;
+
+	/**
+	 * Construct an instance.
+	 * @param reactivePulsarClient the reactive client
+	 * @param defaultConfigCustomizers the optional list of customizers that defines the
+	 * default configuration for each created reader.
+	 */
 	public DefaultReactivePulsarReaderFactory(ReactivePulsarClient reactivePulsarClient,
-			ReactiveMessageReaderSpec readerSpec) {
+			List<ReactiveMessageReaderBuilderCustomizer<T>> defaultConfigCustomizers) {
 		this.reactivePulsarClient = reactivePulsarClient;
-		this.readerSpec = readerSpec;
+		this.defaultConfigCustomizers = defaultConfigCustomizers;
 	}
 
 	@Override
@@ -54,12 +62,19 @@ public class DefaultReactivePulsarReaderFactory<T> implements ReactivePulsarRead
 	public ReactiveMessageReader<T> createReader(Schema<T> schema,
 			List<ReactiveMessageReaderBuilderCustomizer<T>> customizers) {
 
-		ReactiveMessageReaderBuilder<T> reader = this.reactivePulsarClient.messageReader(schema)
-			.applySpec(this.readerSpec);
-		if (!CollectionUtils.isEmpty(customizers)) {
-			customizers.forEach((c) -> c.customize(reader));
+		ReactiveMessageReaderBuilder<T> readerBuilder = this.reactivePulsarClient.messageReader(schema);
+
+		// Apply the default customizers
+		if (!CollectionUtils.isEmpty(this.defaultConfigCustomizers)) {
+			this.defaultConfigCustomizers.forEach((customizer -> customizer.customize(readerBuilder)));
 		}
-		return reader.build();
+
+		// Apply the user specified customizers
+		if (!CollectionUtils.isEmpty(customizers)) {
+			customizers.forEach((c) -> c.customize(readerBuilder));
+		}
+
+		return readerBuilder.build();
 	}
 
 }

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

@@ -23,81 +23,70 @@ import java.util.Objects;
 import org.apache.pulsar.client.api.PulsarClient;
 import org.apache.pulsar.client.api.Schema;
 import org.apache.pulsar.reactive.client.adapter.AdaptedReactivePulsarClientFactory;
-import org.apache.pulsar.reactive.client.api.ImmutableReactiveMessageSenderSpec;
-import org.apache.pulsar.reactive.client.api.MutableReactiveMessageSenderSpec;
 import org.apache.pulsar.reactive.client.api.ReactiveMessageSender;
 import org.apache.pulsar.reactive.client.api.ReactiveMessageSenderBuilder;
 import org.apache.pulsar.reactive.client.api.ReactiveMessageSenderCache;
-import org.apache.pulsar.reactive.client.api.ReactiveMessageSenderSpec;
 import org.apache.pulsar.reactive.client.api.ReactivePulsarClient;
 
 import org.springframework.core.log.LogAccessor;
 import org.springframework.lang.Nullable;
 import org.springframework.pulsar.core.DefaultTopicResolver;
 import org.springframework.pulsar.core.TopicResolver;
+import org.springframework.util.Assert;
 import org.springframework.util.CollectionUtils;
 
 /**
  * Default implementation of {@link ReactivePulsarSenderFactory}.
  *
- * @param <T> reactive sender type.
+ * @param <T> underlying payload type for the reactive sender.
  * @author Christophe Bornet
  * @author Chris Bono
  */
-public class DefaultReactivePulsarSenderFactory<T> implements ReactivePulsarSenderFactory<T> {
+public final class DefaultReactivePulsarSenderFactory<T> implements ReactivePulsarSenderFactory<T> {
 
 	private final LogAccessor logger = new LogAccessor(this.getClass());
 
 	private final ReactivePulsarClient reactivePulsarClient;
 
-	private final ReactiveMessageSenderSpec reactiveMessageSenderSpec;
+	private final TopicResolver topicResolver;
 
 	@Nullable
 	private final ReactiveMessageSenderCache reactiveMessageSenderCache;
 
 	@Nullable
-	private final List<ReactiveMessageSenderBuilderCustomizer<T>> defaultSenderBuilderCustomizers;
+	private String defaultTopic;
 
-	private TopicResolver topicResolver;
+	@Nullable
+	private final List<ReactiveMessageSenderBuilderCustomizer<T>> defaultConfigCustomizers;
 
-	/**
-	 * Construct an instance.
-	 * @param pulsarClient the pulsar client to adapt into a reactive client
-	 * @param reactiveMessageSenderSpec spec that defines the initial settings on the
-	 * created senders
-	 * @param reactiveMessageSenderCache cache used to cache created senders
-	 * @param defaultSenderBuilderCustomizers optional list of sender builder customizers
-	 * to apply to the created senders
-	 */
-	public DefaultReactivePulsarSenderFactory(PulsarClient pulsarClient,
-			@Nullable ReactiveMessageSenderSpec reactiveMessageSenderSpec,
-			@Nullable ReactiveMessageSenderCache reactiveMessageSenderCache,
-			@Nullable List<ReactiveMessageSenderBuilderCustomizer<T>> defaultSenderBuilderCustomizers) {
-		this(AdaptedReactivePulsarClientFactory.create(pulsarClient), reactiveMessageSenderSpec,
-				reactiveMessageSenderCache, defaultSenderBuilderCustomizers, new DefaultTopicResolver());
+	private DefaultReactivePulsarSenderFactory(ReactivePulsarClient reactivePulsarClient, TopicResolver topicResolver,
+			@Nullable ReactiveMessageSenderCache reactiveMessageSenderCache, @Nullable String defaultTopic,
+			@Nullable List<ReactiveMessageSenderBuilderCustomizer<T>> defaultConfigCustomizers) {
+		this.reactivePulsarClient = reactivePulsarClient;
+		this.topicResolver = topicResolver;
+		this.reactiveMessageSenderCache = reactiveMessageSenderCache;
+		this.defaultTopic = defaultTopic;
+		this.defaultConfigCustomizers = defaultConfigCustomizers;
 	}
 
 	/**
-	 * Construct an instance.
-	 * @param reactivePulsarClient the reactive client to use
-	 * @param reactiveMessageSenderSpec spec that defines the initial settings on the
-	 * created senders
-	 * @param reactiveMessageSenderCache cache used to cache created senders
-	 * @param defaultSenderBuilderCustomizers optional list of sender builder customizers
-	 * to apply to the created senders
-	 * @param topicResolver the topic resolver to use
+	 * Create a builder that uses the specified Reactive pulsar client.
+	 * @param reactivePulsarClient the reactive client
+	 * @param <T> underlying payload type for the reactive sender
+	 * @return the newly created builder instance
 	 */
-	public DefaultReactivePulsarSenderFactory(ReactivePulsarClient reactivePulsarClient,
-			@Nullable ReactiveMessageSenderSpec reactiveMessageSenderSpec,
-			@Nullable ReactiveMessageSenderCache reactiveMessageSenderCache,
-			@Nullable List<ReactiveMessageSenderBuilderCustomizer<T>> defaultSenderBuilderCustomizers,
-			TopicResolver topicResolver) {
-		this.reactivePulsarClient = reactivePulsarClient;
-		this.reactiveMessageSenderSpec = new ImmutableReactiveMessageSenderSpec(
-				reactiveMessageSenderSpec != null ? reactiveMessageSenderSpec : new MutableReactiveMessageSenderSpec());
-		this.reactiveMessageSenderCache = reactiveMessageSenderCache;
-		this.topicResolver = topicResolver;
-		this.defaultSenderBuilderCustomizers = defaultSenderBuilderCustomizers;
+	public static <T> Builder<T> builderFor(ReactivePulsarClient reactivePulsarClient) {
+		return new Builder<>(reactivePulsarClient);
+	}
+
+	/**
+	 * Create a builder that adapts the specified pulsar client.
+	 * @param pulsarClient the Pulsar client to adapt into a Reactive client
+	 * @param <T> underlying payload type for the reactive sender
+	 * @return the newly created builder instance
+	 */
+	public static <T> Builder<T> builderFor(PulsarClient pulsarClient) {
+		return new Builder<>(AdaptedReactivePulsarClientFactory.create(pulsarClient));
 	}
 
 	@Override
@@ -121,34 +110,123 @@ public class DefaultReactivePulsarSenderFactory<T> implements ReactivePulsarSend
 	private ReactiveMessageSender<T> doCreateReactiveMessageSender(Schema<T> schema, @Nullable String topic,
 			@Nullable List<ReactiveMessageSenderBuilderCustomizer<T>> customizers) {
 		Objects.requireNonNull(schema, "Schema must be specified");
-		String resolvedTopic = this.topicResolver
-			.resolveTopic(topic, () -> getReactiveMessageSenderSpec().getTopicName())
-			.orElseThrow();
+		String resolvedTopic = this.topicResolver.resolveTopic(topic, () -> getDefaultTopic()).orElseThrow();
 		this.logger.trace(() -> "Creating reactive message sender for '%s' topic".formatted(resolvedTopic));
 
 		ReactiveMessageSenderBuilder<T> sender = this.reactivePulsarClient.messageSender(schema);
-		sender.applySpec(this.reactiveMessageSenderSpec);
 
-		// Apply the default config customizer (preserve the topic)
-		if (!CollectionUtils.isEmpty(this.defaultSenderBuilderCustomizers)) {
-			this.defaultSenderBuilderCustomizers.forEach((customizer -> customizer.customize(sender)));
+		// Apply the default customizers (preserve the topic)
+		if (!CollectionUtils.isEmpty(this.defaultConfigCustomizers)) {
+			this.defaultConfigCustomizers.forEach((customizer -> customizer.customize(sender)));
 		}
 		sender.topic(resolvedTopic);
+
 		if (this.reactiveMessageSenderCache != null) {
 			sender.cache(this.reactiveMessageSenderCache);
 		}
+
+		// Apply the user specified customizers (preserve the topic)
 		if (!CollectionUtils.isEmpty(customizers)) {
 			customizers.forEach((c) -> c.customize(sender));
 		}
-		// make sure the customizer do not override the topic
 		sender.topic(resolvedTopic);
 
 		return sender.build();
 	}
 
 	@Override
-	public ReactiveMessageSenderSpec getReactiveMessageSenderSpec() {
-		return this.reactiveMessageSenderSpec;
+	public String getDefaultTopic() {
+		return this.defaultTopic;
+	}
+
+	/**
+	 * Builder for {@link DefaultReactivePulsarSenderFactory}.
+	 *
+	 * @param <T> the reactive sender type
+	 */
+	public static final class Builder<T> {
+
+		private final ReactivePulsarClient reactivePulsarClient;
+
+		private TopicResolver topicResolver = new DefaultTopicResolver();
+
+		@Nullable
+		private ReactiveMessageSenderCache messageSenderCache;
+
+		@Nullable
+		private String defaultTopic;
+
+		@Nullable
+		private List<ReactiveMessageSenderBuilderCustomizer<T>> defaultConfigCustomizers;
+
+		private Builder(ReactivePulsarClient reactivePulsarClient) {
+			Assert.notNull(reactivePulsarClient, "Reactive client is required");
+			this.reactivePulsarClient = reactivePulsarClient;
+		}
+
+		/**
+		 * Provide the topic resolver to use.
+		 * @param topicResolver the topic resolver to use
+		 * @return this same builder instance
+		 */
+		public Builder<T> withTopicResolver(TopicResolver topicResolver) {
+			this.topicResolver = topicResolver;
+			return this;
+		}
+
+		/**
+		 * Provide the message sender cache to use.
+		 * @param messageSenderCache the message sender cache to use
+		 * @return this same builder instance
+		 */
+		public Builder<T> withMessageSenderCache(ReactiveMessageSenderCache messageSenderCache) {
+			this.messageSenderCache = messageSenderCache;
+			return this;
+		}
+
+		/**
+		 * Provide the default topic to use when one is not specified.
+		 * @param defaultTopic the default topic to use
+		 * @return this same builder instance
+		 */
+		public Builder<T> withDefaultTopic(String defaultTopic) {
+			this.defaultTopic = defaultTopic;
+			return this;
+		}
+
+		/**
+		 * Provide a customizer to apply to the sender builder.
+		 * @param customizer the customizer to apply to the builder before creating
+		 * senders
+		 * @return this same builder instance
+		 */
+		public Builder<T> withDefaultConfigCustomizer(ReactiveMessageSenderBuilderCustomizer<T> customizer) {
+			this.defaultConfigCustomizers = List.of(customizer);
+			return this;
+		}
+
+		/**
+		 * Provide an optional list of sender builder customizers to apply to the builder
+		 * before creating the senders.
+		 * @param customizers optional list of sender builder customizers to apply to the
+		 * builder before creating the senders.
+		 * @return this same builder instance
+		 */
+		public Builder<T> withDefaultConfigCustomizers(List<ReactiveMessageSenderBuilderCustomizer<T>> customizers) {
+			this.defaultConfigCustomizers = customizers;
+			return this;
+		}
+
+		/**
+		 * Construct the sender factory using the specified settings.
+		 * @return pulsar sender factory
+		 */
+		public DefaultReactivePulsarSenderFactory<T> build() {
+			Assert.notNull(this.topicResolver, "Topic resolver is required");
+			return new DefaultReactivePulsarSenderFactory<>(this.reactivePulsarClient, this.topicResolver,
+					this.messageSenderCache, this.defaultTopic, this.defaultConfigCustomizers);
+		}
+
 	}
 
 }

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

@@ -20,7 +20,6 @@ import java.util.List;
 
 import org.apache.pulsar.client.api.Schema;
 import org.apache.pulsar.reactive.client.api.ReactiveMessageSender;
-import org.apache.pulsar.reactive.client.api.ReactiveMessageSenderSpec;
 
 import org.springframework.lang.Nullable;
 
@@ -64,9 +63,10 @@ public interface ReactivePulsarSenderFactory<T> {
 			@Nullable List<ReactiveMessageSenderBuilderCustomizer<T>> customizers);
 
 	/**
-	 * Return the ReactiveMessageSenderSpec to use when creating reactive senders.
-	 * @return the ReactiveMessageSenderSpec
+	 * Get the default topic to use for all created senders.
+	 * @return the default topic to use for all created senders or null if no default set.
 	 */
-	ReactiveMessageSenderSpec getReactiveMessageSenderSpec();
+	@Nullable
+	String getDefaultTopic();
 
 }

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

@@ -157,7 +157,7 @@ public class ReactivePulsarTemplate<T> implements ReactivePulsarOperations<T> {
 	}
 
 	private String resolveTopic(@Nullable String topic, @Nullable Object message) {
-		String defaultTopic = this.reactiveMessageSenderFactory.getReactiveMessageSenderSpec().getTopicName();
+		String defaultTopic = this.reactiveMessageSenderFactory.getDefaultTopic();
 		return this.topicResolver.resolveTopic(topic, message, () -> defaultTopic).orElseThrow();
 	}
 

spring-pulsar-reactive/src/test/java/org/springframework/pulsar/reactive/core/DefaultReactivePulsarConsumerFactoryTests.java

@@ -19,11 +19,11 @@ package org.springframework.pulsar.reactive.core;
 import static org.assertj.core.api.Assertions.assertThat;
 
 import java.util.Collections;
+import java.util.List;
 
 import org.apache.pulsar.client.api.PulsarClient;
 import org.apache.pulsar.client.api.Schema;
 import org.apache.pulsar.reactive.client.adapter.AdaptedReactivePulsarClientFactory;
-import org.apache.pulsar.reactive.client.api.MutableReactiveMessageConsumerSpec;
 import org.apache.pulsar.reactive.client.api.ReactiveMessageConsumer;
 import org.apache.pulsar.reactive.client.api.ReactiveMessageConsumerSpec;
 import org.assertj.core.api.InstanceOfAssertFactories;
@@ -44,7 +44,7 @@ class DefaultReactivePulsarConsumerFactoryTests {
 	@Nested
 	class FactoryCreatedWithoutSpec {
 
-		private org.springframework.pulsar.reactive.core.DefaultReactivePulsarConsumerFactory<String> consumerFactory = new org.springframework.pulsar.reactive.core.DefaultReactivePulsarConsumerFactory<>(
+		private DefaultReactivePulsarConsumerFactory<String> consumerFactory = new DefaultReactivePulsarConsumerFactory<>(
 				AdaptedReactivePulsarClientFactory.create((PulsarClient) null), null);
 
 		@Test
@@ -76,10 +76,9 @@ class DefaultReactivePulsarConsumerFactoryTests {
 
 		@BeforeEach
 		void createConsumerFactory() {
-			MutableReactiveMessageConsumerSpec spec = new MutableReactiveMessageConsumerSpec();
-			spec.setConsumerName("test-consumer");
 			consumerFactory = new DefaultReactivePulsarConsumerFactory<>(
-					AdaptedReactivePulsarClientFactory.create((PulsarClient) null), spec);
+					AdaptedReactivePulsarClientFactory.create((PulsarClient) null),
+					List.of((builder) -> builder.consumerName("test-consumer")));
 		}
 
 		@Test

spring-pulsar-reactive/src/test/java/org/springframework/pulsar/reactive/core/DefaultReactivePulsarReaderFactoryTests.java

@@ -19,11 +19,11 @@ package org.springframework.pulsar.reactive.core;
 import static org.assertj.core.api.Assertions.assertThat;
 
 import java.util.Collections;
+import java.util.List;
 
 import org.apache.pulsar.client.api.PulsarClient;
 import org.apache.pulsar.client.api.Schema;
 import org.apache.pulsar.reactive.client.adapter.AdaptedReactivePulsarClientFactory;
-import org.apache.pulsar.reactive.client.api.MutableReactiveMessageReaderSpec;
 import org.apache.pulsar.reactive.client.api.ReactiveMessageReader;
 import org.apache.pulsar.reactive.client.api.ReactiveMessageReaderSpec;
 import org.assertj.core.api.InstanceOfAssertFactories;
@@ -41,10 +41,9 @@ class DefaultReactivePulsarReaderFactoryTests {
 
 	@Test
 	void createReader() {
-		MutableReactiveMessageReaderSpec spec = new MutableReactiveMessageReaderSpec();
-		spec.setReaderName("test-reader");
 		DefaultReactivePulsarReaderFactory<String> readerFactory = new DefaultReactivePulsarReaderFactory<>(
-				AdaptedReactivePulsarClientFactory.create((PulsarClient) null), spec);
+				AdaptedReactivePulsarClientFactory.create((PulsarClient) null),
+				List.of((builder) -> builder.readerName("test-reader")));
 
 		ReactiveMessageReader<String> reader = readerFactory.createReader(schema);
 
@@ -55,10 +54,9 @@ class DefaultReactivePulsarReaderFactoryTests {
 
 	@Test
 	void createReaderWithCustomizer() {
-		MutableReactiveMessageReaderSpec spec = new MutableReactiveMessageReaderSpec();
-		spec.setReaderName("test-reader");
 		DefaultReactivePulsarReaderFactory<String> readerFactory = new DefaultReactivePulsarReaderFactory<>(
-				AdaptedReactivePulsarClientFactory.create((PulsarClient) null), spec);
+				AdaptedReactivePulsarClientFactory.create((PulsarClient) null),
+				List.of((builder) -> builder.readerName("test-reader")));
 
 		ReactiveMessageReader<String> reader = readerFactory.createReader(schema,
 				Collections.singletonList(builder -> builder.readerName("new-test-reader")));

spring-pulsar-reactive/src/test/java/org/springframework/pulsar/reactive/core/DefaultReactivePulsarSenderFactoryTests.java

@@ -28,19 +28,22 @@ import java.util.Collections;
 import java.util.List;
 
 import org.apache.pulsar.client.api.CompressionType;
+import org.apache.pulsar.client.api.PulsarClient;
 import org.apache.pulsar.client.api.Schema;
 import org.apache.pulsar.reactive.client.adapter.AdaptedReactivePulsarClientFactory;
-import org.apache.pulsar.reactive.client.api.MutableReactiveMessageSenderSpec;
 import org.apache.pulsar.reactive.client.api.ReactiveMessageSender;
 import org.apache.pulsar.reactive.client.api.ReactiveMessageSenderBuilder;
 import org.apache.pulsar.reactive.client.api.ReactiveMessageSenderCache;
 import org.apache.pulsar.reactive.client.api.ReactiveMessageSenderSpec;
+import org.apache.pulsar.reactive.client.api.ReactivePulsarClient;
 import org.assertj.core.api.InstanceOfAssertFactories;
 import org.assertj.core.api.ThrowingConsumer;
 import org.junit.jupiter.api.Nested;
 import org.junit.jupiter.api.Test;
 import org.mockito.InOrder;
 
+import org.springframework.pulsar.core.TopicResolver;
+
 /**
  * Unit tests for {@link DefaultReactivePulsarSenderFactory}.
  *
@@ -58,6 +61,15 @@ class DefaultReactivePulsarSenderFactoryTests {
 		assertThat(sender).extracting("producerCache").isSameAs(cache);
 	}
 
+	@Test
+	void createSenderWithTopicResolver() {
+		var customTopicResolver = mock(TopicResolver.class);
+		var senderFactory = DefaultReactivePulsarSenderFactory.<String>builderFor(mock(ReactivePulsarClient.class))
+			.withTopicResolver(customTopicResolver)
+			.build();
+		assertThat(senderFactory).hasFieldOrPropertyWithValue("topicResolver", customTopicResolver);
+	}
+
 	private void assertThatSenderHasTopic(ReactiveMessageSender<String> sender, String expectedTopic) {
 		assertThatSenderSpecSatisfies(sender,
 				(senderSpec) -> assertThat(senderSpec).extracting(ReactiveMessageSenderSpec::getTopicName)
@@ -71,17 +83,19 @@ class DefaultReactivePulsarSenderFactoryTests {
 	}
 
 	private ReactivePulsarSenderFactory<String> newSenderFactory() {
-		return new DefaultReactivePulsarSenderFactory<>(null, null, null, null);
+		return DefaultReactivePulsarSenderFactory.<String>builderFor(mock(PulsarClient.class)).build();
 	}
 
 	private ReactivePulsarSenderFactory<String> newSenderFactoryWithDefaultTopic(String defaultTopic) {
-		MutableReactiveMessageSenderSpec senderSpec = new MutableReactiveMessageSenderSpec();
-		senderSpec.setTopicName(defaultTopic);
-		return new DefaultReactivePulsarSenderFactory<>(null, senderSpec, null, null);
+		return DefaultReactivePulsarSenderFactory.<String>builderFor(mock(PulsarClient.class))
+			.withDefaultTopic(defaultTopic)
+			.build();
 	}
 
 	private ReactivePulsarSenderFactory<String> newSenderFactoryWithCache(ReactiveMessageSenderCache cache) {
-		return new DefaultReactivePulsarSenderFactory<>(null, null, cache, null);
+		return DefaultReactivePulsarSenderFactory.<String>builderFor(mock(PulsarClient.class))
+			.withMessageSenderCache(cache)
+			.build();
 	}
 
 	@Nested
@@ -189,8 +203,9 @@ class DefaultReactivePulsarSenderFactoryTests {
 
 		private ReactivePulsarSenderFactory<String> newSenderFactoryWithCustomizers(
 				List<ReactiveMessageSenderBuilderCustomizer<String>> customizers) {
-			MutableReactiveMessageSenderSpec senderSpec = new MutableReactiveMessageSenderSpec();
-			return new DefaultReactivePulsarSenderFactory<>(null, senderSpec, null, customizers);
+			return DefaultReactivePulsarSenderFactory.<String>builderFor(mock(PulsarClient.class))
+				.withDefaultConfigCustomizers(customizers)
+				.build();
 		}
 
 	}

spring-pulsar-reactive/src/test/java/org/springframework/pulsar/reactive/core/ReactivePulsarTemplateTests.java

@@ -33,7 +33,6 @@ 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.reactive.client.api.MessageSpec;
-import org.apache.pulsar.reactive.client.api.MutableReactiveMessageSenderSpec;
 import org.assertj.core.api.InstanceOfAssertFactories;
 import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.BeforeEach;
@@ -178,12 +177,9 @@ class ReactivePulsarTemplateTests implements PulsarTestContainerSupport {
 	@ValueSource(booleans = { true, false })
 	void sendMessageWithTopicInferredByTypeMappings(boolean producerFactoryHasDefaultTopic) throws Exception {
 		String topic = "ptt-topicInferred-" + producerFactoryHasDefaultTopic + "-topic";
-		MutableReactiveMessageSenderSpec spec = new MutableReactiveMessageSenderSpec();
-		if (producerFactoryHasDefaultTopic) {
-			spec.setTopicName("fake-topic");
-		}
-		ReactivePulsarSenderFactory<Foo> producerFactory = new DefaultReactivePulsarSenderFactory<>(client, spec, null,
-				null);
+		ReactivePulsarSenderFactory<Foo> producerFactory = DefaultReactivePulsarSenderFactory.<Foo>builderFor(client)
+			.withDefaultTopic(producerFactoryHasDefaultTopic ? "fake-topic" : null)
+			.build();
 		// Topic mappings allows not specifying the topic when sending (nor having
 		// default on producer)
 		DefaultTopicResolver topicResolver = new DefaultTopicResolver();
@@ -198,24 +194,18 @@ class ReactivePulsarTemplateTests implements PulsarTestContainerSupport {
 
 	@Test
 	void sendMessageWithoutTopicFails() {
-		ReactivePulsarSenderFactory<String> senderFactory = new DefaultReactivePulsarSenderFactory<>(client,
-				new MutableReactiveMessageSenderSpec(), null, null);
-		ReactivePulsarTemplate<String> pulsarTemplate = new ReactivePulsarTemplate<>(senderFactory);
+		ReactivePulsarTemplate<String> pulsarTemplate = new ReactivePulsarTemplate<>(
+				DefaultReactivePulsarSenderFactory.<String>builderFor(client).build());
 		assertThatIllegalArgumentException().isThrownBy(() -> pulsarTemplate.send("test-message").subscribe())
 			.withMessage("Topic must be specified when no default topic is configured");
 	}
 
 	private <T> Message<T> sendAndConsume(Consumer<ReactivePulsarTemplate<T>> sendFunction, String topic,
 			Schema<T> schema, @Nullable T expectedValue, Boolean withDefaultTopic) throws Exception {
-		MutableReactiveMessageSenderSpec senderSpec = new MutableReactiveMessageSenderSpec();
-		if (withDefaultTopic) {
-			senderSpec.setTopicName(topic);
-		}
-		ReactivePulsarSenderFactory<T> senderFactory = new DefaultReactivePulsarSenderFactory<>(client, senderSpec,
-				null, null);
-
+		ReactivePulsarSenderFactory<T> senderFactory = DefaultReactivePulsarSenderFactory.<T>builderFor(client)
+			.withDefaultTopic(withDefaultTopic ? topic : null)
+			.build();
 		ReactivePulsarTemplate<T> pulsarTemplate = new ReactivePulsarTemplate<>(senderFactory);
-
 		return sendAndConsume(pulsarTemplate, sendFunction, topic, schema, expectedValue);
 	}
 
@@ -258,10 +248,10 @@ class ReactivePulsarTemplateTests implements PulsarTestContainerSupport {
 		@Test
 		void withSchemaInferredByTypeMappings() throws Exception {
 			String topic = "ptt-schemaInferred-topic";
-			MutableReactiveMessageSenderSpec spec = new MutableReactiveMessageSenderSpec();
-			spec.setTopicName(topic);
-			ReactivePulsarSenderFactory<Foo> producerFactory = new DefaultReactivePulsarSenderFactory<>(client, spec,
-					null, null);
+			ReactivePulsarSenderFactory<Foo> producerFactory = DefaultReactivePulsarSenderFactory
+				.<Foo>builderFor(client)
+				.withDefaultTopic(topic)
+				.build();
 			// Custom schema resolver allows not specifying the schema when sending
 			DefaultSchemaResolver schemaResolver = new DefaultSchemaResolver();
 			schemaResolver.addCustomSchemaMapping(Foo.class, Schema.JSON(Foo.class));
@@ -280,10 +270,10 @@ class ReactivePulsarTemplateTests implements PulsarTestContainerSupport {
 
 		@Test
 		void sendNullWithDefaultTopicFails() {
-			MutableReactiveMessageSenderSpec spec = new MutableReactiveMessageSenderSpec();
-			spec.setTopicName("sendNullWithDefaultTopicFails");
-			ReactivePulsarSenderFactory<String> senderFactory = new DefaultReactivePulsarSenderFactory<>(client, spec,
-					null, null);
+			ReactivePulsarSenderFactory<String> senderFactory = DefaultReactivePulsarSenderFactory
+				.<String>builderFor(client)
+				.withDefaultConfigCustomizer((builder) -> builder.topic("sendNullWithDefaultTopicFails"))
+				.build();
 			ReactivePulsarTemplate<String> pulsarTemplate = new ReactivePulsarTemplate<>(senderFactory);
 			assertThatIllegalArgumentException()
 				.isThrownBy(() -> pulsarTemplate.send((String) null, Schema.STRING).subscribe())
@@ -292,8 +282,9 @@ class ReactivePulsarTemplateTests implements PulsarTestContainerSupport {
 
 		@Test
 		void sendNullWithoutSchemaFails() {
-			ReactivePulsarSenderFactory<String> senderFactory = new DefaultReactivePulsarSenderFactory<>(client,
-					new MutableReactiveMessageSenderSpec(), null, null);
+			ReactivePulsarSenderFactory<String> senderFactory = DefaultReactivePulsarSenderFactory
+				.<String>builderFor(client)
+				.build();
 			ReactivePulsarTemplate<String> pulsarTemplate = new ReactivePulsarTemplate<>(senderFactory);
 			assertThatIllegalArgumentException()
 				.isThrownBy(() -> pulsarTemplate.send("sendNullWithoutSchemaFails", (String) null, null).subscribe())

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

@@ -19,7 +19,6 @@ package org.springframework.pulsar.reactive.listener;
 import static org.assertj.core.api.Assertions.assertThat;
 
 import java.time.Duration;
-import java.util.Collections;
 import java.util.List;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
@@ -32,17 +31,15 @@ import org.apache.pulsar.reactive.client.adapter.AdaptedReactivePulsarClientFact
 import org.apache.pulsar.reactive.client.adapter.DefaultMessageGroupingFunction;
 import org.apache.pulsar.reactive.client.api.MessageResult;
 import org.apache.pulsar.reactive.client.api.MessageSpec;
-import org.apache.pulsar.reactive.client.api.MutableReactiveMessageConsumerSpec;
-import org.apache.pulsar.reactive.client.api.MutableReactiveMessageSenderSpec;
-import org.apache.pulsar.reactive.client.api.ReactiveMessageConsumer;
 import org.apache.pulsar.reactive.client.api.ReactiveMessagePipeline;
 import org.apache.pulsar.reactive.client.api.ReactivePulsarClient;
 import org.assertj.core.api.InstanceOfAssertFactories;
 import org.junit.jupiter.api.Test;
 
-import org.springframework.pulsar.core.DefaultTopicResolver;
+import org.springframework.core.log.LogAccessor;
 import org.springframework.pulsar.reactive.core.DefaultReactivePulsarConsumerFactory;
 import org.springframework.pulsar.reactive.core.DefaultReactivePulsarSenderFactory;
+import org.springframework.pulsar.reactive.core.ReactiveMessageConsumerBuilderCustomizer;
 import org.springframework.pulsar.reactive.core.ReactivePulsarTemplate;
 import org.springframework.pulsar.test.support.PulsarTestContainerSupport;
 
@@ -54,290 +51,294 @@ import reactor.test.StepVerifier;
  * Tests for {@link DefaultReactivePulsarMessageListenerContainer}
  *
  * @author Christophe Bornet
+ * @author Chris Bono
  */
 class DefaultReactivePulsarMessageListenerContainerTests implements PulsarTestContainerSupport {
 
+	private final LogAccessor logger = new LogAccessor(this.getClass());
+
 	@Test
-	void messageHandlerListener() throws Exception {
-		String topic = "drpmlct-012";
-		MutableReactiveMessageConsumerSpec config = new MutableReactiveMessageConsumerSpec();
-		config.setTopicNames(Collections.singletonList(topic));
-		config.setSubscriptionName("drpmlct-sb-012");
-		PulsarClient pulsarClient = PulsarClient.builder()
-			.serviceUrl(PulsarTestContainerSupport.getPulsarBrokerUrl())
-			.build();
-		ReactivePulsarClient reactivePulsarClient = AdaptedReactivePulsarClientFactory.create(pulsarClient);
-		DefaultReactivePulsarConsumerFactory<String> pulsarConsumerFactory = new DefaultReactivePulsarConsumerFactory<>(
-				reactivePulsarClient, config);
+	void oneByOneMessageHandler() throws Exception {
+		var pulsarClient = PulsarClient.builder().serviceUrl(PulsarTestContainerSupport.getPulsarBrokerUrl()).build();
+		ReactivePulsarMessageListenerContainer<String> container = null;
+		try {
+			var reactivePulsarClient = AdaptedReactivePulsarClientFactory.create(pulsarClient);
+			var topic = topicNameForTest("1");
+			var consumerFactory = createAndPrepareConsumerFactory(topic, reactivePulsarClient);
+			var latch = new CountDownLatch(1);
+			var containerProperties = new ReactivePulsarContainerProperties<String>();
+			containerProperties.setSchema(Schema.STRING);
+			containerProperties.setMessageHandler(
+					(ReactivePulsarOneByOneMessageHandler<String>) (msg) -> Mono.fromRunnable(latch::countDown));
+			container = new DefaultReactivePulsarMessageListenerContainer<>(consumerFactory, containerProperties);
+			container.start();
+			createPulsarTemplate(topic, reactivePulsarClient).send("hello john doe").subscribe();
+			assertThat(latch.await(10, TimeUnit.SECONDS)).isTrue();
+		}
+		finally {
+			safeStopContainer(container);
+			pulsarClient.close();
+		}
+	}
+
+	@Test
+	void streamingMessageHandler() throws Exception {
+		var pulsarClient = PulsarClient.builder().serviceUrl(PulsarTestContainerSupport.getPulsarBrokerUrl()).build();
+		ReactivePulsarMessageListenerContainer<String> container = null;
+		try {
+			var reactivePulsarClient = AdaptedReactivePulsarClientFactory.create(pulsarClient);
+			var topic = topicNameForTest("2");
+			var consumerFactory = createAndPrepareConsumerFactory(topic, reactivePulsarClient);
+			var latch = new CountDownLatch(5);
+			var containerProperties = new ReactivePulsarContainerProperties<String>();
+			containerProperties.setSchema(Schema.STRING);
+			containerProperties.setMessageHandler(
+					(ReactivePulsarStreamingHandler<String>) (msg) -> msg.doOnNext((m) -> latch.countDown())
+						.map(MessageResult::acknowledge));
+			container = new DefaultReactivePulsarMessageListenerContainer<>(consumerFactory, containerProperties);
+			container.start();
+			createPulsarTemplate(topic, reactivePulsarClient)
+				.newMessages(Flux.range(0, 5).map(i -> MessageSpec.of("hello john doe" + i)))
+				.send()
+				.subscribe();
+			assertThat(latch.await(10, TimeUnit.SECONDS)).isTrue();
+		}
+		finally {
+			safeStopContainer(container);
+			pulsarClient.close();
+		}
+	}
+
+	@Test
+	void containerPropertiesAreRespected() throws Exception {
+		var pulsarClient = PulsarClient.builder().serviceUrl(PulsarTestContainerSupport.getPulsarBrokerUrl()).build();
+		ReactivePulsarMessageListenerContainer<String> container = null;
+		try {
+			var reactivePulsarClient = AdaptedReactivePulsarClientFactory.create(pulsarClient);
+			var topic = topicNameForTest("3");
+			var consumerFactory = createAndPrepareConsumerFactory(topic, reactivePulsarClient);
+			var latch = new CountDownLatch(1);
+			var containerProperties = new ReactivePulsarContainerProperties<String>();
+			containerProperties.setSchema(Schema.STRING);
+			containerProperties.setMessageHandler(
+					(ReactivePulsarOneByOneMessageHandler<String>) (msg) -> Mono.fromRunnable(latch::countDown));
+			containerProperties.setConcurrency(5);
+			containerProperties.setUseKeyOrderedProcessing(true);
+			containerProperties.setHandlingTimeout(Duration.ofMillis(7));
+			container = new DefaultReactivePulsarMessageListenerContainer<>(consumerFactory, containerProperties);
+			container.start();
+			createPulsarTemplate(topic, reactivePulsarClient).send("hello john doe").subscribe();
+			assertThat(latch.await(10, TimeUnit.SECONDS)).isTrue();
+			assertThat(container).extracting("pipeline", InstanceOfAssertFactories.type(ReactiveMessagePipeline.class))
+				.hasFieldOrPropertyWithValue("concurrency", 5)
+				.hasFieldOrPropertyWithValue("handlingTimeout", Duration.ofMillis(7))
+				.extracting("groupingFunction")
+				.isInstanceOf(DefaultMessageGroupingFunction.class);
+		}
+		finally {
+			safeStopContainer(container);
+			pulsarClient.close();
+		}
+	}
+
+	@Test
+	void createConsumerWithSharedSubTypeOnFactoryWithExclusiveSubType() throws Exception {
+		var pulsarClient = PulsarClient.builder().serviceUrl(PulsarTestContainerSupport.getPulsarBrokerUrl()).build();
+		ReactivePulsarMessageListenerContainer<String> container = null;
+		try {
+			var reactivePulsarClient = AdaptedReactivePulsarClientFactory.create(pulsarClient);
+			var topic = topicNameForTest("4");
+			ReactiveMessageConsumerBuilderCustomizer<String> defaultConfig = (builder) -> {
+				builder.topic(topic);
+				builder.subscriptionName(topic + "-sub");
+			};
+			var consumerFactory = new DefaultReactivePulsarConsumerFactory<>(reactivePulsarClient,
+					List.of(defaultConfig));
+			var containerProperties = new ReactivePulsarContainerProperties<String>();
+			containerProperties.setSchema(Schema.STRING);
+			containerProperties.setMessageHandler((ReactivePulsarOneByOneMessageHandler<String>) (msg) -> Mono.empty());
+			container = new DefaultReactivePulsarMessageListenerContainer<>(consumerFactory, containerProperties);
+			container.start();
+
+			Thread.sleep(2_000);
+			StepVerifier
+				.create(consumerFactory.createConsumer(Schema.STRING,
+						List.of(builder -> builder.subscriptionType(SubscriptionType.Shared)))
+					.consumeNothing())
+				.expectErrorMatches((ex) -> {
+					return true;
+				})
+				.verify(Duration.ofSeconds(10));
+		}
+		finally {
+			safeStopContainer(container);
+			pulsarClient.close();
+		}
+	}
+
+	@Test
+	void createConsumerWithSharedSubTypeOnFactoryWithSharedSubType() throws Exception {
+		var pulsarClient = PulsarClient.builder().serviceUrl(PulsarTestContainerSupport.getPulsarBrokerUrl()).build();
+		ReactivePulsarMessageListenerContainer<String> container = null;
+		try {
+			var reactivePulsarClient = AdaptedReactivePulsarClientFactory.create(pulsarClient);
+			var topic = topicNameForTest("5");
+			ReactiveMessageConsumerBuilderCustomizer<String> defaultConfig = (builder) -> {
+				builder.topic(topic);
+				builder.subscriptionName(topic + "-sub");
+			};
+			var consumerFactory = new DefaultReactivePulsarConsumerFactory<>(reactivePulsarClient,
+					List.of(defaultConfig));
+			var containerProperties = new ReactivePulsarContainerProperties<String>();
+			containerProperties.setSchema(Schema.STRING);
+			containerProperties.setSubscriptionType(SubscriptionType.Shared);
+			containerProperties.setMessageHandler((ReactivePulsarOneByOneMessageHandler<String>) (msg) -> Mono.empty());
+			container = new DefaultReactivePulsarMessageListenerContainer<>(consumerFactory, containerProperties);
+			container.start();
+
+			Thread.sleep(2_000);
+			StepVerifier
+				.create(consumerFactory.createConsumer(Schema.STRING,
+						List.of(builder -> builder.subscriptionType(SubscriptionType.Shared)))
+					.consumeNothing())
+				.expectComplete()
+				.verify(Duration.ofSeconds(10));
+		}
+		finally {
+			safeStopContainer(container);
+			pulsarClient.close();
+		}
+	}
+
+	@Test
+	void containerPropertiesTopicsPattern() throws Exception {
+		var pulsarClient = PulsarClient.builder().serviceUrl(PulsarTestContainerSupport.getPulsarBrokerUrl()).build();
+		ReactivePulsarMessageListenerContainer<String> container = null;
+		try {
+			var reactivePulsarClient = AdaptedReactivePulsarClientFactory.create(pulsarClient);
+			var topic = "drpmlct-6-foo";
+			var subscription = topic + "-sub";
+			ReactiveMessageConsumerBuilderCustomizer<String> customizer = (builder) -> {
+				builder.topic(topic);
+				builder.subscriptionName(topic + "-sub");
+			};
+			var consumerFactory = new DefaultReactivePulsarConsumerFactory<String>(reactivePulsarClient, null);
+			// Ensure subscription is created
+			consumerFactory.createConsumer(Schema.STRING, List.of(customizer))
+				.consumeNothing()
+				.block(Duration.ofSeconds(5));
+			var latch = new CountDownLatch(1);
+			var containerProperties = new ReactivePulsarContainerProperties<String>();
+			containerProperties.setSchema(Schema.STRING);
+			containerProperties.setTopicsPattern("persistent://public/default/drpmlct-6-.*");
+			containerProperties.setSubscriptionName(subscription);
+			containerProperties.setMessageHandler(
+					(ReactivePulsarOneByOneMessageHandler<String>) (msg) -> Mono.fromRunnable(latch::countDown));
+			container = new DefaultReactivePulsarMessageListenerContainer<>(consumerFactory, containerProperties);
+			container.start();
+
+			createPulsarTemplate(topic, reactivePulsarClient).send("hello john doe").subscribe();
+			assertThat(latch.await(10, TimeUnit.SECONDS)).isTrue();
+		}
+		finally {
+			safeStopContainer(container);
+			pulsarClient.close();
+		}
+	}
+
+	@Test
+	void deadLetterTopicCustomizer() throws Exception {
+		var pulsarClient = PulsarClient.builder().serviceUrl(PulsarTestContainerSupport.getPulsarBrokerUrl()).build();
+		ReactivePulsarMessageListenerContainer<String> container = null;
+		try {
+			var reactivePulsarClient = AdaptedReactivePulsarClientFactory.create(pulsarClient);
+			var topic = "drpmlct-7";
+			var deadLetterTopic = topic + "-dlt";
+			ReactiveMessageConsumerBuilderCustomizer<String> defaultConfig = (builder) -> {
+				builder.topic(topic);
+				builder.subscriptionName(topic + "-sub");
+				builder.negativeAckRedeliveryDelay(Duration.ZERO);
+			};
+			var consumerFactory = new DefaultReactivePulsarConsumerFactory<>(reactivePulsarClient,
+					List.of(defaultConfig));
+			var dlqConsumer = consumerFactory.createConsumer(Schema.STRING,
+					List.of((builder) -> builder.topics(List.of(deadLetterTopic))));
+			// Ensure subscriptions are created
+			consumerFactory.createConsumer(Schema.STRING).consumeNothing().block(Duration.ofSeconds(5));
+			dlqConsumer.consumeNothing().block(Duration.ofSeconds(5));
+			var latch = new CountDownLatch(6);
+			var containerProperties = new ReactivePulsarContainerProperties<String>();
+			containerProperties.setSchema(Schema.STRING);
+			containerProperties.setMessageHandler(
+					(ReactivePulsarStreamingHandler<String>) (msg) -> msg.doOnNext((m) -> latch.countDown())
+						.map((m) -> m.getValue().endsWith("4") ? MessageResult.negativeAcknowledge(m)
+								: MessageResult.acknowledge(m)));
+			containerProperties.setSubscriptionType(SubscriptionType.Shared);
+
+			var deadLetterPolicy = DeadLetterPolicy.builder()
+				.maxRedeliverCount(1)
+				.deadLetterTopic(deadLetterTopic)
+				.build();
+			container = new DefaultReactivePulsarMessageListenerContainer<>(consumerFactory, containerProperties);
+			container.setConsumerCustomizer(b -> b.deadLetterPolicy(deadLetterPolicy));
+			container.start();
+
+			var producerFactory = DefaultReactivePulsarSenderFactory.<String>builderFor(reactivePulsarClient)
+				.withDefaultTopic(topic)
+				.withDefaultConfigCustomizer((builder) -> builder.batchingEnabled(false))
+				.build();
+			var pulsarTemplate = new ReactivePulsarTemplate<>(producerFactory);
+			Flux.range(0, 5).map(i -> MessageSpec.of("hello john doe" + i)).as(pulsarTemplate::send).subscribe();
+			assertThat(latch.await(10, TimeUnit.SECONDS)).isTrue();
+
+			var dlqLatch = new CountDownLatch(1);
+			dlqConsumer.consumeOne(message -> {
+				if (message.getValue().endsWith("4")) {
+					dlqLatch.countDown();
+				}
+				return Mono.just(MessageResult.acknowledge(message));
+			}).block();
+			assertThat(dlqLatch.await(10, TimeUnit.SECONDS)).isTrue();
+		}
+		finally {
+			safeStopContainer(container);
+			pulsarClient.close();
+		}
+	}
+
+	private String topicNameForTest(String suffix) {
+		return "drpmlct-" + suffix;
+	}
+
+	private DefaultReactivePulsarConsumerFactory<String> createAndPrepareConsumerFactory(String topic,
+			ReactivePulsarClient reactivePulsarClient) {
+		ReactiveMessageConsumerBuilderCustomizer<String> defaultConfig = (builder) -> {
+			builder.topic(topic);
+			builder.subscriptionName(topic + "-sub");
+		};
+		var consumerFactory = new DefaultReactivePulsarConsumerFactory<>(reactivePulsarClient, List.of(defaultConfig));
 		// Ensure subscription is created
-		pulsarConsumerFactory.createConsumer(Schema.STRING).consumeNothing().block(Duration.ofSeconds(10));
-		CountDownLatch latch = new CountDownLatch(1);
-		ReactivePulsarContainerProperties<String> pulsarContainerProperties = new ReactivePulsarContainerProperties<>();
-		pulsarContainerProperties.setMessageHandler(
-				(ReactivePulsarOneByOneMessageHandler<String>) (msg) -> Mono.fromRunnable(latch::countDown));
-		pulsarContainerProperties.setSchema(Schema.STRING);
-		DefaultReactivePulsarMessageListenerContainer<String> container = new DefaultReactivePulsarMessageListenerContainer<>(
-				pulsarConsumerFactory, pulsarContainerProperties);
-		container.start();
-		MutableReactiveMessageSenderSpec prodConfig = new MutableReactiveMessageSenderSpec();
-		prodConfig.setTopicName(topic);
-		DefaultReactivePulsarSenderFactory<String> pulsarProducerFactory = new DefaultReactivePulsarSenderFactory<>(
-				reactivePulsarClient, prodConfig, null, null, new DefaultTopicResolver());
-		ReactivePulsarTemplate<String> pulsarTemplate = new ReactivePulsarTemplate<>(pulsarProducerFactory);
-		pulsarTemplate.send("hello john doe").subscribe();
-		assertThat(latch.await(10, TimeUnit.SECONDS)).isTrue();
-		container.stop();
-		pulsarClient.close();
+		consumerFactory.createConsumer(Schema.STRING).consumeNothing().block(Duration.ofSeconds(5));
+		return consumerFactory;
 	}
 
-	@Test
-	void streamingHandlerListener() throws Exception {
-		String topic = "drpmlct-013";
-		MutableReactiveMessageConsumerSpec config = new MutableReactiveMessageConsumerSpec();
-		config.setTopicNames(Collections.singletonList(topic));
-		config.setSubscriptionName("drpmlct-sb-013");
-		PulsarClient pulsarClient = PulsarClient.builder()
-			.serviceUrl(PulsarTestContainerSupport.getPulsarBrokerUrl())
+	private ReactivePulsarTemplate<String> createPulsarTemplate(String topic,
+			ReactivePulsarClient reactivePulsarClient) {
+		var producerFactory = DefaultReactivePulsarSenderFactory.<String>builderFor(reactivePulsarClient)
+			.withDefaultTopic(topic)
 			.build();
-		ReactivePulsarClient reactivePulsarClient = AdaptedReactivePulsarClientFactory.create(pulsarClient);
-		DefaultReactivePulsarConsumerFactory<String> pulsarConsumerFactory = new DefaultReactivePulsarConsumerFactory<>(
-				reactivePulsarClient, config);
-		// Ensure subscription is created
-		pulsarConsumerFactory.createConsumer(Schema.STRING).consumeNothing().block(Duration.ofSeconds(10));
-		CountDownLatch latch = new CountDownLatch(5);
-		ReactivePulsarContainerProperties<String> pulsarContainerProperties = new ReactivePulsarContainerProperties<>();
-		pulsarContainerProperties
-			.setMessageHandler((ReactivePulsarStreamingHandler<String>) (msg) -> msg.doOnNext((m) -> latch.countDown())
-				.map(MessageResult::acknowledge));
-		pulsarContainerProperties.setSchema(Schema.STRING);
-		DefaultReactivePulsarMessageListenerContainer<String> container = new DefaultReactivePulsarMessageListenerContainer<>(
-				pulsarConsumerFactory, pulsarContainerProperties);
-		container.start();
-		MutableReactiveMessageSenderSpec prodConfig = new MutableReactiveMessageSenderSpec();
-		prodConfig.setTopicName(topic);
-		DefaultReactivePulsarSenderFactory<String> pulsarProducerFactory = new DefaultReactivePulsarSenderFactory<>(
-				reactivePulsarClient, prodConfig, null, null, new DefaultTopicResolver());
-		ReactivePulsarTemplate<String> pulsarTemplate = new ReactivePulsarTemplate<>(pulsarProducerFactory);
-		Flux.range(0, 5).map(i -> MessageSpec.of("hello john doe" + i)).as(pulsarTemplate::send).subscribe();
-		assertThat(latch.await(10, TimeUnit.SECONDS)).isTrue();
-		container.stop();
-		pulsarClient.close();
+		return new ReactivePulsarTemplate<>(producerFactory);
 	}
 
-	@Test
-	void containerProperties() throws Exception {
-		String topic = "drpmlct-sb-014";
-		String subscriptionName = "drpmlct-sb-014";
-		PulsarClient pulsarClient = PulsarClient.builder()
-			.serviceUrl(PulsarTestContainerSupport.getPulsarBrokerUrl())
-			.build();
-		ReactivePulsarClient reactivePulsarClient = AdaptedReactivePulsarClientFactory.create(pulsarClient);
-		DefaultReactivePulsarConsumerFactory<String> pulsarConsumerFactory = new DefaultReactivePulsarConsumerFactory<>(
-				reactivePulsarClient, null);
-		// Ensure subscription is created
-		pulsarConsumerFactory
-			.createConsumer(Schema.STRING,
-					Collections.singletonList(
-							c -> c.topics(Collections.singletonList(topic)).subscriptionName(subscriptionName)))
-			.consumeNothing()
-			.block(Duration.ofSeconds(10));
-		CountDownLatch latch = new CountDownLatch(1);
-		ReactivePulsarContainerProperties<String> pulsarContainerProperties = new ReactivePulsarContainerProperties<>();
-		pulsarContainerProperties.setMessageHandler(
-				(ReactivePulsarOneByOneMessageHandler<String>) (msg) -> Mono.fromRunnable(latch::countDown));
-		pulsarContainerProperties.setSchema(Schema.STRING);
-		pulsarContainerProperties.setTopics(List.of(topic));
-		pulsarContainerProperties.setSubscriptionName(subscriptionName);
-		pulsarContainerProperties.setConcurrency(5);
-		pulsarContainerProperties.setUseKeyOrderedProcessing(true);
-		pulsarContainerProperties.setHandlingTimeout(Duration.ofMillis(7));
-		DefaultReactivePulsarMessageListenerContainer<String> container = new DefaultReactivePulsarMessageListenerContainer<>(
-				pulsarConsumerFactory, pulsarContainerProperties);
-		container.start();
-		MutableReactiveMessageSenderSpec prodConfig = new MutableReactiveMessageSenderSpec();
-		prodConfig.setTopicName(topic);
-		DefaultReactivePulsarSenderFactory<String> pulsarProducerFactory = new DefaultReactivePulsarSenderFactory<>(
-				reactivePulsarClient, prodConfig, null, null, new DefaultTopicResolver());
-		ReactivePulsarTemplate<String> pulsarTemplate = new ReactivePulsarTemplate<>(pulsarProducerFactory);
-		pulsarTemplate.send("hello john doe").subscribe();
-		assertThat(latch.await(10, TimeUnit.SECONDS)).isTrue();
-
-		assertThat(container).extracting("pipeline", InstanceOfAssertFactories.type(ReactiveMessagePipeline.class))
-			.hasFieldOrPropertyWithValue("concurrency", 5)
-			.hasFieldOrPropertyWithValue("handlingTimeout", Duration.ofMillis(7))
-			.extracting("groupingFunction")
-			.isInstanceOf(DefaultMessageGroupingFunction.class);
-
-		container.stop();
-		pulsarClient.close();
-	}
-
-	@Test
-	void defaultSubscriptionType() throws Exception {
-		String topic = "drpmlct-015";
-		MutableReactiveMessageConsumerSpec config = new MutableReactiveMessageConsumerSpec();
-		config.setTopicNames(Collections.singletonList(topic));
-		config.setSubscriptionName("drpmlct-sb-015");
-		PulsarClient pulsarClient = PulsarClient.builder()
-			.serviceUrl(PulsarTestContainerSupport.getPulsarBrokerUrl())
-			.build();
-		ReactivePulsarClient reactivePulsarClient = AdaptedReactivePulsarClientFactory.create(pulsarClient);
-		DefaultReactivePulsarConsumerFactory<String> pulsarConsumerFactory = new DefaultReactivePulsarConsumerFactory<>(
-				reactivePulsarClient, config);
-
-		ReactivePulsarContainerProperties<String> pulsarContainerProperties = new ReactivePulsarContainerProperties<>();
-		pulsarContainerProperties
-			.setMessageHandler((ReactivePulsarOneByOneMessageHandler<String>) (msg) -> Mono.empty());
-		pulsarContainerProperties.setSchema(Schema.STRING);
-		DefaultReactivePulsarMessageListenerContainer<String> container = new DefaultReactivePulsarMessageListenerContainer<>(
-				pulsarConsumerFactory, pulsarContainerProperties);
-		container.start();
-
-		Thread.sleep(2_000);
-
-		StepVerifier
-			.create(pulsarConsumerFactory.createConsumer(Schema.STRING,
-					Collections.singletonList(c -> c.subscriptionType(SubscriptionType.Shared)))
-				.consumeNothing())
-			.expectError()
-			.verify(Duration.ofSeconds(10));
-
-		container.stop();
-		pulsarClient.close();
-	}
-
-	@Test
-	void containerSubscriptionType() throws Exception {
-		String topic = "drpmlct-016";
-		MutableReactiveMessageConsumerSpec config = new MutableReactiveMessageConsumerSpec();
-		config.setTopicNames(Collections.singletonList(topic));
-		config.setSubscriptionName("drpmlct-sb-016");
-		PulsarClient pulsarClient = PulsarClient.builder()
-			.serviceUrl(PulsarTestContainerSupport.getPulsarBrokerUrl())
-			.build();
-		ReactivePulsarClient reactivePulsarClient = AdaptedReactivePulsarClientFactory.create(pulsarClient);
-		DefaultReactivePulsarConsumerFactory<String> pulsarConsumerFactory = new DefaultReactivePulsarConsumerFactory<>(
-				reactivePulsarClient, config);
-
-		ReactivePulsarContainerProperties<String> pulsarContainerProperties = new ReactivePulsarContainerProperties<>();
-		pulsarContainerProperties
-			.setMessageHandler((ReactivePulsarOneByOneMessageHandler<String>) (msg) -> Mono.empty());
-		pulsarContainerProperties.setSchema(Schema.STRING);
-		pulsarContainerProperties.setSubscriptionType(SubscriptionType.Shared);
-		DefaultReactivePulsarMessageListenerContainer<String> container = new DefaultReactivePulsarMessageListenerContainer<>(
-				pulsarConsumerFactory, pulsarContainerProperties);
-		container.start();
-
-		Thread.sleep(2_000);
-
-		StepVerifier
-			.create(pulsarConsumerFactory.createConsumer(Schema.STRING,
-					Collections.singletonList(c -> c.subscriptionType(SubscriptionType.Shared)))
-				.consumeNothing())
-			.expectComplete()
-			.verify(Duration.ofSeconds(10));
-
-		container.stop();
-		pulsarClient.close();
-	}
-
-	@Test
-	void containerTopicsPattern() throws Exception {
-		String topic = "drpmlct-017-foo";
-		String subscriptionName = "drpmlct-sb-017";
-		PulsarClient pulsarClient = PulsarClient.builder()
-			.serviceUrl(PulsarTestContainerSupport.getPulsarBrokerUrl())
-			.build();
-		ReactivePulsarClient reactivePulsarClient = AdaptedReactivePulsarClientFactory.create(pulsarClient);
-		DefaultReactivePulsarConsumerFactory<String> pulsarConsumerFactory = new DefaultReactivePulsarConsumerFactory<>(
-				reactivePulsarClient, null);
-		// Ensure subscription is created
-		pulsarConsumerFactory
-			.createConsumer(Schema.STRING,
-					Collections.singletonList(
-							c -> c.topics(Collections.singletonList(topic)).subscriptionName(subscriptionName)))
-			.consumeNothing()
-			.block(Duration.ofSeconds(10));
-		CountDownLatch latch = new CountDownLatch(1);
-		ReactivePulsarContainerProperties<String> pulsarContainerProperties = new ReactivePulsarContainerProperties<>();
-		pulsarContainerProperties.setMessageHandler(
-				(ReactivePulsarOneByOneMessageHandler<String>) (msg) -> Mono.fromRunnable(latch::countDown));
-		pulsarContainerProperties.setSchema(Schema.STRING);
-		pulsarContainerProperties.setTopicsPattern("persistent://public/default/drpmlct-017-.*");
-		pulsarContainerProperties.setSubscriptionName(subscriptionName);
-		DefaultReactivePulsarMessageListenerContainer<String> container = new DefaultReactivePulsarMessageListenerContainer<>(
-				pulsarConsumerFactory, pulsarContainerProperties);
-		container.start();
-		MutableReactiveMessageSenderSpec prodConfig = new MutableReactiveMessageSenderSpec();
-		prodConfig.setTopicName(topic);
-		DefaultReactivePulsarSenderFactory<String> pulsarProducerFactory = new DefaultReactivePulsarSenderFactory<>(
-				reactivePulsarClient, prodConfig, null, null, new DefaultTopicResolver());
-		ReactivePulsarTemplate<String> pulsarTemplate = new ReactivePulsarTemplate<>(pulsarProducerFactory);
-		pulsarTemplate.send("hello john doe").subscribe();
-		assertThat(latch.await(10, TimeUnit.SECONDS)).isTrue();
-
-		container.stop();
-		pulsarClient.close();
-	}
-
-	@Test
-	void consumerCustomizer() throws Exception {
-		String topic = "drpmlct-018";
-		String deadLetterTopic = "drpmlct-018-dlq-topic";
-		MutableReactiveMessageConsumerSpec config = new MutableReactiveMessageConsumerSpec();
-		config.setTopicNames(Collections.singletonList(topic));
-		config.setSubscriptionName("drpmlct-sb-018");
-		config.setNegativeAckRedeliveryDelay(Duration.ZERO);
-		PulsarClient pulsarClient = PulsarClient.builder()
-			.serviceUrl(PulsarTestContainerSupport.getPulsarBrokerUrl())
-			.build();
-		ReactivePulsarClient reactivePulsarClient = AdaptedReactivePulsarClientFactory.create(pulsarClient);
-		DefaultReactivePulsarConsumerFactory<String> pulsarConsumerFactory = new DefaultReactivePulsarConsumerFactory<>(
-				reactivePulsarClient, config);
-		ReactiveMessageConsumer<String> dlqConsumer = pulsarConsumerFactory.createConsumer(Schema.STRING,
-				Collections.singletonList(b -> b.topics(Collections.singletonList(deadLetterTopic))));
-
-		// Ensure subscriptions are created
-		pulsarConsumerFactory.createConsumer(Schema.STRING).consumeNothing().block(Duration.ofSeconds(10));
-		dlqConsumer.consumeNothing().block(Duration.ofSeconds(10));
-
-		CountDownLatch latch = new CountDownLatch(6);
-		ReactivePulsarContainerProperties<String> pulsarContainerProperties = new ReactivePulsarContainerProperties<>();
-		pulsarContainerProperties
-			.setMessageHandler((ReactivePulsarStreamingHandler<String>) (msg) -> msg.doOnNext((m) -> latch.countDown())
-				.map((m) -> m.getValue().endsWith("4") ? MessageResult.negativeAcknowledge(m)
-						: MessageResult.acknowledge(m)));
-		pulsarContainerProperties.setSchema(Schema.STRING);
-		pulsarContainerProperties.setSubscriptionType(SubscriptionType.Shared);
-		DefaultReactivePulsarMessageListenerContainer<String> container = new DefaultReactivePulsarMessageListenerContainer<>(
-				pulsarConsumerFactory, pulsarContainerProperties);
-		DeadLetterPolicy deadLetterPolicy = DeadLetterPolicy.builder()
-			.maxRedeliverCount(1)
-			.deadLetterTopic(deadLetterTopic)
-			.build();
-		container.setConsumerCustomizer(b -> b.deadLetterPolicy(deadLetterPolicy));
-		container.start();
-		MutableReactiveMessageSenderSpec prodConfig = new MutableReactiveMessageSenderSpec();
-		prodConfig.setBatchingEnabled(false);
-		prodConfig.setTopicName(topic);
-		DefaultReactivePulsarSenderFactory<String> pulsarProducerFactory = new DefaultReactivePulsarSenderFactory<>(
-				reactivePulsarClient, prodConfig, null, null, new DefaultTopicResolver());
-		ReactivePulsarTemplate<String> pulsarTemplate = new ReactivePulsarTemplate<>(pulsarProducerFactory);
-		Flux.range(0, 5).map(i -> MessageSpec.of("hello john doe" + i)).as(pulsarTemplate::send).subscribe();
-		assertThat(latch.await(10, TimeUnit.SECONDS)).isTrue();
-
-		CountDownLatch dlqLatch = new CountDownLatch(1);
-		dlqConsumer.consumeOne(message -> {
-			if (message.getValue().endsWith("4")) {
-				dlqLatch.countDown();
+	private void safeStopContainer(ReactivePulsarMessageListenerContainer<?> container) {
+		try {
+			if (container != null) {
+				container.stop();
 			}
-			return Mono.just(MessageResult.acknowledge(message));
-		}).block();
-
-		assertThat(dlqLatch.await(10, TimeUnit.SECONDS)).isTrue();
-
-		container.stop();
-		pulsarClient.close();
+		}
+		catch (Exception ex) {
+			logger.warn(ex, "Failed to stop container %s: %s".formatted(container, ex.getMessage()));
+		}
 	}
 
 }

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

@@ -20,6 +20,7 @@ import static org.assertj.core.api.Assertions.assertThat;
 
 import java.nio.charset.StandardCharsets;
 import java.time.Duration;
+import java.util.Collections;
 import java.util.List;
 import java.util.Objects;
 import java.util.concurrent.BlockingQueue;
@@ -44,7 +45,6 @@ import org.apache.pulsar.common.schema.KeyValueEncodingType;
 import org.apache.pulsar.common.schema.SchemaType;
 import org.apache.pulsar.reactive.client.adapter.AdaptedReactivePulsarClientFactory;
 import org.apache.pulsar.reactive.client.api.MessageResult;
-import org.apache.pulsar.reactive.client.api.MutableReactiveMessageConsumerSpec;
 import org.apache.pulsar.reactive.client.api.ReactivePulsarClient;
 import org.junit.jupiter.api.Nested;
 import org.junit.jupiter.api.Test;
@@ -124,7 +124,7 @@ public class ReactivePulsarListenerTests implements PulsarTestContainerSupport {
 
 		@Bean
 		public ReactivePulsarConsumerFactory<String> pulsarConsumerFactory(ReactivePulsarClient pulsarClient) {
-			return new DefaultReactivePulsarConsumerFactory<>(pulsarClient, new MutableReactiveMessageConsumerSpec());
+			return new DefaultReactivePulsarConsumerFactory<>(pulsarClient, Collections.emptyList());
 		}
 
 		@Bean