From a88dc541c70f515d5505568cd73aa8f392541468 Mon Sep 17 00:00:00 2001 From: Chris Bono Date: Sat, 6 May 2023 09:04:59 -0500 Subject: [PATCH] Use builder to autoconfigure PulsarReaderFactory (#400) --- .../PulsarAutoConfiguration.java | 2 +- .../autoconfigure/PulsarProperties.java | 31 ++++---- .../PulsarAutoConfigurationTests.java | 72 ++++++++++--------- .../autoconfigure/PulsarPropertiesTests.java | 60 +++++++++++----- .../core/DefaultPulsarReaderFactory.java | 47 +++++++++--- .../core/DefaultPulsarReaderFactoryTests.java | 61 +++++++++++++++- ...aultPulsarMessageReaderContainerTests.java | 14 ++-- .../pulsar/reader/PulsarReaderTests.java | 3 +- 8 files changed, 196 insertions(+), 94 deletions(-) diff --git a/spring-pulsar-spring-boot-autoconfigure/src/main/java/org/springframework/pulsar/autoconfigure/PulsarAutoConfiguration.java b/spring-pulsar-spring-boot-autoconfigure/src/main/java/org/springframework/pulsar/autoconfigure/PulsarAutoConfiguration.java index 00afebdd..6111ce17 100644 --- a/spring-pulsar-spring-boot-autoconfigure/src/main/java/org/springframework/pulsar/autoconfigure/PulsarAutoConfiguration.java +++ b/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()); } } diff --git a/spring-pulsar-spring-boot-autoconfigure/src/main/java/org/springframework/pulsar/autoconfigure/PulsarProperties.java b/spring-pulsar-spring-boot-autoconfigure/src/main/java/org/springframework/pulsar/autoconfigure/PulsarProperties.java index 5fdb9dc4..c9decd8e 100644 --- a/spring-pulsar-spring-boot-autoconfigure/src/main/java/org/springframework/pulsar/autoconfigure/PulsarProperties.java +++ b/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 buildReaderProperties() { - return new HashMap<>(this.reader.buildProperties()); - } - public static class Template { /** @@ -1290,21 +1287,17 @@ public class PulsarProperties { this.resetIncludeHead = resetIncludeHead; } - public Map 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()); + }; } } diff --git a/spring-pulsar-spring-boot-autoconfigure/src/test/java/org/springframework/pulsar/autoconfigure/PulsarAutoConfigurationTests.java b/spring-pulsar-spring-boot-autoconfigure/src/test/java/org/springframework/pulsar/autoconfigure/PulsarAutoConfigurationTests.java index 6c9add11..1b961191 100644 --- a/spring-pulsar-spring-boot-autoconfigure/src/test/java/org/springframework/pulsar/autoconfigure/PulsarAutoConfigurationTests.java +++ b/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 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 { diff --git a/spring-pulsar-spring-boot-autoconfigure/src/test/java/org/springframework/pulsar/autoconfigure/PulsarPropertiesTests.java b/spring-pulsar-spring-boot-autoconfigure/src/test/java/org/springframework/pulsar/autoconfigure/PulsarPropertiesTests.java index c22d9d78..b1f56147 100644 --- a/spring-pulsar-spring-boot-autoconfigure/src/test/java/org/springframework/pulsar/autoconfigure/PulsarPropertiesTests.java +++ b/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 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 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(); } } diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/core/DefaultPulsarReaderFactory.java b/spring-pulsar/src/main/java/org/springframework/pulsar/core/DefaultPulsarReaderFactory.java index fe3f076e..f07d450a 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/core/DefaultPulsarReaderFactory.java +++ b/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 implements PulsarReaderFactory { private final PulsarClient pulsarClient; - private final Map readerConfig; + @Nullable + private final ReaderBuilderCustomizer 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 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 defaultConfigCustomizer) { this.pulsarClient = pulsarClient; - this.readerConfig = readerConfig; + this.defaultConfigCustomizer = defaultConfigCustomizer; } @Override @@ -57,12 +70,19 @@ public class DefaultPulsarReaderFactory implements PulsarReaderFactory { @Nullable List> customizers) throws PulsarClientException { Objects.requireNonNull(schema, "Schema must be specified"); ReaderBuilder 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 implements PulsarReaderFactory { return readerBuilder.create(); } + private void replaceTopicsOnBuilder(ReaderBuilder builder, Collection topics) { + var builderImpl = (ReaderBuilderImpl) builder; + builderImpl.getConf().setTopicNames(new HashSet<>(topics)); + } + } diff --git a/spring-pulsar/src/test/java/org/springframework/pulsar/core/DefaultPulsarReaderFactoryTests.java b/spring-pulsar/src/test/java/org/springframework/pulsar/core/DefaultPulsarReaderFactoryTests.java index cb624ca2..09f41211 100644 --- a/spring-pulsar/src/test/java/org/springframework/pulsar/core/DefaultPulsarReaderFactoryTests.java +++ b/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 message; + try (Reader reader = pulsarReaderFactory.createReader(null, null, Schema.STRING, + Collections.emptyList())) { + PulsarProducerFactory pulsarProducerFactory = new DefaultPulsarProducerFactory<>(pulsarClient, + "basic-pulsar-reader-topic"); + PulsarTemplate 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 message; + try (Reader reader = pulsarReaderFactory.createReader(List.of("basic-pulsar-reader-topic"), + MessageId.earliest, Schema.STRING, Collections.emptyList())) { + + PulsarProducerFactory pulsarProducerFactory = new DefaultPulsarProducerFactory<>(pulsarClient, + "basic-pulsar-reader-topic"); + PulsarTemplate 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 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(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 diff --git a/spring-pulsar/src/test/java/org/springframework/pulsar/reader/DefaultPulsarMessageReaderContainerTests.java b/spring-pulsar/src/test/java/org/springframework/pulsar/reader/DefaultPulsarMessageReaderContainerTests.java index 4505c977..3af04a72 100644 --- a/spring-pulsar/src/test/java/org/springframework/pulsar/reader/DefaultPulsarMessageReaderContainerTests.java +++ b/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 pulsarReaderFactory = new DefaultPulsarReaderFactory<>(pulsarClient, config); + DefaultPulsarReaderFactory 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.emptyMap(); - DefaultPulsarReaderFactory pulsarReaderFactory = new DefaultPulsarReaderFactory<>(pulsarClient, config); + DefaultPulsarReaderFactory 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.emptyMap(); - var readerFactory = new DefaultPulsarReaderFactory(pulsarClient, readerConfig); + var readerFactory = new DefaultPulsarReaderFactory(pulsarClient); DefaultPulsarMessageReaderContainer container = null; try { container = new DefaultPulsarMessageReaderContainer<>(readerFactory, containerProps); diff --git a/spring-pulsar/src/test/java/org/springframework/pulsar/reader/PulsarReaderTests.java b/spring-pulsar/src/test/java/org/springframework/pulsar/reader/PulsarReaderTests.java index 63f55af8..92618e72 100644 --- a/spring-pulsar/src/test/java/org/springframework/pulsar/reader/PulsarReaderTests.java +++ b/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