Add support for Apache Pulsar

Add support for Apache Pulsar using the Spring for Apache Pulsar
project.

See gh-34763

Co-authored-by: Phillip Webb <pwebb@vmware.com>
This commit is contained in:
Chris Bono
2023-08-30 16:39:37 -07:00
committed by Phillip Webb
parent 8f78acd548
commit 6e7b845bdf
56 changed files with 5261 additions and 3 deletions

View File

@@ -179,6 +179,8 @@ dependencies {
optional("org.springframework.data:spring-data-redis")
optional("org.springframework.graphql:spring-graphql")
optional("org.springframework.hateoas:spring-hateoas")
optional("org.springframework.pulsar:spring-pulsar")
optional("org.springframework.pulsar:spring-pulsar-reactive")
optional("org.springframework.security:spring-security-acl")
optional("org.springframework.security:spring-security-config")
optional("org.springframework.security:spring-security-data") {
@@ -255,6 +257,7 @@ dependencies {
testImplementation("org.testcontainers:junit-jupiter")
testImplementation("org.testcontainers:mongodb")
testImplementation("org.testcontainers:neo4j")
testImplementation("org.testcontainers:pulsar")
testImplementation("org.testcontainers:testcontainers")
testImplementation("org.yaml:snakeyaml")

View File

@@ -0,0 +1,49 @@
/*
* Copyright 2012-2023 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* https://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.boot.autoconfigure.pulsar;
import org.apache.pulsar.client.api.DeadLetterPolicy;
import org.apache.pulsar.client.api.DeadLetterPolicy.DeadLetterPolicyBuilder;
import org.springframework.boot.context.properties.PropertyMapper;
import org.springframework.util.Assert;
/**
* Helper class used to map {@link PulsarProperties.Consumer.DeadLetterPolicy dead letter
* policy properties}.
*
* @author Chris Bono
* @author Phillip Webb
*/
final class DeadLetterPolicyMapper {
private DeadLetterPolicyMapper() {
}
static DeadLetterPolicy map(PulsarProperties.Consumer.DeadLetterPolicy policy) {
Assert.state(policy.getMaxRedeliverCount() > 0,
"Pulsar DeadLetterPolicy must have a positive 'max-redelivery-count' property value");
PropertyMapper map = PropertyMapper.get().alwaysApplyingWhenNonNull();
DeadLetterPolicyBuilder builder = DeadLetterPolicy.builder();
map.from(policy::getMaxRedeliverCount).to(builder::maxRedeliverCount);
map.from(policy::getRetryLetterTopic).to(builder::retryLetterTopic);
map.from(policy::getDeadLetterTopic).to(builder::deadLetterTopic);
map.from(policy::getInitialSubscriptionName).to(builder::initialSubscriptionName);
return builder.build();
}
}

View File

@@ -0,0 +1,196 @@
/*
* Copyright 2012-2023 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* https://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.boot.autoconfigure.pulsar;
import java.util.ArrayList;
import java.util.List;
import org.apache.pulsar.client.api.ConsumerBuilder;
import org.apache.pulsar.client.api.ProducerBuilder;
import org.apache.pulsar.client.api.PulsarClient;
import org.apache.pulsar.client.api.ReaderBuilder;
import org.apache.pulsar.client.api.interceptor.ProducerInterceptor;
import org.springframework.beans.factory.ObjectProvider;
import org.springframework.boot.autoconfigure.AutoConfiguration;
import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.boot.util.LambdaSafe;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Import;
import org.springframework.pulsar.annotation.EnablePulsar;
import org.springframework.pulsar.config.ConcurrentPulsarListenerContainerFactory;
import org.springframework.pulsar.config.DefaultPulsarReaderContainerFactory;
import org.springframework.pulsar.config.PulsarAnnotationSupportBeanNames;
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.ProducerBuilderCustomizer;
import org.springframework.pulsar.core.PulsarConsumerFactory;
import org.springframework.pulsar.core.PulsarProducerFactory;
import org.springframework.pulsar.core.PulsarReaderFactory;
import org.springframework.pulsar.core.PulsarTemplate;
import org.springframework.pulsar.core.ReaderBuilderCustomizer;
import org.springframework.pulsar.core.SchemaResolver;
import org.springframework.pulsar.core.TopicResolver;
import org.springframework.pulsar.listener.PulsarContainerProperties;
import org.springframework.pulsar.reader.PulsarReaderContainerProperties;
/**
* {@link EnableAutoConfiguration Auto-configuration} for Apache Pulsar.
*
* @author Chris Bono
* @author Soby Chacko
* @author Alexander Preuß
* @author Phillip Webb
* @since 3.2.0
*/
@AutoConfiguration
@ConditionalOnClass({ PulsarClient.class, PulsarTemplate.class })
@Import(PulsarConfiguration.class)
public class PulsarAutoConfiguration {
private PulsarProperties properties;
private PulsarPropertiesMapper propertiesMapper;
PulsarAutoConfiguration(PulsarProperties properties) {
this.properties = properties;
this.propertiesMapper = new PulsarPropertiesMapper(properties);
}
@Bean
@ConditionalOnMissingBean(PulsarProducerFactory.class)
@ConditionalOnProperty(name = "spring.pulsar.producer.cache.enabled", havingValue = "false")
DefaultPulsarProducerFactory<?> pulsarProducerFactory(PulsarClient pulsarClient, TopicResolver topicResolver,
ObjectProvider<ProducerBuilderCustomizer<?>> customizersProvider) {
List<ProducerBuilderCustomizer<Object>> lambdaSafeCustomizers = lambdaSafeProducerBuilderCustomizers(
customizersProvider);
return new DefaultPulsarProducerFactory<>(pulsarClient, this.properties.getProducer().getTopicName(),
lambdaSafeCustomizers, topicResolver);
}
@Bean
@ConditionalOnMissingBean(PulsarProducerFactory.class)
@ConditionalOnProperty(name = "spring.pulsar.producer.cache.enabled", havingValue = "true", matchIfMissing = true)
CachingPulsarProducerFactory<?> cachingPulsarProducerFactory(PulsarClient pulsarClient, TopicResolver topicResolver,
ObjectProvider<ProducerBuilderCustomizer<?>> customizersProvider) {
PulsarProperties.Producer.Cache cacheProperties = this.properties.getProducer().getCache();
List<ProducerBuilderCustomizer<Object>> lambdaSafeCustomizers = lambdaSafeProducerBuilderCustomizers(
customizersProvider);
return new CachingPulsarProducerFactory<>(pulsarClient, this.properties.getProducer().getTopicName(),
lambdaSafeCustomizers, topicResolver, cacheProperties.getExpireAfterAccess(),
cacheProperties.getMaximumSize(), cacheProperties.getInitialCapacity());
}
private List<ProducerBuilderCustomizer<Object>> lambdaSafeProducerBuilderCustomizers(
ObjectProvider<ProducerBuilderCustomizer<?>> customizersProvider) {
List<ProducerBuilderCustomizer<?>> customizers = new ArrayList<>();
customizers.add(this.propertiesMapper::customizeProducerBuilder);
customizers.addAll(customizersProvider.orderedStream().toList());
return List.of((builder) -> applyProducerBuilderCustomizers(customizers, builder));
}
@SuppressWarnings("unchecked")
private void applyProducerBuilderCustomizers(List<ProducerBuilderCustomizer<?>> customizers,
ProducerBuilder<?> builder) {
LambdaSafe.callbacks(ProducerBuilderCustomizer.class, customizers, builder)
.invoke((customizer) -> customizer.customize(builder));
}
@Bean
@ConditionalOnMissingBean
PulsarTemplate<?> pulsarTemplate(PulsarProducerFactory<?> pulsarProducerFactory,
ObjectProvider<ProducerInterceptor> producerInterceptors, SchemaResolver schemaResolver,
TopicResolver topicResolver) {
return new PulsarTemplate<>(pulsarProducerFactory, producerInterceptors.orderedStream().toList(),
schemaResolver, topicResolver, this.properties.getTemplate().isObservationsEnabled());
}
@Bean
@ConditionalOnMissingBean(PulsarConsumerFactory.class)
DefaultPulsarConsumerFactory<Object> pulsarConsumerFactory(PulsarClient pulsarClient,
ObjectProvider<ConsumerBuilderCustomizer<?>> customizersProvider) {
List<ConsumerBuilderCustomizer<?>> customizers = new ArrayList<>();
customizers.add(this.propertiesMapper::customizeConsumerBuilder);
customizers.addAll(customizersProvider.orderedStream().toList());
List<ConsumerBuilderCustomizer<Object>> lambdaSafeCustomizers = List
.of((builder) -> applyConsumerBuilderCustomizers(customizers, builder));
return new DefaultPulsarConsumerFactory<>(pulsarClient, lambdaSafeCustomizers);
}
@SuppressWarnings("unchecked")
private void applyConsumerBuilderCustomizers(List<ConsumerBuilderCustomizer<?>> customizers,
ConsumerBuilder<?> builder) {
LambdaSafe.callbacks(ConsumerBuilderCustomizer.class, customizers, builder)
.invoke((customizer) -> customizer.customize(builder));
}
@Bean
@ConditionalOnMissingBean(name = "pulsarListenerContainerFactory")
ConcurrentPulsarListenerContainerFactory<Object> pulsarListenerContainerFactory(
PulsarConsumerFactory<Object> pulsarConsumerFactory, SchemaResolver schemaResolver,
TopicResolver topicResolver) {
PulsarContainerProperties containerProperties = new PulsarContainerProperties();
containerProperties.setSchemaResolver(schemaResolver);
containerProperties.setTopicResolver(topicResolver);
this.propertiesMapper.customizeContainerProperties(containerProperties);
return new ConcurrentPulsarListenerContainerFactory<>(pulsarConsumerFactory, containerProperties);
}
@Bean
@ConditionalOnMissingBean(PulsarReaderFactory.class)
DefaultPulsarReaderFactory<?> pulsarReaderFactory(PulsarClient pulsarClient,
ObjectProvider<ReaderBuilderCustomizer<?>> customizersProvider) {
List<ReaderBuilderCustomizer<?>> customizers = new ArrayList<>();
customizers.add(this.propertiesMapper::customizeReaderBuilder);
customizers.addAll(customizersProvider.orderedStream().toList());
List<ReaderBuilderCustomizer<Object>> lambdaSafeCustomizers = List
.of((builder) -> applyReaderBuilderCustomizers(customizers, builder));
return new DefaultPulsarReaderFactory<>(pulsarClient, lambdaSafeCustomizers);
}
@SuppressWarnings("unchecked")
private void applyReaderBuilderCustomizers(List<ReaderBuilderCustomizer<?>> customizers, ReaderBuilder<?> builder) {
LambdaSafe.callbacks(ReaderBuilderCustomizer.class, customizers, builder)
.invoke((customizer) -> customizer.customize(builder));
}
@Bean
@ConditionalOnMissingBean(name = "pulsarReaderContainerFactory")
DefaultPulsarReaderContainerFactory<?> pulsarReaderContainerFactory(PulsarReaderFactory<?> pulsarReaderFactory,
SchemaResolver schemaResolver) {
PulsarReaderContainerProperties readerContainerProperties = new PulsarReaderContainerProperties();
readerContainerProperties.setSchemaResolver(schemaResolver);
this.propertiesMapper.customizeReaderContainerProperties(readerContainerProperties);
return new DefaultPulsarReaderContainerFactory<>(pulsarReaderFactory, readerContainerProperties);
}
@Configuration(proxyBeanMethods = false)
@EnablePulsar
@ConditionalOnMissingBean(name = { PulsarAnnotationSupportBeanNames.PULSAR_LISTENER_ANNOTATION_PROCESSOR_BEAN_NAME,
PulsarAnnotationSupportBeanNames.PULSAR_READER_ANNOTATION_PROCESSOR_BEAN_NAME })
static class EnablePulsarConfiguration {
}
}

View File

@@ -0,0 +1,173 @@
/*
* Copyright 2012-2023 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* https://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.boot.autoconfigure.pulsar;
import java.util.ArrayList;
import java.util.List;
import org.apache.pulsar.client.admin.PulsarAdminBuilder;
import org.apache.pulsar.client.api.ClientBuilder;
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.common.schema.SchemaType;
import org.springframework.beans.factory.ObjectProvider;
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.boot.autoconfigure.pulsar.PulsarProperties.Defaults.SchemaInfo;
import org.springframework.boot.autoconfigure.pulsar.PulsarProperties.Defaults.TypeMapping;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.boot.util.LambdaSafe;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.pulsar.core.DefaultPulsarClientFactory;
import org.springframework.pulsar.core.DefaultSchemaResolver;
import org.springframework.pulsar.core.DefaultTopicResolver;
import org.springframework.pulsar.core.PulsarAdminBuilderCustomizer;
import org.springframework.pulsar.core.PulsarAdministration;
import org.springframework.pulsar.core.PulsarClientBuilderCustomizer;
import org.springframework.pulsar.core.PulsarClientFactory;
import org.springframework.pulsar.core.SchemaResolver;
import org.springframework.pulsar.core.SchemaResolver.SchemaResolverCustomizer;
import org.springframework.pulsar.core.TopicResolver;
import org.springframework.pulsar.function.PulsarFunction;
import org.springframework.pulsar.function.PulsarFunctionAdministration;
import org.springframework.pulsar.function.PulsarSink;
import org.springframework.pulsar.function.PulsarSource;
/**
* Common configuration used by both {@link PulsarAutoConfiguration} and
* {@link PulsarReactiveAutoConfiguration}. A separate configuration class is used so that
* {@link PulsarAutoConfiguration} can be excluded for reactive only application.
*
* @author Chris Bono
* @author Phillip Webb
*/
@Configuration(proxyBeanMethods = false)
@EnableConfigurationProperties(PulsarProperties.class)
class PulsarConfiguration {
private final PulsarProperties properties;
private final PulsarPropertiesMapper propertiesMapper;
PulsarConfiguration(PulsarProperties properties) {
this.properties = properties;
this.propertiesMapper = new PulsarPropertiesMapper(properties);
}
@Bean
@ConditionalOnMissingBean(PulsarClientFactory.class)
DefaultPulsarClientFactory pulsarClientFactory(ObjectProvider<PulsarClientBuilderCustomizer> customizersProvider) {
List<PulsarClientBuilderCustomizer> allCustomizers = new ArrayList<>();
allCustomizers.add(this.propertiesMapper::customizeClientBuilder);
allCustomizers.addAll(customizersProvider.orderedStream().toList());
DefaultPulsarClientFactory clientFactory = new DefaultPulsarClientFactory(
(clientBuilder) -> applyClientBuilderCustomizers(allCustomizers, clientBuilder));
return clientFactory;
}
private void applyClientBuilderCustomizers(List<PulsarClientBuilderCustomizer> customizers,
ClientBuilder clientBuilder) {
customizers.forEach((customizer) -> customizer.customize(clientBuilder));
}
@Bean
@ConditionalOnMissingBean
PulsarClient pulsarClient(PulsarClientFactory clientFactory) throws PulsarClientException {
return clientFactory.createClient();
}
@Bean
@ConditionalOnMissingBean
PulsarAdministration pulsarAdministration(
ObjectProvider<PulsarAdminBuilderCustomizer> pulsarAdminBuilderCustomizers) {
List<PulsarAdminBuilderCustomizer> allCustomizers = new ArrayList<>();
allCustomizers.add(this.propertiesMapper::customizeAdminBuilder);
allCustomizers.addAll(pulsarAdminBuilderCustomizers.orderedStream().toList());
return new PulsarAdministration((adminBuilder) -> applyAdminBuilderCustomizers(allCustomizers, adminBuilder));
}
private void applyAdminBuilderCustomizers(List<PulsarAdminBuilderCustomizer> customizers,
PulsarAdminBuilder adminBuilder) {
customizers.forEach((customizer) -> customizer.customize(adminBuilder));
}
@Bean
@ConditionalOnMissingBean(SchemaResolver.class)
DefaultSchemaResolver pulsarSchemaResolver(ObjectProvider<SchemaResolverCustomizer<?>> schemaResolverCustomizers) {
DefaultSchemaResolver schemaResolver = new DefaultSchemaResolver();
addCustomSchemaMappings(schemaResolver, this.properties.getDefaults().getTypeMappings());
applySchemaResolverCustomizers(schemaResolverCustomizers.orderedStream().toList(), schemaResolver);
return schemaResolver;
}
private void addCustomSchemaMappings(DefaultSchemaResolver schemaResolver, List<TypeMapping> typeMappings) {
if (typeMappings != null) {
typeMappings.forEach((typeMapping) -> addCustomSchemaMapping(schemaResolver, typeMapping));
}
}
private void addCustomSchemaMapping(DefaultSchemaResolver schemaResolver, TypeMapping typeMapping) {
SchemaInfo schemaInfo = typeMapping.schemaInfo();
if (schemaInfo != null) {
Class<?> messageType = typeMapping.messageType();
SchemaType schemaType = schemaInfo.schemaType();
Class<?> messageKeyType = schemaInfo.messageKeyType();
Schema<?> schema = schemaResolver.resolveSchema(schemaType, messageType, messageKeyType).orElseThrow();
schemaResolver.addCustomSchemaMapping(typeMapping.messageType(), schema);
}
}
@SuppressWarnings("unchecked")
private void applySchemaResolverCustomizers(List<SchemaResolverCustomizer<?>> customizers,
DefaultSchemaResolver schemaResolver) {
LambdaSafe.callbacks(SchemaResolverCustomizer.class, customizers, schemaResolver)
.invoke((customizer) -> customizer.customize(schemaResolver));
}
@Bean
@ConditionalOnMissingBean(TopicResolver.class)
DefaultTopicResolver pulsarTopicResolver() {
DefaultTopicResolver topicResolver = new DefaultTopicResolver();
List<TypeMapping> typeMappings = this.properties.getDefaults().getTypeMappings();
if (typeMappings != null) {
typeMappings.forEach((typeMapping) -> addCustomTopicMapping(topicResolver, typeMapping));
}
return topicResolver;
}
private void addCustomTopicMapping(DefaultTopicResolver topicResolver, TypeMapping typeMapping) {
String topicName = typeMapping.topicName();
if (topicName != null) {
topicResolver.addCustomTopicMapping(typeMapping.messageType(), topicName);
}
}
@Bean
@ConditionalOnMissingBean
@ConditionalOnProperty(name = "spring.pulsar.function.enabled", havingValue = "true", matchIfMissing = true)
PulsarFunctionAdministration pulsarFunctionAdministration(PulsarAdministration pulsarAdministration,
ObjectProvider<PulsarFunction> pulsarFunctions, ObjectProvider<PulsarSink> pulsarSinks,
ObjectProvider<PulsarSource> pulsarSources) {
PulsarProperties.Function properties = this.properties.getFunction();
return new PulsarFunctionAdministration(pulsarAdministration, pulsarFunctions, pulsarSinks, pulsarSources,
properties.isFailFast(), properties.isPropagateFailures(), properties.isPropagateStopFailures());
}
}

View File

@@ -0,0 +1,890 @@
/*
* Copyright 2012-2023 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* https://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.boot.autoconfigure.pulsar;
import java.time.Duration;
import java.util.ArrayList;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.regex.Pattern;
import org.apache.pulsar.client.api.CompressionType;
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.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.common.schema.SchemaType;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.boot.context.properties.NestedConfigurationProperty;
import org.springframework.util.Assert;
/**
* Configuration properties Apache Pulsar.
*
* @author Chris Bono
* @author Phillip Webb
* @since 3.2.0
*/
@ConfigurationProperties("spring.pulsar")
public class PulsarProperties {
private final Client client = new Client();
private final Admin admin = new Admin();
private final Defaults defaults = new Defaults();
private final Function function = new Function();
private final Producer producer = new Producer();
private final Consumer consumer = new Consumer();
private final Listener listener = new Listener();
private final Reader reader = new Reader();
private final Template template = new Template();
public Client getClient() {
return this.client;
}
public Admin getAdmin() {
return this.admin;
}
public Defaults getDefaults() {
return this.defaults;
}
public Producer getProducer() {
return this.producer;
}
public Consumer getConsumer() {
return this.consumer;
}
public Listener getListener() {
return this.listener;
}
public Reader getReader() {
return this.reader;
}
public Function getFunction() {
return this.function;
}
public Template getTemplate() {
return this.template;
}
public static class Client {
/**
* Pulsar service URL in the format '(pulsar|pulsar+ssl)://host:port'.
*/
private String serviceUrl = "pulsar://localhost:6650";
/**
* Client operation timeout.
*/
private Duration operationTimeout = Duration.ofSeconds(30);
/**
* Client lookup timeout.
*/
private Duration lookupTimeout = Duration.ofMillis(-1); // FIXME
/**
* Duration to wait for a connection to a broker to be established.
*/
private Duration connectionTimeout = Duration.ofSeconds(10);
/**
* Authentication settings.
*/
private final Authentication authentication = new Authentication();
public String getServiceUrl() {
return this.serviceUrl;
}
public void setServiceUrl(String serviceUrl) {
this.serviceUrl = serviceUrl;
}
public Duration getOperationTimeout() {
return this.operationTimeout;
}
public void setOperationTimeout(Duration operationTimeout) {
this.operationTimeout = operationTimeout;
}
public Duration getLookupTimeout() {
return this.lookupTimeout;
}
public void setLookupTimeout(Duration lookupTimeout) {
this.lookupTimeout = lookupTimeout;
}
public Duration getConnectionTimeout() {
return this.connectionTimeout;
}
public void setConnectionTimeout(Duration connectionTimeout) {
this.connectionTimeout = connectionTimeout;
}
public Authentication getAuthentication() {
return this.authentication;
}
}
public static class Admin {
/**
* Pulsar web URL for the admin endpoint in the format '(http|https)://host:port'.
*/
private String serviceUrl = "http://localhost:8080";
/**
* Duration to wait for a connection to server to be established.
*/
private Duration connectionTimeout = Duration.ofMinutes(1);
/**
* Server response read time out for any request.
*/
private Duration readTimeout = Duration.ofMinutes(1);
/**
* Server request time out for any request.
*/
private Duration requestTimeout = Duration.ofMinutes(5);
/**
* Authentication settings.
*/
private final Authentication authentication = new Authentication();
public String getServiceUrl() {
return this.serviceUrl;
}
public void setServiceUrl(String serviceUrl) {
this.serviceUrl = serviceUrl;
}
public Duration getConnectionTimeout() {
return this.connectionTimeout;
}
public void setConnectionTimeout(Duration connectionTimeout) {
this.connectionTimeout = connectionTimeout;
}
public Duration getReadTimeout() {
return this.readTimeout;
}
public void setReadTimeout(Duration readTimeout) {
this.readTimeout = readTimeout;
}
public Duration getRequestTimeout() {
return this.requestTimeout;
}
public void setRequestTimeout(Duration requestTimeout) {
this.requestTimeout = requestTimeout;
}
public Authentication getAuthentication() {
return this.authentication;
}
}
public static class Defaults {
/**
* List of mappings from message type to topic name and schema info to use as a
* defaults when a topic name and/or schema is not explicitly specified when
* producing or consuming messages of the mapped type.
*/
private List<TypeMapping> typeMappings = new ArrayList<>();
public List<TypeMapping> getTypeMappings() {
return this.typeMappings;
}
public void setTypeMappings(List<TypeMapping> typeMappings) {
this.typeMappings = typeMappings;
}
/**
* A mapping from message type to topic and/or schema info to use (at least one of
* {@code topicName} or {@code schemaInfo} must be specified.
*
* @param messageType the message type
* @param topicName the topic name
* @param schemaInfo the schema info
*/
public record TypeMapping(Class<?> messageType, String topicName, SchemaInfo schemaInfo) {
public TypeMapping {
Assert.notNull(messageType, "messageType must not be null");
Assert.isTrue(topicName != null || schemaInfo != null,
"At least one of topicName or schemaInfo must not be null");
}
}
/**
* Represents a schema - holds enough information to construct an actual schema
* instance.
*
* @param schemaType schema type
* @param messageKeyType message key type (required for key value type)
*/
public record SchemaInfo(SchemaType schemaType, Class<?> messageKeyType) {
public SchemaInfo {
Assert.notNull(schemaType, "schemaType must not be null");
Assert.isTrue(schemaType != SchemaType.NONE, "schemaType 'NONE' not supported");
Assert.isTrue(messageKeyType == null || schemaType == SchemaType.KEY_VALUE,
"messageKeyType can only be set when schemaType is KEY_VALUE");
}
}
}
public static class Function {
/**
* Whether to stop processing further function creates/updates when a failure
* occurs.
*/
private boolean failFast = true;
/**
* Whether to throw an exception if any failure is encountered during server
* startup while creating/updating functions.
*/
private boolean propagateFailures = true;
/**
* Whether to throw an exception if any failure is encountered during server
* shutdown while enforcing stop policy on functions.
*/
private boolean propagateStopFailures = false;
public boolean isFailFast() {
return this.failFast;
}
public void setFailFast(boolean failFast) {
this.failFast = failFast;
}
public boolean isPropagateFailures() {
return this.propagateFailures;
}
public void setPropagateFailures(boolean propagateFailures) {
this.propagateFailures = propagateFailures;
}
public boolean isPropagateStopFailures() {
return this.propagateStopFailures;
}
public void setPropagateStopFailures(boolean propagateStopFailures) {
this.propagateStopFailures = propagateStopFailures;
}
}
public static class Producer {
/**
* Name for the producer. If not assigned, a unique name is generated.
*/
private String name;
/**
* Topic the producer will publish to.
*/
private String topicName;
/**
* Time before a message has to be acknowledged by the broker.
*/
private Duration sendTimeout = Duration.ofSeconds(30);
/**
* Message routing mode for a partitioned producer.
*/
private MessageRoutingMode messageRoutingMode = MessageRoutingMode.RoundRobinPartition;
/**
* Message hashing scheme to choose the partition to which the message is
* published.
*/
private HashingScheme hashingScheme = HashingScheme.JavaStringHash;
/**
* Whether to automatically batch messages.
*/
private boolean batchingEnabled = true;
/**
* Whether to split large-size messages into multiple chunks.
*/
private boolean chunkingEnabled;
/**
* Message compression type.
*/
private CompressionType compressionType;
/**
* Type of access to the topic the producer requires.
*/
private ProducerAccessMode accessMode = ProducerAccessMode.Shared;
private final Cache cache = new Cache();
public String getName() {
return this.name;
}
public void setName(String name) {
this.name = name;
}
public String getTopicName() {
return this.topicName;
}
public void setTopicName(String topicName) {
this.topicName = topicName;
}
public Duration getSendTimeout() {
return this.sendTimeout;
}
public void setSendTimeout(Duration sendTimeout) {
this.sendTimeout = sendTimeout;
}
public MessageRoutingMode getMessageRoutingMode() {
return this.messageRoutingMode;
}
public void setMessageRoutingMode(MessageRoutingMode messageRoutingMode) {
this.messageRoutingMode = messageRoutingMode;
}
public HashingScheme getHashingScheme() {
return this.hashingScheme;
}
public void setHashingScheme(HashingScheme hashingScheme) {
this.hashingScheme = hashingScheme;
}
public boolean isBatchingEnabled() {
return this.batchingEnabled;
}
public void setBatchingEnabled(boolean batchingEnabled) {
this.batchingEnabled = batchingEnabled;
}
public boolean isChunkingEnabled() {
return this.chunkingEnabled;
}
public void setChunkingEnabled(boolean chunkingEnabled) {
this.chunkingEnabled = chunkingEnabled;
}
public CompressionType getCompressionType() {
return this.compressionType;
}
public void setCompressionType(CompressionType compressionType) {
this.compressionType = compressionType;
}
public ProducerAccessMode getAccessMode() {
return this.accessMode;
}
public void setAccessMode(ProducerAccessMode accessMode) {
this.accessMode = accessMode;
}
public Cache getCache() {
return this.cache;
}
public static class Cache {
/**
* Time period to expire unused entries in the cache.
*/
private Duration expireAfterAccess = Duration.ofMinutes(1);
/**
* Maximum size of cache (entries).
*/
private long maximumSize = 1000L;
/**
* Initial size of cache.
*/
private int initialCapacity = 50;
public Duration getExpireAfterAccess() {
return this.expireAfterAccess;
}
public void setExpireAfterAccess(Duration expireAfterAccess) {
this.expireAfterAccess = expireAfterAccess;
}
public long getMaximumSize() {
return this.maximumSize;
}
public void setMaximumSize(long maximumSize) {
this.maximumSize = maximumSize;
}
public int getInitialCapacity() {
return this.initialCapacity;
}
public void setInitialCapacity(int initialCapacity) {
this.initialCapacity = initialCapacity;
}
}
}
public static class Consumer {
/**
* Consumer name to identify a particular consumer from the topic stats.
*/
private String name;
/**
* Topics the consumer subscribes to.
*/
private List<String> topics;
/**
* Pattern for topics the consumer subscribes to.
*/
private Pattern topicsPattern;
/**
* Priority level for shared subscription consumers.
*/
private int priorityLevel = 0;
/**
* Whether to read messages from the compacted topic rather than the full message
* backlog.
*/
private boolean readCompacted = false;
/**
* Dead letter policy to use.
*/
@NestedConfigurationProperty
private DeadLetterPolicy deadLetterPolicy;
/**
* Consumer subscription properties.
*/
private final Subscription subscription = new Subscription();
/**
* Whether to auto retry messages.
*/
private boolean retryEnable = false;
public String getName() {
return this.name;
}
public void setName(String name) {
this.name = name;
}
public Consumer.Subscription getSubscription() {
return this.subscription;
}
public List<String> getTopics() {
return this.topics;
}
public void setTopics(List<String> topics) {
this.topics = topics;
}
public Pattern getTopicsPattern() {
return this.topicsPattern;
}
public void setTopicsPattern(Pattern topicsPattern) {
this.topicsPattern = topicsPattern;
}
public int getPriorityLevel() {
return this.priorityLevel;
}
public void setPriorityLevel(int priorityLevel) {
this.priorityLevel = priorityLevel;
}
public boolean isReadCompacted() {
return this.readCompacted;
}
public void setReadCompacted(boolean readCompacted) {
this.readCompacted = readCompacted;
}
public DeadLetterPolicy getDeadLetterPolicy() {
return this.deadLetterPolicy;
}
public void setDeadLetterPolicy(DeadLetterPolicy deadLetterPolicy) {
this.deadLetterPolicy = deadLetterPolicy;
}
public boolean isRetryEnable() {
return this.retryEnable;
}
public void setRetryEnable(boolean retryEnable) {
this.retryEnable = retryEnable;
}
public static class Subscription {
/**
* Subscription name for the consumer.
*/
private String name;
/**
* Position where to initialize a newly created subscription.
*/
private SubscriptionInitialPosition initialPosition = SubscriptionInitialPosition.Latest;
/**
* Subscription mode to be used when subscribing to the topic.
*/
private SubscriptionMode mode = SubscriptionMode.Durable;
/**
* Determines which type of topics (persistent, non-persistent, or all) the
* consumer should be subscribed to when using pattern subscriptions.
*/
private RegexSubscriptionMode topicsMode = RegexSubscriptionMode.PersistentOnly;
/**
* Subscription type to be used when subscribing to a topic.
*/
private SubscriptionType type = SubscriptionType.Exclusive;
public String getName() {
return this.name;
}
public void setName(String name) {
this.name = name;
}
public SubscriptionInitialPosition getInitialPosition() {
return this.initialPosition;
}
public void setInitialPosition(SubscriptionInitialPosition initialPosition) {
this.initialPosition = initialPosition;
}
public SubscriptionMode getMode() {
return this.mode;
}
public void setMode(SubscriptionMode mode) {
this.mode = mode;
}
public RegexSubscriptionMode getTopicsMode() {
return this.topicsMode;
}
public void setTopicsMode(RegexSubscriptionMode topicsMode) {
this.topicsMode = topicsMode;
}
public SubscriptionType getType() {
return this.type;
}
public void setType(SubscriptionType type) {
this.type = type;
}
}
public static class DeadLetterPolicy {
/**
* Maximum number of times that a message will be redelivered before being
* sent to the dead letter queue.
*/
private int maxRedeliverCount;
/**
* Name of the retry topic where the failing messages will be sent.
*/
private String retryLetterTopic;
/**
* Name of the dead topic where the failing messages will be sent.
*/
private String deadLetterTopic;
/**
* Name of the initial subscription of the dead letter topic. When not set,
* the initial subscription will not be created. However, when the property is
* set then the broker's 'allowAutoSubscriptionCreation' must be enabled or
* the DLQ producer will fail.
*/
private String initialSubscriptionName;
public int getMaxRedeliverCount() {
return this.maxRedeliverCount;
}
public void setMaxRedeliverCount(int maxRedeliverCount) {
this.maxRedeliverCount = maxRedeliverCount;
}
public String getRetryLetterTopic() {
return this.retryLetterTopic;
}
public void setRetryLetterTopic(String retryLetterTopic) {
this.retryLetterTopic = retryLetterTopic;
}
public String getDeadLetterTopic() {
return this.deadLetterTopic;
}
public void setDeadLetterTopic(String deadLetterTopic) {
this.deadLetterTopic = deadLetterTopic;
}
public String getInitialSubscriptionName() {
return this.initialSubscriptionName;
}
public void setInitialSubscriptionName(String initialSubscriptionName) {
this.initialSubscriptionName = initialSubscriptionName;
}
}
}
public static class Listener {
/**
* SchemaType of the consumed messages.
*/
private SchemaType schemaType;
/**
* Whether to record observations for when the Observations API is available and
* the client supports it.
*/
private boolean observationEnabled = true;
public SchemaType getSchemaType() {
return this.schemaType;
}
public void setSchemaType(SchemaType schemaType) {
this.schemaType = schemaType;
}
public boolean isObservationEnabled() {
return this.observationEnabled;
}
public void setObservationEnabled(boolean observationEnabled) {
this.observationEnabled = observationEnabled;
}
}
public static class Reader {
/**
* Reader name.
*/
private String name;
/**
* Topis the reader subscribes to.
*/
private List<String> topics;
/**
* Subscription name.
*/
private String subscriptionName;
/**
* Prefix of subscription role.
*/
private String subscriptionRolePrefix;
/**
* Whether to read messages from a compacted topic rather than a full message
* backlog of a topic.
*/
private boolean readCompacted;
public String getName() {
return this.name;
}
public void setName(String name) {
this.name = name;
}
public List<String> getTopics() {
return this.topics;
}
public void setTopics(List<String> topics) {
this.topics = topics;
}
public String getSubscriptionName() {
return this.subscriptionName;
}
public void setSubscriptionName(String subscriptionName) {
this.subscriptionName = subscriptionName;
}
public String getSubscriptionRolePrefix() {
return this.subscriptionRolePrefix;
}
public void setSubscriptionRolePrefix(String subscriptionRolePrefix) {
this.subscriptionRolePrefix = subscriptionRolePrefix;
}
public boolean isReadCompacted() {
return this.readCompacted;
}
public void setReadCompacted(boolean readCompacted) {
this.readCompacted = readCompacted;
}
}
public static class Template {
/**
* Whether to record observations for when the Observations API is available.
*/
private boolean observationsEnabled = true;
public boolean isObservationsEnabled() {
return this.observationsEnabled;
}
public void setObservationsEnabled(boolean observationsEnabled) {
this.observationsEnabled = observationsEnabled;
}
}
public static class Authentication {
/**
* Fully qualified class name of the authentication plugin.
*/
private String pluginClassName;
/**
* Authentication parameter(s) as a map of parameter names to parameter values.
*/
private Map<String, String> param = new LinkedHashMap<>();
public String getPluginClassName() {
return this.pluginClassName;
}
public void setPluginClassName(String pluginClassName) {
this.pluginClassName = pluginClassName;
}
public Map<String, String> getParam() {
return this.param;
}
public void setParam(Map<String, String> param) {
this.param = param;
}
}
}

View File

@@ -0,0 +1,166 @@
/*
* Copyright 2012-2023 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* https://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.boot.autoconfigure.pulsar;
import java.time.Duration;
import java.util.ArrayList;
import java.util.Map;
import java.util.concurrent.TimeUnit;
import java.util.function.BiConsumer;
import java.util.function.Consumer;
import org.apache.pulsar.client.admin.PulsarAdminBuilder;
import org.apache.pulsar.client.api.ClientBuilder;
import org.apache.pulsar.client.api.ConsumerBuilder;
import org.apache.pulsar.client.api.ProducerBuilder;
import org.apache.pulsar.client.api.PulsarClientException.UnsupportedAuthenticationException;
import org.apache.pulsar.client.api.ReaderBuilder;
import org.springframework.boot.context.properties.PropertyMapper;
import org.springframework.pulsar.listener.PulsarContainerProperties;
import org.springframework.pulsar.reader.PulsarReaderContainerProperties;
import org.springframework.util.StringUtils;
/**
* Helper class used to map {@link PulsarProperties} to various builder customizers.
*
* @author Chris Bono
* @author Phillip Webb
*/
final class PulsarPropertiesMapper {
private final PulsarProperties properties;
PulsarPropertiesMapper(PulsarProperties properties) {
this.properties = properties;
}
void customizeClientBuilder(ClientBuilder clientBuilder) {
PulsarProperties.Client properties = this.properties.getClient();
PropertyMapper map = PropertyMapper.get().alwaysApplyingWhenNonNull();
map.from(properties::getServiceUrl).to(clientBuilder::serviceUrl);
map.from(properties::getConnectionTimeout).to(timeoutProperty(clientBuilder::connectionTimeout));
map.from(properties::getOperationTimeout).to(timeoutProperty(clientBuilder::operationTimeout));
map.from(properties::getLookupTimeout).to(timeoutProperty(clientBuilder::lookupTimeout));
customizeAuthentication(clientBuilder::authentication, properties.getAuthentication());
}
void customizeAdminBuilder(PulsarAdminBuilder adminBuilder) {
PulsarProperties.Admin properties = this.properties.getAdmin();
PropertyMapper map = PropertyMapper.get().alwaysApplyingWhenNonNull();
map.from(properties::getServiceUrl).to(adminBuilder::serviceHttpUrl);
map.from(properties::getConnectionTimeout).to(timeoutProperty(adminBuilder::connectionTimeout));
map.from(properties::getReadTimeout).to(timeoutProperty(adminBuilder::readTimeout));
map.from(properties::getRequestTimeout).to(timeoutProperty(adminBuilder::requestTimeout));
customizeAuthentication(adminBuilder::authentication, properties.getAuthentication());
}
private void customizeAuthentication(AuthenticationConsumer authentication,
PulsarProperties.Authentication properties) {
if (StringUtils.hasText(properties.getPluginClassName())) {
try {
authentication.accept(properties.getPluginClassName(), properties.getParam());
}
catch (UnsupportedAuthenticationException ex) {
throw new IllegalStateException("Unable to configure Pulsar authentication", ex);
}
}
}
<T> void customizeProducerBuilder(ProducerBuilder<T> producerBuilder) {
PulsarProperties.Producer properties = this.properties.getProducer();
PropertyMapper map = PropertyMapper.get().alwaysApplyingWhenNonNull();
map.from(properties::getName).to(producerBuilder::producerName);
map.from(properties::getTopicName).to(producerBuilder::topic);
map.from(properties::getSendTimeout).to(timeoutProperty(producerBuilder::sendTimeout));
map.from(properties::getMessageRoutingMode).to(producerBuilder::messageRoutingMode);
map.from(properties::getHashingScheme).to(producerBuilder::hashingScheme);
map.from(properties::isBatchingEnabled).to(producerBuilder::enableBatching);
map.from(properties::isChunkingEnabled).to(producerBuilder::enableChunking);
map.from(properties::getCompressionType).to(producerBuilder::compressionType);
map.from(properties::getAccessMode).to(producerBuilder::accessMode);
}
<T> void customizeConsumerBuilder(ConsumerBuilder<T> consumerBuilder) {
PulsarProperties.Consumer properties = this.properties.getConsumer();
PropertyMapper map = PropertyMapper.get().alwaysApplyingWhenNonNull();
map.from(properties::getName).to(consumerBuilder::consumerName);
map.from(properties::getTopics).as(ArrayList::new).to(consumerBuilder::topics);
map.from(properties::getTopicsPattern).to(consumerBuilder::topicsPattern);
map.from(properties::getPriorityLevel).to(consumerBuilder::priorityLevel);
map.from(properties::isReadCompacted).to(consumerBuilder::readCompacted);
map.from(properties::getDeadLetterPolicy).as(DeadLetterPolicyMapper::map).to(consumerBuilder::deadLetterPolicy);
map.from(properties::isRetryEnable).to(consumerBuilder::enableRetry);
customizeConsumerBuilderSubscription(consumerBuilder);
}
private void customizeConsumerBuilderSubscription(ConsumerBuilder<?> consumerBuilder) {
PulsarProperties.Consumer.Subscription properties = this.properties.getConsumer().getSubscription();
PropertyMapper map = PropertyMapper.get().alwaysApplyingWhenNonNull();
map.from(properties::getName).to(consumerBuilder::subscriptionName);
map.from(properties::getInitialPosition).to(consumerBuilder::subscriptionInitialPosition);
map.from(properties::getMode).to(consumerBuilder::subscriptionMode);
map.from(properties::getTopicsMode).to(consumerBuilder::subscriptionTopicsMode);
map.from(properties::getType).to(consumerBuilder::subscriptionType);
}
void customizeContainerProperties(PulsarContainerProperties containerProperties) {
customizePulsarContainerConsumerSubscriptionProperties(containerProperties);
customizePulsarContainerListenerProperties(containerProperties);
}
private void customizePulsarContainerConsumerSubscriptionProperties(PulsarContainerProperties containerProperties) {
PulsarProperties.Consumer.Subscription properties = this.properties.getConsumer().getSubscription();
PropertyMapper map = PropertyMapper.get().alwaysApplyingWhenNonNull();
map.from(properties::getType).to(containerProperties::setSubscriptionType);
}
private void customizePulsarContainerListenerProperties(PulsarContainerProperties containerProperties) {
PulsarProperties.Listener properties = this.properties.getListener();
PropertyMapper map = PropertyMapper.get().alwaysApplyingWhenNonNull();
map.from(properties::getSchemaType).to(containerProperties::setSchemaType);
map.from(properties::isObservationEnabled).to(containerProperties::setObservationEnabled);
}
<T> void customizeReaderBuilder(ReaderBuilder<T> readerBuilder) {
PulsarProperties.Reader properties = this.properties.getReader();
PropertyMapper map = PropertyMapper.get().alwaysApplyingWhenNonNull();
map.from(properties::getName).to(readerBuilder::readerName);
map.from(properties::getTopics).to(readerBuilder::topics);
map.from(properties::getSubscriptionName).to(readerBuilder::subscriptionName);
map.from(properties::getSubscriptionRolePrefix).to(readerBuilder::subscriptionRolePrefix);
map.from(properties::isReadCompacted).to(readerBuilder::readCompacted);
}
void customizeReaderContainerProperties(PulsarReaderContainerProperties readerContainerProperties) {
PulsarProperties.Reader properties = this.properties.getReader();
PropertyMapper map = PropertyMapper.get().alwaysApplyingWhenNonNull();
map.from(properties::getTopics).to(readerContainerProperties::setTopics);
}
private Consumer<Duration> timeoutProperty(BiConsumer<Integer, TimeUnit> setter) {
return (duration) -> setter.accept((int) duration.toMillis(), TimeUnit.MILLISECONDS);
}
private interface AuthenticationConsumer {
void accept(String authPluginClassName, Map<String, String> authParams)
throws UnsupportedAuthenticationException;
}
}

View File

@@ -0,0 +1,201 @@
/*
* Copyright 2012-2023 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* https://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.boot.autoconfigure.pulsar;
import java.time.Duration;
import java.util.ArrayList;
import java.util.List;
import org.apache.pulsar.client.api.PulsarClient;
import org.apache.pulsar.reactive.client.adapter.AdaptedReactivePulsarClientFactory;
import org.apache.pulsar.reactive.client.adapter.ProducerCacheProvider;
import org.apache.pulsar.reactive.client.api.ReactiveMessageConsumerBuilder;
import org.apache.pulsar.reactive.client.api.ReactiveMessageReaderBuilder;
import org.apache.pulsar.reactive.client.api.ReactiveMessageSenderBuilder;
import org.apache.pulsar.reactive.client.api.ReactiveMessageSenderCache;
import org.apache.pulsar.reactive.client.api.ReactivePulsarClient;
import org.apache.pulsar.reactive.client.producercache.CaffeineShadedProducerCacheProvider;
import org.springframework.beans.factory.ObjectProvider;
import org.springframework.boot.autoconfigure.AutoConfiguration;
import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.boot.util.LambdaSafe;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Import;
import org.springframework.pulsar.config.PulsarAnnotationSupportBeanNames;
import org.springframework.pulsar.core.SchemaResolver;
import org.springframework.pulsar.core.TopicResolver;
import org.springframework.pulsar.reactive.config.DefaultReactivePulsarListenerContainerFactory;
import org.springframework.pulsar.reactive.config.annotation.EnableReactivePulsar;
import org.springframework.pulsar.reactive.core.DefaultReactivePulsarConsumerFactory;
import org.springframework.pulsar.reactive.core.DefaultReactivePulsarReaderFactory;
import org.springframework.pulsar.reactive.core.DefaultReactivePulsarSenderFactory;
import org.springframework.pulsar.reactive.core.ReactiveMessageConsumerBuilderCustomizer;
import org.springframework.pulsar.reactive.core.ReactiveMessageReaderBuilderCustomizer;
import org.springframework.pulsar.reactive.core.ReactiveMessageSenderBuilderCustomizer;
import org.springframework.pulsar.reactive.core.ReactivePulsarConsumerFactory;
import org.springframework.pulsar.reactive.core.ReactivePulsarReaderFactory;
import org.springframework.pulsar.reactive.core.ReactivePulsarSenderFactory;
import org.springframework.pulsar.reactive.core.ReactivePulsarTemplate;
import org.springframework.pulsar.reactive.listener.ReactivePulsarContainerProperties;
/**
* {@link EnableAutoConfiguration Auto-configuration} for Spring for Apache Pulsar
* Reactive.
*
* @author Chris Bono
* @author Christophe Bornet
* @since 3.2.0
*/
@AutoConfiguration(after = PulsarAutoConfiguration.class)
@ConditionalOnClass({ PulsarClient.class, ReactivePulsarClient.class, ReactivePulsarTemplate.class })
@Import(PulsarConfiguration.class)
public class PulsarReactiveAutoConfiguration {
private final PulsarProperties properties;
private final PulsarReactivePropertiesMapper propertiesMapper;
PulsarReactiveAutoConfiguration(PulsarProperties properties) {
this.properties = properties;
this.propertiesMapper = new PulsarReactivePropertiesMapper(properties);
}
@Bean
@ConditionalOnMissingBean
ReactivePulsarClient reactivePulsarClient(PulsarClient pulsarClient) {
return AdaptedReactivePulsarClientFactory.create(pulsarClient);
}
@Bean
@ConditionalOnMissingBean(ProducerCacheProvider.class)
@ConditionalOnClass(CaffeineShadedProducerCacheProvider.class)
@ConditionalOnProperty(name = "spring.pulsar.producer.cache.enabled", havingValue = "true", matchIfMissing = true)
CaffeineShadedProducerCacheProvider reactivePulsarProducerCacheProvider() {
PulsarProperties.Producer.Cache properties = this.properties.getProducer().getCache();
return new CaffeineShadedProducerCacheProvider(properties.getExpireAfterAccess(), Duration.ofMinutes(10),
properties.getMaximumSize(), properties.getInitialCapacity());
}
@Bean
@ConditionalOnMissingBean
@ConditionalOnProperty(name = "spring.pulsar.producer.cache.enabled", havingValue = "true", matchIfMissing = true)
ReactiveMessageSenderCache reactivePulsarMessageSenderCache(
ObjectProvider<ProducerCacheProvider> producerCacheProvider) {
return reactivePulsarMessageSenderCache(producerCacheProvider.getIfAvailable());
}
private ReactiveMessageSenderCache reactivePulsarMessageSenderCache(ProducerCacheProvider producerCacheProvider) {
return (producerCacheProvider != null) ? AdaptedReactivePulsarClientFactory.createCache(producerCacheProvider)
: AdaptedReactivePulsarClientFactory.createCache();
}
@Bean
@ConditionalOnMissingBean(ReactivePulsarSenderFactory.class)
DefaultReactivePulsarSenderFactory<?> reactivePulsarSenderFactory(ReactivePulsarClient reactivePulsarClient,
ObjectProvider<ReactiveMessageSenderCache> reactiveMessageSenderCache, TopicResolver topicResolver,
ObjectProvider<ReactiveMessageSenderBuilderCustomizer<?>> customizersProvider) {
List<ReactiveMessageSenderBuilderCustomizer<?>> customizers = new ArrayList<>();
customizers.add(this.propertiesMapper::customizeMessageSenderBuilder);
customizers.addAll(customizersProvider.orderedStream().toList());
List<ReactiveMessageSenderBuilderCustomizer<Object>> lambdaSafeCustomizers = List
.of((builder) -> applyMessageSenderBuilderCustomizers(customizers, builder));
return DefaultReactivePulsarSenderFactory.builderFor(reactivePulsarClient)
.withDefaultConfigCustomizers(lambdaSafeCustomizers)
.withMessageSenderCache(reactiveMessageSenderCache.getIfAvailable())
.withTopicResolver(topicResolver)
.build();
}
@SuppressWarnings("unchecked")
private void applyMessageSenderBuilderCustomizers(List<ReactiveMessageSenderBuilderCustomizer<?>> customizers,
ReactiveMessageSenderBuilder<?> builder) {
LambdaSafe.callbacks(ReactiveMessageSenderBuilderCustomizer.class, customizers, builder)
.invoke((customizer) -> customizer.customize(builder));
}
@Bean
@ConditionalOnMissingBean(ReactivePulsarConsumerFactory.class)
DefaultReactivePulsarConsumerFactory<?> reactivePulsarConsumerFactory(
ReactivePulsarClient pulsarReactivePulsarClient,
ObjectProvider<ReactiveMessageConsumerBuilderCustomizer<?>> customizersProvider) {
List<ReactiveMessageConsumerBuilderCustomizer<?>> customizers = new ArrayList<>();
customizers.add(this.propertiesMapper::customizeMessageConsumerBuilder);
customizers.addAll(customizersProvider.orderedStream().toList());
List<ReactiveMessageConsumerBuilderCustomizer<Object>> lambdaSafeCustomizers = List
.of((builder) -> applyMessageConsumerBuilderCustomizers(customizers, builder));
return new DefaultReactivePulsarConsumerFactory<>(pulsarReactivePulsarClient, lambdaSafeCustomizers);
}
@SuppressWarnings("unchecked")
private void applyMessageConsumerBuilderCustomizers(List<ReactiveMessageConsumerBuilderCustomizer<?>> customizers,
ReactiveMessageConsumerBuilder<?> builder) {
LambdaSafe.callbacks(ReactiveMessageConsumerBuilderCustomizer.class, customizers, builder)
.invoke((customizer) -> customizer.customize(builder));
}
@Bean
@ConditionalOnMissingBean(name = "reactivePulsarListenerContainerFactory")
DefaultReactivePulsarListenerContainerFactory<?> reactivePulsarListenerContainerFactory(
ReactivePulsarConsumerFactory<Object> reactivePulsarConsumerFactory, SchemaResolver schemaResolver,
TopicResolver topicResolver) {
ReactivePulsarContainerProperties<Object> containerProperties = new ReactivePulsarContainerProperties<>();
containerProperties.setSchemaResolver(schemaResolver);
containerProperties.setTopicResolver(topicResolver);
this.propertiesMapper.customizeContainerProperties(containerProperties);
return new DefaultReactivePulsarListenerContainerFactory<>(reactivePulsarConsumerFactory, containerProperties);
}
@Bean
@ConditionalOnMissingBean(ReactivePulsarReaderFactory.class)
DefaultReactivePulsarReaderFactory<?> reactivePulsarReaderFactory(ReactivePulsarClient reactivePulsarClient,
ObjectProvider<ReactiveMessageReaderBuilderCustomizer<?>> customizersProvider) {
List<ReactiveMessageReaderBuilderCustomizer<?>> customizers = new ArrayList<>();
customizers.add(this.propertiesMapper::customizeMessageReaderBuilder);
customizers.addAll(customizersProvider.orderedStream().toList());
List<ReactiveMessageReaderBuilderCustomizer<Object>> lambdaSafeCustomizers = List
.of((builder) -> applyMessageReaderBuilderCustomizers(customizers, builder));
return new DefaultReactivePulsarReaderFactory<>(reactivePulsarClient, lambdaSafeCustomizers);
}
@SuppressWarnings("unchecked")
private void applyMessageReaderBuilderCustomizers(List<ReactiveMessageReaderBuilderCustomizer<?>> customizers,
ReactiveMessageReaderBuilder<?> builder) {
LambdaSafe.callbacks(ReactiveMessageReaderBuilderCustomizer.class, customizers, builder)
.invoke((customizer) -> customizer.customize(builder));
}
@Bean
@ConditionalOnMissingBean
ReactivePulsarTemplate<?> pulsarReactiveTemplate(ReactivePulsarSenderFactory<?> reactivePulsarSenderFactory,
SchemaResolver schemaResolver, TopicResolver topicResolver) {
return new ReactivePulsarTemplate<>(reactivePulsarSenderFactory, schemaResolver, topicResolver);
}
@Configuration(proxyBeanMethods = false)
@EnableReactivePulsar
@ConditionalOnMissingBean(
name = PulsarAnnotationSupportBeanNames.REACTIVE_PULSAR_LISTENER_ANNOTATION_PROCESSOR_BEAN_NAME)
static class EnableReactivePulsarConfiguration {
}
}

View File

@@ -0,0 +1,108 @@
/*
* Copyright 2012-2023 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* https://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.boot.autoconfigure.pulsar;
import java.util.ArrayList;
import org.apache.pulsar.reactive.client.api.ReactiveMessageConsumerBuilder;
import org.apache.pulsar.reactive.client.api.ReactiveMessageReaderBuilder;
import org.apache.pulsar.reactive.client.api.ReactiveMessageSenderBuilder;
import org.springframework.boot.context.properties.PropertyMapper;
import org.springframework.pulsar.reactive.listener.ReactivePulsarContainerProperties;
/**
* Helper class used to map reactive {@link PulsarProperties} to various builder
* customizers.
*
* @author Chris Bono
* @author Phillip Webb
*/
final class PulsarReactivePropertiesMapper {
private final PulsarProperties properties;
PulsarReactivePropertiesMapper(PulsarProperties properties) {
this.properties = properties;
}
<T> void customizeMessageSenderBuilder(ReactiveMessageSenderBuilder<T> builder) {
PulsarProperties.Producer properties = this.properties.getProducer();
PropertyMapper map = PropertyMapper.get().alwaysApplyingWhenNonNull();
map.from(properties::getName).to(builder::producerName);
map.from(properties::getTopicName).to(builder::topic);
map.from(properties::getSendTimeout).to(builder::sendTimeout);
map.from(properties::getMessageRoutingMode).to(builder::messageRoutingMode);
map.from(properties::getHashingScheme).to(builder::hashingScheme);
map.from(properties::isBatchingEnabled).to(builder::batchingEnabled);
map.from(properties::isChunkingEnabled).to(builder::chunkingEnabled);
map.from(properties::getCompressionType).to(builder::compressionType);
map.from(properties::getAccessMode).to(builder::accessMode);
}
<T> void customizeMessageConsumerBuilder(ReactiveMessageConsumerBuilder<T> builder) {
PulsarProperties.Consumer properties = this.properties.getConsumer();
PropertyMapper map = PropertyMapper.get().alwaysApplyingWhenNonNull();
map.from(properties::getName).to(builder::consumerName);
map.from(properties::getTopics).as(ArrayList::new).to(builder::topics);
map.from(properties::getTopicsPattern).to(builder::topicsPattern);
map.from(properties::getPriorityLevel).to(builder::priorityLevel);
map.from(properties::isReadCompacted).to(builder::readCompacted);
map.from(properties::getDeadLetterPolicy).as(DeadLetterPolicyMapper::map).to(builder::deadLetterPolicy);
map.from(properties::isRetryEnable).to(builder::retryLetterTopicEnable);
customizerMessageConsumerBuilderSubscription(builder);
}
private <T> void customizerMessageConsumerBuilderSubscription(ReactiveMessageConsumerBuilder<T> builder) {
PulsarProperties.Consumer.Subscription properties = this.properties.getConsumer().getSubscription();
PropertyMapper map = PropertyMapper.get().alwaysApplyingWhenNonNull();
map.from(properties::getName).to(builder::subscriptionName);
map.from(properties::getInitialPosition).to(builder::subscriptionInitialPosition);
map.from(properties::getMode).to(builder::subscriptionMode);
map.from(properties::getTopicsMode).to(builder::topicsPatternSubscriptionMode);
map.from(properties::getType).to(builder::subscriptionType);
}
<T> void customizeContainerProperties(ReactivePulsarContainerProperties<T> containerProperties) {
customizePulsarContainerConsumerSubscriptionProperties(containerProperties);
customizePulsarContainerListenerProperties(containerProperties);
}
private void customizePulsarContainerConsumerSubscriptionProperties(
ReactivePulsarContainerProperties<?> containerProperties) {
PulsarProperties.Consumer.Subscription properties = this.properties.getConsumer().getSubscription();
PropertyMapper map = PropertyMapper.get().alwaysApplyingWhenNonNull();
map.from(properties::getType).to(containerProperties::setSubscriptionType);
}
private void customizePulsarContainerListenerProperties(ReactivePulsarContainerProperties<?> containerProperties) {
PulsarProperties.Listener properties = this.properties.getListener();
PropertyMapper map = PropertyMapper.get().alwaysApplyingWhenNonNull();
map.from(properties::getSchemaType).to(containerProperties::setSchemaType);
}
void customizeMessageReaderBuilder(ReactiveMessageReaderBuilder<?> builder) {
PulsarProperties.Reader properties = this.properties.getReader();
PropertyMapper map = PropertyMapper.get().alwaysApplyingWhenNonNull();
map.from(properties::getName).to(builder::readerName);
map.from(properties::getTopics).to(builder::topics);
map.from(properties::getSubscriptionName).to(builder::subscriptionName);
map.from(properties::getSubscriptionRolePrefix).to(builder::generatedSubscriptionNamePrefix);
map.from(properties::isReadCompacted).to(builder::readCompacted);
}
}

View File

@@ -0,0 +1,20 @@
/*
* Copyright 2012-2023 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* https://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
/**
* Auto-configuration for Spring for Apache Pulsar.
*/
package org.springframework.boot.autoconfigure.pulsar;

View File

@@ -2003,6 +2003,18 @@
"name": "spring.neo4j.uri",
"defaultValue": "bolt://localhost:7687"
},
{
"name": "spring.pulsar.function.enabled",
"type": "java.lang.Boolean",
"description": "Whether to enable function support.",
"defaultValue": true
},
{
"name": "spring.pulsar.producer.cache.enabled",
"type": "java.lang.Boolean",
"description": "Whether to enable caching in the PulsarProducerFactory.",
"defaultValue": true
},
{
"name": "spring.quartz.jdbc.comment-prefix",
"defaultValue": [

View File

@@ -94,6 +94,8 @@ org.springframework.boot.autoconfigure.mustache.MustacheAutoConfiguration
org.springframework.boot.autoconfigure.neo4j.Neo4jAutoConfiguration
org.springframework.boot.autoconfigure.netty.NettyAutoConfiguration
org.springframework.boot.autoconfigure.orm.jpa.HibernateJpaAutoConfiguration
org.springframework.boot.autoconfigure.pulsar.PulsarAutoConfiguration
org.springframework.boot.autoconfigure.pulsar.PulsarReactiveAutoConfiguration
org.springframework.boot.autoconfigure.quartz.QuartzAutoConfiguration
org.springframework.boot.autoconfigure.r2dbc.R2dbcAutoConfiguration
org.springframework.boot.autoconfigure.r2dbc.R2dbcTransactionManagerAutoConfiguration

View File

@@ -0,0 +1,106 @@
/*
* Copyright 2012-2023 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* https://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.boot.autoconfigure.pulsar;
import java.util.List;
import java.util.function.BiConsumer;
import org.assertj.core.api.AssertDelegateTarget;
import org.mockito.InOrder;
import org.springframework.test.util.ReflectionTestUtils;
import static org.mockito.Mockito.inOrder;
import static org.mockito.Mockito.mock;
/**
* Test utility used to check customizers are called correctly.
*
* @param <C> the customizer type
* @param <T> the target class that is customized
* @author Phillip Webb
* @author Chris Bono
*/
final class Customizers<C, T> {
private final BiConsumer<C, T> customizeAction;
private final Class<T> targetClass;
@SuppressWarnings("unchecked")
private Customizers(Class<?> targetClass, BiConsumer<C, T> customizeAction) {
this.customizeAction = customizeAction;
this.targetClass = (Class<T>) targetClass;
}
/**
* Create an instance by getting the value from a field.
* @param source the source to extract the customizers from
* @param fieldName the field name
* @return a new {@link CustomizersAssert} instance
*/
@SuppressWarnings("unchecked")
CustomizersAssert fromField(Object source, String fieldName) {
return new CustomizersAssert(ReflectionTestUtils.getField(source, fieldName));
}
/**
* Create a new {@link Customizers} instance.
* @param <C> the customizer class
* @param <T> the target class that is customized
* @param targetClass the target class that is customized
* @param customizeAction the customizer action to take
* @return a new {@link Customizers} instance
*/
static <C, T> Customizers<C, T> of(Class<?> targetClass, BiConsumer<C, T> customizeAction) {
return new Customizers<>(targetClass, customizeAction);
}
/**
* Assertions that can be applied to customizers.
*/
final class CustomizersAssert implements AssertDelegateTarget {
private final List<C> customizers;
@SuppressWarnings("unchecked")
private CustomizersAssert(Object customizers) {
this.customizers = (customizers instanceof List) ? (List<C>) customizers : List.of((C) customizers);
}
/**
* Assert that the customize method is called in a specified order. It is expected
* that each customizer has set a unique value so the expected values can be used
* as a verify step.
* @param <V> the value type
* @param call the call the customizer makes
* @param expectedValues the expected values
*/
@SuppressWarnings("unchecked")
<V> void callsInOrder(BiConsumer<T, V> call, V... expectedValues) {
T target = mock(Customizers.this.targetClass);
BiConsumer<C, T> customizeAction = Customizers.this.customizeAction;
this.customizers.forEach((customizer) -> customizeAction.accept(customizer, target));
InOrder ordered = inOrder(target);
for (V expectedValue : expectedValues) {
call.accept(ordered.verify(target), expectedValue);
}
}
}
}

View File

@@ -0,0 +1,55 @@
/*
* Copyright 2012-2023 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* https://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.boot.autoconfigure.pulsar;
import org.apache.pulsar.client.api.DeadLetterPolicy;
import org.junit.jupiter.api.Test;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatIllegalStateException;
/**
* Tests for {@link DeadLetterPolicyMapper}.
*
* @author Chris Bono
* @author Phillip Webb
*/
class DeadLetterPolicyMapperTests {
@Test
void map() {
PulsarProperties.Consumer.DeadLetterPolicy properties = new PulsarProperties.Consumer.DeadLetterPolicy();
properties.setMaxRedeliverCount(100);
properties.setRetryLetterTopic("my-retry-topic");
properties.setDeadLetterTopic("my-dlt-topic");
properties.setInitialSubscriptionName("my-initial-subscription");
DeadLetterPolicy policy = DeadLetterPolicyMapper.map(properties);
assertThat(policy.getMaxRedeliverCount()).isEqualTo(100);
assertThat(policy.getRetryLetterTopic()).isEqualTo("my-retry-topic");
assertThat(policy.getDeadLetterTopic()).isEqualTo("my-dlt-topic");
assertThat(policy.getInitialSubscriptionName()).isEqualTo("my-initial-subscription");
}
@Test
void mapWhenMaxRedeliverCountIsNotPositiveThrowsException() {
PulsarProperties.Consumer.DeadLetterPolicy properties = new PulsarProperties.Consumer.DeadLetterPolicy();
properties.setMaxRedeliverCount(0);
assertThatIllegalStateException().isThrownBy(() -> DeadLetterPolicyMapper.map(properties))
.withMessage("Pulsar DeadLetterPolicy must have a positive 'max-redelivery-count' property value");
}
}

View File

@@ -0,0 +1,118 @@
/*
* Copyright 2012-2023 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* https://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.boot.autoconfigure.pulsar;
import java.time.Duration;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import org.apache.pulsar.client.api.PulsarClientException;
import org.junit.jupiter.api.Test;
import org.testcontainers.containers.PulsarContainer;
import org.testcontainers.junit.jupiter.Container;
import org.testcontainers.junit.jupiter.Testcontainers;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.ImportAutoConfiguration;
import org.springframework.boot.autoconfigure.http.HttpMessageConvertersAutoConfiguration;
import org.springframework.boot.autoconfigure.jackson.JacksonAutoConfiguration;
import org.springframework.boot.autoconfigure.web.servlet.DispatcherServletAutoConfiguration;
import org.springframework.boot.autoconfigure.web.servlet.ServletWebServerFactoryAutoConfiguration;
import org.springframework.boot.autoconfigure.web.servlet.WebMvcAutoConfiguration;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.boot.test.context.SpringBootTest.WebEnvironment;
import org.springframework.boot.test.web.client.TestRestTemplate;
import org.springframework.boot.testsupport.testcontainers.DockerImageNames;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Import;
import org.springframework.pulsar.annotation.PulsarListener;
import org.springframework.pulsar.core.PulsarTemplate;
import org.springframework.test.context.DynamicPropertyRegistry;
import org.springframework.test.context.DynamicPropertySource;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;
import static org.assertj.core.api.Assertions.assertThat;
/**
* Integration tests for {@link PulsarAutoConfiguration}.
*
* @author Chris Bono
* @author Phillip Webb
*/
@SpringBootTest(webEnvironment = WebEnvironment.RANDOM_PORT)
@Testcontainers(disabledWithoutDocker = true)
class PulsarAutoConfigurationIntegrationTests {
@Container
private static final PulsarContainer PULSAR_CONTAINER = new PulsarContainer(DockerImageNames.pulsar())
.withStartupAttempts(2)
.withStartupTimeout(Duration.ofMinutes(3));
private static final CountDownLatch LISTEN_LATCH = new CountDownLatch(1);
private static final String TOPIC = "pacit-hello-topic";
@DynamicPropertySource
static void pulsarProperties(DynamicPropertyRegistry registry) {
registry.add("spring.pulsar.client.service-url", PULSAR_CONTAINER::getPulsarBrokerUrl);
registry.add("spring.pulsar.admin.service-url", PULSAR_CONTAINER::getHttpServiceUrl);
}
@Test
void appStartsWithAutoConfiguredSpringPulsarComponents(
@Autowired(required = false) PulsarTemplate<String> pulsarTemplate) {
assertThat(pulsarTemplate).isNotNull();
}
@Test
void templateCanBeAccessedDuringWebRequest(@Autowired TestRestTemplate restTemplate) throws InterruptedException {
assertThat(restTemplate.getForObject("/hello", String.class)).startsWith("Hello World -> ");
assertThat(LISTEN_LATCH.await(5, TimeUnit.SECONDS)).isTrue();
}
@Configuration(proxyBeanMethods = false)
@ImportAutoConfiguration({ DispatcherServletAutoConfiguration.class, ServletWebServerFactoryAutoConfiguration.class,
WebMvcAutoConfiguration.class, HttpMessageConvertersAutoConfiguration.class, JacksonAutoConfiguration.class,
PulsarAutoConfiguration.class, PulsarReactiveAutoConfiguration.class })
@Import(TestWebController.class)
static class TestConfiguration {
@PulsarListener(subscriptionName = TOPIC + "-sub", topics = TOPIC)
void listen(String ignored) {
LISTEN_LATCH.countDown();
}
}
@RestController
static class TestWebController {
private final PulsarTemplate<String> pulsarTemplate;
TestWebController(PulsarTemplate<String> pulsarTemplate) {
this.pulsarTemplate = pulsarTemplate;
}
@GetMapping("/hello")
String sayHello() throws PulsarClientException {
return "Hello World -> " + this.pulsarTemplate.send(TOPIC, "hello");
}
}
}

View File

@@ -0,0 +1,519 @@
/*
* Copyright 2012-2023 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* https://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.boot.autoconfigure.pulsar;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.TimeUnit;
import com.github.benmanes.caffeine.cache.Caffeine;
import org.apache.pulsar.client.api.ConsumerBuilder;
import org.apache.pulsar.client.api.ProducerBuilder;
import org.apache.pulsar.client.api.PulsarClient;
import org.apache.pulsar.client.api.ReaderBuilder;
import org.apache.pulsar.client.api.interceptor.ProducerInterceptor;
import org.apache.pulsar.common.schema.SchemaType;
import org.junit.jupiter.api.Nested;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.ValueSource;
import org.springframework.boot.autoconfigure.AutoConfigurations;
import org.springframework.boot.test.context.FilteredClassLoader;
import org.springframework.boot.test.context.TestConfiguration;
import org.springframework.boot.test.context.runner.ApplicationContextRunner;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.core.annotation.Order;
import org.springframework.pulsar.annotation.PulsarBootstrapConfiguration;
import org.springframework.pulsar.annotation.PulsarListenerAnnotationBeanPostProcessor;
import org.springframework.pulsar.annotation.PulsarReaderAnnotationBeanPostProcessor;
import org.springframework.pulsar.config.ConcurrentPulsarListenerContainerFactory;
import org.springframework.pulsar.config.DefaultPulsarReaderContainerFactory;
import org.springframework.pulsar.config.PulsarListenerContainerFactory;
import org.springframework.pulsar.config.PulsarListenerEndpointRegistry;
import org.springframework.pulsar.config.PulsarReaderEndpointRegistry;
import org.springframework.pulsar.core.CachingPulsarProducerFactory;
import org.springframework.pulsar.core.ConsumerBuilderCustomizer;
import org.springframework.pulsar.core.DefaultPulsarClientFactory;
import org.springframework.pulsar.core.DefaultPulsarConsumerFactory;
import org.springframework.pulsar.core.DefaultPulsarProducerFactory;
import org.springframework.pulsar.core.DefaultPulsarReaderFactory;
import org.springframework.pulsar.core.DefaultSchemaResolver;
import org.springframework.pulsar.core.DefaultTopicResolver;
import org.springframework.pulsar.core.ProducerBuilderCustomizer;
import org.springframework.pulsar.core.PulsarAdministration;
import org.springframework.pulsar.core.PulsarConsumerFactory;
import org.springframework.pulsar.core.PulsarProducerFactory;
import org.springframework.pulsar.core.PulsarReaderFactory;
import org.springframework.pulsar.core.PulsarTemplate;
import org.springframework.pulsar.core.ReaderBuilderCustomizer;
import org.springframework.pulsar.core.SchemaResolver;
import org.springframework.pulsar.core.TopicResolver;
import static org.assertj.core.api.Assertions.assertThat;
import static org.mockito.Mockito.mock;
/**
* Tests for {@link PulsarAutoConfiguration}.
*
* @author Chris Bono
* @author Alexander Preuß
* @author Soby Chacko
* @author Phillip Webb
*/
class PulsarAutoConfigurationTests {
private static final String INTERNAL_PULSAR_LISTENER_ANNOTATION_PROCESSOR = "org.springframework.pulsar.config.internalPulsarListenerAnnotationProcessor";
private static final String INTERNAL_PULSAR_READER_ANNOTATION_PROCESSOR = "org.springframework.pulsar.config.internalPulsarReaderAnnotationProcessor";
private final ApplicationContextRunner contextRunner = new ApplicationContextRunner()
.withConfiguration(AutoConfigurations.of(PulsarAutoConfiguration.class))
.withBean(PulsarClient.class, () -> mock(PulsarClient.class));
@Test
void whenPulsarNotOnClasspathAutoConfigurationIsSkipped() {
new ApplicationContextRunner().withConfiguration(AutoConfigurations.of(PulsarAutoConfiguration.class))
.withClassLoader(new FilteredClassLoader(PulsarClient.class))
.run((context) -> assertThat(context).doesNotHaveBean(PulsarAutoConfiguration.class));
}
@Test
void whenSpringPulsarNotOnClasspathAutoConfigurationIsSkipped() {
this.contextRunner.withClassLoader(new FilteredClassLoader(PulsarTemplate.class))
.run((context) -> assertThat(context).doesNotHaveBean(PulsarAutoConfiguration.class));
}
@Test
void whenCustomPulsarListenerAnnotationProcessorDefinedAutoConfigurationIsSkipped() {
this.contextRunner.withBean(INTERNAL_PULSAR_LISTENER_ANNOTATION_PROCESSOR, String.class, () -> "bean")
.run((context) -> assertThat(context).doesNotHaveBean(PulsarBootstrapConfiguration.class));
}
@Test
void whenCustomPulsarReaderAnnotationProcessorDefinedAutoConfigurationIsSkipped() {
this.contextRunner.withBean(INTERNAL_PULSAR_READER_ANNOTATION_PROCESSOR, String.class, () -> "bean")
.run((context) -> assertThat(context).doesNotHaveBean(PulsarBootstrapConfiguration.class));
}
@Test
void autoConfiguresBeans() {
this.contextRunner.run((context) -> assertThat(context).hasSingleBean(PulsarConfiguration.class)
.hasSingleBean(DefaultPulsarClientFactory.class)
.hasSingleBean(PulsarClient.class)
.hasSingleBean(PulsarAdministration.class)
.hasSingleBean(DefaultSchemaResolver.class)
.hasSingleBean(DefaultTopicResolver.class)
.hasSingleBean(CachingPulsarProducerFactory.class)
.hasSingleBean(PulsarTemplate.class)
.hasSingleBean(DefaultPulsarConsumerFactory.class)
.hasSingleBean(ConcurrentPulsarListenerContainerFactory.class)
.hasSingleBean(DefaultPulsarReaderFactory.class)
.hasSingleBean(DefaultPulsarReaderContainerFactory.class)
.hasSingleBean(PulsarListenerAnnotationBeanPostProcessor.class)
.hasSingleBean(PulsarListenerEndpointRegistry.class)
.hasSingleBean(PulsarReaderAnnotationBeanPostProcessor.class)
.hasSingleBean(PulsarReaderEndpointRegistry.class));
}
@Nested
class ProducerFactoryTests {
private final ApplicationContextRunner contextRunner = PulsarAutoConfigurationTests.this.contextRunner;
@Test
@SuppressWarnings("unchecked")
void whenHasUserDefinedBeanDoesNotAutoConfigureBean() {
PulsarProducerFactory<String> producerFactory = mock(PulsarProducerFactory.class);
this.contextRunner
.withBean("customPulsarProducerFactory", PulsarProducerFactory.class, () -> producerFactory)
.run((context) -> assertThat(context).getBean(PulsarProducerFactory.class).isSameAs(producerFactory));
}
@Test
void whenNoPropertiesUsesCachingPulsarProducerFactory() {
this.contextRunner.run((context) -> assertThat(context).getBean(PulsarProducerFactory.class)
.isExactlyInstanceOf(CachingPulsarProducerFactory.class));
}
@Test
void whenCachingDisabledUsesDefaultPulsarProducerFactory() {
this.contextRunner.withPropertyValues("spring.pulsar.producer.cache.enabled=false")
.run((context) -> assertThat(context).getBean(PulsarProducerFactory.class)
.isExactlyInstanceOf(DefaultPulsarProducerFactory.class));
}
@Test
void whenCachingEnabledUsesCachingPulsarProducerFactory() {
this.contextRunner.withPropertyValues("spring.pulsar.producer.cache.enabled=true")
.run((context) -> assertThat(context).getBean(PulsarProducerFactory.class)
.isExactlyInstanceOf(CachingPulsarProducerFactory.class));
}
@Test
void whenCachingEnabledAndCaffeineNotOnClasspathStillUsesCaffeine() {
this.contextRunner.withClassLoader(new FilteredClassLoader(Caffeine.class))
.withPropertyValues("spring.pulsar.producer.cache.enabled=true")
.run((context) -> {
assertThat(context).getBean(CachingPulsarProducerFactory.class)
.extracting("producerCache")
.extracting(Object::getClass)
.extracting(Class::getName)
.isEqualTo("org.springframework.pulsar.core.CaffeineCacheProvider");
assertThat(context).getBean(CachingPulsarProducerFactory.class)
.extracting("producerCache.cache")
.extracting(Object::getClass)
.extracting(Class::getName)
.asString()
.startsWith("org.springframework.pulsar.shade.com.github.benmanes.caffeine.cache.");
});
}
@Test
void whenCustomCachingPropertiesCreatesConfiguredBean() {
this.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).getBean(CachingPulsarProducerFactory.class)
.extracting("producerCache.cache.cache")
.hasFieldOrPropertyWithValue("maximum", 5150L)
.hasFieldOrPropertyWithValue("expiresAfterAccessNanos", TimeUnit.SECONDS.toNanos(100)));
}
@Test
void whenHasTopicNamePropertyCreatesConfiguredBean() {
this.contextRunner.withPropertyValues("spring.pulsar.producer.topic-name=my-topic")
.run((context) -> assertThat(context).getBean(DefaultPulsarProducerFactory.class)
.hasFieldOrPropertyWithValue("defaultTopic", "my-topic"));
}
@Test
void injectsExpectedBeans() {
this.contextRunner
.withPropertyValues("spring.pulsar.producer.topic-name=my-topic",
"spring.pulsar.producer.cache.enabled=false")
.run((context) -> assertThat(context).getBean(DefaultPulsarProducerFactory.class)
.hasFieldOrPropertyWithValue("pulsarClient", context.getBean(PulsarClient.class))
.hasFieldOrPropertyWithValue("topicResolver", context.getBean(TopicResolver.class)));
}
@ParameterizedTest
@ValueSource(booleans = { true, false })
<T> void whenHasUserDefinedCustomizersAppliesInCorrectOrder(boolean cachingEnabled) {
this.contextRunner
.withPropertyValues("spring.pulsar.producer.cache.enabled=" + cachingEnabled,
"spring.pulsar.producer.name=fromPropsCustomizer")
.withUserConfiguration(ProducerBuilderCustomizersConfig.class)
.run((context) -> {
DefaultPulsarProducerFactory<?> producerFactory = context
.getBean(DefaultPulsarProducerFactory.class);
Customizers<ProducerBuilderCustomizer<T>, ProducerBuilder<T>> customizers = Customizers
.of(ProducerBuilder.class, ProducerBuilderCustomizer::customize);
assertThat(customizers.fromField(producerFactory, "defaultConfigCustomizers")).callsInOrder(
ProducerBuilder::producerName, "fromPropsCustomizer", "fromCustomizer1", "fromCustomizer2");
});
}
@TestConfiguration(proxyBeanMethods = false)
static class ProducerBuilderCustomizersConfig {
@Bean
@Order(200)
ProducerBuilderCustomizer<?> customizerFoo() {
return (builder) -> builder.producerName("fromCustomizer2");
}
@Bean
@Order(100)
ProducerBuilderCustomizer<?> customizerBar() {
return (builder) -> builder.producerName("fromCustomizer1");
}
}
}
@Nested
class TemplateTests {
private final ApplicationContextRunner contextRunner = PulsarAutoConfigurationTests.this.contextRunner;
@Test
@SuppressWarnings("unchecked")
void whenHasUserDefinedBeanDoesNotAutoConfigureBean() {
PulsarTemplate<String> template = mock(PulsarTemplate.class);
this.contextRunner.withBean("customPulsarTemplate", PulsarTemplate.class, () -> template)
.run((context) -> assertThat(context).getBean(PulsarTemplate.class).isSameAs(template));
}
@Test
void injectsExpectedBeans() {
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).getBean(PulsarTemplate.class)
.hasFieldOrPropertyWithValue("producerFactory", producerFactory)
.hasFieldOrPropertyWithValue("schemaResolver", schemaResolver)
.hasFieldOrPropertyWithValue("topicResolver", topicResolver));
}
@Test
void whenHasUseDefinedProducerInterceptorInjectsBean() {
ProducerInterceptor interceptor = mock(ProducerInterceptor.class);
this.contextRunner.withBean("customProducerInterceptor", ProducerInterceptor.class, () -> interceptor)
.run((context) -> assertThat(context).getBean(PulsarTemplate.class)
.extracting("interceptors")
.asList()
.contains(interceptor));
}
@Test
void whenHasUseDefinedProducerInterceptorsInjectsBeansInCorrectOrder() {
this.contextRunner.withUserConfiguration(InterceptorTestConfiguration.class)
.run((context) -> assertThat(context).getBean(PulsarTemplate.class)
.extracting("interceptors")
.asList()
.containsExactly(context.getBean("interceptorBar"), context.getBean("interceptorFoo")));
}
@Test
void whenNoPropertiesEnablesObservation() {
this.contextRunner.run((context) -> assertThat(context).getBean(PulsarTemplate.class)
.hasFieldOrPropertyWithValue("observationEnabled", true));
}
@Test
void whenObservationsEnabledEnablesObservation() {
this.contextRunner.withPropertyValues("spring.pulsar.template.observations-enabled=true")
.run((context) -> assertThat(context).getBean(PulsarTemplate.class)
.hasFieldOrPropertyWithValue("observationEnabled", true));
}
@Test
void whenObservationsDisabledDoesNotEnableObservation() {
this.contextRunner.withPropertyValues("spring.pulsar.template.observations-enabled=false")
.run((context) -> assertThat(context).getBean(PulsarTemplate.class)
.hasFieldOrPropertyWithValue("observationEnabled", false));
}
@Configuration(proxyBeanMethods = false)
static class InterceptorTestConfiguration {
@Bean
@Order(200)
ProducerInterceptor interceptorFoo() {
return mock(ProducerInterceptor.class);
}
@Bean
@Order(100)
ProducerInterceptor interceptorBar() {
return mock(ProducerInterceptor.class);
}
}
}
@Nested
class ConsumerFactoryTests {
private final ApplicationContextRunner contextRunner = PulsarAutoConfigurationTests.this.contextRunner;
@Test
@SuppressWarnings("unchecked")
void whenHasUserDefinedBeanDoesNotAutoConfigureBean() {
PulsarConsumerFactory<String> consumerFactory = mock(PulsarConsumerFactory.class);
this.contextRunner
.withBean("customPulsarConsumerFactory", PulsarConsumerFactory.class, () -> consumerFactory)
.run((context) -> assertThat(context).getBean(PulsarConsumerFactory.class).isSameAs(consumerFactory));
}
@Test
void injectsExpectedBeans() {
this.contextRunner.run((context) -> assertThat(context).getBean(DefaultPulsarConsumerFactory.class)
.hasFieldOrPropertyWithValue("pulsarClient", context.getBean(PulsarClient.class)));
}
@Test
<T> void whenHasUserDefinedCustomizersAppliesInCorrectOrder() {
this.contextRunner.withPropertyValues("spring.pulsar.consumer.name=fromPropsCustomizer")
.withUserConfiguration(ConsumerBuilderCustomizersConfig.class)
.run((context) -> {
DefaultPulsarConsumerFactory<?> consumerFactory = context
.getBean(DefaultPulsarConsumerFactory.class);
Customizers<ConsumerBuilderCustomizer<T>, ConsumerBuilder<T>> customizers = Customizers
.of(ConsumerBuilder.class, ConsumerBuilderCustomizer::customize);
assertThat(customizers.fromField(consumerFactory, "defaultConfigCustomizers")).callsInOrder(
ConsumerBuilder::consumerName, "fromPropsCustomizer", "fromCustomizer1", "fromCustomizer2");
});
}
@TestConfiguration(proxyBeanMethods = false)
static class ConsumerBuilderCustomizersConfig {
@Bean
@Order(200)
ConsumerBuilderCustomizer<?> customizerFoo() {
return (builder) -> builder.consumerName("fromCustomizer2");
}
@Bean
@Order(100)
ConsumerBuilderCustomizer<?> customizerBar() {
return (builder) -> builder.consumerName("fromCustomizer1");
}
}
}
@Nested
class ListenerTests {
private final ApplicationContextRunner contextRunner = PulsarAutoConfigurationTests.this.contextRunner;
@Test
void whenHasUserDefinedListenerContainerFactoryBeanDoesNotAutoConfigureBean() {
PulsarListenerContainerFactory listenerContainerFactory = mock(PulsarListenerContainerFactory.class);
this.contextRunner
.withBean("pulsarListenerContainerFactory", PulsarListenerContainerFactory.class,
() -> listenerContainerFactory)
.run((context) -> assertThat(context).getBean(PulsarListenerContainerFactory.class)
.isSameAs(listenerContainerFactory));
}
@Test
@SuppressWarnings("rawtypes")
void injectsExpectedBeans() {
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).getBean(ConcurrentPulsarListenerContainerFactory.class)
.hasFieldOrPropertyWithValue("consumerFactory", consumerFactory)
.extracting(ConcurrentPulsarListenerContainerFactory::getContainerProperties)
.hasFieldOrPropertyWithValue("schemaResolver", schemaResolver)
.hasFieldOrPropertyWithValue("topicResolver", topicResolver));
}
@Test
@SuppressWarnings("unchecked")
void whenHasUserDefinedListenerAnnotationBeanPostProcessorBeanDoesNotAutoConfigureBean() {
PulsarListenerAnnotationBeanPostProcessor<String> listenerAnnotationBeanPostProcessor = mock(
PulsarListenerAnnotationBeanPostProcessor.class);
this.contextRunner
.withBean("org.springframework.pulsar.config.internalPulsarListenerAnnotationProcessor",
PulsarListenerAnnotationBeanPostProcessor.class, () -> listenerAnnotationBeanPostProcessor)
.run((context) -> assertThat(context).getBean(PulsarListenerAnnotationBeanPostProcessor.class)
.isSameAs(listenerAnnotationBeanPostProcessor));
}
@Test
void whenHasCustomProperties() {
List<String> properties = new ArrayList<>();
properties.add("spring.pulsar.listener.schema-type=avro");
this.contextRunner.withPropertyValues(properties.toArray(String[]::new)).run((context) -> {
ConcurrentPulsarListenerContainerFactory<?> factory = context
.getBean(ConcurrentPulsarListenerContainerFactory.class);
assertThat(factory.getContainerProperties().getSchemaType()).isEqualTo(SchemaType.AVRO);
});
}
@Test
void whenNoPropertiesEnablesObservation() {
this.contextRunner
.run((context) -> assertThat(context).getBean(ConcurrentPulsarListenerContainerFactory.class)
.hasFieldOrPropertyWithValue("containerProperties.observationEnabled", true));
}
@Test
void whenObservationsEnabledEnablesObservation() {
this.contextRunner.withPropertyValues("spring.pulsar.listener.observation-enabled=true")
.run((context) -> assertThat(context).getBean(ConcurrentPulsarListenerContainerFactory.class)
.hasFieldOrPropertyWithValue("containerProperties.observationEnabled", true));
}
@Test
void whenObservationsDisabledDoesNotEnableObservation() {
this.contextRunner.withPropertyValues("spring.pulsar.listener.observation-enabled=false")
.run((context) -> assertThat(context).getBean(ConcurrentPulsarListenerContainerFactory.class)
.hasFieldOrPropertyWithValue("containerProperties.observationEnabled", false));
}
}
@Nested
class ReaderFactoryTests {
private final ApplicationContextRunner contextRunner = PulsarAutoConfigurationTests.this.contextRunner;
@Test
@SuppressWarnings("unchecked")
void whenHasUserDefinedBeanDoesNotAutoConfigureBean() {
PulsarReaderFactory<String> readerFactory = mock(PulsarReaderFactory.class);
this.contextRunner.withBean("customPulsarReaderFactory", PulsarReaderFactory.class, () -> readerFactory)
.run((context) -> assertThat(context).getBean(PulsarReaderFactory.class).isSameAs(readerFactory));
}
@Test
void injectsExpectedBeans() {
this.contextRunner.run((context) -> assertThat(context).getBean(DefaultPulsarReaderFactory.class)
.hasFieldOrPropertyWithValue("pulsarClient", context.getBean(PulsarClient.class)));
}
@Test
<T> void whenHasUserDefinedCustomizersAppliesInCorrectOrder() {
this.contextRunner.withPropertyValues("spring.pulsar.reader.name=fromPropsCustomizer")
.withUserConfiguration(ReaderBuilderCustomizersConfig.class)
.run((context) -> {
DefaultPulsarReaderFactory<?> readerFactory = context.getBean(DefaultPulsarReaderFactory.class);
Customizers<ReaderBuilderCustomizer<T>, ReaderBuilder<T>> customizers = Customizers
.of(ReaderBuilder.class, ReaderBuilderCustomizer::customize);
assertThat(customizers.fromField(readerFactory, "defaultConfigCustomizers")).callsInOrder(
ReaderBuilder::readerName, "fromPropsCustomizer", "fromCustomizer1", "fromCustomizer2");
});
}
@TestConfiguration(proxyBeanMethods = false)
static class ReaderBuilderCustomizersConfig {
@Bean
@Order(200)
ReaderBuilderCustomizer<?> customizerFoo() {
return (builder) -> builder.readerName("fromCustomizer2");
}
@Bean
@Order(100)
ReaderBuilderCustomizer<?> customizerBar() {
return (builder) -> builder.readerName("fromCustomizer1");
}
}
}
}

View File

@@ -0,0 +1,318 @@
/*
* Copyright 2012-2023 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* https://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.boot.autoconfigure.pulsar;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.function.Consumer;
import org.apache.pulsar.client.admin.PulsarAdminBuilder;
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.common.schema.KeyValueEncodingType;
import org.assertj.core.api.InstanceOfAssertFactories;
import org.assertj.core.api.InstanceOfAssertFactory;
import org.assertj.core.api.MapAssert;
import org.junit.jupiter.api.Nested;
import org.junit.jupiter.api.Test;
import org.springframework.boot.autoconfigure.AutoConfigurations;
import org.springframework.boot.test.context.TestConfiguration;
import org.springframework.boot.test.context.runner.ApplicationContextRunner;
import org.springframework.context.annotation.Bean;
import org.springframework.core.annotation.Order;
import org.springframework.pulsar.core.DefaultPulsarClientFactory;
import org.springframework.pulsar.core.DefaultSchemaResolver;
import org.springframework.pulsar.core.DefaultTopicResolver;
import org.springframework.pulsar.core.PulsarAdminBuilderCustomizer;
import org.springframework.pulsar.core.PulsarAdministration;
import org.springframework.pulsar.core.PulsarClientBuilderCustomizer;
import org.springframework.pulsar.core.PulsarClientFactory;
import org.springframework.pulsar.core.SchemaResolver;
import org.springframework.pulsar.core.SchemaResolver.SchemaResolverCustomizer;
import org.springframework.pulsar.core.TopicResolver;
import org.springframework.pulsar.function.PulsarFunctionAdministration;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.entry;
import static org.mockito.Mockito.mock;
/**
* Tests for {@link PulsarConfiguration}.
*
* @author Chris Bono
* @author Alexander Preuß
* @author Soby Chacko
* @author Phillip Webb
*/
class PulsarConfigurationTests {
private final ApplicationContextRunner contextRunner = new ApplicationContextRunner()
.withConfiguration(AutoConfigurations.of(PulsarConfiguration.class))
.withBean(PulsarClient.class, () -> mock(PulsarClient.class));
@Nested
class ClientTests {
@Test
void whenHasUserDefinedClientFactoryBeanDoesNotAutoConfigureBean() {
PulsarClientFactory customFactory = mock(PulsarClientFactory.class);
new ApplicationContextRunner().withConfiguration(AutoConfigurations.of(PulsarConfiguration.class))
.withBean("customPulsarClientFactory", PulsarClientFactory.class, () -> customFactory)
.run((context) -> assertThat(context).getBean(PulsarClientFactory.class).isSameAs(customFactory));
}
@Test
void whenHasUserDefinedClientBeanDoesNotAutoConfigureBean() {
PulsarClient customClient = mock(PulsarClient.class);
new ApplicationContextRunner().withConfiguration(AutoConfigurations.of(PulsarConfiguration.class))
.withBean("customPulsarClient", PulsarClient.class, () -> customClient)
.run((context) -> assertThat(context).getBean(PulsarClient.class).isSameAs(customClient));
}
@Test
void whenHasUserDefinedCustomizersAppliesInCorrectOrder() {
PulsarConfigurationTests.this.contextRunner
.withUserConfiguration(PulsarClientBuilderCustomizersConfig.class)
.withPropertyValues("spring.pulsar.client.service-url=fromPropsCustomizer")
.run((context) -> {
DefaultPulsarClientFactory clientFactory = context.getBean(DefaultPulsarClientFactory.class);
Customizers<PulsarClientBuilderCustomizer, ClientBuilder> customizers = Customizers
.of(ClientBuilder.class, PulsarClientBuilderCustomizer::customize);
assertThat(customizers.fromField(clientFactory, "customizer")).callsInOrder(
ClientBuilder::serviceUrl, "fromPropsCustomizer", "fromCustomizer1", "fromCustomizer2");
});
}
@TestConfiguration(proxyBeanMethods = false)
static class PulsarClientBuilderCustomizersConfig {
@Bean
@Order(200)
PulsarClientBuilderCustomizer customizerFoo() {
return (builder) -> builder.serviceUrl("fromCustomizer2");
}
@Bean
@Order(100)
PulsarClientBuilderCustomizer customizerBar() {
return (builder) -> builder.serviceUrl("fromCustomizer1");
}
}
}
@Nested
class AdministrationTests {
private final ApplicationContextRunner contextRunner = PulsarConfigurationTests.this.contextRunner;
@Test
void whenHasUserDefinedBeanDoesNotAutoConfigureBean() {
PulsarAdministration pulsarAdministration = mock(PulsarAdministration.class);
this.contextRunner
.withBean("customPulsarAdministration", PulsarAdministration.class, () -> pulsarAdministration)
.run((context) -> assertThat(context).getBean(PulsarAdministration.class)
.isSameAs(pulsarAdministration));
}
@Test
void whenHasUserDefinedCustomizersAppliesInCorrectOrder() {
this.contextRunner.withUserConfiguration(PulsarAdminBuilderCustomizersConfig.class)
.withPropertyValues("spring.pulsar.admin.service-url=fromPropsCustomizer")
.run((context) -> {
PulsarAdministration pulsarAdmin = context.getBean(PulsarAdministration.class);
Customizers<PulsarAdminBuilderCustomizer, PulsarAdminBuilder> customizers = Customizers
.of(PulsarAdminBuilder.class, PulsarAdminBuilderCustomizer::customize);
assertThat(customizers.fromField(pulsarAdmin, "adminCustomizers")).callsInOrder(
PulsarAdminBuilder::serviceHttpUrl, "fromPropsCustomizer", "fromCustomizer1",
"fromCustomizer2");
});
}
@TestConfiguration(proxyBeanMethods = false)
static class PulsarAdminBuilderCustomizersConfig {
@Bean
@Order(200)
PulsarAdminBuilderCustomizer customizerFoo() {
return (builder) -> builder.serviceHttpUrl("fromCustomizer2");
}
@Bean
@Order(100)
PulsarAdminBuilderCustomizer customizerBar() {
return (builder) -> builder.serviceHttpUrl("fromCustomizer1");
}
}
}
@Nested
class SchemaResolverTests {
@SuppressWarnings("rawtypes")
private static final InstanceOfAssertFactory<Map, MapAssert<Class, Schema>> CLASS_SCHEMA_MAP = InstanceOfAssertFactories
.map(Class.class, Schema.class);
private final ApplicationContextRunner contextRunner = PulsarConfigurationTests.this.contextRunner;
@Test
void whenHasUserDefinedBeanDoesNotAutoConfigureBean() {
SchemaResolver schemaResolver = mock(SchemaResolver.class);
this.contextRunner.withBean("customSchemaResolver", SchemaResolver.class, () -> schemaResolver)
.run((context) -> assertThat(context).getBean(SchemaResolver.class).isSameAs(schemaResolver));
}
@Test
void whenHasUserDefinedSchemaResolverCustomizer() {
SchemaResolverCustomizer<DefaultSchemaResolver> customizer = (schemaResolver) -> schemaResolver
.addCustomSchemaMapping(TestRecord.class, Schema.STRING);
this.contextRunner.withBean("schemaResolverCustomizer", SchemaResolverCustomizer.class, () -> customizer)
.run((context) -> assertThat(context).getBean(DefaultSchemaResolver.class)
.extracting(DefaultSchemaResolver::getCustomSchemaMappings, InstanceOfAssertFactories.MAP)
.containsEntry(TestRecord.class, Schema.STRING));
}
@Test
void whenHasDefaultsTypeMappingForPrimitiveAddsToSchemaResolver() {
List<String> properties = new ArrayList<>();
properties.add("spring.pulsar.defaults.type-mappings[0].message-type=" + TestRecord.CLASS_NAME);
properties.add("spring.pulsar.defaults.type-mappings[0].schema-info.schema-type=STRING");
this.contextRunner.withPropertyValues(properties.toArray(String[]::new))
.run((context) -> assertThat(context).getBean(DefaultSchemaResolver.class)
.extracting(DefaultSchemaResolver::getCustomSchemaMappings, InstanceOfAssertFactories.MAP)
.containsOnly(entry(TestRecord.class, Schema.STRING)));
}
@Test
void whenHasDefaultsTypeMappingForStructAddsToSchemaResolver() {
List<String> properties = new ArrayList<>();
properties.add("spring.pulsar.defaults.type-mappings[0].message-type=" + TestRecord.CLASS_NAME);
properties.add("spring.pulsar.defaults.type-mappings[0].schema-info.schema-type=JSON");
Schema<?> expectedSchema = Schema.JSON(TestRecord.class);
this.contextRunner.withPropertyValues(properties.toArray(String[]::new))
.run((context) -> assertThat(context).getBean(DefaultSchemaResolver.class)
.extracting(DefaultSchemaResolver::getCustomSchemaMappings, CLASS_SCHEMA_MAP)
.hasEntrySatisfying(TestRecord.class, schemaEqualTo(expectedSchema)));
}
@Test
void whenHasDefaultsTypeMappingForKeyValueAddsToSchemaResolver() {
List<String> properties = new ArrayList<>();
properties.add("spring.pulsar.defaults.type-mappings[0].message-type=" + TestRecord.CLASS_NAME);
properties.add("spring.pulsar.defaults.type-mappings[0].schema-info.schema-type=key-value");
properties.add("spring.pulsar.defaults.type-mappings[0].schema-info.message-key-type=java.lang.String");
Schema<?> expectedSchema = Schema.KeyValue(Schema.STRING, Schema.JSON(TestRecord.class),
KeyValueEncodingType.INLINE);
this.contextRunner.withPropertyValues(properties.toArray(String[]::new))
.run((context) -> assertThat(context).getBean(DefaultSchemaResolver.class)
.extracting(DefaultSchemaResolver::getCustomSchemaMappings, CLASS_SCHEMA_MAP)
.hasEntrySatisfying(TestRecord.class, schemaEqualTo(expectedSchema)));
}
@SuppressWarnings("rawtypes")
private Consumer<Schema> schemaEqualTo(Schema<?> expected) {
return (actual) -> assertThat(actual.getSchemaInfo()).isEqualTo(expected.getSchemaInfo());
}
}
@Nested
class TopicResolverTests {
private final ApplicationContextRunner contextRunner = PulsarConfigurationTests.this.contextRunner;
@Test
void whenHasUserDefinedBeanDoesNotAutoConfigureBean() {
TopicResolver topicResolver = mock(TopicResolver.class);
this.contextRunner.withBean("customTopicResolver", TopicResolver.class, () -> topicResolver)
.run((context) -> assertThat(context).getBean(TopicResolver.class).isSameAs(topicResolver));
}
@Test
void whenHasDefaultsTypeMappingAddsToSchemaResolver() {
List<String> properties = new ArrayList<>();
properties.add("spring.pulsar.defaults.type-mappings[0].message-type=" + TestRecord.CLASS_NAME);
properties.add("spring.pulsar.defaults.type-mappings[0].topic-name=foo-topic");
properties.add("spring.pulsar.defaults.type-mappings[1].message-type=java.lang.String");
properties.add("spring.pulsar.defaults.type-mappings[1].topic-name=string-topic");
this.contextRunner.withPropertyValues(properties.toArray(String[]::new))
.run((context) -> assertThat(context).getBean(TopicResolver.class)
.asInstanceOf(InstanceOfAssertFactories.type(DefaultTopicResolver.class))
.extracting(DefaultTopicResolver::getCustomTopicMappings, InstanceOfAssertFactories.MAP)
.containsOnly(entry(TestRecord.class, "foo-topic"), entry(String.class, "string-topic")));
}
}
@Nested
class FunctionAdministrationTests {
private final ApplicationContextRunner contextRunner = PulsarConfigurationTests.this.contextRunner;
@Test
void whenNoPropertiesAddsFunctionAdministrationBean() {
this.contextRunner.run((context) -> assertThat(context).getBean(PulsarFunctionAdministration.class)
.hasFieldOrPropertyWithValue("failFast", Boolean.TRUE)
.hasFieldOrPropertyWithValue("propagateFailures", Boolean.TRUE)
.hasFieldOrPropertyWithValue("propagateStopFailures", Boolean.FALSE)
.hasNoNullFieldsOrProperties() // ensures object providers set
.extracting("pulsarAdministration")
.isSameAs(context.getBean(PulsarAdministration.class)));
}
@Test
void whenHasFunctionPropertiesAppliesPropertiesToBean() {
List<String> properties = new ArrayList<>();
properties.add("spring.pulsar.function.fail-fast=false");
properties.add("spring.pulsar.function.propagate-failures=false");
properties.add("spring.pulsar.function.propagate-stop-failures=true");
this.contextRunner.withPropertyValues(properties.toArray(String[]::new))
.run((context) -> assertThat(context).getBean(PulsarFunctionAdministration.class)
.hasFieldOrPropertyWithValue("failFast", Boolean.FALSE)
.hasFieldOrPropertyWithValue("propagateFailures", Boolean.FALSE)
.hasFieldOrPropertyWithValue("propagateStopFailures", Boolean.TRUE));
}
@Test
void whenHasFunctionDisabledPropertyDoesNotCreateBean() {
this.contextRunner.withPropertyValues("spring.pulsar.function.enabled=false")
.run((context) -> assertThat(context).doesNotHaveBean(PulsarFunctionAdministration.class));
}
@Test
void whenHasCustomFunctionAdministrationBean() {
PulsarFunctionAdministration functionAdministration = mock(PulsarFunctionAdministration.class);
this.contextRunner.withBean(PulsarFunctionAdministration.class, () -> functionAdministration)
.run((context) -> assertThat(context).getBean(PulsarFunctionAdministration.class)
.isSameAs(functionAdministration));
}
}
record TestRecord() {
private static final String CLASS_NAME = TestRecord.class.getName();
}
}

View File

@@ -0,0 +1,190 @@
/*
* Copyright 2012-2023 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* https://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.boot.autoconfigure.pulsar;
import java.time.Duration;
import java.util.List;
import java.util.Map;
import java.util.concurrent.TimeUnit;
import java.util.regex.Pattern;
import org.apache.pulsar.client.admin.PulsarAdminBuilder;
import org.apache.pulsar.client.api.ClientBuilder;
import org.apache.pulsar.client.api.CompressionType;
import org.apache.pulsar.client.api.ConsumerBuilder;
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.ProducerBuilder;
import org.apache.pulsar.client.api.PulsarClientException.UnsupportedAuthenticationException;
import org.apache.pulsar.client.api.ReaderBuilder;
import org.apache.pulsar.client.api.SubscriptionType;
import org.apache.pulsar.common.schema.SchemaType;
import org.junit.jupiter.api.Test;
import org.springframework.boot.autoconfigure.pulsar.PulsarProperties.Consumer;
import org.springframework.pulsar.listener.PulsarContainerProperties;
import static org.assertj.core.api.Assertions.assertThat;
import static org.mockito.BDDMockito.then;
import static org.mockito.Mockito.mock;
/**
* Tests for {@link PulsarPropertiesMapper}.
*
* @author Chris Bono
* @author Phillip Webb
*/
class PulsarPropertiesMapperTests {
@Test
void customizeClientBuilderWhenHasNoAuthentication() {
PulsarProperties properties = new PulsarProperties();
properties.getClient().setServiceUrl("https://example.com");
properties.getClient().setConnectionTimeout(Duration.ofSeconds(1));
properties.getClient().setOperationTimeout(Duration.ofSeconds(2));
properties.getClient().setLookupTimeout(Duration.ofSeconds(3));
ClientBuilder builder = mock(ClientBuilder.class);
new PulsarPropertiesMapper(properties).customizeClientBuilder(builder);
then(builder).should().serviceUrl("https://example.com");
then(builder).should().connectionTimeout(1000, TimeUnit.MILLISECONDS);
then(builder).should().operationTimeout(2000, TimeUnit.MILLISECONDS);
then(builder).should().lookupTimeout(3000, TimeUnit.MILLISECONDS);
}
@Test
void customizeClientBuilderWhenHasAuthentication() throws UnsupportedAuthenticationException {
PulsarProperties properties = new PulsarProperties();
Map<String, String> params = Map.of("param", "name");
properties.getClient().getAuthentication().setPluginClassName("myclass");
properties.getClient().getAuthentication().setParam(params);
ClientBuilder builder = mock(ClientBuilder.class);
new PulsarPropertiesMapper(properties).customizeClientBuilder(builder);
then(builder).should().authentication("myclass", params);
}
@Test
void customizeAdminBuilderWhenHasNoAuthentication() {
PulsarProperties properties = new PulsarProperties();
properties.getAdmin().setServiceUrl("https://example.com");
properties.getAdmin().setConnectionTimeout(Duration.ofSeconds(1));
properties.getAdmin().setReadTimeout(Duration.ofSeconds(2));
properties.getAdmin().setRequestTimeout(Duration.ofSeconds(3));
PulsarAdminBuilder builder = mock(PulsarAdminBuilder.class);
new PulsarPropertiesMapper(properties).customizeAdminBuilder(builder);
then(builder).should().serviceHttpUrl("https://example.com");
then(builder).should().connectionTimeout(1000, TimeUnit.MILLISECONDS);
then(builder).should().readTimeout(2000, TimeUnit.MILLISECONDS);
then(builder).should().requestTimeout(3000, TimeUnit.MILLISECONDS);
}
@Test
void customizeAdminBuilderWhenHasAuthentication() throws UnsupportedAuthenticationException {
PulsarProperties properties = new PulsarProperties();
Map<String, String> params = Map.of("param", "name");
properties.getAdmin().getAuthentication().setPluginClassName("myclass");
properties.getAdmin().getAuthentication().setParam(params);
PulsarAdminBuilder builder = mock(PulsarAdminBuilder.class);
new PulsarPropertiesMapper(properties).customizeAdminBuilder(builder);
then(builder).should().authentication("myclass", params);
}
@Test
@SuppressWarnings("unchecked")
void customizeProducerBuilder() {
PulsarProperties properties = new PulsarProperties();
properties.getProducer().setName("name");
properties.getProducer().setTopicName("topicname");
properties.getProducer().setSendTimeout(Duration.ofSeconds(1));
properties.getProducer().setMessageRoutingMode(MessageRoutingMode.RoundRobinPartition);
properties.getProducer().setHashingScheme(HashingScheme.JavaStringHash);
properties.getProducer().setBatchingEnabled(false);
properties.getProducer().setChunkingEnabled(true);
properties.getProducer().setCompressionType(CompressionType.SNAPPY);
properties.getProducer().setAccessMode(ProducerAccessMode.Exclusive);
ProducerBuilder<Object> builder = mock(ProducerBuilder.class);
new PulsarPropertiesMapper(properties).customizeProducerBuilder(builder);
then(builder).should().producerName("name");
then(builder).should().topic("topicname");
then(builder).should().sendTimeout(1000, TimeUnit.MILLISECONDS);
then(builder).should().messageRoutingMode(MessageRoutingMode.RoundRobinPartition);
then(builder).should().hashingScheme(HashingScheme.JavaStringHash);
then(builder).should().enableBatching(false);
then(builder).should().enableChunking(true);
then(builder).should().compressionType(CompressionType.SNAPPY);
then(builder).should().accessMode(ProducerAccessMode.Exclusive);
}
@Test
@SuppressWarnings("unchecked")
void customizeConsumerBuilder() {
PulsarProperties properties = new PulsarProperties();
List<String> topics = List.of("mytopic");
Pattern topisPattern = Pattern.compile("my-pattern");
properties.getConsumer().setName("name");
properties.getConsumer().setTopics(topics);
properties.getConsumer().setTopicsPattern(topisPattern);
properties.getConsumer().setPriorityLevel(123);
properties.getConsumer().setReadCompacted(true);
Consumer.DeadLetterPolicy deadLetterPolicy = new Consumer.DeadLetterPolicy();
deadLetterPolicy.setDeadLetterTopic("my-dlt");
deadLetterPolicy.setMaxRedeliverCount(1);
properties.getConsumer().setDeadLetterPolicy(deadLetterPolicy);
ConsumerBuilder<Object> builder = mock(ConsumerBuilder.class);
new PulsarPropertiesMapper(properties).customizeConsumerBuilder(builder);
then(builder).should().consumerName("name");
then(builder).should().topics(topics);
then(builder).should().topicsPattern(topisPattern);
then(builder).should().priorityLevel(123);
then(builder).should().readCompacted(true);
then(builder).should().deadLetterPolicy(new DeadLetterPolicy(1, null, "my-dlt", null));
}
@Test
void customizeContainerProperties() {
PulsarProperties properties = new PulsarProperties();
properties.getConsumer().getSubscription().setType(SubscriptionType.Shared);
properties.getListener().setSchemaType(SchemaType.AVRO);
properties.getListener().setObservationEnabled(false);
PulsarContainerProperties containerProperties = new PulsarContainerProperties("my-topic-pattern");
new PulsarPropertiesMapper(properties).customizeContainerProperties(containerProperties);
assertThat(containerProperties.getSubscriptionType()).isEqualTo(SubscriptionType.Shared);
assertThat(containerProperties.getSchemaType()).isEqualTo(SchemaType.AVRO);
assertThat(containerProperties.isObservationEnabled()).isFalse();
}
@Test
@SuppressWarnings("unchecked")
void customizeReaderBuilder() {
PulsarProperties properties = new PulsarProperties();
List<String> topics = List.of("mytopic");
properties.getReader().setName("name");
properties.getReader().setTopics(topics);
properties.getReader().setSubscriptionName("subname");
properties.getReader().setSubscriptionRolePrefix("subroleprefix");
properties.getReader().setReadCompacted(true);
ReaderBuilder<Object> builder = mock(ReaderBuilder.class);
new PulsarPropertiesMapper(properties).customizeReaderBuilder(builder);
then(builder).should().readerName("name");
then(builder).should().topics(topics);
then(builder).should().subscriptionName("subname");
then(builder).should().subscriptionRolePrefix("subroleprefix");
then(builder).should().readCompacted(true);
}
}

View File

@@ -0,0 +1,365 @@
/*
* Copyright 2012-2023 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* https://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.boot.autoconfigure.pulsar;
import java.time.Duration;
import java.util.HashMap;
import java.util.Map;
import org.apache.pulsar.client.api.CompressionType;
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.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.common.schema.SchemaType;
import org.junit.jupiter.api.Nested;
import org.junit.jupiter.api.Test;
import org.springframework.boot.autoconfigure.pulsar.PulsarProperties.Defaults.SchemaInfo;
import org.springframework.boot.autoconfigure.pulsar.PulsarProperties.Defaults.TypeMapping;
import org.springframework.boot.context.properties.bind.BindException;
import org.springframework.boot.context.properties.bind.Binder;
import org.springframework.boot.context.properties.source.MapConfigurationPropertySource;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatExceptionOfType;
/**
* Tests for {@link PulsarProperties}.
*
* @author Chris Bono
* @author Christophe Bornet
* @author Soby Chacko
* @author Phillip Webb
*/
class PulsarPropertiesTests {
private PulsarProperties bindPropeties(Map<String, String> map) {
return new Binder(new MapConfigurationPropertySource(map)).bind("spring.pulsar", PulsarProperties.class).get();
}
@Nested
class ClientProperties {
@Test
void bind() {
Map<String, String> map = new HashMap<>();
map.put("spring.pulsar.client.service-url", "my-service-url");
map.put("spring.pulsar.client.operation-timeout", "1s");
map.put("spring.pulsar.client.lookup-timeout", "2s");
map.put("spring.pulsar.client.connection-timeout", "12s");
PulsarProperties.Client properties = bindPropeties(map).getClient();
assertThat(properties.getServiceUrl()).isEqualTo("my-service-url");
assertThat(properties.getOperationTimeout()).isEqualTo(Duration.ofMillis(1000));
assertThat(properties.getLookupTimeout()).isEqualTo(Duration.ofMillis(2000));
assertThat(properties.getConnectionTimeout()).isEqualTo(Duration.ofMillis(12000));
}
@Test
void bindAuthentication() {
Map<String, String> map = new HashMap<>();
map.put("spring.pulsar.client.authentication.plugin-class-name", "com.example.MyAuth");
map.put("spring.pulsar.client.authentication.param.token", "1234");
PulsarProperties.Client properties = bindPropeties(map).getClient();
assertThat(properties.getAuthentication().getPluginClassName()).isEqualTo("com.example.MyAuth");
assertThat(properties.getAuthentication().getParam()).containsEntry("token", "1234");
}
}
@Nested
class AdminProperties {
private final String authPluginClassName = "org.apache.pulsar.client.impl.auth.AuthenticationToken";
private final String authToken = "1234";
@Test
void bind() {
Map<String, String> map = new HashMap<>();
map.put("spring.pulsar.admin.service-url", "my-service-url");
map.put("spring.pulsar.admin.connection-timeout", "12s");
map.put("spring.pulsar.admin.read-timeout", "13s");
map.put("spring.pulsar.admin.request-timeout", "14s");
PulsarProperties.Admin properties = bindPropeties(map).getAdmin();
assertThat(properties.getServiceUrl()).isEqualTo("my-service-url");
assertThat(properties.getConnectionTimeout()).isEqualTo(Duration.ofSeconds(12));
assertThat(properties.getReadTimeout()).isEqualTo(Duration.ofSeconds(13));
assertThat(properties.getRequestTimeout()).isEqualTo(Duration.ofSeconds(14));
}
@Test
void bindAuthentication() {
Map<String, String> map = new HashMap<>();
map.put("spring.pulsar.admin.authentication.plugin-class-name", this.authPluginClassName);
map.put("spring.pulsar.admin.authentication.param.token", this.authToken);
PulsarProperties.Admin properties = bindPropeties(map).getAdmin();
assertThat(properties.getAuthentication().getPluginClassName()).isEqualTo(this.authPluginClassName);
assertThat(properties.getAuthentication().getParam()).containsEntry("token", this.authToken);
}
}
@Nested
class DefaultsProperties {
@Test
void bindWhenNoTypeMappings() {
assertThat(new PulsarProperties().getDefaults().getTypeMappings()).isEmpty();
}
@Test
void bindWhenTypeMappingsWithTopicsOnly() {
Map<String, String> map = new HashMap<>();
map.put("spring.pulsar.defaults.type-mappings[0].message-type", TestMessage.class.getName());
map.put("spring.pulsar.defaults.type-mappings[0].topic-name", "foo-topic");
map.put("spring.pulsar.defaults.type-mappings[1].message-type", String.class.getName());
map.put("spring.pulsar.defaults.type-mappings[1].topic-name", "string-topic");
PulsarProperties.Defaults properties = bindPropeties(map).getDefaults();
TypeMapping expectedTopic1 = new TypeMapping(TestMessage.class, "foo-topic", null);
TypeMapping expectedTopic2 = new TypeMapping(String.class, "string-topic", null);
assertThat(properties.getTypeMappings()).containsExactly(expectedTopic1, expectedTopic2);
}
@Test
void bindWhenTypeMappingsWithSchemaOnly() {
Map<String, String> map = new HashMap<>();
map.put("spring.pulsar.defaults.type-mappings[0].message-type", TestMessage.class.getName());
map.put("spring.pulsar.defaults.type-mappings[0].schema-info.schema-type", "JSON");
PulsarProperties.Defaults properties = bindPropeties(map).getDefaults();
TypeMapping expected = new TypeMapping(TestMessage.class, null, new SchemaInfo(SchemaType.JSON, null));
assertThat(properties.getTypeMappings()).containsExactly(expected);
}
@Test
void bindWhenTypeMappingsWithTopicAndSchema() {
Map<String, String> map = new HashMap<>();
map.put("spring.pulsar.defaults.type-mappings[0].message-type", TestMessage.class.getName());
map.put("spring.pulsar.defaults.type-mappings[0].topic-name", "foo-topic");
map.put("spring.pulsar.defaults.type-mappings[0].schema-info.schema-type", "JSON");
PulsarProperties.Defaults properties = bindPropeties(map).getDefaults();
TypeMapping expected = new TypeMapping(TestMessage.class, "foo-topic",
new SchemaInfo(SchemaType.JSON, null));
assertThat(properties.getTypeMappings()).containsExactly(expected);
}
@Test
void bindWhenTypeMappingsWithKeyValueSchema() {
Map<String, String> map = new HashMap<>();
map.put("spring.pulsar.defaults.type-mappings[0].message-type", TestMessage.class.getName());
map.put("spring.pulsar.defaults.type-mappings[0].schema-info.schema-type", "KEY_VALUE");
map.put("spring.pulsar.defaults.type-mappings[0].schema-info.message-key-type", String.class.getName());
PulsarProperties.Defaults properties = bindPropeties(map).getDefaults();
TypeMapping expected = new TypeMapping(TestMessage.class, null,
new SchemaInfo(SchemaType.KEY_VALUE, String.class));
assertThat(properties.getTypeMappings()).containsExactly(expected);
}
@Test
void bindWhenNoSchemaThrowsException() {
Map<String, String> map = new HashMap<>();
map.put("spring.pulsar.defaults.type-mappings[0].message-type", TestMessage.class.getName());
map.put("spring.pulsar.defaults.type-mappings[0].schema-info.message-key-type", String.class.getName());
assertThatExceptionOfType(BindException.class).isThrownBy(() -> bindPropeties(map))
.havingRootCause()
.withMessageContaining("schemaType must not be null");
}
@Test
void bindWhenSchemaTypeNoneThrowsException() {
Map<String, String> map = new HashMap<>();
map.put("spring.pulsar.defaults.type-mappings[0].message-type", TestMessage.class.getName());
map.put("spring.pulsar.defaults.type-mappings[0].schema-info.schema-type", "NONE");
assertThatExceptionOfType(BindException.class).isThrownBy(() -> bindPropeties(map))
.havingRootCause()
.withMessageContaining("schemaType 'NONE' not supported");
}
@Test
void bindWhenMessageKeyTypeSetOnNonKeyValueSchemaThrowsException() {
Map<String, String> map = new HashMap<>();
map.put("spring.pulsar.defaults.type-mappings[0].message-type", TestMessage.class.getName());
map.put("spring.pulsar.defaults.type-mappings[0].schema-info.schema-type", "JSON");
map.put("spring.pulsar.defaults.type-mappings[0].schema-info.message-key-type", String.class.getName());
assertThatExceptionOfType(BindException.class).isThrownBy(() -> bindPropeties(map))
.havingRootCause()
.withMessageContaining("messageKeyType can only be set when schemaType is KEY_VALUE");
}
record TestMessage(String value) {
}
}
@Nested
class FunctionProperties {
@Test
void defaults() {
PulsarProperties.Function properties = new PulsarProperties.Function();
assertThat(properties.isFailFast()).isTrue();
assertThat(properties.isPropagateFailures()).isTrue();
assertThat(properties.isPropagateStopFailures()).isFalse();
}
@Test
void bind() {
Map<String, String> props = new HashMap<>();
props.put("spring.pulsar.function.fail-fast", "false");
props.put("spring.pulsar.function.propagate-failures", "false");
props.put("spring.pulsar.function.propagate-stop-failures", "true");
PulsarProperties.Function properties = bindPropeties(props).getFunction();
assertThat(properties.isFailFast()).isFalse();
assertThat(properties.isPropagateFailures()).isFalse();
assertThat(properties.isPropagateStopFailures()).isTrue();
}
}
@Nested
class ProducerProperties {
@Test
void bind() {
Map<String, String> map = new HashMap<>();
map.put("spring.pulsar.producer.name", "my-producer");
map.put("spring.pulsar.producer.topic-name", "my-topic");
map.put("spring.pulsar.producer.send-timeout", "2s");
map.put("spring.pulsar.producer.message-routing-mode", "custompartition");
map.put("spring.pulsar.producer.hashing-scheme", "murmur3_32hash");
map.put("spring.pulsar.producer.batching-enabled", "false");
map.put("spring.pulsar.producer.chunking-enabled", "true");
map.put("spring.pulsar.producer.compression-type", "lz4");
map.put("spring.pulsar.producer.access-mode", "exclusive");
map.put("spring.pulsar.producer.cache.expire-after-access", "2s");
map.put("spring.pulsar.producer.cache.maximum-size", "3");
map.put("spring.pulsar.producer.cache.initial-capacity", "5");
PulsarProperties.Producer properties = bindPropeties(map).getProducer();
assertThat(properties.getName()).isEqualTo("my-producer");
assertThat(properties.getTopicName()).isEqualTo("my-topic");
assertThat(properties.getSendTimeout()).isEqualTo(Duration.ofSeconds(2));
assertThat(properties.getMessageRoutingMode()).isEqualTo(MessageRoutingMode.CustomPartition);
assertThat(properties.getHashingScheme()).isEqualTo(HashingScheme.Murmur3_32Hash);
assertThat(properties.isBatchingEnabled()).isFalse();
assertThat(properties.isChunkingEnabled()).isTrue();
assertThat(properties.getCompressionType()).isEqualTo(CompressionType.LZ4);
assertThat(properties.getAccessMode()).isEqualTo(ProducerAccessMode.Exclusive);
assertThat(properties.getCache().getExpireAfterAccess()).isEqualTo(Duration.ofSeconds(2));
assertThat(properties.getCache().getMaximumSize()).isEqualTo(3);
assertThat(properties.getCache().getInitialCapacity()).isEqualTo(5);
}
}
@Nested
class ConsumerPropertiesTests {
@Test
void bind() {
Map<String, String> map = new HashMap<>();
map.put("spring.pulsar.consumer.name", "my-consumer");
map.put("spring.pulsar.consumer.subscription.initial-position", "earliest");
map.put("spring.pulsar.consumer.subscription.mode", "nondurable");
map.put("spring.pulsar.consumer.subscription.name", "my-subscription");
map.put("spring.pulsar.consumer.subscription.topics-mode", "all-topics");
map.put("spring.pulsar.consumer.subscription.type", "shared");
map.put("spring.pulsar.consumer.topics[0]", "my-topic");
map.put("spring.pulsar.consumer.topics-pattern", "my-pattern");
map.put("spring.pulsar.consumer.priority-level", "8");
map.put("spring.pulsar.consumer.read-compacted", "true");
map.put("spring.pulsar.consumer.dead-letter-policy.max-redeliver-count", "4");
map.put("spring.pulsar.consumer.dead-letter-policy.retry-letter-topic", "my-retry-topic");
map.put("spring.pulsar.consumer.dead-letter-policy.dead-letter-topic", "my-dlt-topic");
map.put("spring.pulsar.consumer.dead-letter-policy.initial-subscription-name", "my-initial-subscription");
map.put("spring.pulsar.consumer.retry-enable", "true");
PulsarProperties.Consumer properties = bindPropeties(map).getConsumer();
assertThat(properties.getName()).isEqualTo("my-consumer");
assertThat(properties.getSubscription()).satisfies((subscription) -> {
assertThat(subscription.getName()).isEqualTo("my-subscription");
assertThat(subscription.getType()).isEqualTo(SubscriptionType.Shared);
assertThat(subscription.getMode()).isEqualTo(SubscriptionMode.NonDurable);
assertThat(subscription.getInitialPosition()).isEqualTo(SubscriptionInitialPosition.Earliest);
assertThat(subscription.getTopicsMode()).isEqualTo(RegexSubscriptionMode.AllTopics);
});
assertThat(properties.getTopics()).containsExactly("my-topic");
assertThat(properties.getTopicsPattern().toString()).isEqualTo("my-pattern");
assertThat(properties.getPriorityLevel()).isEqualTo(8);
assertThat(properties.isReadCompacted()).isTrue();
assertThat(properties.getDeadLetterPolicy()).satisfies((policy) -> {
assertThat(policy.getMaxRedeliverCount()).isEqualTo(4);
assertThat(policy.getRetryLetterTopic()).isEqualTo("my-retry-topic");
assertThat(policy.getDeadLetterTopic()).isEqualTo("my-dlt-topic");
assertThat(policy.getInitialSubscriptionName()).isEqualTo("my-initial-subscription");
});
assertThat(properties.isRetryEnable()).isTrue();
}
}
@Nested
class ListenerProperties {
@Test
void bind() {
Map<String, String> map = new HashMap<>();
map.put("spring.pulsar.listener.schema-type", "avro");
map.put("spring.pulsar.listener.observation-enabled", "false");
PulsarProperties.Listener properties = bindPropeties(map).getListener();
assertThat(properties.getSchemaType()).isEqualTo(SchemaType.AVRO);
assertThat(properties.isObservationEnabled()).isFalse();
}
}
@Nested
class ReaderProperties {
@Test
void bind() {
Map<String, String> map = new HashMap<>();
map.put("spring.pulsar.reader.name", "my-reader");
map.put("spring.pulsar.reader.topics", "my-topic");
map.put("spring.pulsar.reader.subscription-name", "my-subscription");
map.put("spring.pulsar.reader.subscription-role-prefix", "sub-role");
map.put("spring.pulsar.reader.read-compacted", "true");
PulsarProperties.Reader properties = bindPropeties(map).getReader();
assertThat(properties.getName()).isEqualTo("my-reader");
assertThat(properties.getTopics()).containsExactly("my-topic");
assertThat(properties.getSubscriptionName()).isEqualTo("my-subscription");
assertThat(properties.getSubscriptionRolePrefix()).isEqualTo("sub-role");
assertThat(properties.isReadCompacted()).isTrue();
}
}
@Nested
class TemplateProperties {
@Test
void bind() {
Map<String, String> map = new HashMap<>();
map.put("spring.pulsar.template.observations-enabled", "false");
PulsarProperties.Template properties = bindPropeties(map).getTemplate();
assertThat(properties.isObservationsEnabled()).isFalse();
}
}
}

View File

@@ -0,0 +1,473 @@
/*
* Copyright 2012-2023 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* https://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.boot.autoconfigure.pulsar;
import java.time.Duration;
import java.util.ArrayList;
import java.util.List;
import java.util.function.Supplier;
import com.github.benmanes.caffeine.cache.Caffeine;
import org.apache.pulsar.client.api.PulsarClient;
import org.apache.pulsar.common.schema.SchemaType;
import org.apache.pulsar.reactive.client.adapter.ProducerCacheProvider;
import org.apache.pulsar.reactive.client.api.ReactiveMessageConsumerBuilder;
import org.apache.pulsar.reactive.client.api.ReactiveMessageReaderBuilder;
import org.apache.pulsar.reactive.client.api.ReactiveMessageSenderBuilder;
import org.apache.pulsar.reactive.client.api.ReactiveMessageSenderCache;
import org.apache.pulsar.reactive.client.api.ReactivePulsarClient;
import org.apache.pulsar.reactive.client.producercache.CaffeineShadedProducerCacheProvider;
import org.assertj.core.api.AbstractObjectAssert;
import org.assertj.core.api.InstanceOfAssertFactories;
import org.junit.jupiter.api.Nested;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.ValueSource;
import org.springframework.boot.autoconfigure.AutoConfigurations;
import org.springframework.boot.test.context.FilteredClassLoader;
import org.springframework.boot.test.context.TestConfiguration;
import org.springframework.boot.test.context.assertj.AssertableApplicationContext;
import org.springframework.boot.test.context.runner.ApplicationContextRunner;
import org.springframework.context.annotation.Bean;
import org.springframework.core.annotation.Order;
import org.springframework.pulsar.core.DefaultSchemaResolver;
import org.springframework.pulsar.core.DefaultTopicResolver;
import org.springframework.pulsar.core.PulsarAdministration;
import org.springframework.pulsar.core.SchemaResolver;
import org.springframework.pulsar.core.TopicResolver;
import org.springframework.pulsar.reactive.config.DefaultReactivePulsarListenerContainerFactory;
import org.springframework.pulsar.reactive.config.ReactivePulsarListenerContainerFactory;
import org.springframework.pulsar.reactive.config.ReactivePulsarListenerEndpointRegistry;
import org.springframework.pulsar.reactive.config.annotation.ReactivePulsarBootstrapConfiguration;
import org.springframework.pulsar.reactive.config.annotation.ReactivePulsarListenerAnnotationBeanPostProcessor;
import org.springframework.pulsar.reactive.core.DefaultReactivePulsarConsumerFactory;
import org.springframework.pulsar.reactive.core.DefaultReactivePulsarReaderFactory;
import org.springframework.pulsar.reactive.core.DefaultReactivePulsarSenderFactory;
import org.springframework.pulsar.reactive.core.ReactiveMessageConsumerBuilderCustomizer;
import org.springframework.pulsar.reactive.core.ReactiveMessageReaderBuilderCustomizer;
import org.springframework.pulsar.reactive.core.ReactiveMessageSenderBuilderCustomizer;
import org.springframework.pulsar.reactive.core.ReactivePulsarConsumerFactory;
import org.springframework.pulsar.reactive.core.ReactivePulsarReaderFactory;
import org.springframework.pulsar.reactive.core.ReactivePulsarSenderFactory;
import org.springframework.pulsar.reactive.core.ReactivePulsarTemplate;
import org.springframework.pulsar.reactive.listener.ReactivePulsarContainerProperties;
import static org.assertj.core.api.Assertions.assertThat;
import static org.mockito.Mockito.mock;
/**
* Tests for {@link PulsarReactiveAutoConfiguration}.
*
* @author Chris Bono
* @author Christophe Bornet
* @author Phillip Webb
*/
class PulsarReactiveAutoConfigurationTests {
private static final String INTERNAL_PULSAR_LISTENER_ANNOTATION_PROCESSOR = "org.springframework.pulsar.config.internalReactivePulsarListenerAnnotationProcessor";
private final ApplicationContextRunner contextRunner = new ApplicationContextRunner()
.withConfiguration(AutoConfigurations.of(PulsarReactiveAutoConfiguration.class))
.withBean(PulsarClient.class, () -> mock(PulsarClient.class));
@Test
void whenPulsarNotOnClasspathAutoConfigurationIsSkipped() {
new ApplicationContextRunner().withConfiguration(AutoConfigurations.of(PulsarReactiveAutoConfiguration.class))
.withClassLoader(new FilteredClassLoader(PulsarClient.class))
.run((context) -> assertThat(context).doesNotHaveBean(PulsarReactiveAutoConfiguration.class));
}
@Test
void whenReactivePulsarNotOnClasspathAutoConfigurationIsSkipped() {
this.contextRunner.withClassLoader(new FilteredClassLoader(ReactivePulsarClient.class))
.run((context) -> assertThat(context).doesNotHaveBean(PulsarReactiveAutoConfiguration.class));
}
@Test
void whenReactiveSpringPulsarNotOnClasspathAutoConfigurationIsSkipped() {
this.contextRunner.withClassLoader(new FilteredClassLoader(ReactivePulsarTemplate.class))
.run((context) -> assertThat(context).doesNotHaveBean(PulsarReactiveAutoConfiguration.class));
}
@Test
void whenCustomPulsarListenerAnnotationProcessorDefinedAutoConfigurationIsSkipped() {
this.contextRunner.withBean(INTERNAL_PULSAR_LISTENER_ANNOTATION_PROCESSOR, String.class, () -> "bean")
.run((context) -> assertThat(context).doesNotHaveBean(ReactivePulsarBootstrapConfiguration.class));
}
@Test
void autoConfiguresBeans() {
this.contextRunner.run((context) -> assertThat(context).hasSingleBean(PulsarConfiguration.class)
.hasSingleBean(PulsarClient.class)
.hasSingleBean(PulsarAdministration.class)
.hasSingleBean(DefaultSchemaResolver.class)
.hasSingleBean(DefaultTopicResolver.class)
.hasSingleBean(ReactivePulsarClient.class)
.hasSingleBean(CaffeineShadedProducerCacheProvider.class)
.hasSingleBean(ReactiveMessageSenderCache.class)
.hasSingleBean(DefaultReactivePulsarSenderFactory.class)
.hasSingleBean(ReactivePulsarTemplate.class)
.hasSingleBean(DefaultReactivePulsarConsumerFactory.class)
.hasSingleBean(DefaultReactivePulsarListenerContainerFactory.class)
.hasSingleBean(ReactivePulsarListenerAnnotationBeanPostProcessor.class)
.hasSingleBean(ReactivePulsarListenerEndpointRegistry.class));
}
@Test
@SuppressWarnings("rawtypes")
void injectsExpectedBeansIntoReactivePulsarClient() {
this.contextRunner.run((context) -> {
PulsarClient pulsarClient = context.getBean(PulsarClient.class);
assertThat(context).hasNotFailed()
.getBean(ReactivePulsarClient.class)
.extracting("reactivePulsarResourceAdapter")
.extracting("pulsarClientSupplier", InstanceOfAssertFactories.type(Supplier.class))
.extracting(Supplier::get)
.isSameAs(pulsarClient);
});
}
@ParameterizedTest
@ValueSource(classes = { ReactivePulsarClient.class, ProducerCacheProvider.class, ReactiveMessageSenderCache.class,
ReactivePulsarSenderFactory.class, ReactivePulsarConsumerFactory.class, ReactivePulsarReaderFactory.class,
ReactivePulsarTemplate.class })
<T> void whenHasUserDefinedBeanDoesNotAutoConfigureBean(Class<T> beanClass) {
T bean = mock(beanClass);
this.contextRunner.withBean(beanClass.getName(), beanClass, () -> bean)
.run((context) -> assertThat(context).getBean(beanClass).isSameAs(bean));
}
@Nested
class SenderFactoryTests {
private final ApplicationContextRunner contextRunner = PulsarReactiveAutoConfigurationTests.this.contextRunner;
@Test
void injectsExpectedBeans() {
ReactivePulsarClient client = mock(ReactivePulsarClient.class);
ReactiveMessageSenderCache cache = mock(ReactiveMessageSenderCache.class);
this.contextRunner.withPropertyValues("spring.pulsar.producer.topic-name=test-topic")
.withBean("customReactivePulsarClient", ReactivePulsarClient.class, () -> client)
.withBean("customReactiveMessageSenderCache", ReactiveMessageSenderCache.class, () -> cache)
.run((context) -> {
DefaultReactivePulsarSenderFactory<?> senderFactory = context
.getBean(DefaultReactivePulsarSenderFactory.class);
assertThat(senderFactory)
.extracting("reactivePulsarClient", InstanceOfAssertFactories.type(ReactivePulsarClient.class))
.isSameAs(client);
assertThat(senderFactory)
.extracting("reactiveMessageSenderCache",
InstanceOfAssertFactories.type(ReactiveMessageSenderCache.class))
.isSameAs(cache);
assertThat(senderFactory)
.extracting("topicResolver", InstanceOfAssertFactories.type(TopicResolver.class))
.isSameAs(context.getBean(TopicResolver.class));
});
}
@Test
void injectsExpectedBeansIntoReactiveMessageSenderCache() {
ProducerCacheProvider provider = mock(ProducerCacheProvider.class);
this.contextRunner.withBean("customProducerCacheProvider", ProducerCacheProvider.class, () -> provider)
.run((context) -> assertThat(context).getBean(ReactiveMessageSenderCache.class)
.extracting("cacheProvider", InstanceOfAssertFactories.type(ProducerCacheProvider.class))
.isSameAs(provider));
}
@Test
<T> void whenHasUserDefinedCustomizersAppliesInCorrectOrder() {
this.contextRunner.withPropertyValues("spring.pulsar.producer.name=fromPropsCustomizer")
.withUserConfiguration(ReactiveMessageSenderBuilderCustomizerConfig.class)
.run((context) -> {
DefaultReactivePulsarSenderFactory<?> producerFactory = context
.getBean(DefaultReactivePulsarSenderFactory.class);
Customizers<ReactiveMessageSenderBuilderCustomizer<T>, ReactiveMessageSenderBuilder<T>> customizers = Customizers
.of(ReactiveMessageSenderBuilder.class, ReactiveMessageSenderBuilderCustomizer::customize);
assertThat(customizers.fromField(producerFactory, "defaultConfigCustomizers")).callsInOrder(
ReactiveMessageSenderBuilder::producerName, "fromPropsCustomizer", "fromCustomizer1",
"fromCustomizer2");
});
}
@TestConfiguration(proxyBeanMethods = false)
static class ReactiveMessageSenderBuilderCustomizerConfig {
@Bean
@Order(200)
ReactiveMessageSenderBuilderCustomizer<?> customizerFoo() {
return (builder) -> builder.producerName("fromCustomizer2");
}
@Bean
@Order(100)
ReactiveMessageSenderBuilderCustomizer<?> customizerBar() {
return (builder) -> builder.producerName("fromCustomizer1");
}
}
}
@Nested
class TemplateTests {
private final ApplicationContextRunner contextRunner = PulsarReactiveAutoConfigurationTests.this.contextRunner;
@Test
@SuppressWarnings("rawtypes")
void injectsExpectedBeans() {
ReactivePulsarSenderFactory senderFactory = mock(ReactivePulsarSenderFactory.class);
SchemaResolver schemaResolver = mock(SchemaResolver.class);
this.contextRunner
.withBean("customReactivePulsarSenderFactory", ReactivePulsarSenderFactory.class, () -> senderFactory)
.withBean("schemaResolver", SchemaResolver.class, () -> schemaResolver)
.run((context) -> assertThat(context).getBean(ReactivePulsarTemplate.class).satisfies((template) -> {
assertThat(template).extracting("reactiveMessageSenderFactory").isSameAs(senderFactory);
assertThat(template).extracting("schemaResolver").isSameAs(schemaResolver);
}));
}
}
@Nested
class ConsumerFactoryTests {
private final ApplicationContextRunner contextRunner = PulsarReactiveAutoConfigurationTests.this.contextRunner;
@Test
void injectsExpectedBeans() {
ReactivePulsarClient client = mock(ReactivePulsarClient.class);
this.contextRunner.withBean("customReactivePulsarClient", ReactivePulsarClient.class, () -> client)
.run((context) -> {
ReactivePulsarConsumerFactory<?> consumerFactory = context
.getBean(DefaultReactivePulsarConsumerFactory.class);
assertThat(consumerFactory)
.extracting("reactivePulsarClient", InstanceOfAssertFactories.type(ReactivePulsarClient.class))
.isSameAs(client);
});
}
@Test
<T> void whenHasUserDefinedCustomizersAppliesInCorrectOrder() {
this.contextRunner.withPropertyValues("spring.pulsar.consumer.name=fromPropsCustomizer")
.withUserConfiguration(ReactiveMessageConsumerBuilderCustomizerConfig.class)
.run((context) -> {
DefaultReactivePulsarConsumerFactory<?> consumerFactory = context
.getBean(DefaultReactivePulsarConsumerFactory.class);
Customizers<ReactiveMessageConsumerBuilderCustomizer<T>, ReactiveMessageConsumerBuilder<T>> customizers = Customizers
.of(ReactiveMessageConsumerBuilder.class, ReactiveMessageConsumerBuilderCustomizer::customize);
assertThat(customizers.fromField(consumerFactory, "defaultConfigCustomizers")).callsInOrder(
ReactiveMessageConsumerBuilder::consumerName, "fromPropsCustomizer", "fromCustomizer1",
"fromCustomizer2");
});
}
@TestConfiguration(proxyBeanMethods = false)
static class ReactiveMessageConsumerBuilderCustomizerConfig {
@Bean
@Order(200)
ReactiveMessageConsumerBuilderCustomizer<?> customizerFoo() {
return (builder) -> builder.consumerName("fromCustomizer2");
}
@Bean
@Order(100)
ReactiveMessageConsumerBuilderCustomizer<?> customizerBar() {
return (builder) -> builder.consumerName("fromCustomizer1");
}
}
}
@Nested
class ListenerTests {
private final ApplicationContextRunner contextRunner = PulsarReactiveAutoConfigurationTests.this.contextRunner;
@Test
void whenHasUserDefinedBeanDoesNotAutoConfigureBean() {
ReactivePulsarListenerContainerFactory<?> listenerContainerFactory = mock(
ReactivePulsarListenerContainerFactory.class);
this.contextRunner
.withBean("reactivePulsarListenerContainerFactory", ReactivePulsarListenerContainerFactory.class,
() -> listenerContainerFactory)
.run((context) -> assertThat(context).getBean(ReactivePulsarListenerContainerFactory.class)
.isSameAs(listenerContainerFactory));
}
@Test
void whenHasUserDefinedReactivePulsarListenerAnnotationBeanPostProcessorDoesNotAutoConfigureBean() {
ReactivePulsarListenerAnnotationBeanPostProcessor<?> listenerAnnotationBeanPostProcessor = mock(
ReactivePulsarListenerAnnotationBeanPostProcessor.class);
this.contextRunner.withBean(INTERNAL_PULSAR_LISTENER_ANNOTATION_PROCESSOR,
ReactivePulsarListenerAnnotationBeanPostProcessor.class, () -> listenerAnnotationBeanPostProcessor)
.run((context) -> assertThat(context).getBean(ReactivePulsarListenerAnnotationBeanPostProcessor.class)
.isSameAs(listenerAnnotationBeanPostProcessor));
}
@Test
void whenHasCustomProperties() {
List<String> properties = new ArrayList<>();
properties.add("spring.pulsar.listener.schema-type=avro");
this.contextRunner.withPropertyValues(properties.toArray(String[]::new)).run((context) -> {
DefaultReactivePulsarListenerContainerFactory<?> factory = context
.getBean(DefaultReactivePulsarListenerContainerFactory.class);
assertThat(factory.getContainerProperties().getSchemaType()).isEqualTo(SchemaType.AVRO);
});
}
@Test
void injectsExpectedBeans() {
ReactivePulsarConsumerFactory<?> consumerFactory = mock(ReactivePulsarConsumerFactory.class);
SchemaResolver schemaResolver = mock(SchemaResolver.class);
this.contextRunner
.withBean("customReactivePulsarConsumerFactory", ReactivePulsarConsumerFactory.class,
() -> consumerFactory)
.withBean("schemaResolver", SchemaResolver.class, () -> schemaResolver)
.run((context) -> {
DefaultReactivePulsarListenerContainerFactory<?> containerFactory = context
.getBean(DefaultReactivePulsarListenerContainerFactory.class);
assertThat(containerFactory).extracting("consumerFactory").isSameAs(consumerFactory);
assertThat(containerFactory)
.extracting(DefaultReactivePulsarListenerContainerFactory::getContainerProperties)
.extracting(ReactivePulsarContainerProperties::getSchemaResolver)
.isSameAs(schemaResolver);
});
}
}
@Nested
class ReaderFactoryTests {
private final ApplicationContextRunner contextRunner = PulsarReactiveAutoConfigurationTests.this.contextRunner;
@Test
void injectsExpectedBeans() {
ReactivePulsarClient client = mock(ReactivePulsarClient.class);
this.contextRunner.withPropertyValues("spring.pulsar.reader.name=test-reader")
.withBean("customReactivePulsarClient", ReactivePulsarClient.class, () -> client)
.run((context) -> {
DefaultReactivePulsarReaderFactory<?> readerFactory = context
.getBean(DefaultReactivePulsarReaderFactory.class);
assertThat(readerFactory)
.extracting("reactivePulsarClient", InstanceOfAssertFactories.type(ReactivePulsarClient.class))
.isSameAs(client);
});
}
@Test
<T> void whenHasUserDefinedCustomizersAppliesInCorrectOrder() {
this.contextRunner.withPropertyValues("spring.pulsar.reader.name=fromPropsCustomizer")
.withUserConfiguration(ReactiveMessageReaderBuilderCustomizerConfig.class)
.run((context) -> {
DefaultReactivePulsarReaderFactory<?> readerFactory = context
.getBean(DefaultReactivePulsarReaderFactory.class);
Customizers<ReactiveMessageReaderBuilderCustomizer<T>, ReactiveMessageReaderBuilder<T>> customizers = Customizers
.of(ReactiveMessageReaderBuilder.class, ReactiveMessageReaderBuilderCustomizer::customize);
assertThat(customizers.fromField(readerFactory, "defaultConfigCustomizers")).callsInOrder(
ReactiveMessageReaderBuilder::readerName, "fromPropsCustomizer", "fromCustomizer1",
"fromCustomizer2");
});
}
@TestConfiguration(proxyBeanMethods = false)
static class ReactiveMessageReaderBuilderCustomizerConfig {
@Bean
@Order(200)
ReactiveMessageReaderBuilderCustomizer<?> customizerFoo() {
return (builder) -> builder.readerName("fromCustomizer2");
}
@Bean
@Order(100)
ReactiveMessageReaderBuilderCustomizer<?> customizerBar() {
return (builder) -> builder.readerName("fromCustomizer1");
}
}
}
@Nested
class SenderCacheAutoConfigurationTests {
private final ApplicationContextRunner contextRunner = PulsarReactiveAutoConfigurationTests.this.contextRunner;
@Test
void whenNoPropertiesEnablesCaching() {
this.contextRunner.run(this::assertCaffeineProducerCacheProvider);
}
@Test
void whenCachingEnabledEnablesCaching() {
this.contextRunner.withPropertyValues("spring.pulsar.producer.cache.enabled=true")
.run(this::assertCaffeineProducerCacheProvider);
}
@Test
void whenCachingDisabledDoesNotEnableCaching() {
this.contextRunner.withPropertyValues("spring.pulsar.producer.cache.enabled=false")
.run((context) -> assertThat(context).doesNotHaveBean(ProducerCacheProvider.class)
.doesNotHaveBean(ReactiveMessageSenderCache.class));
}
@Test
void whenCachingEnabledAndCaffeineNotOnClasspathStillUsesCaffeine() {
// The reactive client shades Caffeine - it should still be used
this.contextRunner.withClassLoader(new FilteredClassLoader(Caffeine.class))
.withPropertyValues("spring.pulsar.producer.cache.enabled=true")
.run(this::assertCaffeineProducerCacheProvider);
}
@Test
void whenCachingEnabledAndNoCacheProviderAvailable() {
// The reactive client uses a shaded caffeine cache provider as its internal
// cache
this.contextRunner.withClassLoader(new FilteredClassLoader(CaffeineShadedProducerCacheProvider.class))
.withPropertyValues("spring.pulsar.producer.cache.enabled=true")
.run((context) -> assertThat(context).doesNotHaveBean(ProducerCacheProvider.class)
.getBean(ReactiveMessageSenderCache.class)
.extracting("cacheProvider")
.isExactlyInstanceOf(CaffeineShadedProducerCacheProvider.class));
}
@Test
void whenCustomCachingPropertiesCreatesConfiguredBean() {
this.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) -> assertCaffeineProducerCacheProvider(context).extracting("cache.cache")
.hasFieldOrPropertyWithValue("expiresAfterAccessNanos", Duration.ofSeconds(100).toNanos())
.hasFieldOrPropertyWithValue("maximum", 5150L));
}
private AbstractObjectAssert<?, ProducerCacheProvider> assertCaffeineProducerCacheProvider(
AssertableApplicationContext context) {
return assertThat(context).hasSingleBean(ReactiveMessageSenderCache.class)
.getBean(ProducerCacheProvider.class)
.isExactlyInstanceOf(CaffeineShadedProducerCacheProvider.class);
}
}
}

View File

@@ -0,0 +1,146 @@
/*
* Copyright 2012-2023 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* https://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.boot.autoconfigure.pulsar;
import java.time.Duration;
import java.util.List;
import java.util.regex.Pattern;
import org.apache.pulsar.client.api.CompressionType;
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.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.common.schema.SchemaType;
import org.apache.pulsar.reactive.client.api.ReactiveMessageConsumerBuilder;
import org.apache.pulsar.reactive.client.api.ReactiveMessageReaderBuilder;
import org.apache.pulsar.reactive.client.api.ReactiveMessageSenderBuilder;
import org.junit.jupiter.api.Test;
import org.springframework.boot.autoconfigure.pulsar.PulsarProperties.Consumer;
import org.springframework.boot.autoconfigure.pulsar.PulsarProperties.Consumer.Subscription;
import org.springframework.pulsar.reactive.listener.ReactivePulsarContainerProperties;
import static org.assertj.core.api.Assertions.assertThat;
import static org.mockito.BDDMockito.then;
import static org.mockito.Mockito.mock;
/**
* Tests for {@link PulsarReactivePropertiesMapper}.
*
* @author Chris Bono
* @author Phillip Webb
*/
class PulsarReactivePropertiesMapperTests {
@Test
@SuppressWarnings("unchecked")
void customizeMessageSenderBuilder() {
PulsarProperties properties = new PulsarProperties();
properties.getProducer().setName("name");
properties.getProducer().setTopicName("topicname");
properties.getProducer().setSendTimeout(Duration.ofSeconds(1));
properties.getProducer().setMessageRoutingMode(MessageRoutingMode.RoundRobinPartition);
properties.getProducer().setHashingScheme(HashingScheme.JavaStringHash);
properties.getProducer().setBatchingEnabled(false);
properties.getProducer().setChunkingEnabled(true);
properties.getProducer().setCompressionType(CompressionType.SNAPPY);
properties.getProducer().setAccessMode(ProducerAccessMode.Exclusive);
ReactiveMessageSenderBuilder<Object> builder = mock(ReactiveMessageSenderBuilder.class);
new PulsarReactivePropertiesMapper(properties).customizeMessageSenderBuilder(builder);
then(builder).should().producerName("name");
then(builder).should().topic("topicname");
then(builder).should().sendTimeout(Duration.ofSeconds(1));
then(builder).should().messageRoutingMode(MessageRoutingMode.RoundRobinPartition);
then(builder).should().hashingScheme(HashingScheme.JavaStringHash);
then(builder).should().batchingEnabled(false);
then(builder).should().chunkingEnabled(true);
then(builder).should().compressionType(CompressionType.SNAPPY);
then(builder).should().accessMode(ProducerAccessMode.Exclusive);
}
@Test
@SuppressWarnings("unchecked")
void customizeMessageConsumerBuilder() {
PulsarProperties properties = new PulsarProperties();
List<String> topics = List.of("mytopic");
Pattern topisPattern = Pattern.compile("my-pattern");
properties.getConsumer().setName("name");
properties.getConsumer().setTopics(topics);
properties.getConsumer().setTopicsPattern(topisPattern);
properties.getConsumer().setPriorityLevel(123);
properties.getConsumer().setReadCompacted(true);
Consumer.DeadLetterPolicy deadLetterPolicy = new Consumer.DeadLetterPolicy();
deadLetterPolicy.setDeadLetterTopic("my-dlt");
deadLetterPolicy.setMaxRedeliverCount(1);
properties.getConsumer().setDeadLetterPolicy(deadLetterPolicy);
properties.getConsumer().setRetryEnable(false);
Subscription subscriptionProperties = properties.getConsumer().getSubscription();
subscriptionProperties.setName("subname");
subscriptionProperties.setInitialPosition(SubscriptionInitialPosition.Earliest);
subscriptionProperties.setMode(SubscriptionMode.NonDurable);
subscriptionProperties.setTopicsMode(RegexSubscriptionMode.NonPersistentOnly);
subscriptionProperties.setType(SubscriptionType.Key_Shared);
ReactiveMessageConsumerBuilder<Object> builder = mock(ReactiveMessageConsumerBuilder.class);
new PulsarReactivePropertiesMapper(properties).customizeMessageConsumerBuilder(builder);
then(builder).should().consumerName("name");
then(builder).should().topics(topics);
then(builder).should().topicsPattern(topisPattern);
then(builder).should().priorityLevel(123);
then(builder).should().readCompacted(true);
then(builder).should().deadLetterPolicy(new DeadLetterPolicy(1, null, "my-dlt", null));
then(builder).should().retryLetterTopicEnable(false);
then(builder).should().subscriptionName("subname");
then(builder).should().subscriptionInitialPosition(SubscriptionInitialPosition.Earliest);
then(builder).should().subscriptionMode(SubscriptionMode.NonDurable);
then(builder).should().topicsPatternSubscriptionMode(RegexSubscriptionMode.NonPersistentOnly);
then(builder).should().subscriptionType(SubscriptionType.Key_Shared);
}
@Test
void customizeContainerProperties() {
PulsarProperties properties = new PulsarProperties();
properties.getConsumer().getSubscription().setType(SubscriptionType.Shared);
properties.getListener().setSchemaType(SchemaType.AVRO);
ReactivePulsarContainerProperties<Object> containerProperties = new ReactivePulsarContainerProperties<>();
new PulsarReactivePropertiesMapper(properties).customizeContainerProperties(containerProperties);
assertThat(containerProperties.getSubscriptionType()).isEqualTo(SubscriptionType.Shared);
assertThat(containerProperties.getSchemaType()).isEqualTo(SchemaType.AVRO);
}
@Test
@SuppressWarnings("unchecked")
void customizeMessageReaderBuilder() {
List<String> topics = List.of("my-topic");
PulsarProperties properties = new PulsarProperties();
properties.getReader().setName("name");
properties.getReader().setTopics(topics);
properties.getReader().setSubscriptionName("subname");
properties.getReader().setSubscriptionRolePrefix("srp");
ReactiveMessageReaderBuilder<Object> builder = mock(ReactiveMessageReaderBuilder.class);
new PulsarReactivePropertiesMapper(properties).customizeMessageReaderBuilder(builder);
then(builder).should().readerName("name");
then(builder).should().topics(topics);
then(builder).should().subscriptionName("subname");
then(builder).should().generatedSubscriptionNamePrefix("srp");
}
}