From 82458872b52024f5464e677bf21763ccf564c0d2 Mon Sep 17 00:00:00 2001 From: Chris Bono Date: Fri, 5 May 2023 19:55:07 -0500 Subject: [PATCH] Use builder to autoconfigure PulsarConsumerFactory (#399) - Move consumer props 'toMap' into PulsarBinderUtils --- .../ConsumerConfigProperties.java | 95 +-- .../PulsarAutoConfiguration.java | 3 +- .../autoconfigure/PulsarProperties.java | 4 - .../PulsarAutoConfigurationTests.java | 740 +++++++++--------- .../autoconfigure/PulsarPropertiesTests.java | 90 +-- .../stream/binder/PulsarBinderUtils.java | 52 ++ .../binder/PulsarMessageChannelBinder.java | 7 +- ...sarBinderConfigurationPropertiesTests.java | 2 +- .../binder/PulsarBinderIntegrationTests.java | 9 +- .../stream/binder/PulsarBinderTests.java | 4 +- .../stream/binder/PulsarBinderUtilsTests.java | 110 +++ .../PulsarExtendedBindingPropertiesTests.java | 6 +- .../core/DefaultPulsarConsumerFactory.java | 45 +- .../pulsar/core/PulsarConsumerFactory.java | 6 - .../core/ConsumerAcknowledgmentTests.java | 74 +- .../DefaultPulsarConsumerFactoryTests.java | 13 +- .../pulsar/core/FailoverConsumerTests.java | 14 +- .../core/SharedSubscriptionConsumerTests.java | 14 +- ...efaultPulsarConsumerErrorHandlerTests.java | 66 +- ...ltPulsarMessageListenerContainerTests.java | 78 +- .../pulsar/listener/PulsarListenerTests.java | 7 +- .../ObservationIntegrationTests.java | 3 +- .../pulsar/observation/ObservationTests.java | 3 +- 23 files changed, 806 insertions(+), 639 deletions(-) diff --git a/spring-pulsar-spring-boot-autoconfigure/src/main/java/org/springframework/pulsar/autoconfigure/ConsumerConfigProperties.java b/spring-pulsar-spring-boot-autoconfigure/src/main/java/org/springframework/pulsar/autoconfigure/ConsumerConfigProperties.java index 2551a972..e7f6bc08 100644 --- a/spring-pulsar-spring-boot-autoconfigure/src/main/java/org/springframework/pulsar/autoconfigure/ConsumerConfigProperties.java +++ b/spring-pulsar-spring-boot-autoconfigure/src/main/java/org/springframework/pulsar/autoconfigure/ConsumerConfigProperties.java @@ -17,11 +17,13 @@ package org.springframework.pulsar.autoconfigure; import java.time.Duration; +import java.util.ArrayList; import java.util.HashMap; import java.util.Map; import java.util.Set; import java.util.SortedMap; import java.util.TreeMap; +import java.util.concurrent.TimeUnit; import java.util.regex.Pattern; import org.apache.pulsar.client.api.ConsumerCryptoFailureAction; @@ -35,7 +37,7 @@ import org.apache.pulsar.client.api.SubscriptionType; import org.springframework.boot.context.properties.NestedConfigurationProperty; import org.springframework.boot.context.properties.PropertyMapper; import org.springframework.lang.Nullable; -import org.springframework.pulsar.autoconfigure.PulsarProperties.Properties; +import org.springframework.pulsar.core.ConsumerBuilderCustomizer; /** * Configuration properties used to specify Pulsar consumers. @@ -483,51 +485,52 @@ public class ConsumerConfigProperties { this.expireTimeOfIncompleteChunkedMessage = expireTimeOfIncompleteChunkedMessage; } - public Map buildProperties() { - PulsarProperties.Properties properties = new Properties(); - - PropertyMapper map = PropertyMapper.get().alwaysApplyingWhenNonNull(); - - map.from(this::getTopics).to(properties.in("topicNames")); - map.from(this::getTopicsPattern).to(properties.in("topicsPattern")); - map.from(this::getSubscriptionName).to(properties.in("subscriptionName")); - map.from(this::getSubscriptionType).to(properties.in("subscriptionType")); - map.from(this::getSubscriptionProperties).to(properties.in("subscriptionProperties")); - map.from(this::getSubscriptionMode).to(properties.in("subscriptionMode")); - map.from(this::getReceiverQueueSize).to(properties.in("receiverQueueSize")); - map.from(this::getAcknowledgementsGroupTime).as(it -> it.toNanos() / 1000) - .to(properties.in("acknowledgementsGroupTimeMicros")); - map.from(this::getNegativeAckRedeliveryDelay).as(it -> it.toNanos() / 1000) - .to(properties.in("negativeAckRedeliveryDelayMicros")); - map.from(this::getMaxTotalReceiverQueueSizeAcrossPartitions) - .to(properties.in("maxTotalReceiverQueueSizeAcrossPartitions")); - map.from(this::getConsumerName).to(properties.in("consumerName")); - map.from(this::getAckTimeout).as(Duration::toMillis).to(properties.in("ackTimeoutMillis")); - map.from(this::getTickDuration).as(Duration::toMillis).to(properties.in("tickDurationMillis")); - map.from(this::getPriorityLevel).to(properties.in("priorityLevel")); - map.from(this::getCryptoFailureAction).to(properties.in("cryptoFailureAction")); - map.from(this::getProperties).to(properties.in("properties")); - map.from(this::getReadCompacted).to(properties.in("readCompacted")); - map.from(this::getSubscriptionInitialPosition).to(properties.in("subscriptionInitialPosition")); - map.from(this::getPatternAutoDiscoveryPeriod).to(properties.in("patternAutoDiscoveryPeriod")); - map.from(this::getRegexSubscriptionMode).to(properties.in("regexSubscriptionMode")); - map.from(this::getDeadLetterPolicy).to(properties.in("deadLetterPolicy")); - map.from(this::getRetryEnable).to(properties.in("retryEnable")); - map.from(this::getAutoUpdatePartitions).to(properties.in("autoUpdatePartitions")); - map.from(this::getAutoUpdatePartitionsInterval).as(Duration::toSeconds) - .to(properties.in("autoUpdatePartitionsIntervalSeconds")); - map.from(this::getReplicateSubscriptionState).to(properties.in("replicateSubscriptionState")); - map.from(this::getResetIncludeHead).to(properties.in("resetIncludeHead")); - map.from(this::getBatchIndexAckEnabled).to(properties.in("batchIndexAckEnabled")); - map.from(this::getAckReceiptEnabled).to(properties.in("ackReceiptEnabled")); - map.from(this::getPoolMessages).to(properties.in("poolMessages")); - map.from(this::getStartPaused).to(properties.in("startPaused")); - map.from(this::getAutoAckOldestChunkedMessageOnQueueFull) - .to(properties.in("autoAckOldestChunkedMessageOnQueueFull")); - map.from(this::getMaxPendingChunkedMessage).to(properties.in("maxPendingChunkedMessage")); - map.from(this::getExpireTimeOfIncompleteChunkedMessage).as(Duration::toMillis) - .to(properties.in("expireTimeOfIncompleteChunkedMessageMillis")); - return properties; + @SuppressWarnings("deprecation") + public ConsumerBuilderCustomizer toConsumerBuilderCustomizer() { + return (consumerBuilder) -> { + PropertyMapper map = PropertyMapper.get().alwaysApplyingWhenNonNull(); + map.from(this::getTopics).as(ArrayList::new).to(consumerBuilder::topics); + map.from(this::getTopicsPattern).to(consumerBuilder::topicsPattern); + map.from(this::getSubscriptionName).to(consumerBuilder::subscriptionName); + map.from(this::getSubscriptionType).to(consumerBuilder::subscriptionType); + map.from(this::getSubscriptionProperties).to(consumerBuilder::subscriptionProperties); + map.from(this::getSubscriptionMode).to(consumerBuilder::subscriptionMode); + map.from(this::getReceiverQueueSize).to(consumerBuilder::receiverQueueSize); + map.from(this::getAcknowledgementsGroupTime).as(Duration::toMillis).to(consumerBuilder, + (cb, val) -> cb.acknowledgmentGroupTime(val, TimeUnit.MILLISECONDS)); + map.from(this::getNegativeAckRedeliveryDelay).as(Duration::toMillis).to(consumerBuilder, + (cb, val) -> cb.negativeAckRedeliveryDelay(val, TimeUnit.MILLISECONDS)); + map.from(this::getMaxTotalReceiverQueueSizeAcrossPartitions) + .to(consumerBuilder::maxTotalReceiverQueueSizeAcrossPartitions); + map.from(this::getConsumerName).to(consumerBuilder::consumerName); + map.from(this::getAckTimeout).as(Duration::toMillis).to(consumerBuilder, + (cb, val) -> cb.ackTimeout(val, TimeUnit.MILLISECONDS)); + map.from(this::getTickDuration).as(Duration::toMillis).to(consumerBuilder, + (cb, val) -> cb.ackTimeoutTickTime(val, TimeUnit.MILLISECONDS)); + map.from(this::getPriorityLevel).to(consumerBuilder::priorityLevel); + map.from(this::getCryptoFailureAction).to(consumerBuilder::cryptoFailureAction); + map.from(this::getProperties).to(consumerBuilder::properties); + map.from(this::getReadCompacted).to(consumerBuilder::readCompacted); + map.from(this::getSubscriptionInitialPosition).to(consumerBuilder::subscriptionInitialPosition); + map.from(this::getPatternAutoDiscoveryPeriod).to(consumerBuilder::patternAutoDiscoveryPeriod); + map.from(this::getRegexSubscriptionMode).to(consumerBuilder::subscriptionTopicsMode); + map.from(this::getDeadLetterPolicy).to(consumerBuilder::deadLetterPolicy); + map.from(this::getRetryEnable).to(consumerBuilder::enableRetry); + map.from(this::getAutoUpdatePartitions).to(consumerBuilder::autoUpdatePartitions); + map.from(this::getAutoUpdatePartitionsInterval).asInt(Duration::toMillis).to(consumerBuilder, + (cb, val) -> cb.autoUpdatePartitionsInterval(val, TimeUnit.MILLISECONDS)); + map.from(this::getReplicateSubscriptionState).to(consumerBuilder::replicateSubscriptionState); + map.from(this::getResetIncludeHead).whenTrue().to((b) -> consumerBuilder.startMessageIdInclusive()); + map.from(this::getBatchIndexAckEnabled).to(consumerBuilder::enableBatchIndexAcknowledgment); + map.from(this::getAckReceiptEnabled).to(consumerBuilder::isAckReceiptEnabled); + map.from(this::getPoolMessages).to(consumerBuilder::poolMessages); + map.from(this::getStartPaused).to(consumerBuilder::startPaused); + map.from(this::getAutoAckOldestChunkedMessageOnQueueFull) + .to(consumerBuilder::autoAckOldestChunkedMessageOnQueueFull); + map.from(this::getMaxPendingChunkedMessage).to(consumerBuilder::maxPendingChunkedMessage); + map.from(this::getExpireTimeOfIncompleteChunkedMessage).as(Duration::toMillis).to(consumerBuilder, + (cb, val) -> cb.expireTimeOfIncompleteChunkedMessage(val, TimeUnit.MILLISECONDS)); + }; } } 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 7db51909..00afebdd 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 @@ -150,7 +150,8 @@ public class PulsarAutoConfiguration { @Bean @ConditionalOnMissingBean public PulsarConsumerFactory pulsarConsumerFactory(PulsarClient pulsarClient) { - return new DefaultPulsarConsumerFactory<>(pulsarClient, this.properties.buildConsumerProperties()); + return new DefaultPulsarConsumerFactory<>(pulsarClient, + this.properties.getConsumer().toConsumerBuilderCustomizer()); } @Bean 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 c4dbc82a..5fdb9dc4 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 @@ -106,10 +106,6 @@ public class PulsarProperties { return this.defaults; } - public Map buildConsumerProperties() { - return new HashMap<>(this.consumer.buildProperties()); - } - public Map buildAdminProperties() { return new HashMap<>(this.admin.buildProperties()); } 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 0e9dae9e..6c9add11 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 @@ -31,7 +31,6 @@ import java.util.concurrent.TimeUnit; import org.apache.pulsar.client.api.ClientBuilder; import org.apache.pulsar.client.api.PulsarClient; import org.apache.pulsar.client.api.Schema; -import org.apache.pulsar.client.api.SubscriptionInitialPosition; import org.apache.pulsar.client.api.SubscriptionType; import org.apache.pulsar.client.api.interceptor.ProducerInterceptor; import org.apache.pulsar.common.schema.KeyValueEncodingType; @@ -56,6 +55,8 @@ import org.springframework.pulsar.config.ConcurrentPulsarListenerContainerFactor import org.springframework.pulsar.config.PulsarListenerContainerFactory; import org.springframework.pulsar.config.PulsarListenerEndpointRegistry; import org.springframework.pulsar.core.CachingPulsarProducerFactory; +import org.springframework.pulsar.core.ConsumerBuilderCustomizer; +import org.springframework.pulsar.core.DefaultPulsarConsumerFactory; import org.springframework.pulsar.core.DefaultPulsarProducerFactory; import org.springframework.pulsar.core.DefaultPulsarReaderFactory; import org.springframework.pulsar.core.DefaultSchemaResolver; @@ -122,180 +123,406 @@ class PulsarAutoConfigurationTests { .hasSingleBean(DefaultSchemaResolver.class).hasSingleBean(DefaultTopicResolver.class)); } - @Test - void customPulsarClientBuilderConfigurerIsRespected() { - var customConfigurer = new PulsarClientBuilderConfigurer(new PulsarProperties(), Collections.emptyList()); - this.contextRunner - .withBean("customPulsarClientConfigurer", PulsarClientBuilderConfigurer.class, () -> customConfigurer) - .run((context) -> assertThat(context).getBean(PulsarClientBuilderConfigurer.class) - .isSameAs(customConfigurer)); - } + @Nested + class ProducerFactoryTests { - @Test - void customPulsarClientIsRespected() { - var customClient = mock(PulsarClient.class); - this.contextRunner.withBean("customPulsarClient", PulsarClient.class, () -> customClient) - .run((context) -> assertThat(context).getBean(PulsarClient.class).isSameAs(customClient)); - } - - @Test - void customSchemaResolverIsRespected() { - SchemaResolver customSchemaResolver = mock(SchemaResolver.class); - this.contextRunner.withBean("customSchemaResolver", SchemaResolver.class, () -> customSchemaResolver) - .run((context) -> assertThat(context).hasNotFailed().getBean(SchemaResolver.class) - .isSameAs(customSchemaResolver)); - } - - @Test - void defaultSchemaResolverCanBeCustomized() { - record Foo() { + @Test + void customPulsarProducerFactoryIsRespected() { + PulsarProducerFactory producerFactory = mock(PulsarProducerFactory.class); + contextRunner.withBean("customPulsarProducerFactory", PulsarProducerFactory.class, () -> producerFactory) + .run((context) -> assertThat(context).hasNotFailed().getBean(PulsarProducerFactory.class) + .isSameAs(producerFactory)); } - SchemaResolverCustomizer customizer = (sr) -> sr.addCustomSchemaMapping(Foo.class, - Schema.STRING); - this.contextRunner.withBean("schemaResolverCustomizer", SchemaResolverCustomizer.class, () -> customizer) - .run((context) -> assertThat(context).hasNotFailed().getBean(DefaultSchemaResolver.class) - .extracting(DefaultSchemaResolver::getCustomSchemaMappings, InstanceOfAssertFactories.MAP) - .containsEntry(Foo.class, Schema.STRING)); + + @Test + void cachingProducerFactoryEnabledByDefault() { + contextRunner.run((context) -> assertHasProducerFactoryOfType(CachingPulsarProducerFactory.class, context)); + } + + @Test + void nonCachingProducerFactoryCanBeEnabled() { + contextRunner.withPropertyValues("spring.pulsar.producer.cache.enabled=false") + .run((context -> assertHasProducerFactoryOfType(DefaultPulsarProducerFactory.class, context))); + } + + @Test + void cachingProducerFactoryCanBeEnabled() { + contextRunner.withPropertyValues("spring.pulsar.producer.cache.enabled=true") + .run((context -> assertHasProducerFactoryOfType(CachingPulsarProducerFactory.class, context))); + } + + @Test + void cachingEnabledAndCaffeineNotOnClasspath() { + contextRunner.withClassLoader(new FilteredClassLoader(Caffeine.class)) + .withPropertyValues("spring.pulsar.producer.cache.enabled=true") + .run((context -> assertHasProducerFactoryOfType(CachingPulsarProducerFactory.class, context))); + } + + @Test + void cachingProducerFactoryCanBeConfigured() { + contextRunner + .withPropertyValues("spring.pulsar.producer.cache.expire-after-access=100s", + "spring.pulsar.producer.cache.maximum-size=5150", + "spring.pulsar.producer.cache.initial-capacity=200") + .run((context -> assertThat(context).hasNotFailed().getBean(PulsarProducerFactory.class) + .extracting("producerCache.cache.cache").hasFieldOrPropertyWithValue("maximum", 5150L) + .hasFieldOrPropertyWithValue("expiresAfterAccessNanos", TimeUnit.SECONDS.toNanos(100)))); + } + + @Test + void beansAreInjectedInNonCachingProducerFactory() { + contextRunner.withUserConfiguration(SpyCustomizersConfig.class) + .withPropertyValues("spring.pulsar.producer.topic-name=foo-topic", + "spring.pulsar.producer.cache.enabled=false") + .run((context) -> assertThat(context).getBean(DefaultPulsarProducerFactory.class) + .hasFieldOrPropertyWithValue("defaultTopic", "foo-topic") + .hasFieldOrPropertyWithValue("defaultConfigCustomizer", + SpyCustomizersConfig.testProducerCustomizer) + .hasFieldOrPropertyWithValue("pulsarClient", context.getBean(PulsarClient.class)) + .hasFieldOrPropertyWithValue("topicResolver", context.getBean(TopicResolver.class))); + } + + @Test + void beansAreInjectedInCachingProducerFactory() { + contextRunner.withUserConfiguration(SpyCustomizersConfig.class) + .withPropertyValues("spring.pulsar.producer.topic-name=foo-topic", + "spring.pulsar.producer.cache.enabled=true") + .run((context) -> assertThat(context).getBean(CachingPulsarProducerFactory.class) + .hasFieldOrPropertyWithValue("defaultTopic", "foo-topic") + .hasFieldOrPropertyWithValue("defaultConfigCustomizer", + SpyCustomizersConfig.testProducerCustomizer) + .hasFieldOrPropertyWithValue("pulsarClient", context.getBean(PulsarClient.class)) + .hasFieldOrPropertyWithValue("topicResolver", context.getBean(TopicResolver.class))); + } + + private void assertHasProducerFactoryOfType(Class producerFactoryType, + AssertableApplicationContext context) { + assertThat(context).hasNotFailed().hasSingleBean(PulsarProducerFactory.class) + .getBean(PulsarProducerFactory.class).isExactlyInstanceOf(producerFactoryType); + } + } - @Test - void customTopicResolverIsRespected() { - TopicResolver customTopicResolver = mock(TopicResolver.class); - this.contextRunner.withBean("customTopicResolver", TopicResolver.class, () -> customTopicResolver) - .run((context) -> assertThat(context).hasNotFailed().getBean(TopicResolver.class) - .isSameAs(customTopicResolver)); + @Nested + class ConsumerFactoryTests { + + @Test + void customPulsarConsumerFactoryIsRespected() { + PulsarConsumerFactory consumerFactory = mock(PulsarConsumerFactory.class); + contextRunner.withBean("customPulsarConsumerFactory", PulsarConsumerFactory.class, () -> consumerFactory) + .run((context) -> assertThat(context).getBean(PulsarConsumerFactory.class) + .isSameAs(consumerFactory)); + } + + @Test + void beansAreInjectedInConsumerFactory() { + contextRunner.withUserConfiguration(SpyCustomizersConfig.class) + .run((context) -> assertThat(context).getBean(DefaultPulsarConsumerFactory.class) + .hasFieldOrPropertyWithValue("pulsarClient", context.getBean(PulsarClient.class)) + .hasFieldOrPropertyWithValue("defaultConfigCustomizer", + SpyCustomizersConfig.testConsumerCustomizer)); + } + } - @Test - void customPulsarProducerFactoryIsRespected() { - PulsarProducerFactory producerFactory = mock(PulsarProducerFactory.class); - this.contextRunner.withBean("customPulsarProducerFactory", PulsarProducerFactory.class, () -> producerFactory) - .run((context) -> assertThat(context).hasNotFailed().getBean(PulsarProducerFactory.class) - .isSameAs(producerFactory)); + /* + * 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. + */ + @TestConfiguration(proxyBeanMethods = false) + static class SpyCustomizersConfig { + + @SuppressWarnings("rawtypes") + static ProducerBuilderCustomizer testProducerCustomizer = (producerBuilder) -> { + }; + + @SuppressWarnings("rawtypes") + static ConsumerBuilderCustomizer testConsumerCustomizer = (consumerBuilder) -> { + }; + + @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 spyPulsarProps = spy(pulsarProps); + when(spyPulsarProps.getProducer()).thenReturn(producerProps); + when(spyPulsarProps.getConsumer()).thenReturn(consumerProps); + return spyPulsarProps; + } + } - @Test - void customPulsarTemplateIsRespected() { - PulsarTemplate template = mock(PulsarTemplate.class); - this.contextRunner.withBean("customPulsarTemplate", PulsarTemplate.class, () -> template) - .run((context) -> assertThat(context).hasNotFailed().getBean(PulsarTemplate.class).isSameAs(template)); + @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))); + } + } - @Test - void beansAreInjectedInPulsarTemplate() { - PulsarProducerFactory producerFactory = mock(PulsarProducerFactory.class); - SchemaResolver schemaResolver = mock(SchemaResolver.class); - TopicResolver topicResolver = mock(TopicResolver.class); - this.contextRunner.withBean("customPulsarProducerFactory", PulsarProducerFactory.class, () -> producerFactory) - .withBean("schemaResolver", SchemaResolver.class, () -> schemaResolver) - .withBean("topicResolver", TopicResolver.class, () -> topicResolver) - .run((context -> assertThat(context).hasNotFailed().getBean(PulsarTemplate.class) - .hasFieldOrPropertyWithValue("producerFactory", producerFactory) - .hasFieldOrPropertyWithValue("schemaResolver", schemaResolver) - .hasFieldOrPropertyWithValue("topicResolver", topicResolver))); + @Nested + class SchemaAndTopicResolversTests { + + @Test + void customSchemaResolverIsRespected() { + SchemaResolver customSchemaResolver = mock(SchemaResolver.class); + contextRunner.withBean("customSchemaResolver", SchemaResolver.class, () -> customSchemaResolver) + .run((context) -> assertThat(context).hasNotFailed().getBean(SchemaResolver.class) + .isSameAs(customSchemaResolver)); + } + + @Test + void defaultSchemaResolverCanBeCustomized() { + record Foo() { + } + SchemaResolverCustomizer customizer = (sr) -> sr.addCustomSchemaMapping(Foo.class, + Schema.STRING); + contextRunner.withBean("schemaResolverCustomizer", SchemaResolverCustomizer.class, () -> customizer) + .run((context) -> assertThat(context).hasNotFailed().getBean(DefaultSchemaResolver.class) + .extracting(DefaultSchemaResolver::getCustomSchemaMappings, InstanceOfAssertFactories.MAP) + .containsEntry(Foo.class, Schema.STRING)); + } + + @Test + void customTopicResolverIsRespected() { + TopicResolver customTopicResolver = mock(TopicResolver.class); + contextRunner.withBean("customTopicResolver", TopicResolver.class, () -> customTopicResolver) + .run((context) -> assertThat(context).hasNotFailed().getBean(TopicResolver.class) + .isSameAs(customTopicResolver)); + } + } - @Test - void customPulsarConsumerFactoryIsRespected() { - PulsarConsumerFactory consumerFactory = mock(PulsarConsumerFactory.class); - this.contextRunner.withBean("customPulsarConsumerFactory", PulsarConsumerFactory.class, () -> consumerFactory) - .run((context) -> assertThat(context).hasNotFailed().getBean(PulsarConsumerFactory.class) - .isSameAs(consumerFactory)); + @Nested + class PulsarTemplateTests { + + @Test + void customPulsarTemplateIsRespected() { + PulsarTemplate template = mock(PulsarTemplate.class); + contextRunner.withBean("customPulsarTemplate", PulsarTemplate.class, () -> template).run( + (context) -> assertThat(context).hasNotFailed().getBean(PulsarTemplate.class).isSameAs(template)); + } + + @Test + void beansAreInjectedInPulsarTemplate() { + PulsarProducerFactory producerFactory = mock(PulsarProducerFactory.class); + SchemaResolver schemaResolver = mock(SchemaResolver.class); + TopicResolver topicResolver = mock(TopicResolver.class); + contextRunner.withBean("customPulsarProducerFactory", PulsarProducerFactory.class, () -> producerFactory) + .withBean("schemaResolver", SchemaResolver.class, () -> schemaResolver) + .withBean("topicResolver", TopicResolver.class, () -> topicResolver) + .run((context -> assertThat(context).hasNotFailed().getBean(PulsarTemplate.class) + .hasFieldOrPropertyWithValue("producerFactory", producerFactory) + .hasFieldOrPropertyWithValue("schemaResolver", schemaResolver) + .hasFieldOrPropertyWithValue("topicResolver", topicResolver))); + } + + @Test + void customProducerInterceptorIsUsedInPulsarTemplate() { + ProducerInterceptor interceptor = mock(ProducerInterceptor.class); + contextRunner.withBean("customProducerInterceptor", ProducerInterceptor.class, () -> interceptor) + .run((context -> assertThat(context).hasNotFailed().getBean(PulsarTemplate.class) + .extracting("interceptors") + .asInstanceOf(InstanceOfAssertFactories.list(ProducerInterceptor.class)) + .contains(interceptor))); + } + + @Test + void customProducerInterceptorsOrderedProperly() { + contextRunner.withUserConfiguration(InterceptorTestConfiguration.class) + .run((context -> assertThat(context).hasNotFailed().getBean(PulsarTemplate.class) + .extracting("interceptors") + .asInstanceOf(InstanceOfAssertFactories.list(ProducerInterceptor.class)) + .containsExactly(InterceptorTestConfiguration.interceptorBar, + InterceptorTestConfiguration.interceptorFoo))); + } + + @Configuration(proxyBeanMethods = false) + static class InterceptorTestConfiguration { + + static ProducerInterceptor interceptorFoo = mock(ProducerInterceptor.class); + static ProducerInterceptor interceptorBar = mock(ProducerInterceptor.class); + + @Bean + @Order(200) + ProducerInterceptor interceptorFoo() { + return interceptorFoo; + } + + @Bean + @Order(100) + ProducerInterceptor interceptorBar() { + return interceptorBar; + } + + } + } - @Test - void pulsarConsumerFactoryWithEnumPropertyValue() { - this.contextRunner.withPropertyValues("spring.pulsar.consumer.subscription-initial-position=earliest") - .run((context -> assertThat(context).hasNotFailed().getBean(PulsarConsumerFactory.class) - .extracting("consumerConfig").hasFieldOrPropertyWithValue("subscriptionInitialPosition", - SubscriptionInitialPosition.Earliest))); - } + @Nested + class PulsarListenerTests { - @Test - void customPulsarListenerContainerFactoryIsRespected() { - PulsarListenerContainerFactory listenerContainerFactory = mock(PulsarListenerContainerFactory.class); - this.contextRunner - .withBean("pulsarListenerContainerFactory", PulsarListenerContainerFactory.class, - () -> listenerContainerFactory) - .run((context) -> assertThat(context).hasNotFailed().getBean(PulsarListenerContainerFactory.class) - .isSameAs(listenerContainerFactory)); - } + @Test + void customPulsarListenerContainerFactoryIsRespected() { + PulsarListenerContainerFactory listenerContainerFactory = mock(PulsarListenerContainerFactory.class); + contextRunner + .withBean("pulsarListenerContainerFactory", PulsarListenerContainerFactory.class, + () -> listenerContainerFactory) + .run((context) -> assertThat(context).hasNotFailed().getBean(PulsarListenerContainerFactory.class) + .isSameAs(listenerContainerFactory)); + } - @Test - void beansAreInjectedInPulsarListenerContainerFactory() { - PulsarConsumerFactory consumerFactory = mock(PulsarConsumerFactory.class); - SchemaResolver schemaResolver = mock(SchemaResolver.class); - TopicResolver topicResolver = mock(TopicResolver.class); - this.contextRunner.withBean("pulsarConsumerFactory", PulsarConsumerFactory.class, () -> consumerFactory) - .withBean("schemaResolver", SchemaResolver.class, () -> schemaResolver) - .withBean("topicResolver", TopicResolver.class, () -> topicResolver) - .run((context -> assertThat(context).hasNotFailed() - .getBean(ConcurrentPulsarListenerContainerFactory.class) - .hasFieldOrPropertyWithValue("consumerFactory", consumerFactory) - .extracting(ConcurrentPulsarListenerContainerFactory::getContainerProperties) - .hasFieldOrPropertyWithValue("schemaResolver", schemaResolver) - .hasFieldOrPropertyWithValue("topicResolver", topicResolver))); - } - - @Test - void customPulsarListenerAnnotationBeanPostProcessorIsRespected() { - PulsarListenerAnnotationBeanPostProcessor listenerAnnotationBeanPostProcessor = mock( - PulsarListenerAnnotationBeanPostProcessor.class); - this.contextRunner - .withBean("org.springframework.pulsar.config.internalPulsarListenerAnnotationProcessor", - PulsarListenerAnnotationBeanPostProcessor.class, () -> listenerAnnotationBeanPostProcessor) - .run((context) -> assertThat(context).hasNotFailed() - .getBean(PulsarListenerAnnotationBeanPostProcessor.class) - .isSameAs(listenerAnnotationBeanPostProcessor)); - } - - @Test - void customPulsarAdministrationIsRespected() { - PulsarAdministration pulsarAdministration = mock(PulsarAdministration.class); - this.contextRunner - .withBean("customPulsarAdministration", PulsarAdministration.class, () -> pulsarAdministration) - .run((context) -> assertThat(context).hasNotFailed().getBean(PulsarAdministration.class) - .isSameAs(pulsarAdministration)); - } - - @Test - void customProducerInterceptorIsUsedInPulsarTemplate() { - ProducerInterceptor interceptor = mock(ProducerInterceptor.class); - this.contextRunner.withBean("customProducerInterceptor", ProducerInterceptor.class, () -> interceptor) - .run((context -> assertThat(context).hasNotFailed().getBean(PulsarTemplate.class) - .extracting("interceptors") - .asInstanceOf(InstanceOfAssertFactories.list(ProducerInterceptor.class)) - .contains(interceptor))); - } - - @Test - void customProducerInterceptorsOrderedProperly() { - this.contextRunner.withUserConfiguration(InterceptorTestConfiguration.class) - .run((context -> assertThat(context).hasNotFailed().getBean(PulsarTemplate.class) - .extracting("interceptors") - .asInstanceOf(InstanceOfAssertFactories.list(ProducerInterceptor.class)) - .containsExactly(InterceptorTestConfiguration.interceptorBar, - InterceptorTestConfiguration.interceptorFoo))); - } - - @Test - void listenerPropertiesAreHonored() { - contextRunner - .withPropertyValues("spring.pulsar.listener.ack-mode=manual", "spring.pulsar.listener.schema-type=avro", - "spring.pulsar.listener.max-num-messages=10", "spring.pulsar.listener.max-num-bytes=101B", - "spring.pulsar.listener.batch-timeout=50ms", "spring.pulsar.consumer.subscription-type=shared") - .run((context -> { - AbstractObjectAssert properties = assertThat(context).hasNotFailed() + @Test + void beansAreInjectedInPulsarListenerContainerFactory() { + PulsarConsumerFactory consumerFactory = mock(PulsarConsumerFactory.class); + SchemaResolver schemaResolver = mock(SchemaResolver.class); + TopicResolver topicResolver = mock(TopicResolver.class); + contextRunner.withBean("pulsarConsumerFactory", PulsarConsumerFactory.class, () -> consumerFactory) + .withBean("schemaResolver", SchemaResolver.class, () -> schemaResolver) + .withBean("topicResolver", TopicResolver.class, () -> topicResolver) + .run((context -> assertThat(context).hasNotFailed() .getBean(ConcurrentPulsarListenerContainerFactory.class) - .extracting(ConcurrentPulsarListenerContainerFactory::getContainerProperties); - properties.extracting(PulsarContainerProperties::getAckMode).isEqualTo(AckMode.MANUAL); - properties.extracting(PulsarContainerProperties::getSchemaType).isEqualTo(SchemaType.AVRO); - properties.extracting(PulsarContainerProperties::getMaxNumMessages).isEqualTo(10); - properties.extracting(PulsarContainerProperties::getMaxNumBytes).isEqualTo(101); - properties.extracting(PulsarContainerProperties::getBatchTimeoutMillis).isEqualTo(50); - properties.extracting(PulsarContainerProperties::getSubscriptionType) - .isEqualTo(SubscriptionType.Shared); - })); + .hasFieldOrPropertyWithValue("consumerFactory", consumerFactory) + .extracting(ConcurrentPulsarListenerContainerFactory::getContainerProperties) + .hasFieldOrPropertyWithValue("schemaResolver", schemaResolver) + .hasFieldOrPropertyWithValue("topicResolver", topicResolver))); + } + + @Test + void customPulsarListenerAnnotationBeanPostProcessorIsRespected() { + PulsarListenerAnnotationBeanPostProcessor listenerAnnotationBeanPostProcessor = mock( + PulsarListenerAnnotationBeanPostProcessor.class); + contextRunner + .withBean("org.springframework.pulsar.config.internalPulsarListenerAnnotationProcessor", + PulsarListenerAnnotationBeanPostProcessor.class, () -> listenerAnnotationBeanPostProcessor) + .run((context) -> assertThat(context).hasNotFailed() + .getBean(PulsarListenerAnnotationBeanPostProcessor.class) + .isSameAs(listenerAnnotationBeanPostProcessor)); + } + + @Test + void listenerPropertiesAreHonored() { + contextRunner.withPropertyValues("spring.pulsar.listener.ack-mode=manual", + "spring.pulsar.listener.schema-type=avro", "spring.pulsar.listener.max-num-messages=10", + "spring.pulsar.listener.max-num-bytes=101B", "spring.pulsar.listener.batch-timeout=50ms", + "spring.pulsar.consumer.subscription-type=shared").run((context -> { + AbstractObjectAssert properties = assertThat(context) + .hasNotFailed().getBean(ConcurrentPulsarListenerContainerFactory.class) + .extracting(ConcurrentPulsarListenerContainerFactory::getContainerProperties); + properties.extracting(PulsarContainerProperties::getAckMode).isEqualTo(AckMode.MANUAL); + properties.extracting(PulsarContainerProperties::getSchemaType).isEqualTo(SchemaType.AVRO); + properties.extracting(PulsarContainerProperties::getMaxNumMessages).isEqualTo(10); + properties.extracting(PulsarContainerProperties::getMaxNumBytes).isEqualTo(101); + properties.extracting(PulsarContainerProperties::getBatchTimeoutMillis).isEqualTo(50); + properties.extracting(PulsarContainerProperties::getSubscriptionType) + .isEqualTo(SubscriptionType.Shared); + })); + } + + } + + @Nested + class PulsarClientTests { + + @Test + void customPulsarClientIsRespected() { + var customClient = mock(PulsarClient.class); + contextRunner.withBean("customPulsarClient", PulsarClient.class, () -> customClient) + .run((context) -> assertThat(context).getBean(PulsarClient.class).isSameAs(customClient)); + } + + @Test + void customPulsarClientBuilderConfigurerIsRespected() { + var customConfigurer = new PulsarClientBuilderConfigurer(new PulsarProperties(), Collections.emptyList()); + contextRunner + .withBean("customPulsarClientConfigurer", PulsarClientBuilderConfigurer.class, + () -> customConfigurer) + .run((context) -> assertThat(context).getBean(PulsarClientBuilderConfigurer.class) + .isSameAs(customConfigurer)); + } + + @Test + void clientConfigurerWithNoUserDefinedCustomizers() { + contextRunner.run((context) -> assertThat(context).getBean(PulsarClientBuilderConfigurer.class) + .hasFieldOrPropertyWithValue("customizers", Collections.emptyList())); + } + + @Test + void clientConfigurerWithUserDefinedCustomizers() { + contextRunner.withUserConfiguration(ClientCustomizersTestConfiguration.class) + .run((context) -> assertThat(context).getBean(PulsarClientBuilderConfigurer.class) + .extracting("customizers", InstanceOfAssertFactories.LIST) + .containsExactly(ClientCustomizersTestConfiguration.clientCustomizerBar, + ClientCustomizersTestConfiguration.clientCustomizerFoo)); + } + + @Test + void clientConfigurerIsApplied() { + var clientConfigurer = spy( + new PulsarClientBuilderConfigurer(new PulsarProperties(), Collections.emptyList())); + contextRunner.withBean("clientConfigurer", PulsarClientBuilderConfigurer.class, () -> clientConfigurer) + .run((context) -> verify(clientConfigurer).configure(any(ClientBuilder.class))); + } + + @Configuration(proxyBeanMethods = false) + static class ClientCustomizersTestConfiguration { + + static PulsarClientBuilderCustomizer clientCustomizerFoo = mock(PulsarClientBuilderCustomizer.class); + static PulsarClientBuilderCustomizer clientCustomizerBar = mock(PulsarClientBuilderCustomizer.class); + + @Bean + @Order(200) + PulsarClientBuilderCustomizer clientCustomizerFoo() { + return clientCustomizerFoo; + } + + @Bean + @Order(100) + PulsarClientBuilderCustomizer clientCustomizerBar() { + return clientCustomizerBar; + } + + } + + } + + @Nested + class PulsarAdministrationTests { + + @Test + void customPulsarAdministrationIsRespected() { + PulsarAdministration pulsarAdministration = mock(PulsarAdministration.class); + contextRunner.withBean("customPulsarAdministration", PulsarAdministration.class, () -> pulsarAdministration) + .run((context) -> assertThat(context).hasNotFailed().getBean(PulsarAdministration.class) + .isSameAs(pulsarAdministration)); + } + } @Nested @@ -368,35 +595,7 @@ class PulsarAutoConfigurationTests { } @Nested - class ClientAutoConfigurationTests { - - @Test - void clientConfigurerWithNoUserDefinedCustomizers() { - contextRunner.run((context) -> assertThat(context).getBean(PulsarClientBuilderConfigurer.class) - .hasFieldOrPropertyWithValue("customizers", Collections.emptyList())); - } - - @Test - void clientConfigurerWithUserDefinedCustomizers() { - contextRunner.withUserConfiguration(ClientCustomizersTestConfiguration.class) - .run((context) -> assertThat(context).getBean(PulsarClientBuilderConfigurer.class) - .extracting("customizers", InstanceOfAssertFactories.LIST) - .containsExactly(ClientCustomizersTestConfiguration.clientCustomizerBar, - ClientCustomizersTestConfiguration.clientCustomizerFoo)); - } - - @Test - void clientConfigurerIsApplied() { - var clientConfigurer = spy( - new PulsarClientBuilderConfigurer(new PulsarProperties(), Collections.emptyList())); - contextRunner.withBean("clientConfigurer", PulsarClientBuilderConfigurer.class, () -> clientConfigurer) - .run((context) -> verify(clientConfigurer).configure(any(ClientBuilder.class))); - } - - } - - @Nested - class FunctionAutoConfigurationTests { + class FunctionTests { @Test void functionSupportEnabledByDefault() { @@ -437,7 +636,7 @@ class PulsarAutoConfigurationTests { } @Nested - class ObservationAutoConfigurationTests { + class ObservationTests { @Test void templateObservationsEnabledByDefault() { @@ -481,169 +680,4 @@ class PulsarAutoConfigurationTests { } - @Nested - class ProducerFactoryAutoConfigurationTests { - - @Test - void cachingProducerFactoryEnabledByDefault() { - contextRunner.run((context) -> assertHasProducerFactoryOfType(CachingPulsarProducerFactory.class, context)); - } - - @Test - void nonCachingProducerFactoryCanBeEnabled() { - contextRunner.withPropertyValues("spring.pulsar.producer.cache.enabled=false") - .run((context -> assertHasProducerFactoryOfType(DefaultPulsarProducerFactory.class, context))); - } - - @Test - void cachingProducerFactoryCanBeEnabled() { - contextRunner.withPropertyValues("spring.pulsar.producer.cache.enabled=true") - .run((context -> assertHasProducerFactoryOfType(CachingPulsarProducerFactory.class, context))); - } - - @Test - void cachingEnabledAndCaffeineNotOnClasspath() { - contextRunner.withClassLoader(new FilteredClassLoader(Caffeine.class)) - .withPropertyValues("spring.pulsar.producer.cache.enabled=true") - .run((context -> assertHasProducerFactoryOfType(CachingPulsarProducerFactory.class, context))); - } - - @Test - void cachingProducerFactoryCanBeConfigured() { - contextRunner - .withPropertyValues("spring.pulsar.producer.cache.expire-after-access=100s", - "spring.pulsar.producer.cache.maximum-size=5150", - "spring.pulsar.producer.cache.initial-capacity=200") - .run((context -> assertThat(context).hasNotFailed().getBean(PulsarProducerFactory.class) - .extracting("producerCache.cache.cache").hasFieldOrPropertyWithValue("maximum", 5150L) - .hasFieldOrPropertyWithValue("expiresAfterAccessNanos", TimeUnit.SECONDS.toNanos(100)))); - } - - @Test - void beansAreInjectedInNonCachingProducerFactory() { - contextRunner.withUserConfiguration(ProducerCustomizerConfig.class) - .withPropertyValues("spring.pulsar.producer.topic-name=foo-topic", - "spring.pulsar.producer.cache.enabled=false") - .run((context) -> assertThat(context).getBean(DefaultPulsarProducerFactory.class) - .hasFieldOrPropertyWithValue("defaultTopic", "foo-topic") - .hasFieldOrPropertyWithValue("defaultConfigCustomizer", - ProducerCustomizerConfig.testCustomizer) - .hasFieldOrPropertyWithValue("pulsarClient", context.getBean(PulsarClient.class)) - .hasFieldOrPropertyWithValue("topicResolver", context.getBean(TopicResolver.class))); - } - - @Test - void beansAreInjectedInCachingProducerFactory() { - contextRunner.withUserConfiguration(ProducerCustomizerConfig.class) - .withPropertyValues("spring.pulsar.producer.topic-name=foo-topic", - "spring.pulsar.producer.cache.enabled=true") - .run((context) -> assertThat(context).getBean(CachingPulsarProducerFactory.class) - .hasFieldOrPropertyWithValue("defaultTopic", "foo-topic") - .hasFieldOrPropertyWithValue("defaultConfigCustomizer", - ProducerCustomizerConfig.testCustomizer) - .hasFieldOrPropertyWithValue("pulsarClient", context.getBean(PulsarClient.class)) - .hasFieldOrPropertyWithValue("topicResolver", context.getBean(TopicResolver.class))); - } - - private void assertHasProducerFactoryOfType(Class producerFactoryType, - AssertableApplicationContext context) { - assertThat(context).hasNotFailed().hasSingleBean(PulsarProducerFactory.class) - .getBean(PulsarProducerFactory.class).isExactlyInstanceOf(producerFactoryType); - } - - } - - @Nested - class ReaderFactoryAutoConfigurationTests { - - @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))); - } - - } - - @Configuration(proxyBeanMethods = false) - static class InterceptorTestConfiguration { - - static ProducerInterceptor interceptorFoo = mock(ProducerInterceptor.class); - static ProducerInterceptor interceptorBar = mock(ProducerInterceptor.class); - - @Bean - @Order(200) - ProducerInterceptor interceptorFoo() { - return interceptorFoo; - } - - @Bean - @Order(100) - ProducerInterceptor interceptorBar() { - return interceptorBar; - } - - } - - @Configuration(proxyBeanMethods = false) - static class ClientCustomizersTestConfiguration { - - static PulsarClientBuilderCustomizer clientCustomizerFoo = mock(PulsarClientBuilderCustomizer.class); - static PulsarClientBuilderCustomizer clientCustomizerBar = mock(PulsarClientBuilderCustomizer.class); - - @Bean - @Order(200) - PulsarClientBuilderCustomizer clientCustomizerFoo() { - return clientCustomizerFoo; - } - - @Bean - @Order(100) - PulsarClientBuilderCustomizer clientCustomizerBar() { - return clientCustomizerBar; - } - - } - - /* - * Use '@TestConfiguration' and exact name of the PulsarProperties bean that is - * created via the '@EnableConfigurationProperties' on the actual auto-config in order - * to 'replace' the PulsarProperties bean - all so we can make sure the returned - * producer builder customizer is the one we expect. - */ - @TestConfiguration(proxyBeanMethods = false) - static class ProducerCustomizerConfig { - - @SuppressWarnings("rawtypes") - static ProducerBuilderCustomizer testCustomizer = (producerBuilder) -> { - }; - - @Bean(name = "spring.pulsar-org.springframework.pulsar.autoconfigure.PulsarProperties") - PulsarProperties pulsarProperties() { - var pulsarProps = new PulsarProperties(); - var producerProps = spy(pulsarProps.getProducer()); - when(producerProps.toProducerBuilderCustomizer()).thenReturn(testCustomizer); - var spyPulsarProps = spy(pulsarProps); - when(spyPulsarProps.getProducer()).thenReturn(producerProps); - return spyPulsarProps; - } - - } - } 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 3315fd90..c22d9d78 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 @@ -30,7 +30,6 @@ import java.util.Map; import org.apache.pulsar.client.admin.PulsarAdmin; import org.apache.pulsar.client.api.CompressionType; import org.apache.pulsar.client.api.ConsumerCryptoFailureAction; -import org.apache.pulsar.client.api.DeadLetterPolicy; import org.apache.pulsar.client.api.HashingScheme; import org.apache.pulsar.client.api.MessageRoutingMode; import org.apache.pulsar.client.api.ProducerAccessMode; @@ -41,7 +40,6 @@ 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.ConsumerConfigurationData; import org.apache.pulsar.client.impl.conf.ReaderConfigurationData; import org.apache.pulsar.common.schema.SchemaType; import org.assertj.core.api.InstanceOfAssertFactories; @@ -445,7 +443,7 @@ public class PulsarPropertiesTests { @Test void consumerProperties() { - Map props = new HashMap<>(); + var props = new HashMap(); props.put("spring.pulsar.consumer.topics[0]", "my-topic"); props.put("spring.pulsar.consumer.topics-pattern", "my-pattern"); props.put("spring.pulsar.consumer.subscription-name", "my-subscription"); @@ -484,53 +482,47 @@ public class PulsarPropertiesTests { props.put("spring.pulsar.consumer.expire-time-of-incomplete-chunked-message", "12s"); bind(props); - Map consumerProps = properties.buildConsumerProperties(); - // Verify that the props can be loaded in a ConsumerBuilder - assertThatNoException().isThrownBy(() -> ConfigurationDataUtils.loadData(consumerProps, - new ConsumerConfigurationData<>(), ConsumerConfigurationData.class)); - - assertThat(consumerProps) - .hasEntrySatisfying("topicNames", - topics -> assertThat(topics) - .asInstanceOf(InstanceOfAssertFactories.collection(String.class)) - .containsExactly("my-topic")) - .hasEntrySatisfying("topicsPattern", p -> assertThat(p.toString()).isEqualTo("my-pattern")) - .containsEntry("subscriptionName", "my-subscription") - .containsEntry("subscriptionType", SubscriptionType.Shared) - .hasEntrySatisfying("subscriptionProperties", - properties -> assertThat(properties) - .asInstanceOf(InstanceOfAssertFactories.map(String.class, String.class)) - .containsEntry("my-sub-prop", "my-sub-prop-value")) - .containsEntry("subscriptionMode", SubscriptionMode.NonDurable) - .containsEntry("receiverQueueSize", 1).containsEntry("acknowledgementsGroupTimeMicros", 2_000_000L) - .containsEntry("negativeAckRedeliveryDelayMicros", 3_000_000L) - .containsEntry("maxTotalReceiverQueueSizeAcrossPartitions", 5) - .containsEntry("consumerName", "my-consumer").containsEntry("ackTimeoutMillis", 6_000L) - .containsEntry("tickDurationMillis", 7_000L).containsEntry("priorityLevel", 8) - .containsEntry("cryptoFailureAction", ConsumerCryptoFailureAction.DISCARD) - .hasEntrySatisfying("properties", - properties -> assertThat(properties) - .asInstanceOf(InstanceOfAssertFactories.map(String.class, String.class)) - .containsEntry("my-prop", "my-prop-value")) - .containsEntry("readCompacted", true) - .containsEntry("subscriptionInitialPosition", SubscriptionInitialPosition.Earliest) - .containsEntry("patternAutoDiscoveryPeriod", 9) - .containsEntry("regexSubscriptionMode", RegexSubscriptionMode.AllTopics) - .hasEntrySatisfying("deadLetterPolicy", dlp -> { - DeadLetterPolicy deadLetterPolicy = (DeadLetterPolicy) dlp; - assertThat(deadLetterPolicy.getMaxRedeliverCount()).isEqualTo(4); - assertThat(deadLetterPolicy.getRetryLetterTopic()).isEqualTo("my-retry-topic"); - assertThat(deadLetterPolicy.getDeadLetterTopic()).isEqualTo("my-dlt-topic"); - assertThat(deadLetterPolicy.getInitialSubscriptionName()).isEqualTo("my-initial-subscription"); - }).containsEntry("retryEnable", true).containsEntry("autoUpdatePartitions", false) - .containsEntry("autoUpdatePartitionsIntervalSeconds", 10L) - .containsEntry("replicateSubscriptionState", true).containsEntry("resetIncludeHead", true) - .containsEntry("batchIndexAckEnabled", true).containsEntry("ackReceiptEnabled", true) - .containsEntry("poolMessages", true).containsEntry("startPaused", true) - .containsEntry("autoAckOldestChunkedMessageOnQueueFull", false) - .containsEntry("maxPendingChunkedMessage", 11) - .containsEntry("expireTimeOfIncompleteChunkedMessageMillis", 12_000L); + var consumerProps = properties.getConsumer(); + assertThat(consumerProps.getTopics()).containsExactly("my-topic"); + assertThat(consumerProps.getTopicsPattern().toString()).isEqualTo("my-pattern"); + assertThat(consumerProps.getSubscriptionName()).isEqualTo("my-subscription"); + assertThat(consumerProps.getSubscriptionType()).isEqualTo(SubscriptionType.Shared); + assertThat(consumerProps.getSubscriptionProperties()) + .containsExactly(entry("my-sub-prop", "my-sub-prop-value")); + assertThat(consumerProps.getSubscriptionMode()).isEqualTo(SubscriptionMode.NonDurable); + assertThat(consumerProps.getReceiverQueueSize()).isEqualTo(1); + assertThat(consumerProps.getAcknowledgementsGroupTime()).isEqualTo(Duration.ofMillis(2_000)); + assertThat(consumerProps.getNegativeAckRedeliveryDelay()).isEqualTo(Duration.ofMillis(3_000)); + assertThat(consumerProps.getMaxTotalReceiverQueueSizeAcrossPartitions()).isEqualTo(5); + assertThat(consumerProps.getConsumerName()).isEqualTo("my-consumer"); + assertThat(consumerProps.getAckTimeout()).isEqualTo(Duration.ofMillis(6_000)); + assertThat(consumerProps.getTickDuration()).isEqualTo(Duration.ofMillis(7_000)); + assertThat(consumerProps.getPriorityLevel()).isEqualTo(8); + assertThat(consumerProps.getCryptoFailureAction()).isEqualTo(ConsumerCryptoFailureAction.DISCARD); + assertThat(consumerProps.getProperties()).containsExactly(entry("my-prop", "my-prop-value")); + assertThat(consumerProps.getReadCompacted()).isTrue(); + assertThat(consumerProps.getSubscriptionInitialPosition()).isEqualTo(SubscriptionInitialPosition.Earliest); + assertThat(consumerProps.getPatternAutoDiscoveryPeriod()).isEqualTo(9); + assertThat(consumerProps.getRegexSubscriptionMode()).isEqualTo(RegexSubscriptionMode.AllTopics); + assertThat(consumerProps.getDeadLetterPolicy()).satisfies(dlp -> { + assertThat(dlp.getMaxRedeliverCount()).isEqualTo(4); + assertThat(dlp.getRetryLetterTopic()).isEqualTo("my-retry-topic"); + assertThat(dlp.getDeadLetterTopic()).isEqualTo("my-dlt-topic"); + assertThat(dlp.getInitialSubscriptionName()).isEqualTo("my-initial-subscription"); + }); + assertThat(consumerProps.getRetryEnable()).isTrue(); + assertThat(consumerProps.getAutoUpdatePartitions()).isFalse(); + assertThat(consumerProps.getAutoUpdatePartitionsInterval()).isEqualTo(Duration.ofMillis(10_000)); + assertThat(consumerProps.getReplicateSubscriptionState()).isTrue(); + assertThat(consumerProps.getResetIncludeHead()).isTrue(); + assertThat(consumerProps.getBatchIndexAckEnabled()).isTrue(); + assertThat(consumerProps.getAckReceiptEnabled()).isTrue(); + assertThat(consumerProps.getPoolMessages()).isTrue(); + assertThat(consumerProps.getStartPaused()).isTrue(); + assertThat(consumerProps.getAutoAckOldestChunkedMessageOnQueueFull()).isFalse(); + assertThat(consumerProps.getMaxPendingChunkedMessage()).isEqualTo(11); + assertThat(consumerProps.getExpireTimeOfIncompleteChunkedMessage()).isEqualTo(Duration.ofMillis(12_000)); } } diff --git a/spring-pulsar-spring-cloud-stream-binder/src/main/java/org/springframework/pulsar/spring/cloud/stream/binder/PulsarBinderUtils.java b/spring-pulsar-spring-cloud-stream-binder/src/main/java/org/springframework/pulsar/spring/cloud/stream/binder/PulsarBinderUtils.java index 8acd5e89..6c6e06a3 100644 --- a/spring-pulsar-spring-cloud-stream-binder/src/main/java/org/springframework/pulsar/spring/cloud/stream/binder/PulsarBinderUtils.java +++ b/spring-pulsar-spring-cloud-stream-binder/src/main/java/org/springframework/pulsar/spring/cloud/stream/binder/PulsarBinderUtils.java @@ -25,6 +25,7 @@ import java.util.UUID; import org.springframework.boot.context.properties.PropertyMapper; import org.springframework.cloud.stream.provisioning.ConsumerDestination; import org.springframework.core.log.LogAccessor; +import org.springframework.pulsar.autoconfigure.ConsumerConfigProperties; import org.springframework.pulsar.autoconfigure.ProducerConfigProperties; import org.springframework.pulsar.spring.cloud.stream.binder.properties.PulsarConsumerProperties; import org.springframework.util.StringUtils; @@ -144,6 +145,57 @@ final class PulsarBinderUtils { return properties; } + /** + * Gets a map representation of a {@link ConsumerConfigProperties}. + * @param consumerProps the consumer props + * @return map representation of consumer props where each entry is a field and its + * associated value + */ + static Map convertConsumerPropertiesToMap(ConsumerConfigProperties consumerProps) { + var properties = new PulsarBinderUtils.Properties(); + var map = PropertyMapper.get().alwaysApplyingWhenNonNull(); + map.from(consumerProps::getTopics).to(properties.in("topicNames")); + map.from(consumerProps::getTopicsPattern).to(properties.in("topicsPattern")); + map.from(consumerProps::getSubscriptionName).to(properties.in("subscriptionName")); + map.from(consumerProps::getSubscriptionType).to(properties.in("subscriptionType")); + map.from(consumerProps::getSubscriptionProperties).to(properties.in("subscriptionProperties")); + map.from(consumerProps::getSubscriptionMode).to(properties.in("subscriptionMode")); + map.from(consumerProps::getReceiverQueueSize).to(properties.in("receiverQueueSize")); + map.from(consumerProps::getAcknowledgementsGroupTime).as(it -> it.toNanos() / 1000) + .to(properties.in("acknowledgementsGroupTimeMicros")); + map.from(consumerProps::getNegativeAckRedeliveryDelay).as(it -> it.toNanos() / 1000) + .to(properties.in("negativeAckRedeliveryDelayMicros")); + map.from(consumerProps::getMaxTotalReceiverQueueSizeAcrossPartitions) + .to(properties.in("maxTotalReceiverQueueSizeAcrossPartitions")); + map.from(consumerProps::getConsumerName).to(properties.in("consumerName")); + map.from(consumerProps::getAckTimeout).as(Duration::toMillis).to(properties.in("ackTimeoutMillis")); + map.from(consumerProps::getTickDuration).as(Duration::toMillis).to(properties.in("tickDurationMillis")); + map.from(consumerProps::getPriorityLevel).to(properties.in("priorityLevel")); + map.from(consumerProps::getCryptoFailureAction).to(properties.in("cryptoFailureAction")); + map.from(consumerProps::getProperties).to(properties.in("properties")); + map.from(consumerProps::getReadCompacted).to(properties.in("readCompacted")); + map.from(consumerProps::getSubscriptionInitialPosition).to(properties.in("subscriptionInitialPosition")); + map.from(consumerProps::getPatternAutoDiscoveryPeriod).to(properties.in("patternAutoDiscoveryPeriod")); + map.from(consumerProps::getRegexSubscriptionMode).to(properties.in("regexSubscriptionMode")); + map.from(consumerProps::getDeadLetterPolicy).to(properties.in("deadLetterPolicy")); + map.from(consumerProps::getRetryEnable).to(properties.in("retryEnable")); + map.from(consumerProps::getAutoUpdatePartitions).to(properties.in("autoUpdatePartitions")); + map.from(consumerProps::getAutoUpdatePartitionsInterval).as(Duration::toSeconds) + .to(properties.in("autoUpdatePartitionsIntervalSeconds")); + map.from(consumerProps::getReplicateSubscriptionState).to(properties.in("replicateSubscriptionState")); + map.from(consumerProps::getResetIncludeHead).to(properties.in("resetIncludeHead")); + map.from(consumerProps::getBatchIndexAckEnabled).to(properties.in("batchIndexAckEnabled")); + map.from(consumerProps::getAckReceiptEnabled).to(properties.in("ackReceiptEnabled")); + map.from(consumerProps::getPoolMessages).to(properties.in("poolMessages")); + map.from(consumerProps::getStartPaused).to(properties.in("startPaused")); + map.from(consumerProps::getAutoAckOldestChunkedMessageOnQueueFull) + .to(properties.in("autoAckOldestChunkedMessageOnQueueFull")); + map.from(consumerProps::getMaxPendingChunkedMessage).to(properties.in("maxPendingChunkedMessage")); + map.from(consumerProps::getExpireTimeOfIncompleteChunkedMessage).as(Duration::toMillis) + .to(properties.in("expireTimeOfIncompleteChunkedMessageMillis")); + return properties; + } + static class Properties extends HashMap { java.util.function.Consumer in(String key) { diff --git a/spring-pulsar-spring-cloud-stream-binder/src/main/java/org/springframework/pulsar/spring/cloud/stream/binder/PulsarMessageChannelBinder.java b/spring-pulsar-spring-cloud-stream-binder/src/main/java/org/springframework/pulsar/spring/cloud/stream/binder/PulsarMessageChannelBinder.java index c31e3f81..004b30e2 100644 --- a/spring-pulsar-spring-cloud-stream-binder/src/main/java/org/springframework/pulsar/spring/cloud/stream/binder/PulsarMessageChannelBinder.java +++ b/spring-pulsar-spring-cloud-stream-binder/src/main/java/org/springframework/pulsar/spring/cloud/stream/binder/PulsarMessageChannelBinder.java @@ -166,9 +166,10 @@ public class PulsarMessageChannelBinder extends var subscriptionName = PulsarBinderUtils.subscriptionName(properties.getExtension(), destination); containerProperties.setSubscriptionName(subscriptionName); - var baseConsumerProps = new ConsumerConfigProperties().buildProperties(); - var binderConsumerProps = this.binderConfigProps.getConsumer().buildProperties(); - var bindingConsumerProps = properties.getExtension().buildProperties(); + var baseConsumerProps = PulsarBinderUtils.convertConsumerPropertiesToMap(new ConsumerConfigProperties()); + var binderConsumerProps = PulsarBinderUtils + .convertConsumerPropertiesToMap(this.binderConfigProps.getConsumer()); + var bindingConsumerProps = PulsarBinderUtils.convertConsumerPropertiesToMap(properties.getExtension()); var mergedConsumerProps = PulsarBinderUtils.mergePropertiesWithPrecedence(baseConsumerProps, binderConsumerProps, bindingConsumerProps); containerProperties.getPulsarConsumerProperties().putAll(mergedConsumerProps); diff --git a/spring-pulsar-spring-cloud-stream-binder/src/test/java/org/springframework/pulsar/spring/cloud/stream/binder/PulsarBinderConfigurationPropertiesTests.java b/spring-pulsar-spring-cloud-stream-binder/src/test/java/org/springframework/pulsar/spring/cloud/stream/binder/PulsarBinderConfigurationPropertiesTests.java index 7ff3c387..af054db4 100644 --- a/spring-pulsar-spring-cloud-stream-binder/src/test/java/org/springframework/pulsar/spring/cloud/stream/binder/PulsarBinderConfigurationPropertiesTests.java +++ b/spring-pulsar-spring-cloud-stream-binder/src/test/java/org/springframework/pulsar/spring/cloud/stream/binder/PulsarBinderConfigurationPropertiesTests.java @@ -98,7 +98,7 @@ public class PulsarBinderConfigurationPropertiesTests { props.put("spring.cloud.stream.pulsar.binder.consumer.receiver-queue-size", "1"); bind(props); - Map consumerProps = properties.getConsumer().buildProperties(); + Map consumerProps = PulsarBinderUtils.convertConsumerPropertiesToMap(properties.getConsumer()); // Verify that the props can be loaded in a ConsumerBuilder assertThatNoException().isThrownBy(() -> ConfigurationDataUtils.loadData(consumerProps, diff --git a/spring-pulsar-spring-cloud-stream-binder/src/test/java/org/springframework/pulsar/spring/cloud/stream/binder/PulsarBinderIntegrationTests.java b/spring-pulsar-spring-cloud-stream-binder/src/test/java/org/springframework/pulsar/spring/cloud/stream/binder/PulsarBinderIntegrationTests.java index f8d41366..9d641e2d 100644 --- a/spring-pulsar-spring-cloud-stream-binder/src/test/java/org/springframework/pulsar/spring/cloud/stream/binder/PulsarBinderIntegrationTests.java +++ b/spring-pulsar-spring-cloud-stream-binder/src/test/java/org/springframework/pulsar/spring/cloud/stream/binder/PulsarBinderIntegrationTests.java @@ -649,10 +649,13 @@ class PulsarBinderIntegrationTests implements PulsarTestContainerSupport { topicResolver); } + @SuppressWarnings("unchecked") @Bean public PulsarConsumerFactory pulsarConsumerFactory(PulsarClient pulsarClient, PulsarProperties pulsarProperties) { - return new TrackingConsumerFactory(pulsarClient, pulsarProperties.buildConsumerProperties()); + var customizer = (ConsumerBuilderCustomizer) pulsarProperties.getConsumer() + .toConsumerBuilderCustomizer(); + return new TrackingConsumerFactory(pulsarClient, customizer); } } @@ -683,8 +686,8 @@ class PulsarBinderIntegrationTests implements PulsarTestContainerSupport { List> consumersCreated = new ArrayList<>(); - TrackingConsumerFactory(PulsarClient pulsarClient, Map consumerConfig) { - super(pulsarClient, consumerConfig); + TrackingConsumerFactory(PulsarClient pulsarClient, ConsumerBuilderCustomizer defaultConsumerConfig) { + super(pulsarClient, defaultConsumerConfig); } @Override diff --git a/spring-pulsar-spring-cloud-stream-binder/src/test/java/org/springframework/pulsar/spring/cloud/stream/binder/PulsarBinderTests.java b/spring-pulsar-spring-cloud-stream-binder/src/test/java/org/springframework/pulsar/spring/cloud/stream/binder/PulsarBinderTests.java index 49b07b3a..8313daac 100644 --- a/spring-pulsar-spring-cloud-stream-binder/src/test/java/org/springframework/pulsar/spring/cloud/stream/binder/PulsarBinderTests.java +++ b/spring-pulsar-spring-cloud-stream-binder/src/test/java/org/springframework/pulsar/spring/cloud/stream/binder/PulsarBinderTests.java @@ -106,8 +106,8 @@ public class PulsarBinderTests extends var provisioner = new PulsarTopicProvisioner(pulsarAdministration, configProps); var producerFactory = new DefaultPulsarProducerFactory<>(pulsarClient); var pulsarTemplate = new PulsarTemplate<>(producerFactory); - var config = Map.of("subscriptionInitialPosition", SubscriptionInitialPosition.Earliest); - var consumerFactory = new DefaultPulsarConsumerFactory<>(pulsarClient, config); + var consumerFactory = new DefaultPulsarConsumerFactory<>(pulsarClient, + (consumerBuilder -> consumerBuilder.subscriptionInitialPosition(SubscriptionInitialPosition.Earliest))); if (this.binder == null) { this.binder = new PulsarTestBinder(provisioner, pulsarTemplate, consumerFactory, configProps, new DefaultSchemaResolver(), JsonPulsarHeaderMapper.builder().build()); diff --git a/spring-pulsar-spring-cloud-stream-binder/src/test/java/org/springframework/pulsar/spring/cloud/stream/binder/PulsarBinderUtilsTests.java b/spring-pulsar-spring-cloud-stream-binder/src/test/java/org/springframework/pulsar/spring/cloud/stream/binder/PulsarBinderUtilsTests.java index 95d2839c..ba7be643 100644 --- a/spring-pulsar-spring-cloud-stream-binder/src/test/java/org/springframework/pulsar/spring/cloud/stream/binder/PulsarBinderUtilsTests.java +++ b/spring-pulsar-spring-cloud-stream-binder/src/test/java/org/springframework/pulsar/spring/cloud/stream/binder/PulsarBinderUtilsTests.java @@ -28,11 +28,18 @@ import java.util.Map; import java.util.stream.Stream; import org.apache.pulsar.client.api.CompressionType; +import org.apache.pulsar.client.api.ConsumerCryptoFailureAction; +import org.apache.pulsar.client.api.DeadLetterPolicy; import org.apache.pulsar.client.api.HashingScheme; import org.apache.pulsar.client.api.MessageRoutingMode; import org.apache.pulsar.client.api.ProducerAccessMode; import org.apache.pulsar.client.api.ProducerCryptoFailureAction; +import org.apache.pulsar.client.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.ConsumerConfigurationData; import org.apache.pulsar.client.impl.conf.ProducerConfigurationData; import org.assertj.core.api.InstanceOfAssertFactories; import org.junit.jupiter.api.Nested; @@ -46,6 +53,7 @@ import org.springframework.boot.context.properties.bind.Binder; import org.springframework.boot.context.properties.source.ConfigurationPropertySource; import org.springframework.boot.context.properties.source.MapConfigurationPropertySource; import org.springframework.cloud.stream.provisioning.ConsumerDestination; +import org.springframework.pulsar.autoconfigure.ConsumerConfigProperties; import org.springframework.pulsar.autoconfigure.ProducerConfigProperties; import org.springframework.pulsar.spring.cloud.stream.binder.properties.PulsarConsumerProperties; @@ -211,4 +219,106 @@ public class PulsarBinderUtilsTests { } + @Nested + class ConvertedConsumerPropertiesTests { + + private final ConsumerConfigProperties properties = new ConsumerConfigProperties(); + + private void bind(Map map) { + ConfigurationPropertySource source = new MapConfigurationPropertySource(map); + new Binder(source).bind("spring.pulsar.consumer", Bindable.ofInstance(this.properties)); + } + + @Test + void consumerPropertiesToMap() { + Map props = new HashMap<>(); + props.put("spring.pulsar.consumer.topics[0]", "my-topic"); + props.put("spring.pulsar.consumer.topics-pattern", "my-pattern"); + props.put("spring.pulsar.consumer.subscription-name", "my-subscription"); + props.put("spring.pulsar.consumer.subscription-type", "shared"); + props.put("spring.pulsar.consumer.subscription-properties[my-sub-prop]", "my-sub-prop-value"); + props.put("spring.pulsar.consumer.subscription-mode", "nondurable"); + props.put("spring.pulsar.consumer.receiver-queue-size", "1"); + props.put("spring.pulsar.consumer.acknowledgements-group-time", "2s"); + props.put("spring.pulsar.consumer.negative-ack-redelivery-delay", "3s"); + props.put("spring.pulsar.consumer.max-total-receiver-queue-size-across-partitions", "5"); + props.put("spring.pulsar.consumer.consumer-name", "my-consumer"); + props.put("spring.pulsar.consumer.ack-timeout", "6s"); + props.put("spring.pulsar.consumer.tick-duration", "7s"); + props.put("spring.pulsar.consumer.priority-level", "8"); + props.put("spring.pulsar.consumer.crypto-failure-action", "discard"); + props.put("spring.pulsar.consumer.properties[my-prop]", "my-prop-value"); + props.put("spring.pulsar.consumer.read-compacted", "true"); + props.put("spring.pulsar.consumer.subscription-initial-position", "earliest"); + props.put("spring.pulsar.consumer.pattern-auto-discovery-period", "9"); + props.put("spring.pulsar.consumer.regex-subscription-mode", "all-topics"); + props.put("spring.pulsar.consumer.dead-letter-policy.max-redeliver-count", "4"); + props.put("spring.pulsar.consumer.dead-letter-policy.retry-letter-topic", "my-retry-topic"); + props.put("spring.pulsar.consumer.dead-letter-policy.dead-letter-topic", "my-dlt-topic"); + props.put("spring.pulsar.consumer.dead-letter-policy.initial-subscription-name", "my-initial-subscription"); + props.put("spring.pulsar.consumer.retry-enable", "true"); + props.put("spring.pulsar.consumer.auto-update-partitions", "false"); + props.put("spring.pulsar.consumer.auto-update-partitions-interval", "10s"); + props.put("spring.pulsar.consumer.replicate-subscription-state", "true"); + props.put("spring.pulsar.consumer.reset-include-head", "true"); + props.put("spring.pulsar.consumer.batch-index-ack-enabled", "true"); + props.put("spring.pulsar.consumer.ack-receipt-enabled", "true"); + props.put("spring.pulsar.consumer.pool-messages", "true"); + props.put("spring.pulsar.consumer.start-paused", "true"); + props.put("spring.pulsar.consumer.auto-ack-oldest-chunked-message-on-queue-full", "false"); + props.put("spring.pulsar.consumer.max-pending-chunked-message", "11"); + props.put("spring.pulsar.consumer.expire-time-of-incomplete-chunked-message", "12s"); + + bind(props); + Map consumerProps = PulsarBinderUtils.convertConsumerPropertiesToMap(properties); + + // Verify that the props can be loaded in a ConsumerBuilder + assertThatNoException().isThrownBy(() -> ConfigurationDataUtils.loadData(consumerProps, + new ConsumerConfigurationData<>(), ConsumerConfigurationData.class)); + + assertThat(consumerProps) + .hasEntrySatisfying("topicNames", + topics -> assertThat(topics) + .asInstanceOf(InstanceOfAssertFactories.collection(String.class)) + .containsExactly("my-topic")) + .hasEntrySatisfying("topicsPattern", p -> assertThat(p.toString()).isEqualTo("my-pattern")) + .containsEntry("subscriptionName", "my-subscription") + .containsEntry("subscriptionType", SubscriptionType.Shared) + .hasEntrySatisfying("subscriptionProperties", + properties -> assertThat(properties) + .asInstanceOf(InstanceOfAssertFactories.map(String.class, String.class)) + .containsEntry("my-sub-prop", "my-sub-prop-value")) + .containsEntry("subscriptionMode", SubscriptionMode.NonDurable) + .containsEntry("receiverQueueSize", 1).containsEntry("acknowledgementsGroupTimeMicros", 2_000_000L) + .containsEntry("negativeAckRedeliveryDelayMicros", 3_000_000L) + .containsEntry("maxTotalReceiverQueueSizeAcrossPartitions", 5) + .containsEntry("consumerName", "my-consumer").containsEntry("ackTimeoutMillis", 6_000L) + .containsEntry("tickDurationMillis", 7_000L).containsEntry("priorityLevel", 8) + .containsEntry("cryptoFailureAction", ConsumerCryptoFailureAction.DISCARD) + .hasEntrySatisfying("properties", + properties -> assertThat(properties) + .asInstanceOf(InstanceOfAssertFactories.map(String.class, String.class)) + .containsEntry("my-prop", "my-prop-value")) + .containsEntry("readCompacted", true) + .containsEntry("subscriptionInitialPosition", SubscriptionInitialPosition.Earliest) + .containsEntry("patternAutoDiscoveryPeriod", 9) + .containsEntry("regexSubscriptionMode", RegexSubscriptionMode.AllTopics) + .hasEntrySatisfying("deadLetterPolicy", dlp -> { + DeadLetterPolicy deadLetterPolicy = (DeadLetterPolicy) dlp; + assertThat(deadLetterPolicy.getMaxRedeliverCount()).isEqualTo(4); + assertThat(deadLetterPolicy.getRetryLetterTopic()).isEqualTo("my-retry-topic"); + assertThat(deadLetterPolicy.getDeadLetterTopic()).isEqualTo("my-dlt-topic"); + assertThat(deadLetterPolicy.getInitialSubscriptionName()).isEqualTo("my-initial-subscription"); + }).containsEntry("retryEnable", true).containsEntry("autoUpdatePartitions", false) + .containsEntry("autoUpdatePartitionsIntervalSeconds", 10L) + .containsEntry("replicateSubscriptionState", true).containsEntry("resetIncludeHead", true) + .containsEntry("batchIndexAckEnabled", true).containsEntry("ackReceiptEnabled", true) + .containsEntry("poolMessages", true).containsEntry("startPaused", true) + .containsEntry("autoAckOldestChunkedMessageOnQueueFull", false) + .containsEntry("maxPendingChunkedMessage", 11) + .containsEntry("expireTimeOfIncompleteChunkedMessageMillis", 12_000L); + } + + } + } diff --git a/spring-pulsar-spring-cloud-stream-binder/src/test/java/org/springframework/pulsar/spring/cloud/stream/binder/PulsarExtendedBindingPropertiesTests.java b/spring-pulsar-spring-cloud-stream-binder/src/test/java/org/springframework/pulsar/spring/cloud/stream/binder/PulsarExtendedBindingPropertiesTests.java index 5abf6254..b00b1a01 100644 --- a/spring-pulsar-spring-cloud-stream-binder/src/test/java/org/springframework/pulsar/spring/cloud/stream/binder/PulsarExtendedBindingPropertiesTests.java +++ b/spring-pulsar-spring-cloud-stream-binder/src/test/java/org/springframework/pulsar/spring/cloud/stream/binder/PulsarExtendedBindingPropertiesTests.java @@ -96,7 +96,8 @@ public class PulsarExtendedBindingPropertiesTests { bind(props); assertThat(properties.getBindings()).containsOnlyKeys("my-foo"); - Map consumerProps = properties.getExtendedConsumerProperties("my-foo").buildProperties(); + Map consumerProps = PulsarBinderUtils + .convertConsumerPropertiesToMap(properties.getExtendedConsumerProperties("my-foo")); // Verify that the props can be loaded in a ConsumerBuilder assertThatNoException().isThrownBy(() -> ConfigurationDataUtils.loadData(consumerProps, new ConsumerConfigurationData<>(), ConsumerConfigurationData.class)); @@ -122,7 +123,8 @@ public class PulsarExtendedBindingPropertiesTests { bind(props); - var bindingConsumerProps = properties.getExtendedConsumerProperties("my-foo").buildProperties(); + var bindingConsumerProps = PulsarBinderUtils + .convertConsumerPropertiesToMap(properties.getExtendedConsumerProperties("my-foo")); PulsarContainerProperties pulsarContainerProperties = new PulsarContainerProperties(); pulsarContainerProperties.getPulsarConsumerProperties().putAll(bindingConsumerProps); diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/core/DefaultPulsarConsumerFactory.java b/spring-pulsar/src/main/java/org/springframework/pulsar/core/DefaultPulsarConsumerFactory.java index d4e7184c..96c3cd66 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/core/DefaultPulsarConsumerFactory.java +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/core/DefaultPulsarConsumerFactory.java @@ -18,7 +18,6 @@ package org.springframework.pulsar.core; import java.util.Collection; import java.util.Collections; -import java.util.HashMap; import java.util.HashSet; import java.util.List; import java.util.Map; @@ -30,6 +29,7 @@ import org.apache.pulsar.client.api.ConsumerBuilder; import org.apache.pulsar.client.api.PulsarClient; import org.apache.pulsar.client.api.PulsarClientException; import org.apache.pulsar.client.api.Schema; +import org.apache.pulsar.client.impl.ConsumerBuilderImpl; import org.springframework.lang.Nullable; import org.springframework.util.CollectionUtils; @@ -41,22 +41,25 @@ import org.springframework.util.CollectionUtils; * @author Soby Chacko * @author Alexander Preuß * @author Christophe Bornet + * @author Chris Bono */ public class DefaultPulsarConsumerFactory implements PulsarConsumerFactory { - private final Map consumerConfig; - private final PulsarClient pulsarClient; + @Nullable + private final ConsumerBuilderCustomizer defaultConfigCustomizer; + /** * Construct a consumer factory instance. * @param pulsarClient the client used to consume - * @param consumerConfig default configuration to apply to the created consumer or - * empty map to use no default configuration + * @param defaultConfigCustomizer the default configuration to apply to the consumers + * or null to use no default configuration */ - public DefaultPulsarConsumerFactory(PulsarClient pulsarClient, Map consumerConfig) { + public DefaultPulsarConsumerFactory(PulsarClient pulsarClient, + ConsumerBuilderCustomizer defaultConfigCustomizer) { this.pulsarClient = pulsarClient; - this.consumerConfig = Collections.unmodifiableMap(consumerConfig); + this.defaultConfigCustomizer = defaultConfigCustomizer; } @Override @@ -72,25 +75,35 @@ public class DefaultPulsarConsumerFactory implements PulsarConsumerFactory @Nullable List> customizers) throws PulsarClientException { Objects.requireNonNull(schema, "Schema must be specified"); ConsumerBuilder consumerBuilder = this.pulsarClient.newConsumer(schema); - Map config = new HashMap<>(this.consumerConfig); - if (topics != null) { - config.put("topicNames", new HashSet<>(topics)); + + // Apply the default config customizer (preserve the topic) + if (this.defaultConfigCustomizer != null) { + this.defaultConfigCustomizer.customize(consumerBuilder); } - if (metadataProperties != null) { - config.put("properties", new TreeMap<>(metadataProperties)); + if (topics != null) { + replaceTopicsOnBuilder(consumerBuilder, topics); } if (subscriptionName != null) { - config.put("subscriptionName", subscriptionName); + consumerBuilder.subscriptionName(subscriptionName); + } + if (metadataProperties != null) { + replaceMetadataPropertiesOnBuilder(consumerBuilder, metadataProperties); } - ConsumerBuilderConfigurationUtil.loadConf(consumerBuilder, config); if (!CollectionUtils.isEmpty(customizers)) { customizers.forEach(customizer -> customizer.customize(consumerBuilder)); } return consumerBuilder.subscribe(); } - public Map getConsumerConfig() { - return this.consumerConfig; + private void replaceTopicsOnBuilder(ConsumerBuilder builder, Collection topics) { + var builderImpl = (ConsumerBuilderImpl) builder; + builderImpl.getConf().setTopicNames(new HashSet<>(topics)); + } + + private void replaceMetadataPropertiesOnBuilder(ConsumerBuilder builder, + Map metadataProperties) { + var builderImpl = (ConsumerBuilderImpl) builder; + builderImpl.getConf().setProperties(new TreeMap<>(metadataProperties)); } } diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/core/PulsarConsumerFactory.java b/spring-pulsar/src/main/java/org/springframework/pulsar/core/PulsarConsumerFactory.java index c3bc1c08..ec2b529f 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/core/PulsarConsumerFactory.java +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/core/PulsarConsumerFactory.java @@ -85,10 +85,4 @@ public interface PulsarConsumerFactory { @Nullable Map metadataProperties, @Nullable List> customizers) throws PulsarClientException; - /** - * Return the configuration options to use when creating consumers. - * @return the configuration options - */ - Map getConsumerConfig(); - } diff --git a/spring-pulsar/src/test/java/org/springframework/pulsar/core/ConsumerAcknowledgmentTests.java b/spring-pulsar/src/test/java/org/springframework/pulsar/core/ConsumerAcknowledgmentTests.java index a6a751dd..1230ea7a 100644 --- a/spring-pulsar/src/test/java/org/springframework/pulsar/core/ConsumerAcknowledgmentTests.java +++ b/spring-pulsar/src/test/java/org/springframework/pulsar/core/ConsumerAcknowledgmentTests.java @@ -30,9 +30,7 @@ import static org.mockito.Mockito.verify; import java.time.Duration; import java.util.ArrayList; -import java.util.Collections; import java.util.List; -import java.util.Map; import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; @@ -61,14 +59,10 @@ class ConsumerAcknowledgmentTests implements PulsarTestContainerSupport { @Test void testRecordAck() throws Exception { - Map config = Map.of("topicNames", Collections.singleton("cons-ack-tests-011"), - "subscriptionName", "cons-ack-tests-sb-011"); - PulsarClient pulsarClient = PulsarClient.builder().serviceUrl(PulsarTestContainerSupport.getPulsarBrokerUrl()) .build(); - DefaultPulsarConsumerFactory pulsarConsumerFactory = spy( - new DefaultPulsarConsumerFactory<>(pulsarClient, config)); - + DefaultPulsarConsumerFactory pulsarConsumerFactory = spy(new DefaultPulsarConsumerFactory<>( + pulsarClient, defaultConfig("cons-ack-tests-011", "cons-ack-tests-sb-011"))); PulsarContainerProperties pulsarContainerProperties = new PulsarContainerProperties(); pulsarContainerProperties.setMessageListener((PulsarRecordMessageListener) (consumer, msg) -> { }); @@ -98,13 +92,10 @@ class ConsumerAcknowledgmentTests implements PulsarTestContainerSupport { @Test void testBatchAck() throws Exception { - Map config = Map.of("topicNames", Collections.singleton("cons-ack-tests-012"), - "subscriptionName", "cons-ack-tests-sb-012"); PulsarClient pulsarClient = PulsarClient.builder().serviceUrl(PulsarTestContainerSupport.getPulsarBrokerUrl()) .build(); - DefaultPulsarConsumerFactory pulsarConsumerFactory = spy( - new DefaultPulsarConsumerFactory<>(pulsarClient, config)); - + DefaultPulsarConsumerFactory pulsarConsumerFactory = spy(new DefaultPulsarConsumerFactory<>( + pulsarClient, defaultConfig("cons-ack-tests-012", "cons-ack-tests-sb-012"))); PulsarContainerProperties pulsarContainerProperties = new PulsarContainerProperties(); CountDownLatch latch = new CountDownLatch(10); pulsarContainerProperties @@ -131,16 +122,12 @@ class ConsumerAcknowledgmentTests implements PulsarTestContainerSupport { @Test void testBatchAckButSomeRecordsFail() throws Exception { - Map config = Map.of("topicNames", Collections.singleton("cons-ack-tests-013"), - "subscriptionName", "cons-ack-tests-sb-013"); PulsarClient pulsarClient = PulsarClient.builder().serviceUrl(PulsarTestContainerSupport.getPulsarBrokerUrl()) .build(); - DefaultPulsarConsumerFactory pulsarConsumerFactory = spy( - new DefaultPulsarConsumerFactory<>(pulsarClient, config)); - + DefaultPulsarConsumerFactory pulsarConsumerFactory = spy(new DefaultPulsarConsumerFactory<>( + pulsarClient, defaultConfig("cons-ack-tests-013", "cons-ack-tests-sb-013"))); PulsarContainerProperties pulsarContainerProperties = new PulsarContainerProperties(); CountDownLatch latch = new CountDownLatch(10); - pulsarContainerProperties.setMessageListener((PulsarRecordMessageListener) (consumer, msg) -> { latch.countDown(); if (latch.getCount() % 2 == 0) { @@ -194,22 +181,17 @@ class ConsumerAcknowledgmentTests implements PulsarTestContainerSupport { @Test @SuppressWarnings("unchecked") void testManualAckForRecordListener() throws Exception { - Map config = Map.of("topicNames", Collections.singleton("cons-ack-tests-014"), - "subscriptionName", "cons-ack-tests-sb-014"); PulsarClient pulsarClient = PulsarClient.builder().serviceUrl(PulsarTestContainerSupport.getPulsarBrokerUrl()) .build(); - DefaultPulsarConsumerFactory pulsarConsumerFactory = spy( - new DefaultPulsarConsumerFactory<>(pulsarClient, config)); - + DefaultPulsarConsumerFactory pulsarConsumerFactory = spy(new DefaultPulsarConsumerFactory<>( + pulsarClient, defaultConfig("cons-ack-tests-014", "cons-ack-tests-sb-014"))); PulsarContainerProperties pulsarContainerProperties = new PulsarContainerProperties(); List acksObjects = new ArrayList<>(); PulsarAcknowledgingMessageListener pulsarAcknowledgingMessageListener = (consumer, msg, acknowledgement) -> { acksObjects.add(acknowledgement); acknowledgement.acknowledge(); }; - pulsarContainerProperties.setMessageListener(pulsarAcknowledgingMessageListener); - pulsarContainerProperties.setSchema(Schema.STRING); pulsarContainerProperties.setAckMode(AckMode.MANUAL); DefaultPulsarMessageListenerContainer container = new DefaultPulsarMessageListenerContainer<>( @@ -243,25 +225,20 @@ class ConsumerAcknowledgmentTests implements PulsarTestContainerSupport { @Test @SuppressWarnings("unchecked") void testBatchAckForBatchListener() throws Exception { - Map config = Map.of("topicNames", Collections.singleton("cons-ack-tests-015"), - "subscriptionName", "cons-ack-tests-sb-015"); PulsarClient pulsarClient = PulsarClient.builder().serviceUrl(PulsarTestContainerSupport.getPulsarBrokerUrl()) .build(); - DefaultPulsarConsumerFactory pulsarConsumerFactory = spy( - new DefaultPulsarConsumerFactory<>(pulsarClient, config)); - + DefaultPulsarConsumerFactory pulsarConsumerFactory = spy(new DefaultPulsarConsumerFactory<>( + pulsarClient, defaultConfig("cons-ack-tests-015", "cons-ack-tests-sb-015"))); PulsarContainerProperties pulsarContainerProperties = new PulsarContainerProperties(); pulsarContainerProperties.setMaxNumMessages(10); pulsarContainerProperties.setBatchTimeoutMillis(60_000); pulsarContainerProperties.setBatchListener(true); CountDownLatch latch = new CountDownLatch(1); PulsarBatchMessageListener pulsarBatchMessageListener = mock(PulsarBatchMessageListener.class); - doAnswer(invocation -> { latch.countDown(); return null; }).when(pulsarBatchMessageListener).received(any(Consumer.class), any(List.class)); - pulsarContainerProperties.setMessageListener(pulsarBatchMessageListener); pulsarContainerProperties.setSchema(Schema.STRING); DefaultPulsarMessageListenerContainer container = new DefaultPulsarMessageListenerContainer<>( @@ -286,25 +263,20 @@ class ConsumerAcknowledgmentTests implements PulsarTestContainerSupport { @Test @SuppressWarnings("unchecked") void testBatchNackForEntireBatchWhenUsingBatchListener() throws Exception { - Map config = Map.of("topicNames", Collections.singleton("cons-ack-tests-016"), - "subscriptionName", "cons-ack-tests-sb-016"); PulsarClient pulsarClient = PulsarClient.builder().serviceUrl(PulsarTestContainerSupport.getPulsarBrokerUrl()) .build(); - DefaultPulsarConsumerFactory pulsarConsumerFactory = spy( - new DefaultPulsarConsumerFactory<>(pulsarClient, config)); - + DefaultPulsarConsumerFactory pulsarConsumerFactory = spy(new DefaultPulsarConsumerFactory<>( + pulsarClient, defaultConfig("cons-ack-tests-016", "cons-ack-tests-sb-016"))); PulsarContainerProperties pulsarContainerProperties = new PulsarContainerProperties(); pulsarContainerProperties.setMaxNumMessages(10); pulsarContainerProperties.setBatchTimeoutMillis(60_000); pulsarContainerProperties.setBatchListener(true); PulsarBatchMessageListener pulsarBatchMessageListener = mock(PulsarBatchMessageListener.class); CountDownLatch latch = new CountDownLatch(1); - doAnswer(invocation -> { latch.countDown(); throw new RuntimeException(); }).when(pulsarBatchMessageListener).received(any(Consumer.class), any(List.class)); - pulsarContainerProperties.setMessageListener(pulsarBatchMessageListener); pulsarContainerProperties.setSchema(Schema.STRING); DefaultPulsarMessageListenerContainer container = new DefaultPulsarMessageListenerContainer<>( @@ -329,18 +301,14 @@ class ConsumerAcknowledgmentTests implements PulsarTestContainerSupport { @Test void messagesAreProperlyAckdOnContainerStopBeforeExitingListenerThread() throws Exception { - Map config = Map.of("topicNames", Collections.singleton("duplicate-message-test"), - "subscriptionName", "duplicate-sub-1"); PulsarClient pulsarClient = PulsarClient.builder().serviceUrl(PulsarTestContainerSupport.getPulsarBrokerUrl()) .build(); DefaultPulsarConsumerFactory pulsarConsumerFactory = new DefaultPulsarConsumerFactory<>(pulsarClient, - config); - + defaultConfig("duplicate-message-test", "duplicate-sub-1")); PulsarContainerProperties pulsarContainerProperties = new PulsarContainerProperties(); AtomicInteger counter1 = new AtomicInteger(0); - pulsarContainerProperties.setMessageListener((PulsarRecordMessageListener) (consumer, msg) -> { - counter1.getAndIncrement(); - }); + pulsarContainerProperties + .setMessageListener((PulsarRecordMessageListener) (consumer, msg) -> counter1.getAndIncrement()); pulsarContainerProperties.setSchema(Schema.STRING); DefaultPulsarMessageListenerContainer container1 = new DefaultPulsarMessageListenerContainer<>( pulsarConsumerFactory, pulsarContainerProperties); @@ -361,9 +329,8 @@ class ConsumerAcknowledgmentTests implements PulsarTestContainerSupport { container1.stop(); AtomicInteger counter2 = new AtomicInteger(0); - pulsarContainerProperties.setMessageListener((PulsarRecordMessageListener) (consumer, msg) -> { - counter2.getAndIncrement(); - }); + pulsarContainerProperties + .setMessageListener((PulsarRecordMessageListener) (consumer, msg) -> counter2.getAndIncrement()); pulsarContainerProperties.setSchema(Schema.STRING); DefaultPulsarMessageListenerContainer container2 = new DefaultPulsarMessageListenerContainer<>( pulsarConsumerFactory, pulsarContainerProperties); @@ -382,4 +349,11 @@ class ConsumerAcknowledgmentTests implements PulsarTestContainerSupport { pulsarClient.close(); } + private ConsumerBuilderCustomizer defaultConfig(String topicName, String subscriptionName) { + return (consumerBuilder) -> { + consumerBuilder.topic(topicName); + consumerBuilder.subscriptionName(subscriptionName); + }; + } + } diff --git a/spring-pulsar/src/test/java/org/springframework/pulsar/core/DefaultPulsarConsumerFactoryTests.java b/spring-pulsar/src/test/java/org/springframework/pulsar/core/DefaultPulsarConsumerFactoryTests.java index 83016549..23e5795e 100644 --- a/spring-pulsar/src/test/java/org/springframework/pulsar/core/DefaultPulsarConsumerFactoryTests.java +++ b/spring-pulsar/src/test/java/org/springframework/pulsar/core/DefaultPulsarConsumerFactoryTests.java @@ -23,7 +23,6 @@ import static org.mockito.Mockito.inOrder; import static org.mockito.Mockito.mock; import java.util.Collections; -import java.util.HashMap; import java.util.List; import java.util.Map; @@ -72,7 +71,7 @@ class DefaultPulsarConsumerFactoryTests implements PulsarTestContainerSupport { @BeforeEach void createConsumerFactory() { - consumerFactory = new DefaultPulsarConsumerFactory<>(pulsarClient, Collections.emptyMap()); + consumerFactory = new DefaultPulsarConsumerFactory<>(pulsarClient, null); } @Test @@ -170,11 +169,11 @@ class DefaultPulsarConsumerFactoryTests implements PulsarTestContainerSupport { @BeforeEach void createConsumerFactory() { - Map defaultConfig = new HashMap<>(); - defaultConfig.put("topicNames", Collections.singleton(defaultTopic)); - defaultConfig.put("properties", defaultMetadataProperties); - defaultConfig.put("subscriptionName", defaultSubscription); - consumerFactory = new DefaultPulsarConsumerFactory<>(pulsarClient, defaultConfig); + consumerFactory = new DefaultPulsarConsumerFactory<>(pulsarClient, (consumerBuilder) -> { + consumerBuilder.topic(defaultTopic); + consumerBuilder.subscriptionName(defaultSubscription); + consumerBuilder.properties(defaultMetadataProperties); + }); } @Test diff --git a/spring-pulsar/src/test/java/org/springframework/pulsar/core/FailoverConsumerTests.java b/spring-pulsar/src/test/java/org/springframework/pulsar/core/FailoverConsumerTests.java index f6804843..8b2f30fd 100644 --- a/spring-pulsar/src/test/java/org/springframework/pulsar/core/FailoverConsumerTests.java +++ b/spring-pulsar/src/test/java/org/springframework/pulsar/core/FailoverConsumerTests.java @@ -19,8 +19,6 @@ package org.springframework.pulsar.core; import static org.assertj.core.api.Assertions.assertThat; import java.io.Serial; -import java.util.Collections; -import java.util.Map; import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; @@ -48,18 +46,16 @@ class FailoverConsumerTests implements PulsarTestContainerSupport { void testFailOverConsumersOnPartitionedTopic() throws Exception { PulsarAdmin admin = PulsarAdmin.builder().serviceHttpUrl(PulsarTestContainerSupport.getHttpServiceUrl()) .build(); - String topicName = "persistent://public/default/my-part-topic-1"; - int numPartitions = 3; - admin.topics().createPartitionedTopic(topicName, numPartitions); - - Map config = Map.of("topicNames", Collections.singleton("my-part-topic-1"), "subscriptionName", - "my-part-subscription-1"); + admin.topics().createPartitionedTopic(topicName, 3); PulsarClient pulsarClient = PulsarClient.builder().serviceUrl(PulsarTestContainerSupport.getPulsarBrokerUrl()) .build(); DefaultPulsarConsumerFactory pulsarConsumerFactory = new DefaultPulsarConsumerFactory<>(pulsarClient, - config); + (consumerBuilder) -> { + consumerBuilder.topic("my-part-topic-1"); + consumerBuilder.subscriptionName("my-part-subscription-1"); + }); CountDownLatch latch1 = new CountDownLatch(1); CountDownLatch latch2 = new CountDownLatch(1); diff --git a/spring-pulsar/src/test/java/org/springframework/pulsar/core/SharedSubscriptionConsumerTests.java b/spring-pulsar/src/test/java/org/springframework/pulsar/core/SharedSubscriptionConsumerTests.java index 9fd94273..92212f67 100644 --- a/spring-pulsar/src/test/java/org/springframework/pulsar/core/SharedSubscriptionConsumerTests.java +++ b/spring-pulsar/src/test/java/org/springframework/pulsar/core/SharedSubscriptionConsumerTests.java @@ -18,7 +18,6 @@ package org.springframework.pulsar.core; import static org.assertj.core.api.Assertions.assertThat; -import java.util.Collections; import java.util.HashMap; import java.util.Map; import java.util.concurrent.CountDownLatch; @@ -57,9 +56,10 @@ public class SharedSubscriptionConsumerTests implements PulsarTestContainerSuppo try { pulsarClient = PulsarClient.builder().serviceUrl(PulsarTestContainerSupport.getPulsarBrokerUrl()).build(); DefaultPulsarConsumerFactory pulsarConsumerFactory = new DefaultPulsarConsumerFactory<>( - pulsarClient, - Map.of("topicNames", Collections.singleton("shared-subscription-single-msg-test-topic"), - "subscriptionName", "shared-subscription-single-msg-test-sub")); + pulsarClient, (consumerBuilder) -> { + consumerBuilder.topic("shared-subscription-single-msg-test-topic"); + consumerBuilder.subscriptionName("shared-subscription-single-msg-test-sub"); + }); CountDownLatch latch1 = new CountDownLatch(1); CountDownLatch latch2 = new CountDownLatch(1); @@ -113,8 +113,10 @@ public class SharedSubscriptionConsumerTests implements PulsarTestContainerSuppo try { pulsarClient = PulsarClient.builder().serviceUrl(PulsarTestContainerSupport.getPulsarBrokerUrl()).build(); DefaultPulsarConsumerFactory consumerFactory = new DefaultPulsarConsumerFactory<>(pulsarClient, - Map.of("topicNames", Collections.singleton("key-shared-batch-disabled-topic"), "subscriptionName", - "key-shared-batch-disabled-sub")); + (consumerBuilder) -> { + consumerBuilder.topic("key-shared-batch-disabled-topic"); + consumerBuilder.subscriptionName("key-shared-batch-disabled-sub"); + }); CountDownLatch latch = new CountDownLatch(30); diff --git a/spring-pulsar/src/test/java/org/springframework/pulsar/listener/DefaultPulsarConsumerErrorHandlerTests.java b/spring-pulsar/src/test/java/org/springframework/pulsar/listener/DefaultPulsarConsumerErrorHandlerTests.java index e5b6461e..8fdcac40 100644 --- a/spring-pulsar/src/test/java/org/springframework/pulsar/listener/DefaultPulsarConsumerErrorHandlerTests.java +++ b/spring-pulsar/src/test/java/org/springframework/pulsar/listener/DefaultPulsarConsumerErrorHandlerTests.java @@ -28,9 +28,7 @@ import static org.mockito.Mockito.when; import java.time.Duration; import java.util.ArrayList; -import java.util.Collections; import java.util.List; -import java.util.Map; import java.util.concurrent.atomic.AtomicInteger; import org.apache.pulsar.client.api.Consumer; @@ -56,13 +54,13 @@ public class DefaultPulsarConsumerErrorHandlerTests implements PulsarTestContain @Test @SuppressWarnings("unchecked") void happyPathErrorHandlingForRecordMessageListener() throws Exception { - Map config = Map.of("topicNames", Collections.singleton("default-error-handler-tests-1"), - "subscriptionName", "default-error-handler-tests-sub-1"); - PulsarClient pulsarClient = PulsarClient.builder().serviceUrl(PulsarTestContainerSupport.getPulsarBrokerUrl()) .build(); DefaultPulsarConsumerFactory pulsarConsumerFactory = new DefaultPulsarConsumerFactory<>(pulsarClient, - config); + (consumerBuilder) -> { + consumerBuilder.topic("default-error-handler-tests-1"); + consumerBuilder.subscriptionName("default-error-handler-tests-sub-1"); + }); PulsarContainerProperties pulsarContainerProperties = new PulsarContainerProperties(); PulsarRecordMessageListener messageListener = mock(PulsarRecordMessageListener.class); @@ -104,13 +102,13 @@ public class DefaultPulsarConsumerErrorHandlerTests implements PulsarTestContain @Test @SuppressWarnings("unchecked") void errorHandlingForRecordMessageListenerWithTransientError() throws Exception { - Map config = Map.of("topicNames", Collections.singleton("default-error-handler-tests-2"), - "subscriptionName", "default-error-handler-tests-sub-2"); - PulsarClient pulsarClient = PulsarClient.builder().serviceUrl(PulsarTestContainerSupport.getPulsarBrokerUrl()) .build(); DefaultPulsarConsumerFactory pulsarConsumerFactory = new DefaultPulsarConsumerFactory<>(pulsarClient, - config); + (consumerBuilder) -> { + consumerBuilder.topic("default-error-handler-tests-2"); + consumerBuilder.subscriptionName("default-error-handler-tests-sub-2"); + }); PulsarContainerProperties pulsarContainerProperties = new PulsarContainerProperties(); PulsarRecordMessageListener messageListener = mock(PulsarRecordMessageListener.class); @@ -150,13 +148,13 @@ public class DefaultPulsarConsumerErrorHandlerTests implements PulsarTestContain @Test @SuppressWarnings("unchecked") void everyOtherRecordThrowsNonTransientExceptionsRecordMessageListener() throws Exception { - Map config = Map.of("topicNames", Collections.singleton("default-error-handler-tests-3"), - "subscriptionName", "default-error-handler-tests-sub-3"); - PulsarClient pulsarClient = PulsarClient.builder().serviceUrl(PulsarTestContainerSupport.getPulsarBrokerUrl()) .build(); DefaultPulsarConsumerFactory pulsarConsumerFactory = new DefaultPulsarConsumerFactory<>(pulsarClient, - config); + (consumerBuilder) -> { + consumerBuilder.topic("default-error-handler-tests-3"); + consumerBuilder.subscriptionName("default-error-handler-tests-sub-3"); + }); PulsarContainerProperties pulsarContainerProperties = new PulsarContainerProperties(); PulsarRecordMessageListener messageListener = mock(PulsarRecordMessageListener.class); @@ -206,13 +204,13 @@ public class DefaultPulsarConsumerErrorHandlerTests implements PulsarTestContain @Test @SuppressWarnings("unchecked") void batchRecordListenerFirstOneOnlyErrorAndRecover() throws Exception { - Map config = Map.of("topicNames", Collections.singleton("default-error-handler-tests-4"), - "subscriptionName", "default-error-handler-tests-sub-4"); - PulsarClient pulsarClient = PulsarClient.builder().serviceUrl(PulsarTestContainerSupport.getPulsarBrokerUrl()) .build(); DefaultPulsarConsumerFactory pulsarConsumerFactory = new DefaultPulsarConsumerFactory<>(pulsarClient, - config); + (consumerBuilder) -> { + consumerBuilder.topic("default-error-handler-tests-4"); + consumerBuilder.subscriptionName("default-error-handler-tests-sub-4"); + }); PulsarContainerProperties pulsarContainerProperties = new PulsarContainerProperties(); pulsarContainerProperties.setMaxNumMessages(10); @@ -275,13 +273,13 @@ public class DefaultPulsarConsumerErrorHandlerTests implements PulsarTestContain @Test @SuppressWarnings("unchecked") void batchRecordListenerRecordFailsInTheMiddle() throws Exception { - Map config = Map.of("topicNames", Collections.singleton("default-error-handler-tests-5"), - "subscriptionName", "default-error-handler-tests-sub-5"); - PulsarClient pulsarClient = PulsarClient.builder().serviceUrl(PulsarTestContainerSupport.getPulsarBrokerUrl()) .build(); DefaultPulsarConsumerFactory pulsarConsumerFactory = new DefaultPulsarConsumerFactory<>(pulsarClient, - config); + (consumerBuilder) -> { + consumerBuilder.topic("default-error-handler-tests-5"); + consumerBuilder.subscriptionName("default-error-handler-tests-sub-5"); + }); PulsarContainerProperties pulsarContainerProperties = new PulsarContainerProperties(); pulsarContainerProperties.setMaxNumMessages(10); @@ -342,13 +340,13 @@ public class DefaultPulsarConsumerErrorHandlerTests implements PulsarTestContain @Test @SuppressWarnings("unchecked") void batchRecordListenerRecordFailsTwiceInTheMiddle() throws Exception { - Map config = Map.of("topicNames", Collections.singleton("default-error-handler-tests-6"), - "subscriptionName", "default-error-handler-tests-sub-6"); - PulsarClient pulsarClient = PulsarClient.builder().serviceUrl(PulsarTestContainerSupport.getPulsarBrokerUrl()) .build(); DefaultPulsarConsumerFactory pulsarConsumerFactory = new DefaultPulsarConsumerFactory<>(pulsarClient, - config); + (consumerBuilder) -> { + consumerBuilder.topic("default-error-handler-tests-6"); + consumerBuilder.subscriptionName("default-error-handler-tests-sub-6"); + }); PulsarContainerProperties pulsarContainerProperties = new PulsarContainerProperties(); pulsarContainerProperties.setMaxNumMessages(10); @@ -409,13 +407,13 @@ public class DefaultPulsarConsumerErrorHandlerTests implements PulsarTestContain @Test @SuppressWarnings("unchecked") void batchRecordListenerRecordFailsInTheMiddleButTransientError() throws Exception { - Map config = Map.of("topicNames", Collections.singleton("default-error-handler-tests-7"), - "subscriptionName", "default-error-handler-tests-sub-7"); - PulsarClient pulsarClient = PulsarClient.builder().serviceUrl(PulsarTestContainerSupport.getPulsarBrokerUrl()) .build(); DefaultPulsarConsumerFactory pulsarConsumerFactory = new DefaultPulsarConsumerFactory<>(pulsarClient, - config); + (consumerBuilder) -> { + consumerBuilder.topic("default-error-handler-tests-7"); + consumerBuilder.subscriptionName("default-error-handler-tests-sub-7"); + }); PulsarContainerProperties pulsarContainerProperties = new PulsarContainerProperties(); pulsarContainerProperties.setMaxNumMessages(10); @@ -475,13 +473,13 @@ public class DefaultPulsarConsumerErrorHandlerTests implements PulsarTestContain @Test @SuppressWarnings("unchecked") void batchListenerFailsTransientErrorFollowedByNonTransient() throws Exception { - Map config = Map.of("topicNames", Collections.singleton("default-error-handler-tests-8"), - "subscriptionName", "default-error-handler-tests-sub-8"); - PulsarClient pulsarClient = PulsarClient.builder().serviceUrl(PulsarTestContainerSupport.getPulsarBrokerUrl()) .build(); DefaultPulsarConsumerFactory pulsarConsumerFactory = new DefaultPulsarConsumerFactory<>(pulsarClient, - config); + (consumerBuilder) -> { + consumerBuilder.topic("default-error-handler-tests-8"); + consumerBuilder.subscriptionName("default-error-handler-tests-sub-8"); + }); PulsarContainerProperties pulsarContainerProperties = new PulsarContainerProperties(); pulsarContainerProperties.setMaxNumMessages(10); diff --git a/spring-pulsar/src/test/java/org/springframework/pulsar/listener/DefaultPulsarMessageListenerContainerTests.java b/spring-pulsar/src/test/java/org/springframework/pulsar/listener/DefaultPulsarMessageListenerContainerTests.java index 3ff1daf2..b0c13e47 100644 --- a/spring-pulsar/src/test/java/org/springframework/pulsar/listener/DefaultPulsarMessageListenerContainerTests.java +++ b/spring-pulsar/src/test/java/org/springframework/pulsar/listener/DefaultPulsarMessageListenerContainerTests.java @@ -26,9 +26,7 @@ import static org.mockito.Mockito.verify; import java.time.Duration; import java.util.ArrayList; -import java.util.Collections; import java.util.List; -import java.util.Map; import java.util.Set; import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; @@ -65,13 +63,13 @@ class DefaultPulsarMessageListenerContainerTests implements PulsarTestContainerS @Test void basicDefaultConsumer() throws Exception { - Map config = Map.of("topicNames", Collections.singleton("dpmlct-012"), "subscriptionName", - "dpmlct-sb-012"); - PulsarClient pulsarClient = PulsarClient.builder().serviceUrl(PulsarTestContainerSupport.getPulsarBrokerUrl()) .build(); DefaultPulsarConsumerFactory pulsarConsumerFactory = new DefaultPulsarConsumerFactory<>(pulsarClient, - config); + (consumerBuilder) -> { + consumerBuilder.topic("dpmlct-012"); + consumerBuilder.subscriptionName("dpmlct-sb-012"); + }); CountDownLatch latch = new CountDownLatch(1); PulsarContainerProperties pulsarContainerProperties = new PulsarContainerProperties(); pulsarContainerProperties @@ -93,13 +91,13 @@ class DefaultPulsarMessageListenerContainerTests implements PulsarTestContainerS @Disabled @Test void containerPauseAndResumeFeatureUsingWaitAndNotify() throws Exception { - Map config = Map.of("topicNames", Collections.singleton("containerPauseResumeWaitNotify-topic"), - "subscriptionName", "containerPauseResumeWaitNotify-sub"); - PulsarClient pulsarClient = PulsarClient.builder().serviceUrl(PulsarTestContainerSupport.getPulsarBrokerUrl()) .build(); DefaultPulsarConsumerFactory pulsarConsumerFactory = new DefaultPulsarConsumerFactory<>(pulsarClient, - config); + (consumerBuilder) -> { + consumerBuilder.topic("containerPauseResumeWaitNotify-topic"); + consumerBuilder.subscriptionName("containerPauseResumeWaitNotify-sub"); + }); PulsarContainerProperties pulsarContainerProperties = new PulsarContainerProperties(); pulsarContainerProperties.setMessageListener((PulsarRecordMessageListener) (consumer, msg) -> { }); @@ -162,13 +160,14 @@ class DefaultPulsarMessageListenerContainerTests implements PulsarTestContainerS @Test void subscriptionInitialPositionEarliest() throws Exception { - Map config = Map.of("topicNames", Collections.singleton("dpmlct-013"), "subscriptionName", - "dpmlct-sb-013", "subscriptionInitialPosition", SubscriptionInitialPosition.Earliest); - PulsarClient pulsarClient = PulsarClient.builder().serviceUrl(PulsarTestContainerSupport.getPulsarBrokerUrl()) .build(); DefaultPulsarConsumerFactory pulsarConsumerFactory = new DefaultPulsarConsumerFactory<>(pulsarClient, - config); + (consumerBuilder) -> { + consumerBuilder.topic("dpmlct-013"); + consumerBuilder.subscriptionName("dpmlct-sb-013"); + consumerBuilder.subscriptionInitialPosition(SubscriptionInitialPosition.Earliest); + }); CountDownLatch latch = new CountDownLatch(5); PulsarContainerProperties pulsarContainerProperties = new PulsarContainerProperties(); pulsarContainerProperties @@ -192,13 +191,13 @@ class DefaultPulsarMessageListenerContainerTests implements PulsarTestContainerS @Test void subscriptionInitialPositionDefaultLatest() throws Exception { - Map config = Map.of("topicNames", Collections.singleton("dpmlct-014"), "subscriptionName", - "dpmlct-sb-014"); - PulsarClient pulsarClient = PulsarClient.builder().serviceUrl(PulsarTestContainerSupport.getPulsarBrokerUrl()) .build(); DefaultPulsarConsumerFactory pulsarConsumerFactory = new DefaultPulsarConsumerFactory<>(pulsarClient, - config); + (consumerBuilder) -> { + consumerBuilder.topic("dpmlct-014"); + consumerBuilder.subscriptionName("dpmlct-sb-014"); + }); PulsarContainerProperties pulsarContainerProperties = new PulsarContainerProperties(); List messages = new ArrayList<>(); pulsarContainerProperties.setMessageListener( @@ -225,15 +224,16 @@ class DefaultPulsarMessageListenerContainerTests implements PulsarTestContainerS @Test void negativeAckRedeliveryBackoff() throws Exception { - RedeliveryBackoff redeliveryBackoff = MultiplierRedeliveryBackoff.builder().minDelayMs(1000) - .maxDelayMs(5 * 1000).build(); - Map config = Map.of("topicNames", Collections.singleton("dpmlct-015"), "subscriptionName", - "dpmlct-sb-015", "negativeAckRedeliveryBackoff", redeliveryBackoff); - PulsarClient pulsarClient = PulsarClient.builder().serviceUrl(PulsarTestContainerSupport.getPulsarBrokerUrl()) .build(); + RedeliveryBackoff redeliveryBackoff = MultiplierRedeliveryBackoff.builder().minDelayMs(1000) + .maxDelayMs(5 * 1000).build(); DefaultPulsarConsumerFactory pulsarConsumerFactory = spy( - new DefaultPulsarConsumerFactory<>(pulsarClient, config)); + new DefaultPulsarConsumerFactory<>(pulsarClient, (consumerBuilder) -> { + consumerBuilder.topic("dpmlct-015"); + consumerBuilder.subscriptionName("dpmlct-sb-015"); + consumerBuilder.negativeAckRedeliveryBackoff(redeliveryBackoff); + })); CountDownLatch latch = new CountDownLatch(10); PulsarContainerProperties pulsarContainerProperties = new PulsarContainerProperties(); pulsarContainerProperties.setMessageListener((PulsarRecordMessageListener) (consumer, msg) -> { @@ -271,16 +271,17 @@ class DefaultPulsarMessageListenerContainerTests implements PulsarTestContainerS @Test void deadLetterPolicyDefault() throws Exception { - DeadLetterPolicy deadLetterPolicy = DeadLetterPolicy.builder().maxRedeliverCount(1) - .deadLetterTopic("dpmlct-016-dlq-topic").build(); - Map config = Map.of("topicNames", Collections.singleton("dpmlct-016"), "subscriptionName", - "dpmlct-sb-016", "negativeAckRedeliveryDelayMicros", TimeUnit.SECONDS.toMicros(1), "deadLetterPolicy", - deadLetterPolicy); - PulsarClient pulsarClient = PulsarClient.builder().serviceUrl(PulsarTestContainerSupport.getPulsarBrokerUrl()) .build(); + DeadLetterPolicy deadLetterPolicy = DeadLetterPolicy.builder().maxRedeliverCount(1) + .deadLetterTopic("dpmlct-016-dlq-topic").build(); DefaultPulsarConsumerFactory pulsarConsumerFactory = new DefaultPulsarConsumerFactory<>(pulsarClient, - config); + (consumerBuilder) -> { + consumerBuilder.topic("dpmlct-016"); + consumerBuilder.subscriptionName("dpmlct-sb-016"); + consumerBuilder.negativeAckRedeliveryDelay(1L, TimeUnit.SECONDS); + consumerBuilder.deadLetterPolicy(deadLetterPolicy); + }); CountDownLatch dlqLatch = new CountDownLatch(1); CountDownLatch latch = new CountDownLatch(6); @@ -326,16 +327,17 @@ class DefaultPulsarMessageListenerContainerTests implements PulsarTestContainerS @Test void deadLetterPolicyCustom() throws Exception { - DeadLetterPolicy deadLetterPolicy = DeadLetterPolicy.builder().maxRedeliverCount(5).deadLetterTopic("dlq-topic") - .build(); - Map config = Map.of("topicNames", Collections.singleton("dpmlct-017"), "subscriptionName", - "dpmlct-sb-016", "negativeAckRedeliveryDelayMicros", TimeUnit.SECONDS.toMicros(1), "deadLetterPolicy", - deadLetterPolicy); - PulsarClient pulsarClient = PulsarClient.builder().serviceUrl(PulsarTestContainerSupport.getPulsarBrokerUrl()) .build(); + DeadLetterPolicy deadLetterPolicy = DeadLetterPolicy.builder().maxRedeliverCount(5).deadLetterTopic("dlq-topic") + .build(); DefaultPulsarConsumerFactory pulsarConsumerFactory = new DefaultPulsarConsumerFactory<>(pulsarClient, - config); + (consumerBuilder) -> { + consumerBuilder.topic("dpmlct-017"); + consumerBuilder.subscriptionName("dpmlct-sb-017"); + consumerBuilder.negativeAckRedeliveryDelay(1L, TimeUnit.SECONDS); + consumerBuilder.deadLetterPolicy(deadLetterPolicy); + }); CountDownLatch dlqLatch = new CountDownLatch(1); CountDownLatch latch = new CountDownLatch(6); diff --git a/spring-pulsar/src/test/java/org/springframework/pulsar/listener/PulsarListenerTests.java b/spring-pulsar/src/test/java/org/springframework/pulsar/listener/PulsarListenerTests.java index 86321486..472d73f1 100644 --- a/spring-pulsar/src/test/java/org/springframework/pulsar/listener/PulsarListenerTests.java +++ b/spring-pulsar/src/test/java/org/springframework/pulsar/listener/PulsarListenerTests.java @@ -22,7 +22,6 @@ import static org.assertj.core.api.Assertions.assertThatThrownBy; import java.nio.charset.StandardCharsets; import java.time.Duration; import java.util.ArrayList; -import java.util.HashMap; import java.util.List; import java.util.Objects; import java.util.Properties; @@ -118,7 +117,7 @@ public class PulsarListenerTests implements PulsarTestContainerSupport { @Bean public PulsarConsumerFactory pulsarConsumerFactory(PulsarClient pulsarClient) { - return new DefaultPulsarConsumerFactory<>(pulsarClient, new HashMap<>()); + return new DefaultPulsarConsumerFactory<>(pulsarClient, null); } @Bean @@ -1085,9 +1084,7 @@ public class PulsarListenerTests implements PulsarTestContainerSupport { @Bean public ConsumerBuilderCustomizer myCustomizer() { - return cb -> { - cb.subscriptionName("test-changed-subscription-name"); - }; + return cb -> cb.subscriptionName("test-changed-subscription-name"); } } diff --git a/spring-pulsar/src/test/java/org/springframework/pulsar/observation/ObservationIntegrationTests.java b/spring-pulsar/src/test/java/org/springframework/pulsar/observation/ObservationIntegrationTests.java index 35c85e96..c74ed8fd 100644 --- a/spring-pulsar/src/test/java/org/springframework/pulsar/observation/ObservationIntegrationTests.java +++ b/spring-pulsar/src/test/java/org/springframework/pulsar/observation/ObservationIntegrationTests.java @@ -18,7 +18,6 @@ package org.springframework.pulsar.observation; import static org.assertj.core.api.Assertions.assertThat; -import java.util.Collections; import java.util.HashMap; import java.util.List; import java.util.Map; @@ -147,7 +146,7 @@ public class ObservationIntegrationTests extends SampleTestRunner implements Pul @Bean public PulsarConsumerFactory pulsarConsumerFactory(PulsarClient pulsarClient) { - return new DefaultPulsarConsumerFactory<>(pulsarClient, Collections.emptyMap()); + return new DefaultPulsarConsumerFactory<>(pulsarClient, null); } @Bean diff --git a/spring-pulsar/src/test/java/org/springframework/pulsar/observation/ObservationTests.java b/spring-pulsar/src/test/java/org/springframework/pulsar/observation/ObservationTests.java index 5bdf0bd9..dd19949f 100644 --- a/spring-pulsar/src/test/java/org/springframework/pulsar/observation/ObservationTests.java +++ b/spring-pulsar/src/test/java/org/springframework/pulsar/observation/ObservationTests.java @@ -20,7 +20,6 @@ import static org.assertj.core.api.Assertions.assertThat; import static org.awaitility.Awaitility.await; import java.util.Arrays; -import java.util.Collections; import java.util.Deque; import java.util.List; import java.util.concurrent.CountDownLatch; @@ -197,7 +196,7 @@ public class ObservationTests implements PulsarTestContainerSupport { @Bean PulsarConsumerFactory pulsarConsumerFactory(PulsarClient pulsarClient) { - return new DefaultPulsarConsumerFactory<>(pulsarClient, Collections.emptyMap()); + return new DefaultPulsarConsumerFactory<>(pulsarClient, null); } @Bean