Spring/spring-pulsar
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Add caching producer factory

Browse Source 
Fixes #30
This commit is contained in:
Chris Bono
2022-07-22 19:24:49 -05:00
parent 3faddca84a
commit 33fba1d450
15 changed files with 899 additions and 55 deletions
Download Patch File Download Diff File Expand all files Collapse all files

9
build.gradle
View File

@@ -63,7 +63,7 @@ ext {
springBootVersion = '3.0.0-SNAPSHOT' // docs module
springRetryVersion = '1.3.3'
springVersion = '6.0.0-SNAPSHOT'
caffeineVersion = '3.1.1'
idPrefix = 'pulsar'
}
@@ -163,6 +163,9 @@ subprojects { subproject ->
optionalApi "org.assertj:assertj-core:$assertjVersion"
testImplementation "org.testcontainers:pulsar:1.17.2"
testImplementation "org.springframework:spring-test"
}
// enable all compiler warnings; individual projects may customize further
@@ -302,6 +305,8 @@ project ('spring-pulsar') {
api "org.apache.pulsar:pulsar-client-admin:$pulsarVersion"
api "org.apache.pulsar:pulsar-client-admin-api:$pulsarVersion"
api "com.github.ben-manes.caffeine:caffeine:$caffeineVersion"
optionalApi 'com.fasterxml.jackson.core:jackson-core'
optionalApi 'com.fasterxml.jackson.core:jackson-databind'
optionalApi 'com.fasterxml.jackson.datatype:jackson-datatype-jdk8'
@@ -333,6 +338,8 @@ project ('spring-pulsar-boot-autoconfigure') {
api "org.springframework.boot:spring-boot-starter-validation:$springBootVersion"
api project (':spring-pulsar')
testImplementation "org.springframework.boot:spring-boot-starter-test:$springBootVersion"
}
}

2
spring-pulsar-boot-autoconfigure/src/main/java/org/springframework/pulsar/autoconfigure/PulsarAnnotationDrivenConfiguration.java
View File

@@ -70,7 +70,7 @@ public class PulsarAnnotationDrivenConfiguration {
@Configuration(proxyBeanMethods = false)
@EnablePulsar
@ConditionalOnMissingBean(name = PulsarListenerBeanNames.PULSAR_LISTENER_ANNOTATION_PROCESSOR_BEAN_NAME)
static class EnableKafkaConfiguration {
static class EnablePulsarConfiguration {
}

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

@@ -19,13 +19,16 @@ package org.springframework.pulsar.autoconfigure;
import org.apache.pulsar.client.api.PulsarClient;
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.context.properties.EnableConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Import;
import org.springframework.pulsar.config.PulsarClientConfiguration;
import org.springframework.pulsar.config.PulsarClientFactoryBean;
import org.springframework.pulsar.core.CachingPulsarProducerFactory;
import org.springframework.pulsar.core.DefaultPulsarConsumerFactory;
import org.springframework.pulsar.core.DefaultPulsarProducerFactory;
import org.springframework.pulsar.core.PulsarConsumerFactory;
@@ -33,9 +36,10 @@ import org.springframework.pulsar.core.PulsarProducerFactory;
import org.springframework.pulsar.core.PulsarTemplate;
/**
* {@link org.springframework.boot.autoconfigure.EnableAutoConfiguration Auto-configuration} for Apache Pulsar.
* {@link EnableAutoConfiguration Auto-configuration} for Apache Pulsar.
*
* @author Soby Chacko
* @author Chris Bono
*/
@AutoConfiguration
@ConditionalOnClass(PulsarTemplate.class)
@@ -49,24 +53,35 @@ public class PulsarAutoConfiguration {
this.properties = properties;
}
@Bean
@ConditionalOnMissingBean(PulsarClientFactoryBean.class)
public PulsarClientFactoryBean pulsarClientFactoryBean(PulsarClientConfiguration pulsarClientConfiguration) {
return new PulsarClientFactoryBean(pulsarClientConfiguration);
}
@Bean
@ConditionalOnMissingBean(PulsarClientConfiguration.class)
public PulsarClientConfiguration pulsarClientConfiguration() {
return new PulsarClientConfiguration(this.properties.buildClientProperties());
}
@Bean
@ConditionalOnMissingBean(PulsarClientFactoryBean.class)
public PulsarClientFactoryBean pulsarClientFactoryBean(PulsarClientConfiguration pulsarClientConfiguration) {
return new PulsarClientFactoryBean(pulsarClientConfiguration);
}
@Bean
@ConditionalOnMissingBean(PulsarProducerFactory.class)
@ConditionalOnProperty(name = "spring.pulsar.producer.cache.enabled", havingValue = "false")
public PulsarProducerFactory<?> pulsarProducerFactory(PulsarClient pulsarClient) {
return new DefaultPulsarProducerFactory<>(pulsarClient, this.properties.buildProducerProperties());
}
@Bean
@ConditionalOnMissingBean(PulsarProducerFactory.class)
@ConditionalOnProperty(name = "spring.pulsar.producer.cache.enabled", havingValue = "true", matchIfMissing = true)
public PulsarProducerFactory<?> cachingPulsarProducerFactory(PulsarClient pulsarClient) {
return new CachingPulsarProducerFactory<>(pulsarClient, this.properties.buildProducerProperties(),
this.properties.getProducer().getCache().getExpireAfterAccess(),
this.properties.getProducer().getCache().getMaximumSize(),
this.properties.getProducer().getCache().getInitialCapacity());
}
@Bean
@ConditionalOnMissingBean(PulsarTemplate.class)
public PulsarTemplate<?> pulsarTemplate(PulsarProducerFactory<?> pulsarProducerFactory) {
@@ -76,9 +91,6 @@ public class PulsarAutoConfiguration {
@Bean
@ConditionalOnMissingBean(PulsarConsumerFactory.class)
public PulsarConsumerFactory<?> pulsarConsumerFactory(PulsarClient pulsarClient) {
DefaultPulsarConsumerFactory<Object> factory = new DefaultPulsarConsumerFactory<>(pulsarClient,
this.properties.buildConsumerProperties());
return factory;
return new DefaultPulsarConsumerFactory<>(pulsarClient, this.properties.buildConsumerProperties());
}
}

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

@@ -16,6 +16,7 @@
package org.springframework.pulsar.autoconfigure;
import java.time.Duration;
import java.util.HashMap;
import java.util.Map;
import java.util.Set;
@@ -383,6 +384,8 @@ public class PulsarProperties {
private ProducerAccessMode producerAccessMode = ProducerAccessMode.Shared;
private Cache cache = new Cache();
public String getTopicName() {
return this.topicName;
}
@@ -511,6 +514,10 @@ public class PulsarProperties {
this.producerAccessMode = producerAccessMode;
}
public Cache getCache() {
return this.cache;
}
public Map<String, Object> buildProperties() {
PulsarProperties.Properties properties = new Properties();
@@ -535,6 +542,43 @@ public class PulsarProperties {
return properties;
}
}
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 Integer 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 Integer getInitialCapacity() {
return this.initialCapacity;
}
public void setInitialCapacity(Integer initialCapacity) {
this.initialCapacity = initialCapacity;
}
}
public static class Client {

201
spring-pulsar-boot-autoconfigure/src/test/java/org/springframework/pulsar/autoconfigure/PulsarAutoConfigurationTests.java Normal file
View File

@@ -0,0 +1,201 @@
/*
* Copyright 2022 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.pulsar.autoconfigure;
import static org.assertj.core.api.Assertions.assertThat;
import static org.mockito.Mockito.mock;
import java.util.concurrent.TimeUnit;
import org.junit.jupiter.api.Nested;
import org.junit.jupiter.api.Test;
import org.springframework.boot.autoconfigure.AutoConfigurations;
import org.springframework.boot.test.context.FilteredClassLoader;
import org.springframework.boot.test.context.assertj.AssertableApplicationContext;
import org.springframework.boot.test.context.runner.ApplicationContextRunner;
import org.springframework.pulsar.annotation.EnablePulsar;
import org.springframework.pulsar.annotation.PulsarBootstrapConfiguration;
import org.springframework.pulsar.annotation.PulsarListenerAnnotationBeanPostProcessor;
import org.springframework.pulsar.config.DefaultPulsarListenerContainerFactory;
import org.springframework.pulsar.config.PulsarClientConfiguration;
import org.springframework.pulsar.config.PulsarClientFactoryBean;
import org.springframework.pulsar.config.PulsarListenerContainerFactory;
import org.springframework.pulsar.config.PulsarListenerEndpointRegistry;
import org.springframework.pulsar.core.CachingPulsarProducerFactory;
import org.springframework.pulsar.core.DefaultPulsarProducerFactory;
import org.springframework.pulsar.core.PulsarConsumerFactory;
import org.springframework.pulsar.core.PulsarProducerFactory;
import org.springframework.pulsar.core.PulsarTemplate;
import org.springframework.pulsar.listener.DefaultPulsarMessageListenerContainer;
/**
* Autoconfiguration tests for {@link PulsarAutoConfiguration}.
*
* @author Chris Bono
*/
@SuppressWarnings("unchecked")
class PulsarAutoConfigurationTests {
private ApplicationContextRunner contextRunner = new ApplicationContextRunner()
.withConfiguration(AutoConfigurations.of(PulsarAutoConfiguration.class));
@Test
void autoConfigurationSkippedWhenPulsarTemplateNotOnClasspath() {
this.contextRunner.withClassLoader(new FilteredClassLoader(PulsarTemplate.class))
.run((context) -> assertThat(context).hasNotFailed()
.doesNotHaveBean(PulsarAutoConfiguration.class));
}
@Test
void annotationDrivenConfigurationSkippedWhenEnablePulsarAnnotationNotOnClasspath() {
this.contextRunner.withClassLoader(new FilteredClassLoader(EnablePulsar.class))
.run((context) -> assertThat(context).hasNotFailed()
.doesNotHaveBean(PulsarAnnotationDrivenConfiguration.class));
}
@Test
void bootstrapConfigurationSkippedWhenCustomPulsarListenerAnnotationProcessorDefined() {
this.contextRunner
.withBean("org.springframework.pulsar.config.internalPulsarListenerAnnotationProcessor",
String.class, () -> "someFauxBean")
.run((context) -> assertThat(context).hasNotFailed()
.doesNotHaveBean(PulsarBootstrapConfiguration.class));
}
@Test
void defaultBeansAreAutoConfigured() {
this.contextRunner.run((context) -> assertThat(context).hasNotFailed()
.hasSingleBean(PulsarClientConfiguration.class)
.hasSingleBean(PulsarClientFactoryBean.class)
.hasSingleBean(PulsarProducerFactory.class)
.hasSingleBean(PulsarTemplate.class)
.hasSingleBean(PulsarConsumerFactory.class)
.hasSingleBean(DefaultPulsarListenerContainerFactory.class)
.hasSingleBean(PulsarListenerAnnotationBeanPostProcessor.class)
.hasSingleBean(PulsarListenerEndpointRegistry.class));
}
@Test
void customPulsarClientConfigurationIsRespected() {
PulsarClientConfiguration clientConfig = new PulsarClientConfiguration(new PulsarProperties().buildClientProperties());
this.contextRunner
.withBean("customPulsarClientConfiguration", PulsarClientConfiguration.class, () -> clientConfig)
.run((context) -> assertThat(context).hasNotFailed()
.getBean(PulsarClientConfiguration.class).isSameAs(clientConfig));
}
@Test
void customPulsarClientFactoryBeanIsRespected() {
PulsarClientConfiguration clientConfig = new PulsarClientConfiguration(new PulsarProperties().buildClientProperties());
PulsarClientFactoryBean clientFactoryBean = new PulsarClientFactoryBean(clientConfig);
this.contextRunner
.withBean("customPulsarClientFactoryBean", PulsarClientFactoryBean.class, () -> clientFactoryBean)
.run((context) -> assertThat(context)
.getBean("&customPulsarClientFactoryBean", PulsarClientFactoryBean.class)
.isSameAs(clientFactoryBean));
}
@Test
void customPulsarProducerFactoryIsRespected() {
PulsarProducerFactory<String> producerFactory = mock(PulsarProducerFactory.class);
this.contextRunner
.withBean("customPulsarProducerFactory", PulsarProducerFactory.class, () -> producerFactory)
.run((context) -> assertThat(context).hasNotFailed()
.getBean(PulsarProducerFactory.class).isSameAs(producerFactory));
}
@Test
void customPulsarTemplateIsRespected() {
PulsarTemplate<String> template = mock(PulsarTemplate.class);
this.contextRunner
.withBean("customPulsarTemplate", PulsarTemplate.class, () -> template)
.run((context) -> assertThat(context).hasNotFailed()
.getBean(PulsarTemplate.class).isSameAs(template));
}
@Test
void customPulsarConsumerFactoryIsRespected() {
PulsarConsumerFactory<String> consumerFactory = mock(PulsarConsumerFactory.class);
this.contextRunner
.withBean("customPulsarConsumerFactory", PulsarConsumerFactory.class, () -> consumerFactory)
.run((context) -> assertThat(context).hasNotFailed()
.getBean(PulsarConsumerFactory.class).isSameAs(consumerFactory));
}
@Test
void customPulsarListenerContainerFactoryIsRespected() {
PulsarListenerContainerFactory<DefaultPulsarMessageListenerContainer<String>> listenerContainerFactory = mock(PulsarListenerContainerFactory.class);
this.contextRunner
.withBean("pulsarListenerContainerFactory", PulsarListenerContainerFactory.class, () -> listenerContainerFactory)
.run((context) -> assertThat(context).hasNotFailed()
.getBean(PulsarListenerContainerFactory.class).isSameAs(listenerContainerFactory));
}
@Test
void customPulsarListenerAnnotationBeanPostProcessorIsRespected() {
PulsarListenerAnnotationBeanPostProcessor<String, String> listenerAnnotationBeanPostProcessor = mock(PulsarListenerAnnotationBeanPostProcessor.class);
this.contextRunner
.withBean("org.springframework.pulsar.config.internalPulsarListenerAnnotationProcessor",
PulsarListenerAnnotationBeanPostProcessor.class, () -> listenerAnnotationBeanPostProcessor)
.run((context) -> assertThat(context).hasNotFailed()
.getBean(PulsarListenerAnnotationBeanPostProcessor.class).isSameAs(listenerAnnotationBeanPostProcessor));
}
@Nested
class ProducerFactoryAutoConfigurationTests {
@Test
void cachingProducerFactoryEnabledByDefault() {
contextRunner.run((context) -> assertHasProducerFactoryOfType(CachingPulsarProducerFactory.class, context));
}
@Test
void nonCachingProducerFactoryCanBeEnabled() {
contextRunner.withPropertyValues("spring.pulsar.producer.cache.enabled=false")
.run((context -> assertHasProducerFactoryOfType(DefaultPulsarProducerFactory.class, context)));
}
@Test
void cachingProducerFactoryCanBeEnabled() {
contextRunner.withPropertyValues("spring.pulsar.producer.cache.enabled=true")
.run((context -> assertHasProducerFactoryOfType(CachingPulsarProducerFactory.class, context)));
}
@Test
void cachingProducerFactoryCanBeConfigured() {
contextRunner.withPropertyValues(
"spring.pulsar.producer.cache.expire-after-access=100s",
"spring.pulsar.producer.cache.maximum-size=5150",
"spring.pulsar.producer.cache.initial-capacity=200")
.run((context -> assertThat(context)
.hasNotFailed()
.getBean(PulsarProducerFactory.class)
.extracting("producerCache")
.extracting("cache")
.hasFieldOrPropertyWithValue("maximum", 5150L)
.hasFieldOrPropertyWithValue("expiresAfterAccessNanos", TimeUnit.SECONDS.toNanos(100))));
}
private void assertHasProducerFactoryOfType(Class<?> producerFactoryType, AssertableApplicationContext context) {
assertThat(context).hasNotFailed()
.hasSingleBean(PulsarProducerFactory.class).getBean(PulsarProducerFactory.class)
.isExactlyInstanceOf(producerFactoryType);
}
}
}

4
spring-pulsar/src/main/java/org/springframework/pulsar/config/PulsarListenerBeanNames.java
View File

@@ -28,12 +28,12 @@ public abstract class PulsarListenerBeanNames {
* The bean name of the internally managed Pulsar listener annotation processor.
*/
public static final String PULSAR_LISTENER_ANNOTATION_PROCESSOR_BEAN_NAME =
"org.springframework.pulsar.config.internalKafkaListenerAnnotationProcessor";
"org.springframework.pulsar.config.internalPulsarListenerAnnotationProcessor";
/**
* The bean name of the internally managed Pulsar listener endpoint registry.
*/
public static final String PULSAR_LISTENER_ENDPOINT_REGISTRY_BEAN_NAME =
"org.springframework.pulsar.config.internalKafkaListenerEndpointRegistry";
"org.springframework.pulsar.config.internalPulsarListenerEndpointRegistry";
}

168
spring-pulsar/src/main/java/org/springframework/pulsar/core/CachingPulsarProducerFactory.java Normal file
View File

@@ -0,0 +1,168 @@
/*
* Copyright 2022 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.pulsar.core;
import java.time.Duration;
import java.util.Map;
import java.util.concurrent.CompletableFuture;
import java.util.function.Consumer;
import javax.annotation.Nonnull;
import javax.annotation.Nullable;
import org.aopalliance.intercept.MethodInterceptor;
import org.aopalliance.intercept.MethodInvocation;
import org.apache.commons.logging.LogFactory;
import org.apache.pulsar.client.api.MessageRouter;
import org.apache.pulsar.client.api.Producer;
import org.apache.pulsar.client.api.PulsarClient;
import org.apache.pulsar.client.api.PulsarClientException;
import org.apache.pulsar.client.api.Schema;
import org.springframework.aop.framework.AopProxyUtils;
import org.springframework.aop.framework.ProxyFactory;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.core.log.LogAccessor;
import org.springframework.util.Assert;
import com.github.benmanes.caffeine.cache.Cache;
import com.github.benmanes.caffeine.cache.Caffeine;
import com.github.benmanes.caffeine.cache.RemovalListener;
import com.github.benmanes.caffeine.cache.Scheduler;
/**
* A {@link PulsarProducerFactory} that extends the {@link DefaultPulsarProducerFactory default implementation}
* by caching the created producers.
* <p>
* The created producer is wrapped in a proxy so that calls to {@link Producer#close()} do not actually close it.
* The actual close occurs when the producer is evicted from the cache or when {@link DisposableBean#destroy()} is
* invoked.
* <p>
* The proxied producer is cached in an LRU fashion and evicted when it has not been used within a configured time
* period.
*
* @param <T> producer type.
*
* @author Chris Bono
*/
public class CachingPulsarProducerFactory<T> extends DefaultPulsarProducerFactory<T> implements DisposableBean {
private final LogAccessor logger = new LogAccessor(LogFactory.getLog(this.getClass()));
private final Cache<SchemaTopic<T>, Producer<T>> producerCache;
/**
* Construct a caching producer factory with the specified values for the cache configuration.
*
* @param pulsarClient the client used to create the producers
* @param producerConfig the configuration to use when creating a producer
* @param cacheExpireAfterAccess time period to expire unused entries in the cache
* @param cacheMaximumSize maximum size of cache (entries)
* @param cacheInitialCapacity the initial size of cache
*/
public CachingPulsarProducerFactory(PulsarClient pulsarClient, Map<String, Object> producerConfig,
Duration cacheExpireAfterAccess, Long cacheMaximumSize, Integer cacheInitialCapacity) {
super(pulsarClient, producerConfig);
this.producerCache = Caffeine.newBuilder()
.expireAfterAccess(cacheExpireAfterAccess)
.maximumSize(cacheMaximumSize)
.initialCapacity(cacheInitialCapacity)
.scheduler(Scheduler.systemScheduler())
.evictionListener((RemovalListener<SchemaTopic<T>, Producer<T>>) (schemaTopic, producer, cause) -> {
this.logger.debug(() -> String.format("Producer %s evicted from cache due to %s",
ProducerUtils.formatProducer(producer), cause));
closeProducer(producer);
}).build();
}
@Override
public Producer<T> createProducer(String topic, Schema<T> schema, MessageRouter messageRouter) {
final String topicName = ProducerUtils.resolveTopicName(topic, this);
SchemaTopic<T> schemaTopic = new SchemaTopic<>(schema, topicName, messageRouter);
return this.producerCache.get(schemaTopic, (st) -> {
try {
return this.doCreateProducer(st.topicName, st.schema, messageRouter);
}
catch (PulsarClientException ex) {
throw new RuntimeException(ex);
}
});
}
@Override
protected Producer<T> doCreateProducer(String topic, Schema<T> schema, MessageRouter messageRouter) throws PulsarClientException {
Producer<T> producer = super.doCreateProducer(topic, schema, messageRouter);
return wrapProducerWithCloseCallback(producer, (p) -> this.logger.trace(() ->
String.format("Client closed producer %s but will skip actual closing", ProducerUtils.formatProducer(producer))));
}
@SuppressWarnings("unchecked")
private Producer<T> wrapProducerWithCloseCallback(Producer<T> producer, Consumer<Producer<T>> closeCallback) {
ProxyFactory factory = new ProxyFactory(producer);
factory.addAdvice(new MethodInterceptor() {
@Nullable
@Override
public Object invoke(@Nonnull MethodInvocation invocation) throws Throwable {
if (invocation.getMethod().getName().equals("close")) {
closeCallback.accept((Producer<T>) invocation.getThis());
return null;
}
if (invocation.getMethod().getName().equals("closeAsync")) {
closeCallback.accept((Producer<T>) invocation.getThis());
return CompletableFuture.completedFuture(null);
}
return invocation.proceed();
}
});
return (Producer<T>) factory.getProxy();
}
@Override
public void destroy() {
this.producerCache.asMap().forEach((schemaTopic, producer) -> {
this.producerCache.invalidate(schemaTopic);
closeProducer(producer);
});
}
@SuppressWarnings("unchecked")
private void closeProducer(Producer<T> producer) {
Producer<T> actualProducer = (Producer<T>) AopProxyUtils.getSingletonTarget(producer);
if (actualProducer == null) {
this.logger.warn(() -> String.format("Unable to get actual producer for %s - will skip closing it",
ProducerUtils.formatProducer(producer)));
return;
}
ProducerUtils.closeProducerAsync(actualProducer, this.logger);
}
/**
* Holder for a schema, topic and optional message router used as unique identifier of a producer in cache key.
*
* @param schema schema of the message
* @param topicName topic to send the message to
* @param messageRouter router to use to send the topic
*
* @param <T> type of the schema
*/
record SchemaTopic<T>(Schema<T> schema, String topicName, MessageRouter messageRouter) {
public SchemaTopic {
Assert.notNull(schema, () -> "'schema' must be non-null");
Assert.notNull(topicName, () -> "'topicName' must be non-null");
}
}
}

20
spring-pulsar/src/main/java/org/springframework/pulsar/core/DefaultPulsarProducerFactory.java
View File

@@ -27,7 +27,6 @@ import org.apache.pulsar.client.api.PulsarClient;
import org.apache.pulsar.client.api.PulsarClientException;
import org.apache.pulsar.client.api.Schema;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.core.log.LogAccessor;
import org.springframework.util.CollectionUtils;
@@ -39,12 +38,10 @@ import org.springframework.util.CollectionUtils;
* @author Soby Chacko
* @author Chris Bono
*/
public class DefaultPulsarProducerFactory<T> implements PulsarProducerFactory<T>, DisposableBean {
public class DefaultPulsarProducerFactory<T> implements PulsarProducerFactory<T> {
private final LogAccessor logger = new LogAccessor(LogFactory.getLog(this.getClass()));
// TODO add caching of producers per schema/topic w/ ttl
private final Map<String, Object> producerConfig = new HashMap<>();
private final PulsarClient pulsarClient;
@@ -63,17 +60,20 @@ public class DefaultPulsarProducerFactory<T> implements PulsarProducerFactory<T>
@Override
public Producer<T> createProducer(String topic, Schema<T> schema, MessageRouter messageRouter) throws PulsarClientException {
this.logger.trace(() -> String.format("Creating producer for '%s' topic", topic));
return doCreateProducer(topic, schema, messageRouter);
}
protected Producer<T> doCreateProducer(String topic, Schema<T> schema, MessageRouter messageRouter) throws PulsarClientException {
final String resolvedTopic = ProducerUtils.resolveTopicName(topic, this);
this.logger.trace(() -> String.format("Creating producer for '%s' topic", resolvedTopic));
final ProducerBuilder<T> producerBuilder = this.pulsarClient.newProducer(schema);
if (!CollectionUtils.isEmpty(this.producerConfig)) {
producerBuilder.loadConf(this.producerConfig);
}
producerBuilder.topic(resolvedTopic);
if (messageRouter != null) {
producerBuilder.messageRouter(messageRouter);
}
if (topic != null) {
producerBuilder.topic(topic);
}
return producerBuilder.create();
}
@@ -81,8 +81,4 @@ public class DefaultPulsarProducerFactory<T> implements PulsarProducerFactory<T>
public Map<String, Object> getProducerConfig() {
return this.producerConfig;
}
@Override
public void destroy() {
}
}

55
spring-pulsar/src/main/java/org/springframework/pulsar/core/ProducerUtils.java Normal file
View File

@@ -0,0 +1,55 @@
/*
* Copyright 2022 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.pulsar.core;
import java.util.Optional;
import org.apache.pulsar.client.api.Producer;
import org.springframework.core.log.LogAccessor;
import org.springframework.util.StringUtils;
/**
* Common utilities used by producer components.
*
* @author Chris Bono
*/
final class ProducerUtils {
private ProducerUtils() {
}
static <T> String formatProducer(Producer<T> producer) {
return String.format("(%s:%s)", producer.getProducerName(), producer.getTopic());
}
static <T> String resolveTopicName(String userSpecifiedTopic, PulsarProducerFactory<T> producerFactory) {
if (StringUtils.hasText(userSpecifiedTopic)) {
return userSpecifiedTopic;
}
return Optional.ofNullable(producerFactory.getProducerConfig().get("topicName"))
.map(Object::toString)
.orElseThrow(() -> new IllegalArgumentException("Topic must be specified when no default topic is configured"));
}
static <T> void closeProducerAsync(Producer<T> producer, LogAccessor logger) {
producer.closeAsync().exceptionally(e -> {
logger.warn(e, () -> String.format("Failed to close producer %s", ProducerUtils.formatProducer(producer)));
return null;
});
}
}

2
spring-pulsar/src/main/java/org/springframework/pulsar/core/PulsarProducerFactory.java
View File

@@ -24,7 +24,7 @@ import org.apache.pulsar.client.api.PulsarClientException;
import org.apache.pulsar.client.api.Schema;
/**
* The strategy to produce a {@link Producer} instance(s).
* The strategy to create a {@link Producer} instance(s).
*
* @param <T> producer payload type
*

22
spring-pulsar/src/main/java/org/springframework/pulsar/core/PulsarTemplate.java
View File

@@ -16,7 +16,6 @@
package org.springframework.pulsar.core;
import java.util.Optional;
import java.util.concurrent.CompletableFuture;
import org.apache.commons.logging.LogFactory;
@@ -27,7 +26,6 @@ import org.apache.pulsar.client.api.PulsarClientException;
import org.apache.pulsar.client.api.Schema;
import org.springframework.core.log.LogAccessor;
import org.springframework.util.StringUtils;
/**
* A thread-safe template for executing high-level Pulsar operations.
@@ -63,7 +61,7 @@ public class PulsarTemplate<T> implements PulsarOperations<T> {
@Override
public CompletableFuture<MessageId> sendAsync(String topic, T message, MessageRouter messageRouter) throws PulsarClientException {
final String topicName = resolveTopicName(topic);
final String topicName = ProducerUtils.resolveTopicName(topic, this.producerFactory);
this.logger.trace(() -> String.format("Sending msg to '%s' topic", topicName));
final Producer<T> producer = prepareProducerForSend(topic, message, messageRouter);
return producer.sendAsync(message)
@@ -76,28 +74,12 @@ public class PulsarTemplate<T> implements PulsarOperations<T> {
this.logger.error(ex, () -> String.format("Failed to send msg to '%s' topic", topicName));
// TODO fail metrics
}
closeProducerAsync(producer);
ProducerUtils.closeProducerAsync(producer, this.logger);
});
}
private String resolveTopicName(String userSpecifiedTopic) {
if (StringUtils.hasText(userSpecifiedTopic)) {
return userSpecifiedTopic;
}
return Optional.ofNullable(this.producerFactory.getProducerConfig().get("topicName"))
.map(Object::toString)
.orElseThrow(() -> new IllegalArgumentException("Topic must be specified when no default topic is configured"));
}
private Producer<T> prepareProducerForSend(String topic, T message, MessageRouter messageRouter) throws PulsarClientException {
Schema<T> schema = SchemaUtils.getSchema(message);
return this.producerFactory.createProducer(topic, schema, messageRouter);
}
private void closeProducerAsync(Producer<T> producer) {
producer.closeAsync().exceptionally(e -> {
this.logger.warn(e, () -> String.format("Failed to close producer %s:%s", producer.getProducerName(), producer.getTopic()));
return null;
});
}
}

218
spring-pulsar/src/test/java/org/springframework/pulsar/core/CachingPulsarProducerFactoryTests.java Normal file
View File

@@ -0,0 +1,218 @@
/*
* Copyright 2022 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.pulsar.core;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
import static org.mockito.Mockito.spy;
import static org.mockito.Mockito.when;
import java.time.Duration;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import org.apache.pulsar.client.api.MessageRouter;
import org.apache.pulsar.client.api.Producer;
import org.apache.pulsar.client.api.PulsarClient;
import org.apache.pulsar.client.api.PulsarClientException;
import org.apache.pulsar.client.api.Schema;
import org.apache.pulsar.client.impl.schema.StringSchema;
import org.awaitility.Awaitility;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.mockito.Mockito;
import org.springframework.aop.framework.AopProxyUtils;
import org.springframework.pulsar.core.CachingPulsarProducerFactory.SchemaTopic;
import org.springframework.test.util.ReflectionTestUtils;
import org.springframework.util.ObjectUtils;
import com.github.benmanes.caffeine.cache.Cache;
/**
* Tests for {@link CachingPulsarProducerFactory}.
*
* @author Chris Bono
*/
class CachingPulsarProducerFactoryTests extends PulsarProducerFactoryTests {
private List<CachingPulsarProducerFactory<String>> producerFactories;
@BeforeEach
void prepareForTests() {
producerFactories = new ArrayList<>();
}
@AfterEach
void cleanupFromTests() {
producerFactories.forEach(CachingPulsarProducerFactory::destroy);
}
@Test
void createProducerMultipleCalls() throws PulsarClientException {
PulsarProducerFactory<String> producerFactory = producerFactory(pulsarClient, Collections.emptyMap());
SchemaTopic<String> cacheKey = new SchemaTopic<>(schema, "topic1", null);
Producer<String> producer1 = producerFactory.createProducer("topic1", schema);
Producer<String> producer2 = producerFactory.createProducer("topic1", schema);
Producer<String> producer3 = producerFactory.createProducer("topic1", schema);
assertThat(producer1).isSameAs(producer2).isSameAs(producer3);
Cache<SchemaTopic<String>, Producer<String>> producerCache = getAssertedProducerCache(producerFactory, Collections.singletonList(cacheKey));
Producer<String> cachedProducerProxy = producerCache.asMap().get(cacheKey);
assertThat(cachedProducerProxy).isSameAs(producer1);
}
@Test
void cachedProducerIsCloseSafeProxy() throws PulsarClientException {
PulsarProducerFactory<String> producerFactory = producerFactory(pulsarClient, Collections.emptyMap());
Producer<String> proxyProducer = producerFactory.createProducer("topic1", schema);
Producer<String> actualProducer = actualProducerFrom(proxyProducer);
assertThat(actualProducer.isConnected()).isTrue();
proxyProducer.close();
assertThat(actualProducer.isConnected()).isTrue();
actualProducer.close();
assertThat(actualProducer.isConnected()).isFalse();
}
@Test
void createProducerWithMatrixOfCacheKeys() throws PulsarClientException {
String topic1 = "topic1";
String topic2 = "topic2";
Schema<String> schema1 = new StringSchema();
Schema<String> schema2 = new StringSchema();
MessageRouter router1 = Mockito.mock(MessageRouter.class);
MessageRouter router2 = Mockito.mock(MessageRouter.class);
PulsarProducerFactory<String> producerFactory = producerFactory(pulsarClient, Collections.emptyMap());
producerFactory.createProducer(topic1, schema1);
producerFactory.createProducer(topic1, schema1, router1);
producerFactory.createProducer(topic1, schema1, router2);
producerFactory.createProducer(topic1, schema2);
producerFactory.createProducer(topic1, schema2, router1);
producerFactory.createProducer(topic1, schema2, router2);
producerFactory.createProducer(topic2, schema1);
producerFactory.createProducer(topic2, schema1, router1);
producerFactory.createProducer(topic2, schema1, router2);
List<SchemaTopic<String>> expectedCacheKeys = new ArrayList<>();
expectedCacheKeys.add(new SchemaTopic<>(schema1, topic1, null));
expectedCacheKeys.add(new SchemaTopic<>(schema1, topic1, router1));
expectedCacheKeys.add(new SchemaTopic<>(schema1, topic1, router2));
expectedCacheKeys.add(new SchemaTopic<>(schema1, topic2, null));
expectedCacheKeys.add(new SchemaTopic<>(schema1, topic2, router1));
expectedCacheKeys.add(new SchemaTopic<>(schema1, topic2, router2));
expectedCacheKeys.add(new SchemaTopic<>(schema2, topic1, null));
expectedCacheKeys.add(new SchemaTopic<>(schema2, topic1, router1));
expectedCacheKeys.add(new SchemaTopic<>(schema2, topic1, router2));
getAssertedProducerCache(producerFactory, expectedCacheKeys);
}
@Test
void factoryDestroyCleansUpCacheAndClosesProducers() throws PulsarClientException {
CachingPulsarProducerFactory<String> producerFactory = producerFactory(pulsarClient, Collections.emptyMap());
SchemaTopic<String> cacheKey1 = new SchemaTopic<>(schema, "topic1", null);
SchemaTopic<String> cacheKey2 = new SchemaTopic<>(schema, "topic2", null);
Producer<String> actualProducer1 = actualProducerFrom(producerFactory.createProducer("topic1", schema));
Producer<String> actualProducer2 = actualProducerFrom(producerFactory.createProducer("topic2", schema));
Cache<SchemaTopic<String>, Producer<String>> producerCache = getAssertedProducerCache(producerFactory,
Arrays.asList(cacheKey1, cacheKey2));
producerFactory.destroy();
Awaitility.await()
.timeout(Duration.ofSeconds(5L))
.untilAsserted(() -> {
assertThat(producerCache.asMap()).isEmpty();
assertThat(actualProducer1.isConnected()).isFalse();
assertThat(actualProducer2.isConnected()).isFalse();
});
}
@Test
void producerEvictedFromCache() throws PulsarClientException {
CachingPulsarProducerFactory<String> producerFactory = new CachingPulsarProducerFactory<>(pulsarClient,
Collections.emptyMap(), Duration.ofSeconds(3L), 10L, 2);
SchemaTopic<String> cacheKey = new SchemaTopic<>(schema, "topic1", null);
Producer<String> actualProducer = actualProducerFrom(producerFactory.createProducer("topic1", schema));
Cache<SchemaTopic<String>, Producer<String>> producerCache = getAssertedProducerCache(producerFactory,
Collections.singletonList(cacheKey));
Awaitility.await()
.pollDelay(Duration.ofSeconds(5L))
.timeout(Duration.ofSeconds(10L))
.untilAsserted(() -> {
assertThat(producerCache.asMap()).isEmpty();
assertThat(actualProducer.isConnected()).isFalse();
});
}
@Test
void createProducerEncountersException() {
pulsarClient = spy(pulsarClient);
when(this.pulsarClient.newProducer(schema)).thenThrow(new RuntimeException("5150"));
PulsarProducerFactory<String> producerFactory = producerFactory(pulsarClient, Collections.emptyMap());
assertThatThrownBy(() -> producerFactory.createProducer("topic1", schema))
.isInstanceOf(RuntimeException.class)
.hasMessage("5150");
getAssertedProducerCache(producerFactory, Collections.emptyList());
}
@Override
protected void assertProducerHasTopicSchemaAndRouter(Producer<String> producer, String topic, Schema<String> schema, MessageRouter router) {
super.assertProducerHasTopicSchemaAndRouter(actualProducerFrom(producer), topic, schema, router);
}
@SuppressWarnings("unchecked")
private Producer<String> actualProducerFrom(Producer<String> proxyProducer) {
Producer<String> actualProducer = (Producer<String>) AopProxyUtils.getSingletonTarget(proxyProducer);
assertThat(actualProducer).isNotNull();
return actualProducer;
}
@SuppressWarnings("unchecked")
private Cache<SchemaTopic<String>, Producer<String>> getAssertedProducerCache(PulsarProducerFactory<String> producerFactory,
List<SchemaTopic<String>> expectedCacheKeys) {
Cache<SchemaTopic<String>, Producer<String>> producerCache = (Cache<SchemaTopic<String>, Producer<String>>)
ReflectionTestUtils.getField(producerFactory, "producerCache");
assertThat(producerCache).isNotNull();
if (ObjectUtils.isEmpty(expectedCacheKeys)) {
assertThat(producerCache.asMap()).isEmpty();
}
else {
assertThat(producerCache.asMap()).containsOnlyKeys(expectedCacheKeys);
}
return producerCache;
}
@Override
protected CachingPulsarProducerFactory<String> producerFactory(PulsarClient pulsarClient, Map<String, Object> producerConfig) {
CachingPulsarProducerFactory<String> producerFactory = new CachingPulsarProducerFactory<>(pulsarClient,
producerConfig, Duration.ofMinutes(5L), 10L, 2);
producerFactories.add(producerFactory);
return producerFactory;
}
}

54
spring-pulsar/src/test/java/org/springframework/pulsar/core/DefaultPulsarProducerFactoryTests.java Normal file
View File

@@ -0,0 +1,54 @@
/*
* Copyright 2022 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.pulsar.core;
import static org.assertj.core.api.Assertions.assertThat;
import java.util.Collections;
import java.util.Map;
import org.apache.pulsar.client.api.Producer;
import org.apache.pulsar.client.api.PulsarClient;
import org.apache.pulsar.client.api.PulsarClientException;
import org.junit.jupiter.api.Test;
/**
* Tests for {@link DefaultPulsarProducerFactory}.
*
* @author Chris Bono
*/
class DefaultPulsarProducerFactoryTests extends PulsarProducerFactoryTests {
@Test
@SuppressWarnings("unchecked")
void createProducerMultipleCalls() throws PulsarClientException {
Map<String, Object> producerConfig = Collections.emptyMap();
PulsarProducerFactory<String> producerFactory = producerFactory(pulsarClient, producerConfig);
try (Producer<String> producer1 = producerFactory.createProducer("topic1", schema)) {
try (Producer<String> producer2 = producerFactory.createProducer("topic1", schema)) {
try (Producer<String> producer3 = producerFactory.createProducer("topic1", schema)) {
assertThat(producer1).isNotSameAs(producer2).isNotSameAs(producer3);
}
}
}
}
@Override
protected PulsarProducerFactory<String> producerFactory(PulsarClient pulsarClient, Map<String, Object> producerConfig) {
return new DefaultPulsarProducerFactory<>(pulsarClient, producerConfig);
}
}

114
spring-pulsar/src/test/java/org/springframework/pulsar/core/PulsarProducerFactoryTests.java Normal file
View File

@@ -0,0 +1,114 @@
/*
* Copyright 2022 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.pulsar.core;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
import static org.mockito.Mockito.mock;
import java.util.Collections;
import java.util.Map;
import org.apache.pulsar.client.api.MessageRouter;
import org.apache.pulsar.client.api.Producer;
import org.apache.pulsar.client.api.PulsarClient;
import org.apache.pulsar.client.api.PulsarClientException;
import org.apache.pulsar.client.api.Schema;
import org.apache.pulsar.client.impl.conf.ProducerConfigurationData;
import org.assertj.core.api.InstanceOfAssertFactories;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
/**
* Common tests for {@link DefaultPulsarProducerFactory} and {@link CachingPulsarProducerFactory}.
*
* @author Chris Bono
*/
abstract class PulsarProducerFactoryTests extends AbstractContainerBaseTests {
protected final Schema<String> schema = Schema.STRING;
protected PulsarClient pulsarClient;
@BeforeEach
void setup() throws PulsarClientException {
pulsarClient = PulsarClient.builder()
.serviceUrl(getPulsarBrokerUrl())
.build();
}
@Test
void createProducerWithSpecificTopic() throws PulsarClientException {
PulsarProducerFactory<String> producerFactory = producerFactory(pulsarClient, Collections.emptyMap());
try (Producer<String> producer = producerFactory.createProducer("topic1", schema)) {
assertProducerHasTopicSchemaAndRouter(producer, "topic1", schema, null);
}
}
@Test
void createProducerWithSpecificTopicAndMessageRouter() throws PulsarClientException {
PulsarProducerFactory<String> producerFactory = producerFactory(pulsarClient, Collections.emptyMap());
MessageRouter router = mock(MessageRouter.class);
try (Producer<String> producer = producerFactory.createProducer("topic1", schema, router)) {
assertProducerHasTopicSchemaAndRouter(producer, "topic1", schema, router);
}
}
@Test
void createProducerWithDefaultTopic() throws PulsarClientException {
PulsarProducerFactory<String> producerFactory = producerFactory(pulsarClient, Collections.singletonMap("topicName", "topic0"));
try (Producer<String> producer = producerFactory.createProducer(null, schema)) {
assertProducerHasTopicSchemaAndRouter(producer, "topic0", schema, null);
}
}
@Test
void createProducerWithDefaultTopicAndMessageRouter() throws PulsarClientException {
PulsarProducerFactory<String> producerFactory = producerFactory(pulsarClient, Collections.singletonMap("topicName", "topic0"));
MessageRouter router = mock(MessageRouter.class);
try (Producer<String> producer = producerFactory.createProducer(null, schema, router)) {
assertProducerHasTopicSchemaAndRouter(producer, "topic0", schema, router);
}
}
@Test
void createProducerWithNoTopic() {
PulsarProducerFactory<String> producerFactory = producerFactory(pulsarClient, Collections.emptyMap());
assertThatThrownBy(() -> producerFactory.createProducer(null, schema))
.isInstanceOf(IllegalArgumentException.class)
.hasMessage("Topic must be specified when no default topic is configured");
}
protected void assertProducerHasTopicSchemaAndRouter(Producer<String> producer, String topic, Schema<String> schema, MessageRouter router) {
assertThat(producer.getTopic()).isEqualTo(topic);
assertThat(producer).hasFieldOrPropertyWithValue("schema", schema);
assertThat(producer)
.extracting("conf").asInstanceOf(InstanceOfAssertFactories.type(ProducerConfigurationData.class))
.extracting(ProducerConfigurationData::getCustomMessageRouter)
.isSameAs(router);
}
/**
* Subclasses override to provide concrete {@link PulsarProducerFactory} instance.
*
* @param pulsarClient the Pulsar client
* @param producerConfig the Pulsar producers config
* @return a Pulsar producer factory instance to use for the tests
*/
protected abstract PulsarProducerFactory<String> producerFactory(PulsarClient pulsarClient, Map<String, Object> producerConfig);
}

7
src/checkstyle/checkstyle.xml
View File

@@ -148,13 +148,6 @@
value="Line has leading space characters; indentation should be performed with tabs only."/>
<property name="ignoreComments" value="true"/>
</module>
<module name="RegexpSinglelineJava">
<property name="maximum" value="0"/>
<property name="format" value="org\.mockito\..*Mockito\.(when|doThrow|doAnswer)"/>
<property name="message"
value="Please use BDDMockito instead of Mockito.(when|doThrow|doAnswer)."/>
<property name="ignoreComments" value="true"/>
</module>
<module name="RegexpSinglelineJava">
<property name="maximum" value="0"/>
<property name="format" value="org\.junit\.Assert\.assert"/>