Use builder to autoconfigure PulsarReaderFactory (#400)

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

@@ -174,7 +174,7 @@ public class PulsarAutoConfiguration {
 	@Bean
 	@ConditionalOnMissingBean
 	public PulsarReaderFactory<?> pulsarReaderFactory(PulsarClient pulsarClient) {
-		return new DefaultPulsarReaderFactory<>(pulsarClient, this.properties.buildReaderProperties());
+		return new DefaultPulsarReaderFactory<>(pulsarClient, this.properties.getReader().toReaderBuilderCustomizer());
 	}
 
 }

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

@@ -31,6 +31,7 @@ import org.springframework.boot.context.properties.ConfigurationProperties;
 import org.springframework.boot.context.properties.NestedConfigurationProperty;
 import org.springframework.boot.context.properties.PropertyMapper;
 import org.springframework.lang.Nullable;
+import org.springframework.pulsar.core.ReaderBuilderCustomizer;
 import org.springframework.pulsar.listener.AckMode;
 import org.springframework.util.CollectionUtils;
 import org.springframework.util.StringUtils;
@@ -110,10 +111,6 @@ public class PulsarProperties {
 		return new HashMap<>(this.admin.buildProperties());
 	}
 
-	public Map<String, Object> buildReaderProperties() {
-		return new HashMap<>(this.reader.buildProperties());
-	}
-
 	public static class Template {
 
 		/**
@@ -1290,21 +1287,17 @@ public class PulsarProperties {
 			this.resetIncludeHead = resetIncludeHead;
 		}
 
-		public Map<String, Object> buildProperties() {
-
-			PulsarProperties.Properties properties = new Properties();
-
-			PropertyMapper map = PropertyMapper.get().alwaysApplyingWhenNonNull();
-
-			map.from(this::getTopicNames).to(properties.in("topicNames"));
-			map.from(this::getReceiverQueueSize).to(properties.in("receiverQueueSize"));
-			map.from(this::getReaderName).to(properties.in("readerName"));
-			map.from(this::getSubscriptionName).to(properties.in("subscriptionName"));
-			map.from(this::getSubscriptionRolePrefix).to(properties.in("subscriptionRolePrefix"));
-			map.from(this::getReadCompacted).to(properties.in("readCompacted"));
-			map.from(this::getResetIncludeHead).to(properties.in("resetIncludeHead"));
-
-			return properties;
+		public ReaderBuilderCustomizer<?> toReaderBuilderCustomizer() {
+			return (readerBuilder) -> {
+				PropertyMapper map = PropertyMapper.get().alwaysApplyingWhenNonNull();
+				map.from(this::getTopicNames).as(ArrayList::new).to(readerBuilder::topics);
+				map.from(this::getReceiverQueueSize).to(readerBuilder::receiverQueueSize);
+				map.from(this::getReaderName).to(readerBuilder::readerName);
+				map.from(this::getSubscriptionName).to(readerBuilder::subscriptionName);
+				map.from(this::getSubscriptionRolePrefix).to(readerBuilder::subscriptionRolePrefix);
+				map.from(this::getReadCompacted).to(readerBuilder::readCompacted);
+				map.from(this::getResetIncludeHead).whenTrue().to((b) -> readerBuilder.startMessageIdInclusive());
+			};
 		}
 
 	}

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

@@ -25,7 +25,6 @@ import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.when;
 
 import java.util.Collections;
-import java.util.List;
 import java.util.concurrent.TimeUnit;
 
 import org.apache.pulsar.client.api.ClientBuilder;
@@ -68,6 +67,7 @@ import org.springframework.pulsar.core.PulsarConsumerFactory;
 import org.springframework.pulsar.core.PulsarProducerFactory;
 import org.springframework.pulsar.core.PulsarReaderFactory;
 import org.springframework.pulsar.core.PulsarTemplate;
+import org.springframework.pulsar.core.ReaderBuilderCustomizer;
 import org.springframework.pulsar.core.SchemaResolver;
 import org.springframework.pulsar.core.SchemaResolver.SchemaResolverCustomizer;
 import org.springframework.pulsar.core.TopicResolver;
@@ -115,12 +115,13 @@ class PulsarAutoConfigurationTests {
 	@Test
 	void defaultBeansAreAutoConfigured() {
 		this.contextRunner.run((context) -> assertThat(context).hasSingleBean(PulsarClientBuilderConfigurer.class)
-				.hasSingleBean(PulsarClient.class).hasSingleBean(PulsarProducerFactory.class)
-				.hasSingleBean(PulsarTemplate.class).hasSingleBean(PulsarConsumerFactory.class)
+				.hasSingleBean(PulsarClient.class).hasSingleBean(PulsarAdministration.class)
+				.hasSingleBean(PulsarProducerFactory.class).hasSingleBean(PulsarTemplate.class)
+				.hasSingleBean(PulsarConsumerFactory.class).hasSingleBean(PulsarReaderFactory.class)
 				.hasSingleBean(ConcurrentPulsarListenerContainerFactory.class)
 				.hasSingleBean(PulsarListenerAnnotationBeanPostProcessor.class)
-				.hasSingleBean(PulsarListenerEndpointRegistry.class).hasSingleBean(PulsarAdministration.class)
-				.hasSingleBean(DefaultSchemaResolver.class).hasSingleBean(DefaultTopicResolver.class));
+				.hasSingleBean(PulsarListenerEndpointRegistry.class).hasSingleBean(DefaultSchemaResolver.class)
+				.hasSingleBean(DefaultTopicResolver.class));
 	}
 
 	@Nested
@@ -225,11 +226,32 @@ class PulsarAutoConfigurationTests {
 
 	}
 
+	@Nested
+	class ReaderFactoryTests {
+
+		@Test
+		void customPulsarReaderFactoryIsRespected() {
+			PulsarReaderFactory<String> readerFactory = mock(PulsarReaderFactory.class);
+			contextRunner.withBean("customPulsarReaderFactory", PulsarReaderFactory.class, () -> readerFactory)
+					.run((context) -> assertThat(context).getBean(PulsarReaderFactory.class).isSameAs(readerFactory));
+		}
+
+		@Test
+		void beansAreInjectedInReaderFactory() {
+			contextRunner.withUserConfiguration(SpyCustomizersConfig.class)
+					.run((context) -> assertThat(context).getBean(DefaultPulsarReaderFactory.class)
+							.hasFieldOrPropertyWithValue("pulsarClient", context.getBean(PulsarClient.class))
+							.hasFieldOrPropertyWithValue("defaultConfigCustomizer",
+									SpyCustomizersConfig.testReaderCustomizer));
+		}
+
+	}
+
 	/*
 	 * Use '@TestConfiguration' and exact name of the PulsarProperties bean that is
 	 * created via the '@EnableConfigurationProperties' on the actual auto-config in order
 	 * to 'replace' the PulsarProperties bean - all of this effort is to make sure the
-	 * returned producer/consumer builder customizer is the one we expect.
+	 * returned producer/consumer/reader builder customizer is the one we expect.
 	 */
 	@TestConfiguration(proxyBeanMethods = false)
 	static class SpyCustomizersConfig {
@@ -242,50 +264,32 @@ class PulsarAutoConfigurationTests {
 		static ConsumerBuilderCustomizer testConsumerCustomizer = (consumerBuilder) -> {
 		};
 
+		@SuppressWarnings("rawtypes")
+		static ReaderBuilderCustomizer testReaderCustomizer = (readerBuilder) -> {
+		};
+
 		@Bean(name = "spring.pulsar-org.springframework.pulsar.autoconfigure.PulsarProperties")
 		PulsarProperties pulsarProperties() {
 			var pulsarProps = new PulsarProperties();
+
 			var producerProps = spy(pulsarProps.getProducer());
 			when(producerProps.toProducerBuilderCustomizer()).thenReturn(testProducerCustomizer);
+
 			var consumerProps = spy(pulsarProps.getConsumer());
 			when(consumerProps.toConsumerBuilderCustomizer()).thenReturn(testConsumerCustomizer);
 
+			var readerProps = spy(pulsarProps.getReader());
+			when(readerProps.toReaderBuilderCustomizer()).thenReturn(testReaderCustomizer);
+
 			var spyPulsarProps = spy(pulsarProps);
 			when(spyPulsarProps.getProducer()).thenReturn(producerProps);
 			when(spyPulsarProps.getConsumer()).thenReturn(consumerProps);
+			when(spyPulsarProps.getReader()).thenReturn(readerProps);
 			return spyPulsarProps;
 		}
 
 	}
 
-	@Nested
-	class ReaderFactoryTests {
-
-		@Test
-		void readerFactoryIsAutoConfiguredByDefault() {
-			contextRunner.run((context) -> assertThat(context).hasNotFailed().hasSingleBean(PulsarReaderFactory.class)
-					.getBean(PulsarReaderFactory.class).isExactlyInstanceOf(DefaultPulsarReaderFactory.class));
-		}
-
-		@Test
-		void readerFactoryCanBeConfigured() {
-			contextRunner.withPropertyValues("spring.pulsar.reader.topic-names=foo",
-					"spring.pulsar.reader.receiver-queue-size=200", "spring.pulsar.reader.reader-name=test-reader",
-					"spring.pulsar.reader.subscription-name=test-subscription",
-					"spring.pulsar.reader.subscription-role-prefix=test-prefix",
-					"spring.pulsar.reader.read-compacted=true", "spring.pulsar.reader.reset-include-head=true")
-					.run((context -> assertThat(context).hasNotFailed().getBean(PulsarReaderFactory.class)
-							.extracting("readerConfig").hasFieldOrPropertyWithValue("topicNames", List.of("foo"))
-							.hasFieldOrPropertyWithValue("receiverQueueSize", 200)
-							.hasFieldOrPropertyWithValue("readerName", "test-reader")
-							.hasFieldOrPropertyWithValue("subscriptionName", "test-subscription")
-							.hasFieldOrPropertyWithValue("subscriptionRolePrefix", "test-prefix")
-							.hasFieldOrPropertyWithValue("readCompacted", true)
-							.hasFieldOrPropertyWithValue("resetIncludeHead", true)));
-		}
-
-	}
-
 	@Nested
 	class SchemaAndTopicResolversTests {
 

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

@@ -19,12 +19,15 @@ package org.springframework.pulsar.autoconfigure;
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.assertj.core.api.Assertions.assertThatExceptionOfType;
 import static org.assertj.core.api.Assertions.assertThatIllegalArgumentException;
-import static org.assertj.core.api.Assertions.assertThatNoException;
 import static org.assertj.core.api.Assertions.assertThatRuntimeException;
 import static org.assertj.core.api.Assertions.entry;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.verify;
 
 import java.time.Duration;
 import java.util.HashMap;
+import java.util.List;
 import java.util.Map;
 
 import org.apache.pulsar.client.admin.PulsarAdmin;
@@ -35,14 +38,14 @@ import org.apache.pulsar.client.api.MessageRoutingMode;
 import org.apache.pulsar.client.api.ProducerAccessMode;
 import org.apache.pulsar.client.api.ProducerCryptoFailureAction;
 import org.apache.pulsar.client.api.ProxyProtocol;
+import org.apache.pulsar.client.api.ReaderBuilder;
 import org.apache.pulsar.client.api.RegexSubscriptionMode;
 import org.apache.pulsar.client.api.SubscriptionInitialPosition;
 import org.apache.pulsar.client.api.SubscriptionMode;
 import org.apache.pulsar.client.api.SubscriptionType;
-import org.apache.pulsar.client.impl.conf.ConfigurationDataUtils;
-import org.apache.pulsar.client.impl.conf.ReaderConfigurationData;
 import org.apache.pulsar.common.schema.SchemaType;
 import org.assertj.core.api.InstanceOfAssertFactories;
+import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Nested;
 import org.junit.jupiter.api.Test;
 
@@ -556,10 +559,9 @@ public class PulsarPropertiesTests {
 	@Nested
 	class ReaderPropertiesTests {
 
-		@Test
-		void readerProperties() {
+		@BeforeEach
+		void bindProperties() {
 			Map<String, String> props = new HashMap<>();
-
 			props.put("spring.pulsar.reader.topic-names", "my-topic");
 			props.put("spring.pulsar.reader.receiver-queue-size", "100");
 			props.put("spring.pulsar.reader.reader-name", "my-reader");
@@ -568,21 +570,43 @@ public class PulsarPropertiesTests {
 			props.put("spring.pulsar.reader.read-compacted", "true");
 			props.put("spring.pulsar.reader.reset-include-head", "true");
 			bind(props);
+		}
 
-			Map<String, Object> readerProps = properties.buildReaderProperties();
+		@Test
+		void readerProperties() {
+			var readerProps = properties.getReader();
+			assertThat(readerProps.getTopicNames()).containsExactly("my-topic");
+			assertThat(readerProps.getReceiverQueueSize()).isEqualTo(100);
+			assertThat(readerProps.getReaderName()).isEqualTo("my-reader");
+			assertThat(readerProps.getSubscriptionName()).isEqualTo("my-subscription");
+			assertThat(readerProps.getSubscriptionRolePrefix()).isEqualTo("sub-role");
+			assertThat(readerProps.getReadCompacted()).isTrue();
+			assertThat(readerProps.getResetIncludeHead()).isTrue();
+		}
 
-			// Verify that the props can be loaded in a ReaderBuilder
-			assertThatNoException().isThrownBy(() -> ConfigurationDataUtils.loadData(readerProps,
-					new ReaderConfigurationData<>(), ReaderConfigurationData.class));
+		@SuppressWarnings("unchecked")
+		@Test
+		void toReaderCustomizer() {
+			var readerBuilder = mock(ReaderBuilder.class);
+			var customizer = properties.getReader().toReaderBuilderCustomizer();
+			customizer.customize(readerBuilder);
+			verify(readerBuilder).topics(List.of("my-topic"));
+			verify(readerBuilder).receiverQueueSize(100);
+			verify(readerBuilder).readerName("my-reader");
+			verify(readerBuilder).subscriptionName("my-subscription");
+			verify(readerBuilder).subscriptionRolePrefix("sub-role");
+			verify(readerBuilder).readCompacted(true);
+			verify(readerBuilder).startMessageIdInclusive();
+		}
 
-			assertThat(readerProps)
-					.hasEntrySatisfying("topicNames",
-							topics -> assertThat(topics).asInstanceOf(InstanceOfAssertFactories.list(String.class))
-									.containsExactly("my-topic"))
-					.containsEntry("receiverQueueSize", 100).containsEntry("readerName", "my-reader")
-					.containsEntry("subscriptionName", "my-subscription")
-					.containsEntry("subscriptionRolePrefix", "sub-role").containsEntry("readCompacted", true)
-					.containsEntry("resetIncludeHead", true);
+		@SuppressWarnings("unchecked")
+		@Test
+		void toReaderCustomizerResetDoesNotIncludeHead() {
+			properties.getReader().setResetIncludeHead(false);
+			var readerBuilder = mock(ReaderBuilder.class);
+			var customizer = properties.getReader().toReaderBuilderCustomizer();
+			customizer.customize(readerBuilder);
+			verify(readerBuilder, never()).startMessageIdInclusive();
 		}
 
 	}

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/test/java/org/springframework/pulsar/core/DefaultPulsarReaderFactoryTests.java

@@ -64,7 +64,7 @@ public class DefaultPulsarReaderFactoryTests implements PulsarTestContainerSuppo
 
 		@BeforeEach
 		void createReaderFactory() {
-			pulsarReaderFactory = new DefaultPulsarReaderFactory<>(pulsarClient, Collections.emptyMap());
+			pulsarReaderFactory = new DefaultPulsarReaderFactory<>(pulsarClient);
 		}
 
 		@Test
@@ -127,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
@@ -136,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/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;
@@ -68,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");
@@ -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();
@@ -136,8 +137,7 @@ 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);

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

@@ -19,7 +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.HashMap;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
 
@@ -87,7 +86,7 @@ public class PulsarReaderTests implements PulsarTestContainerSupport {
 
 		@Bean
 		public PulsarReaderFactory<?> pulsarReaderFactory(PulsarClient pulsarClient) {
-			return new DefaultPulsarReaderFactory<>(pulsarClient, new HashMap<>());
+			return new DefaultPulsarReaderFactory<>(pulsarClient);
 		}
 
 		@Bean