Use builder to autoconfigure PulsarConsumerFactory (#399)
- Move consumer props 'toMap' into PulsarBinderUtils
This commit is contained in:
Chris Bono
@@ -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<String, Object> 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));
|
||||
};
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -106,10 +106,6 @@ public class PulsarProperties {
|
||||
return this.defaults;
|
||||
}
|
||||
|
||||
public Map<String, Object> buildConsumerProperties() {
|
||||
return new HashMap<>(this.consumer.buildProperties());
|
||||
}
|
||||
|
||||
public Map<String, Object> buildAdminProperties() {
|
||||
return new HashMap<>(this.admin.buildProperties());
|
||||
}
|
||||
|
||||
@@ -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<String> producerFactory = mock(PulsarProducerFactory.class);
|
||||
contextRunner.withBean("customPulsarProducerFactory", PulsarProducerFactory.class, () -> producerFactory)
|
||||
.run((context) -> assertThat(context).hasNotFailed().getBean(PulsarProducerFactory.class)
|
||||
.isSameAs(producerFactory));
|
||||
}
|
||||
SchemaResolverCustomizer<DefaultSchemaResolver> 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<String> 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<String> 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<String> 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<DefaultSchemaResolver> 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<String> 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<String> 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<Object>::getContainerProperties)
|
||||
.hasFieldOrPropertyWithValue("schemaResolver", schemaResolver)
|
||||
.hasFieldOrPropertyWithValue("topicResolver", topicResolver)));
|
||||
}
|
||||
|
||||
@Test
|
||||
void customPulsarListenerAnnotationBeanPostProcessorIsRespected() {
|
||||
PulsarListenerAnnotationBeanPostProcessor<String> 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<?, PulsarContainerProperties> 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<Object>::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<Object>::getContainerProperties)
|
||||
.hasFieldOrPropertyWithValue("schemaResolver", schemaResolver)
|
||||
.hasFieldOrPropertyWithValue("topicResolver", topicResolver)));
|
||||
}
|
||||
|
||||
@Test
|
||||
void customPulsarListenerAnnotationBeanPostProcessorIsRespected() {
|
||||
PulsarListenerAnnotationBeanPostProcessor<String> 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<?, PulsarContainerProperties> properties = assertThat(context)
|
||||
.hasNotFailed().getBean(ConcurrentPulsarListenerContainerFactory.class)
|
||||
.extracting(ConcurrentPulsarListenerContainerFactory<Object>::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;
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
@@ -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<String, String> props = new HashMap<>();
|
||||
var props = new HashMap<String, String>();
|
||||
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<String, Object> 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));
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
@@ -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<String, Object> 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<String, Object> {
|
||||
|
||||
<V> java.util.function.Consumer<V> in(String key) {
|
||||
|
||||
@@ -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);
|
||||
|
||||
@@ -98,7 +98,7 @@ public class PulsarBinderConfigurationPropertiesTests {
|
||||
props.put("spring.cloud.stream.pulsar.binder.consumer.receiver-queue-size", "1");
|
||||
|
||||
bind(props);
|
||||
Map<String, Object> consumerProps = properties.getConsumer().buildProperties();
|
||||
Map<String, Object> consumerProps = PulsarBinderUtils.convertConsumerPropertiesToMap(properties.getConsumer());
|
||||
|
||||
// Verify that the props can be loaded in a ConsumerBuilder
|
||||
assertThatNoException().isThrownBy(() -> ConfigurationDataUtils.loadData(consumerProps,
|
||||
|
||||
@@ -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<String>) pulsarProperties.getConsumer()
|
||||
.toConsumerBuilderCustomizer();
|
||||
return new TrackingConsumerFactory(pulsarClient, customizer);
|
||||
}
|
||||
|
||||
}
|
||||
@@ -683,8 +686,8 @@ class PulsarBinderIntegrationTests implements PulsarTestContainerSupport {
|
||||
|
||||
List<org.apache.pulsar.client.api.Consumer<String>> consumersCreated = new ArrayList<>();
|
||||
|
||||
TrackingConsumerFactory(PulsarClient pulsarClient, Map<String, Object> consumerConfig) {
|
||||
super(pulsarClient, consumerConfig);
|
||||
TrackingConsumerFactory(PulsarClient pulsarClient, ConsumerBuilderCustomizer<String> defaultConsumerConfig) {
|
||||
super(pulsarClient, defaultConsumerConfig);
|
||||
}
|
||||
|
||||
@Override
|
||||
|
||||
@@ -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.<String, Object>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());
|
||||
|
||||
@@ -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<String, String> map) {
|
||||
ConfigurationPropertySource source = new MapConfigurationPropertySource(map);
|
||||
new Binder(source).bind("spring.pulsar.consumer", Bindable.ofInstance(this.properties));
|
||||
}
|
||||
|
||||
@Test
|
||||
void consumerPropertiesToMap() {
|
||||
Map<String, String> 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<String, Object> 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);
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
@@ -96,7 +96,8 @@ public class PulsarExtendedBindingPropertiesTests {
|
||||
bind(props);
|
||||
|
||||
assertThat(properties.getBindings()).containsOnlyKeys("my-foo");
|
||||
Map<String, Object> consumerProps = properties.getExtendedConsumerProperties("my-foo").buildProperties();
|
||||
Map<String, Object> 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);
|
||||
|
||||
|
||||
@@ -18,7 +18,6 @@ package org.springframework.pulsar.core;
|
||||
|
||||
import java.util.Collection;
|
||||
import java.util.Collections;
|
||||
import java.util.HashMap;
|
||||
import java.util.HashSet;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
@@ -30,6 +29,7 @@ import org.apache.pulsar.client.api.ConsumerBuilder;
|
||||
import org.apache.pulsar.client.api.PulsarClient;
|
||||
import org.apache.pulsar.client.api.PulsarClientException;
|
||||
import org.apache.pulsar.client.api.Schema;
|
||||
import org.apache.pulsar.client.impl.ConsumerBuilderImpl;
|
||||
|
||||
import org.springframework.lang.Nullable;
|
||||
import org.springframework.util.CollectionUtils;
|
||||
@@ -41,22 +41,25 @@ import org.springframework.util.CollectionUtils;
|
||||
* @author Soby Chacko
|
||||
* @author Alexander Preuß
|
||||
* @author Christophe Bornet
|
||||
* @author Chris Bono
|
||||
*/
|
||||
public class DefaultPulsarConsumerFactory<T> implements PulsarConsumerFactory<T> {
|
||||
|
||||
private final Map<String, Object> consumerConfig;
|
||||
|
||||
private final PulsarClient pulsarClient;
|
||||
|
||||
@Nullable
|
||||
private final ConsumerBuilderCustomizer<T> defaultConfigCustomizer;
|
||||
|
||||
/**
|
||||
* Construct a consumer factory instance.
|
||||
* @param pulsarClient the client used to consume
|
||||
* @param consumerConfig default configuration to apply to the created consumer or
|
||||
* empty map to use no default configuration
|
||||
* @param defaultConfigCustomizer the default configuration to apply to the consumers
|
||||
* or null to use no default configuration
|
||||
*/
|
||||
public DefaultPulsarConsumerFactory(PulsarClient pulsarClient, Map<String, Object> consumerConfig) {
|
||||
public DefaultPulsarConsumerFactory(PulsarClient pulsarClient,
|
||||
ConsumerBuilderCustomizer<T> defaultConfigCustomizer) {
|
||||
this.pulsarClient = pulsarClient;
|
||||
this.consumerConfig = Collections.unmodifiableMap(consumerConfig);
|
||||
this.defaultConfigCustomizer = defaultConfigCustomizer;
|
||||
}
|
||||
|
||||
@Override
|
||||
@@ -72,25 +75,35 @@ public class DefaultPulsarConsumerFactory<T> implements PulsarConsumerFactory<T>
|
||||
@Nullable List<ConsumerBuilderCustomizer<T>> customizers) throws PulsarClientException {
|
||||
Objects.requireNonNull(schema, "Schema must be specified");
|
||||
ConsumerBuilder<T> consumerBuilder = this.pulsarClient.newConsumer(schema);
|
||||
Map<String, Object> config = new HashMap<>(this.consumerConfig);
|
||||
if (topics != null) {
|
||||
config.put("topicNames", new HashSet<>(topics));
|
||||
|
||||
// Apply the default config customizer (preserve the topic)
|
||||
if (this.defaultConfigCustomizer != null) {
|
||||
this.defaultConfigCustomizer.customize(consumerBuilder);
|
||||
}
|
||||
if (metadataProperties != null) {
|
||||
config.put("properties", new TreeMap<>(metadataProperties));
|
||||
if (topics != null) {
|
||||
replaceTopicsOnBuilder(consumerBuilder, topics);
|
||||
}
|
||||
if (subscriptionName != null) {
|
||||
config.put("subscriptionName", subscriptionName);
|
||||
consumerBuilder.subscriptionName(subscriptionName);
|
||||
}
|
||||
if (metadataProperties != null) {
|
||||
replaceMetadataPropertiesOnBuilder(consumerBuilder, metadataProperties);
|
||||
}
|
||||
ConsumerBuilderConfigurationUtil.loadConf(consumerBuilder, config);
|
||||
if (!CollectionUtils.isEmpty(customizers)) {
|
||||
customizers.forEach(customizer -> customizer.customize(consumerBuilder));
|
||||
}
|
||||
return consumerBuilder.subscribe();
|
||||
}
|
||||
|
||||
public Map<String, Object> getConsumerConfig() {
|
||||
return this.consumerConfig;
|
||||
private void replaceTopicsOnBuilder(ConsumerBuilder<T> builder, Collection<String> topics) {
|
||||
var builderImpl = (ConsumerBuilderImpl<T>) builder;
|
||||
builderImpl.getConf().setTopicNames(new HashSet<>(topics));
|
||||
}
|
||||
|
||||
private void replaceMetadataPropertiesOnBuilder(ConsumerBuilder<T> builder,
|
||||
Map<String, String> metadataProperties) {
|
||||
var builderImpl = (ConsumerBuilderImpl<T>) builder;
|
||||
builderImpl.getConf().setProperties(new TreeMap<>(metadataProperties));
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
@@ -85,10 +85,4 @@ public interface PulsarConsumerFactory<T> {
|
||||
@Nullable Map<String, String> metadataProperties, @Nullable List<ConsumerBuilderCustomizer<T>> customizers)
|
||||
throws PulsarClientException;
|
||||
|
||||
/**
|
||||
* Return the configuration options to use when creating consumers.
|
||||
* @return the configuration options
|
||||
*/
|
||||
Map<String, Object> getConsumerConfig();
|
||||
|
||||
}
|
||||
|
||||
@@ -30,9 +30,7 @@ import static org.mockito.Mockito.verify;
|
||||
|
||||
import java.time.Duration;
|
||||
import java.util.ArrayList;
|
||||
import java.util.Collections;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
import java.util.concurrent.CountDownLatch;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
import java.util.concurrent.atomic.AtomicInteger;
|
||||
@@ -61,14 +59,10 @@ class ConsumerAcknowledgmentTests implements PulsarTestContainerSupport {
|
||||
|
||||
@Test
|
||||
void testRecordAck() throws Exception {
|
||||
Map<String, Object> config = Map.of("topicNames", Collections.singleton("cons-ack-tests-011"),
|
||||
"subscriptionName", "cons-ack-tests-sb-011");
|
||||
|
||||
PulsarClient pulsarClient = PulsarClient.builder().serviceUrl(PulsarTestContainerSupport.getPulsarBrokerUrl())
|
||||
.build();
|
||||
DefaultPulsarConsumerFactory<String> pulsarConsumerFactory = spy(
|
||||
new DefaultPulsarConsumerFactory<>(pulsarClient, config));
|
||||
|
||||
DefaultPulsarConsumerFactory<String> pulsarConsumerFactory = spy(new DefaultPulsarConsumerFactory<>(
|
||||
pulsarClient, defaultConfig("cons-ack-tests-011", "cons-ack-tests-sb-011")));
|
||||
PulsarContainerProperties pulsarContainerProperties = new PulsarContainerProperties();
|
||||
pulsarContainerProperties.setMessageListener((PulsarRecordMessageListener<?>) (consumer, msg) -> {
|
||||
});
|
||||
@@ -98,13 +92,10 @@ class ConsumerAcknowledgmentTests implements PulsarTestContainerSupport {
|
||||
|
||||
@Test
|
||||
void testBatchAck() throws Exception {
|
||||
Map<String, Object> config = Map.of("topicNames", Collections.singleton("cons-ack-tests-012"),
|
||||
"subscriptionName", "cons-ack-tests-sb-012");
|
||||
PulsarClient pulsarClient = PulsarClient.builder().serviceUrl(PulsarTestContainerSupport.getPulsarBrokerUrl())
|
||||
.build();
|
||||
DefaultPulsarConsumerFactory<String> pulsarConsumerFactory = spy(
|
||||
new DefaultPulsarConsumerFactory<>(pulsarClient, config));
|
||||
|
||||
DefaultPulsarConsumerFactory<String> pulsarConsumerFactory = spy(new DefaultPulsarConsumerFactory<>(
|
||||
pulsarClient, defaultConfig("cons-ack-tests-012", "cons-ack-tests-sb-012")));
|
||||
PulsarContainerProperties pulsarContainerProperties = new PulsarContainerProperties();
|
||||
CountDownLatch latch = new CountDownLatch(10);
|
||||
pulsarContainerProperties
|
||||
@@ -131,16 +122,12 @@ class ConsumerAcknowledgmentTests implements PulsarTestContainerSupport {
|
||||
|
||||
@Test
|
||||
void testBatchAckButSomeRecordsFail() throws Exception {
|
||||
Map<String, Object> config = Map.of("topicNames", Collections.singleton("cons-ack-tests-013"),
|
||||
"subscriptionName", "cons-ack-tests-sb-013");
|
||||
PulsarClient pulsarClient = PulsarClient.builder().serviceUrl(PulsarTestContainerSupport.getPulsarBrokerUrl())
|
||||
.build();
|
||||
DefaultPulsarConsumerFactory<String> pulsarConsumerFactory = spy(
|
||||
new DefaultPulsarConsumerFactory<>(pulsarClient, config));
|
||||
|
||||
DefaultPulsarConsumerFactory<String> pulsarConsumerFactory = spy(new DefaultPulsarConsumerFactory<>(
|
||||
pulsarClient, defaultConfig("cons-ack-tests-013", "cons-ack-tests-sb-013")));
|
||||
PulsarContainerProperties pulsarContainerProperties = new PulsarContainerProperties();
|
||||
CountDownLatch latch = new CountDownLatch(10);
|
||||
|
||||
pulsarContainerProperties.setMessageListener((PulsarRecordMessageListener<?>) (consumer, msg) -> {
|
||||
latch.countDown();
|
||||
if (latch.getCount() % 2 == 0) {
|
||||
@@ -194,22 +181,17 @@ class ConsumerAcknowledgmentTests implements PulsarTestContainerSupport {
|
||||
@Test
|
||||
@SuppressWarnings("unchecked")
|
||||
void testManualAckForRecordListener() throws Exception {
|
||||
Map<String, Object> config = Map.of("topicNames", Collections.singleton("cons-ack-tests-014"),
|
||||
"subscriptionName", "cons-ack-tests-sb-014");
|
||||
PulsarClient pulsarClient = PulsarClient.builder().serviceUrl(PulsarTestContainerSupport.getPulsarBrokerUrl())
|
||||
.build();
|
||||
DefaultPulsarConsumerFactory<String> pulsarConsumerFactory = spy(
|
||||
new DefaultPulsarConsumerFactory<>(pulsarClient, config));
|
||||
|
||||
DefaultPulsarConsumerFactory<String> pulsarConsumerFactory = spy(new DefaultPulsarConsumerFactory<>(
|
||||
pulsarClient, defaultConfig("cons-ack-tests-014", "cons-ack-tests-sb-014")));
|
||||
PulsarContainerProperties pulsarContainerProperties = new PulsarContainerProperties();
|
||||
List<Acknowledgement> acksObjects = new ArrayList<>();
|
||||
PulsarAcknowledgingMessageListener<?> pulsarAcknowledgingMessageListener = (consumer, msg, acknowledgement) -> {
|
||||
acksObjects.add(acknowledgement);
|
||||
acknowledgement.acknowledge();
|
||||
};
|
||||
|
||||
pulsarContainerProperties.setMessageListener(pulsarAcknowledgingMessageListener);
|
||||
|
||||
pulsarContainerProperties.setSchema(Schema.STRING);
|
||||
pulsarContainerProperties.setAckMode(AckMode.MANUAL);
|
||||
DefaultPulsarMessageListenerContainer<String> container = new DefaultPulsarMessageListenerContainer<>(
|
||||
@@ -243,25 +225,20 @@ class ConsumerAcknowledgmentTests implements PulsarTestContainerSupport {
|
||||
@Test
|
||||
@SuppressWarnings("unchecked")
|
||||
void testBatchAckForBatchListener() throws Exception {
|
||||
Map<String, Object> config = Map.of("topicNames", Collections.singleton("cons-ack-tests-015"),
|
||||
"subscriptionName", "cons-ack-tests-sb-015");
|
||||
PulsarClient pulsarClient = PulsarClient.builder().serviceUrl(PulsarTestContainerSupport.getPulsarBrokerUrl())
|
||||
.build();
|
||||
DefaultPulsarConsumerFactory<String> pulsarConsumerFactory = spy(
|
||||
new DefaultPulsarConsumerFactory<>(pulsarClient, config));
|
||||
|
||||
DefaultPulsarConsumerFactory<String> pulsarConsumerFactory = spy(new DefaultPulsarConsumerFactory<>(
|
||||
pulsarClient, defaultConfig("cons-ack-tests-015", "cons-ack-tests-sb-015")));
|
||||
PulsarContainerProperties pulsarContainerProperties = new PulsarContainerProperties();
|
||||
pulsarContainerProperties.setMaxNumMessages(10);
|
||||
pulsarContainerProperties.setBatchTimeoutMillis(60_000);
|
||||
pulsarContainerProperties.setBatchListener(true);
|
||||
CountDownLatch latch = new CountDownLatch(1);
|
||||
PulsarBatchMessageListener<?> pulsarBatchMessageListener = mock(PulsarBatchMessageListener.class);
|
||||
|
||||
doAnswer(invocation -> {
|
||||
latch.countDown();
|
||||
return null;
|
||||
}).when(pulsarBatchMessageListener).received(any(Consumer.class), any(List.class));
|
||||
|
||||
pulsarContainerProperties.setMessageListener(pulsarBatchMessageListener);
|
||||
pulsarContainerProperties.setSchema(Schema.STRING);
|
||||
DefaultPulsarMessageListenerContainer<String> container = new DefaultPulsarMessageListenerContainer<>(
|
||||
@@ -286,25 +263,20 @@ class ConsumerAcknowledgmentTests implements PulsarTestContainerSupport {
|
||||
@Test
|
||||
@SuppressWarnings("unchecked")
|
||||
void testBatchNackForEntireBatchWhenUsingBatchListener() throws Exception {
|
||||
Map<String, Object> config = Map.of("topicNames", Collections.singleton("cons-ack-tests-016"),
|
||||
"subscriptionName", "cons-ack-tests-sb-016");
|
||||
PulsarClient pulsarClient = PulsarClient.builder().serviceUrl(PulsarTestContainerSupport.getPulsarBrokerUrl())
|
||||
.build();
|
||||
DefaultPulsarConsumerFactory<String> pulsarConsumerFactory = spy(
|
||||
new DefaultPulsarConsumerFactory<>(pulsarClient, config));
|
||||
|
||||
DefaultPulsarConsumerFactory<String> pulsarConsumerFactory = spy(new DefaultPulsarConsumerFactory<>(
|
||||
pulsarClient, defaultConfig("cons-ack-tests-016", "cons-ack-tests-sb-016")));
|
||||
PulsarContainerProperties pulsarContainerProperties = new PulsarContainerProperties();
|
||||
pulsarContainerProperties.setMaxNumMessages(10);
|
||||
pulsarContainerProperties.setBatchTimeoutMillis(60_000);
|
||||
pulsarContainerProperties.setBatchListener(true);
|
||||
PulsarBatchMessageListener<?> pulsarBatchMessageListener = mock(PulsarBatchMessageListener.class);
|
||||
CountDownLatch latch = new CountDownLatch(1);
|
||||
|
||||
doAnswer(invocation -> {
|
||||
latch.countDown();
|
||||
throw new RuntimeException();
|
||||
}).when(pulsarBatchMessageListener).received(any(Consumer.class), any(List.class));
|
||||
|
||||
pulsarContainerProperties.setMessageListener(pulsarBatchMessageListener);
|
||||
pulsarContainerProperties.setSchema(Schema.STRING);
|
||||
DefaultPulsarMessageListenerContainer<String> container = new DefaultPulsarMessageListenerContainer<>(
|
||||
@@ -329,18 +301,14 @@ class ConsumerAcknowledgmentTests implements PulsarTestContainerSupport {
|
||||
|
||||
@Test
|
||||
void messagesAreProperlyAckdOnContainerStopBeforeExitingListenerThread() throws Exception {
|
||||
Map<String, Object> config = Map.of("topicNames", Collections.singleton("duplicate-message-test"),
|
||||
"subscriptionName", "duplicate-sub-1");
|
||||
PulsarClient pulsarClient = PulsarClient.builder().serviceUrl(PulsarTestContainerSupport.getPulsarBrokerUrl())
|
||||
.build();
|
||||
DefaultPulsarConsumerFactory<String> pulsarConsumerFactory = new DefaultPulsarConsumerFactory<>(pulsarClient,
|
||||
config);
|
||||
|
||||
defaultConfig("duplicate-message-test", "duplicate-sub-1"));
|
||||
PulsarContainerProperties pulsarContainerProperties = new PulsarContainerProperties();
|
||||
AtomicInteger counter1 = new AtomicInteger(0);
|
||||
pulsarContainerProperties.setMessageListener((PulsarRecordMessageListener<?>) (consumer, msg) -> {
|
||||
counter1.getAndIncrement();
|
||||
});
|
||||
pulsarContainerProperties
|
||||
.setMessageListener((PulsarRecordMessageListener<?>) (consumer, msg) -> counter1.getAndIncrement());
|
||||
pulsarContainerProperties.setSchema(Schema.STRING);
|
||||
DefaultPulsarMessageListenerContainer<String> container1 = new DefaultPulsarMessageListenerContainer<>(
|
||||
pulsarConsumerFactory, pulsarContainerProperties);
|
||||
@@ -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<String> container2 = new DefaultPulsarMessageListenerContainer<>(
|
||||
pulsarConsumerFactory, pulsarContainerProperties);
|
||||
@@ -382,4 +349,11 @@ class ConsumerAcknowledgmentTests implements PulsarTestContainerSupport {
|
||||
pulsarClient.close();
|
||||
}
|
||||
|
||||
private <T> ConsumerBuilderCustomizer<T> defaultConfig(String topicName, String subscriptionName) {
|
||||
return (consumerBuilder) -> {
|
||||
consumerBuilder.topic(topicName);
|
||||
consumerBuilder.subscriptionName(subscriptionName);
|
||||
};
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
@@ -23,7 +23,6 @@ import static org.mockito.Mockito.inOrder;
|
||||
import static org.mockito.Mockito.mock;
|
||||
|
||||
import java.util.Collections;
|
||||
import java.util.HashMap;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
|
||||
@@ -72,7 +71,7 @@ class DefaultPulsarConsumerFactoryTests implements PulsarTestContainerSupport {
|
||||
|
||||
@BeforeEach
|
||||
void createConsumerFactory() {
|
||||
consumerFactory = new DefaultPulsarConsumerFactory<>(pulsarClient, Collections.emptyMap());
|
||||
consumerFactory = new DefaultPulsarConsumerFactory<>(pulsarClient, null);
|
||||
}
|
||||
|
||||
@Test
|
||||
@@ -170,11 +169,11 @@ class DefaultPulsarConsumerFactoryTests implements PulsarTestContainerSupport {
|
||||
|
||||
@BeforeEach
|
||||
void createConsumerFactory() {
|
||||
Map<String, Object> defaultConfig = new HashMap<>();
|
||||
defaultConfig.put("topicNames", Collections.singleton(defaultTopic));
|
||||
defaultConfig.put("properties", defaultMetadataProperties);
|
||||
defaultConfig.put("subscriptionName", defaultSubscription);
|
||||
consumerFactory = new DefaultPulsarConsumerFactory<>(pulsarClient, defaultConfig);
|
||||
consumerFactory = new DefaultPulsarConsumerFactory<>(pulsarClient, (consumerBuilder) -> {
|
||||
consumerBuilder.topic(defaultTopic);
|
||||
consumerBuilder.subscriptionName(defaultSubscription);
|
||||
consumerBuilder.properties(defaultMetadataProperties);
|
||||
});
|
||||
}
|
||||
|
||||
@Test
|
||||
|
||||
@@ -19,8 +19,6 @@ package org.springframework.pulsar.core;
|
||||
import static org.assertj.core.api.Assertions.assertThat;
|
||||
|
||||
import java.io.Serial;
|
||||
import java.util.Collections;
|
||||
import java.util.Map;
|
||||
import java.util.concurrent.CountDownLatch;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
|
||||
@@ -48,18 +46,16 @@ class FailoverConsumerTests implements PulsarTestContainerSupport {
|
||||
void testFailOverConsumersOnPartitionedTopic() throws Exception {
|
||||
PulsarAdmin admin = PulsarAdmin.builder().serviceHttpUrl(PulsarTestContainerSupport.getHttpServiceUrl())
|
||||
.build();
|
||||
|
||||
String topicName = "persistent://public/default/my-part-topic-1";
|
||||
int numPartitions = 3;
|
||||
admin.topics().createPartitionedTopic(topicName, numPartitions);
|
||||
|
||||
Map<String, Object> config = Map.of("topicNames", Collections.singleton("my-part-topic-1"), "subscriptionName",
|
||||
"my-part-subscription-1");
|
||||
admin.topics().createPartitionedTopic(topicName, 3);
|
||||
|
||||
PulsarClient pulsarClient = PulsarClient.builder().serviceUrl(PulsarTestContainerSupport.getPulsarBrokerUrl())
|
||||
.build();
|
||||
DefaultPulsarConsumerFactory<String> pulsarConsumerFactory = new DefaultPulsarConsumerFactory<>(pulsarClient,
|
||||
config);
|
||||
(consumerBuilder) -> {
|
||||
consumerBuilder.topic("my-part-topic-1");
|
||||
consumerBuilder.subscriptionName("my-part-subscription-1");
|
||||
});
|
||||
|
||||
CountDownLatch latch1 = new CountDownLatch(1);
|
||||
CountDownLatch latch2 = new CountDownLatch(1);
|
||||
|
||||
@@ -18,7 +18,6 @@ package org.springframework.pulsar.core;
|
||||
|
||||
import static org.assertj.core.api.Assertions.assertThat;
|
||||
|
||||
import java.util.Collections;
|
||||
import java.util.HashMap;
|
||||
import java.util.Map;
|
||||
import java.util.concurrent.CountDownLatch;
|
||||
@@ -57,9 +56,10 @@ public class SharedSubscriptionConsumerTests implements PulsarTestContainerSuppo
|
||||
try {
|
||||
pulsarClient = PulsarClient.builder().serviceUrl(PulsarTestContainerSupport.getPulsarBrokerUrl()).build();
|
||||
DefaultPulsarConsumerFactory<String> pulsarConsumerFactory = new DefaultPulsarConsumerFactory<>(
|
||||
pulsarClient,
|
||||
Map.of("topicNames", Collections.singleton("shared-subscription-single-msg-test-topic"),
|
||||
"subscriptionName", "shared-subscription-single-msg-test-sub"));
|
||||
pulsarClient, (consumerBuilder) -> {
|
||||
consumerBuilder.topic("shared-subscription-single-msg-test-topic");
|
||||
consumerBuilder.subscriptionName("shared-subscription-single-msg-test-sub");
|
||||
});
|
||||
|
||||
CountDownLatch latch1 = new CountDownLatch(1);
|
||||
CountDownLatch latch2 = new CountDownLatch(1);
|
||||
@@ -113,8 +113,10 @@ public class SharedSubscriptionConsumerTests implements PulsarTestContainerSuppo
|
||||
try {
|
||||
pulsarClient = PulsarClient.builder().serviceUrl(PulsarTestContainerSupport.getPulsarBrokerUrl()).build();
|
||||
DefaultPulsarConsumerFactory<String> consumerFactory = new DefaultPulsarConsumerFactory<>(pulsarClient,
|
||||
Map.of("topicNames", Collections.singleton("key-shared-batch-disabled-topic"), "subscriptionName",
|
||||
"key-shared-batch-disabled-sub"));
|
||||
(consumerBuilder) -> {
|
||||
consumerBuilder.topic("key-shared-batch-disabled-topic");
|
||||
consumerBuilder.subscriptionName("key-shared-batch-disabled-sub");
|
||||
});
|
||||
|
||||
CountDownLatch latch = new CountDownLatch(30);
|
||||
|
||||
|
||||
@@ -28,9 +28,7 @@ import static org.mockito.Mockito.when;
|
||||
|
||||
import java.time.Duration;
|
||||
import java.util.ArrayList;
|
||||
import java.util.Collections;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
import java.util.concurrent.atomic.AtomicInteger;
|
||||
|
||||
import org.apache.pulsar.client.api.Consumer;
|
||||
@@ -56,13 +54,13 @@ public class DefaultPulsarConsumerErrorHandlerTests implements PulsarTestContain
|
||||
@Test
|
||||
@SuppressWarnings("unchecked")
|
||||
void happyPathErrorHandlingForRecordMessageListener() throws Exception {
|
||||
Map<String, Object> config = Map.of("topicNames", Collections.singleton("default-error-handler-tests-1"),
|
||||
"subscriptionName", "default-error-handler-tests-sub-1");
|
||||
|
||||
PulsarClient pulsarClient = PulsarClient.builder().serviceUrl(PulsarTestContainerSupport.getPulsarBrokerUrl())
|
||||
.build();
|
||||
DefaultPulsarConsumerFactory<String> pulsarConsumerFactory = new DefaultPulsarConsumerFactory<>(pulsarClient,
|
||||
config);
|
||||
(consumerBuilder) -> {
|
||||
consumerBuilder.topic("default-error-handler-tests-1");
|
||||
consumerBuilder.subscriptionName("default-error-handler-tests-sub-1");
|
||||
});
|
||||
|
||||
PulsarContainerProperties pulsarContainerProperties = new PulsarContainerProperties();
|
||||
PulsarRecordMessageListener<?> messageListener = mock(PulsarRecordMessageListener.class);
|
||||
@@ -104,13 +102,13 @@ public class DefaultPulsarConsumerErrorHandlerTests implements PulsarTestContain
|
||||
@Test
|
||||
@SuppressWarnings("unchecked")
|
||||
void errorHandlingForRecordMessageListenerWithTransientError() throws Exception {
|
||||
Map<String, Object> config = Map.of("topicNames", Collections.singleton("default-error-handler-tests-2"),
|
||||
"subscriptionName", "default-error-handler-tests-sub-2");
|
||||
|
||||
PulsarClient pulsarClient = PulsarClient.builder().serviceUrl(PulsarTestContainerSupport.getPulsarBrokerUrl())
|
||||
.build();
|
||||
DefaultPulsarConsumerFactory<String> pulsarConsumerFactory = new DefaultPulsarConsumerFactory<>(pulsarClient,
|
||||
config);
|
||||
(consumerBuilder) -> {
|
||||
consumerBuilder.topic("default-error-handler-tests-2");
|
||||
consumerBuilder.subscriptionName("default-error-handler-tests-sub-2");
|
||||
});
|
||||
|
||||
PulsarContainerProperties pulsarContainerProperties = new PulsarContainerProperties();
|
||||
PulsarRecordMessageListener<?> messageListener = mock(PulsarRecordMessageListener.class);
|
||||
@@ -150,13 +148,13 @@ public class DefaultPulsarConsumerErrorHandlerTests implements PulsarTestContain
|
||||
@Test
|
||||
@SuppressWarnings("unchecked")
|
||||
void everyOtherRecordThrowsNonTransientExceptionsRecordMessageListener() throws Exception {
|
||||
Map<String, Object> config = Map.of("topicNames", Collections.singleton("default-error-handler-tests-3"),
|
||||
"subscriptionName", "default-error-handler-tests-sub-3");
|
||||
|
||||
PulsarClient pulsarClient = PulsarClient.builder().serviceUrl(PulsarTestContainerSupport.getPulsarBrokerUrl())
|
||||
.build();
|
||||
DefaultPulsarConsumerFactory<Integer> pulsarConsumerFactory = new DefaultPulsarConsumerFactory<>(pulsarClient,
|
||||
config);
|
||||
(consumerBuilder) -> {
|
||||
consumerBuilder.topic("default-error-handler-tests-3");
|
||||
consumerBuilder.subscriptionName("default-error-handler-tests-sub-3");
|
||||
});
|
||||
|
||||
PulsarContainerProperties pulsarContainerProperties = new PulsarContainerProperties();
|
||||
PulsarRecordMessageListener<?> messageListener = mock(PulsarRecordMessageListener.class);
|
||||
@@ -206,13 +204,13 @@ public class DefaultPulsarConsumerErrorHandlerTests implements PulsarTestContain
|
||||
@Test
|
||||
@SuppressWarnings("unchecked")
|
||||
void batchRecordListenerFirstOneOnlyErrorAndRecover() throws Exception {
|
||||
Map<String, Object> config = Map.of("topicNames", Collections.singleton("default-error-handler-tests-4"),
|
||||
"subscriptionName", "default-error-handler-tests-sub-4");
|
||||
|
||||
PulsarClient pulsarClient = PulsarClient.builder().serviceUrl(PulsarTestContainerSupport.getPulsarBrokerUrl())
|
||||
.build();
|
||||
DefaultPulsarConsumerFactory<Integer> pulsarConsumerFactory = new DefaultPulsarConsumerFactory<>(pulsarClient,
|
||||
config);
|
||||
(consumerBuilder) -> {
|
||||
consumerBuilder.topic("default-error-handler-tests-4");
|
||||
consumerBuilder.subscriptionName("default-error-handler-tests-sub-4");
|
||||
});
|
||||
|
||||
PulsarContainerProperties pulsarContainerProperties = new PulsarContainerProperties();
|
||||
pulsarContainerProperties.setMaxNumMessages(10);
|
||||
@@ -275,13 +273,13 @@ public class DefaultPulsarConsumerErrorHandlerTests implements PulsarTestContain
|
||||
@Test
|
||||
@SuppressWarnings("unchecked")
|
||||
void batchRecordListenerRecordFailsInTheMiddle() throws Exception {
|
||||
Map<String, Object> config = Map.of("topicNames", Collections.singleton("default-error-handler-tests-5"),
|
||||
"subscriptionName", "default-error-handler-tests-sub-5");
|
||||
|
||||
PulsarClient pulsarClient = PulsarClient.builder().serviceUrl(PulsarTestContainerSupport.getPulsarBrokerUrl())
|
||||
.build();
|
||||
DefaultPulsarConsumerFactory<Integer> pulsarConsumerFactory = new DefaultPulsarConsumerFactory<>(pulsarClient,
|
||||
config);
|
||||
(consumerBuilder) -> {
|
||||
consumerBuilder.topic("default-error-handler-tests-5");
|
||||
consumerBuilder.subscriptionName("default-error-handler-tests-sub-5");
|
||||
});
|
||||
|
||||
PulsarContainerProperties pulsarContainerProperties = new PulsarContainerProperties();
|
||||
pulsarContainerProperties.setMaxNumMessages(10);
|
||||
@@ -342,13 +340,13 @@ public class DefaultPulsarConsumerErrorHandlerTests implements PulsarTestContain
|
||||
@Test
|
||||
@SuppressWarnings("unchecked")
|
||||
void batchRecordListenerRecordFailsTwiceInTheMiddle() throws Exception {
|
||||
Map<String, Object> config = Map.of("topicNames", Collections.singleton("default-error-handler-tests-6"),
|
||||
"subscriptionName", "default-error-handler-tests-sub-6");
|
||||
|
||||
PulsarClient pulsarClient = PulsarClient.builder().serviceUrl(PulsarTestContainerSupport.getPulsarBrokerUrl())
|
||||
.build();
|
||||
DefaultPulsarConsumerFactory<Integer> pulsarConsumerFactory = new DefaultPulsarConsumerFactory<>(pulsarClient,
|
||||
config);
|
||||
(consumerBuilder) -> {
|
||||
consumerBuilder.topic("default-error-handler-tests-6");
|
||||
consumerBuilder.subscriptionName("default-error-handler-tests-sub-6");
|
||||
});
|
||||
|
||||
PulsarContainerProperties pulsarContainerProperties = new PulsarContainerProperties();
|
||||
pulsarContainerProperties.setMaxNumMessages(10);
|
||||
@@ -409,13 +407,13 @@ public class DefaultPulsarConsumerErrorHandlerTests implements PulsarTestContain
|
||||
@Test
|
||||
@SuppressWarnings("unchecked")
|
||||
void batchRecordListenerRecordFailsInTheMiddleButTransientError() throws Exception {
|
||||
Map<String, Object> config = Map.of("topicNames", Collections.singleton("default-error-handler-tests-7"),
|
||||
"subscriptionName", "default-error-handler-tests-sub-7");
|
||||
|
||||
PulsarClient pulsarClient = PulsarClient.builder().serviceUrl(PulsarTestContainerSupport.getPulsarBrokerUrl())
|
||||
.build();
|
||||
DefaultPulsarConsumerFactory<Integer> pulsarConsumerFactory = new DefaultPulsarConsumerFactory<>(pulsarClient,
|
||||
config);
|
||||
(consumerBuilder) -> {
|
||||
consumerBuilder.topic("default-error-handler-tests-7");
|
||||
consumerBuilder.subscriptionName("default-error-handler-tests-sub-7");
|
||||
});
|
||||
|
||||
PulsarContainerProperties pulsarContainerProperties = new PulsarContainerProperties();
|
||||
pulsarContainerProperties.setMaxNumMessages(10);
|
||||
@@ -475,13 +473,13 @@ public class DefaultPulsarConsumerErrorHandlerTests implements PulsarTestContain
|
||||
@Test
|
||||
@SuppressWarnings("unchecked")
|
||||
void batchListenerFailsTransientErrorFollowedByNonTransient() throws Exception {
|
||||
Map<String, Object> config = Map.of("topicNames", Collections.singleton("default-error-handler-tests-8"),
|
||||
"subscriptionName", "default-error-handler-tests-sub-8");
|
||||
|
||||
PulsarClient pulsarClient = PulsarClient.builder().serviceUrl(PulsarTestContainerSupport.getPulsarBrokerUrl())
|
||||
.build();
|
||||
DefaultPulsarConsumerFactory<Integer> pulsarConsumerFactory = new DefaultPulsarConsumerFactory<>(pulsarClient,
|
||||
config);
|
||||
(consumerBuilder) -> {
|
||||
consumerBuilder.topic("default-error-handler-tests-8");
|
||||
consumerBuilder.subscriptionName("default-error-handler-tests-sub-8");
|
||||
});
|
||||
|
||||
PulsarContainerProperties pulsarContainerProperties = new PulsarContainerProperties();
|
||||
pulsarContainerProperties.setMaxNumMessages(10);
|
||||
|
||||
@@ -26,9 +26,7 @@ import static org.mockito.Mockito.verify;
|
||||
|
||||
import java.time.Duration;
|
||||
import java.util.ArrayList;
|
||||
import java.util.Collections;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
import java.util.Set;
|
||||
import java.util.concurrent.CountDownLatch;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
@@ -65,13 +63,13 @@ class DefaultPulsarMessageListenerContainerTests implements PulsarTestContainerS
|
||||
|
||||
@Test
|
||||
void basicDefaultConsumer() throws Exception {
|
||||
Map<String, Object> config = Map.of("topicNames", Collections.singleton("dpmlct-012"), "subscriptionName",
|
||||
"dpmlct-sb-012");
|
||||
|
||||
PulsarClient pulsarClient = PulsarClient.builder().serviceUrl(PulsarTestContainerSupport.getPulsarBrokerUrl())
|
||||
.build();
|
||||
DefaultPulsarConsumerFactory<String> pulsarConsumerFactory = new DefaultPulsarConsumerFactory<>(pulsarClient,
|
||||
config);
|
||||
(consumerBuilder) -> {
|
||||
consumerBuilder.topic("dpmlct-012");
|
||||
consumerBuilder.subscriptionName("dpmlct-sb-012");
|
||||
});
|
||||
CountDownLatch latch = new CountDownLatch(1);
|
||||
PulsarContainerProperties pulsarContainerProperties = new PulsarContainerProperties();
|
||||
pulsarContainerProperties
|
||||
@@ -93,13 +91,13 @@ class DefaultPulsarMessageListenerContainerTests implements PulsarTestContainerS
|
||||
@Disabled
|
||||
@Test
|
||||
void containerPauseAndResumeFeatureUsingWaitAndNotify() throws Exception {
|
||||
Map<String, Object> config = Map.of("topicNames", Collections.singleton("containerPauseResumeWaitNotify-topic"),
|
||||
"subscriptionName", "containerPauseResumeWaitNotify-sub");
|
||||
|
||||
PulsarClient pulsarClient = PulsarClient.builder().serviceUrl(PulsarTestContainerSupport.getPulsarBrokerUrl())
|
||||
.build();
|
||||
DefaultPulsarConsumerFactory<String> pulsarConsumerFactory = new DefaultPulsarConsumerFactory<>(pulsarClient,
|
||||
config);
|
||||
(consumerBuilder) -> {
|
||||
consumerBuilder.topic("containerPauseResumeWaitNotify-topic");
|
||||
consumerBuilder.subscriptionName("containerPauseResumeWaitNotify-sub");
|
||||
});
|
||||
PulsarContainerProperties pulsarContainerProperties = new PulsarContainerProperties();
|
||||
pulsarContainerProperties.setMessageListener((PulsarRecordMessageListener<?>) (consumer, msg) -> {
|
||||
});
|
||||
@@ -162,13 +160,14 @@ class DefaultPulsarMessageListenerContainerTests implements PulsarTestContainerS
|
||||
|
||||
@Test
|
||||
void subscriptionInitialPositionEarliest() throws Exception {
|
||||
Map<String, Object> config = Map.of("topicNames", Collections.singleton("dpmlct-013"), "subscriptionName",
|
||||
"dpmlct-sb-013", "subscriptionInitialPosition", SubscriptionInitialPosition.Earliest);
|
||||
|
||||
PulsarClient pulsarClient = PulsarClient.builder().serviceUrl(PulsarTestContainerSupport.getPulsarBrokerUrl())
|
||||
.build();
|
||||
DefaultPulsarConsumerFactory<String> pulsarConsumerFactory = new DefaultPulsarConsumerFactory<>(pulsarClient,
|
||||
config);
|
||||
(consumerBuilder) -> {
|
||||
consumerBuilder.topic("dpmlct-013");
|
||||
consumerBuilder.subscriptionName("dpmlct-sb-013");
|
||||
consumerBuilder.subscriptionInitialPosition(SubscriptionInitialPosition.Earliest);
|
||||
});
|
||||
CountDownLatch latch = new CountDownLatch(5);
|
||||
PulsarContainerProperties pulsarContainerProperties = new PulsarContainerProperties();
|
||||
pulsarContainerProperties
|
||||
@@ -192,13 +191,13 @@ class DefaultPulsarMessageListenerContainerTests implements PulsarTestContainerS
|
||||
|
||||
@Test
|
||||
void subscriptionInitialPositionDefaultLatest() throws Exception {
|
||||
Map<String, Object> config = Map.of("topicNames", Collections.singleton("dpmlct-014"), "subscriptionName",
|
||||
"dpmlct-sb-014");
|
||||
|
||||
PulsarClient pulsarClient = PulsarClient.builder().serviceUrl(PulsarTestContainerSupport.getPulsarBrokerUrl())
|
||||
.build();
|
||||
DefaultPulsarConsumerFactory<String> pulsarConsumerFactory = new DefaultPulsarConsumerFactory<>(pulsarClient,
|
||||
config);
|
||||
(consumerBuilder) -> {
|
||||
consumerBuilder.topic("dpmlct-014");
|
||||
consumerBuilder.subscriptionName("dpmlct-sb-014");
|
||||
});
|
||||
PulsarContainerProperties pulsarContainerProperties = new PulsarContainerProperties();
|
||||
List<String> messages = new ArrayList<>();
|
||||
pulsarContainerProperties.setMessageListener(
|
||||
@@ -225,15 +224,16 @@ class DefaultPulsarMessageListenerContainerTests implements PulsarTestContainerS
|
||||
|
||||
@Test
|
||||
void negativeAckRedeliveryBackoff() throws Exception {
|
||||
RedeliveryBackoff redeliveryBackoff = MultiplierRedeliveryBackoff.builder().minDelayMs(1000)
|
||||
.maxDelayMs(5 * 1000).build();
|
||||
Map<String, Object> config = Map.of("topicNames", Collections.singleton("dpmlct-015"), "subscriptionName",
|
||||
"dpmlct-sb-015", "negativeAckRedeliveryBackoff", redeliveryBackoff);
|
||||
|
||||
PulsarClient pulsarClient = PulsarClient.builder().serviceUrl(PulsarTestContainerSupport.getPulsarBrokerUrl())
|
||||
.build();
|
||||
RedeliveryBackoff redeliveryBackoff = MultiplierRedeliveryBackoff.builder().minDelayMs(1000)
|
||||
.maxDelayMs(5 * 1000).build();
|
||||
DefaultPulsarConsumerFactory<String> pulsarConsumerFactory = spy(
|
||||
new DefaultPulsarConsumerFactory<>(pulsarClient, config));
|
||||
new DefaultPulsarConsumerFactory<>(pulsarClient, (consumerBuilder) -> {
|
||||
consumerBuilder.topic("dpmlct-015");
|
||||
consumerBuilder.subscriptionName("dpmlct-sb-015");
|
||||
consumerBuilder.negativeAckRedeliveryBackoff(redeliveryBackoff);
|
||||
}));
|
||||
CountDownLatch latch = new CountDownLatch(10);
|
||||
PulsarContainerProperties pulsarContainerProperties = new PulsarContainerProperties();
|
||||
pulsarContainerProperties.setMessageListener((PulsarRecordMessageListener<?>) (consumer, msg) -> {
|
||||
@@ -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<String, Object> config = Map.of("topicNames", Collections.singleton("dpmlct-016"), "subscriptionName",
|
||||
"dpmlct-sb-016", "negativeAckRedeliveryDelayMicros", TimeUnit.SECONDS.toMicros(1), "deadLetterPolicy",
|
||||
deadLetterPolicy);
|
||||
|
||||
PulsarClient pulsarClient = PulsarClient.builder().serviceUrl(PulsarTestContainerSupport.getPulsarBrokerUrl())
|
||||
.build();
|
||||
DeadLetterPolicy deadLetterPolicy = DeadLetterPolicy.builder().maxRedeliverCount(1)
|
||||
.deadLetterTopic("dpmlct-016-dlq-topic").build();
|
||||
DefaultPulsarConsumerFactory<Integer> pulsarConsumerFactory = new DefaultPulsarConsumerFactory<>(pulsarClient,
|
||||
config);
|
||||
(consumerBuilder) -> {
|
||||
consumerBuilder.topic("dpmlct-016");
|
||||
consumerBuilder.subscriptionName("dpmlct-sb-016");
|
||||
consumerBuilder.negativeAckRedeliveryDelay(1L, TimeUnit.SECONDS);
|
||||
consumerBuilder.deadLetterPolicy(deadLetterPolicy);
|
||||
});
|
||||
|
||||
CountDownLatch dlqLatch = new CountDownLatch(1);
|
||||
CountDownLatch latch = new CountDownLatch(6);
|
||||
@@ -326,16 +327,17 @@ class DefaultPulsarMessageListenerContainerTests implements PulsarTestContainerS
|
||||
|
||||
@Test
|
||||
void deadLetterPolicyCustom() throws Exception {
|
||||
DeadLetterPolicy deadLetterPolicy = DeadLetterPolicy.builder().maxRedeliverCount(5).deadLetterTopic("dlq-topic")
|
||||
.build();
|
||||
Map<String, Object> config = Map.of("topicNames", Collections.singleton("dpmlct-017"), "subscriptionName",
|
||||
"dpmlct-sb-016", "negativeAckRedeliveryDelayMicros", TimeUnit.SECONDS.toMicros(1), "deadLetterPolicy",
|
||||
deadLetterPolicy);
|
||||
|
||||
PulsarClient pulsarClient = PulsarClient.builder().serviceUrl(PulsarTestContainerSupport.getPulsarBrokerUrl())
|
||||
.build();
|
||||
DeadLetterPolicy deadLetterPolicy = DeadLetterPolicy.builder().maxRedeliverCount(5).deadLetterTopic("dlq-topic")
|
||||
.build();
|
||||
DefaultPulsarConsumerFactory<Integer> pulsarConsumerFactory = new DefaultPulsarConsumerFactory<>(pulsarClient,
|
||||
config);
|
||||
(consumerBuilder) -> {
|
||||
consumerBuilder.topic("dpmlct-017");
|
||||
consumerBuilder.subscriptionName("dpmlct-sb-017");
|
||||
consumerBuilder.negativeAckRedeliveryDelay(1L, TimeUnit.SECONDS);
|
||||
consumerBuilder.deadLetterPolicy(deadLetterPolicy);
|
||||
});
|
||||
|
||||
CountDownLatch dlqLatch = new CountDownLatch(1);
|
||||
CountDownLatch latch = new CountDownLatch(6);
|
||||
|
||||
@@ -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<String> myCustomizer() {
|
||||
return cb -> {
|
||||
cb.subscriptionName("test-changed-subscription-name");
|
||||
};
|
||||
return cb -> cb.subscriptionName("test-changed-subscription-name");
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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
|
||||
|
||||
Reference in New Issue
Block a user