diff --git a/build.gradle b/build.gradle index 643f6a2f..1cebea66 100644 --- a/build.gradle +++ b/build.gradle @@ -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" } } diff --git a/spring-pulsar-boot-autoconfigure/src/main/java/org/springframework/pulsar/autoconfigure/PulsarAnnotationDrivenConfiguration.java b/spring-pulsar-boot-autoconfigure/src/main/java/org/springframework/pulsar/autoconfigure/PulsarAnnotationDrivenConfiguration.java index 959727cd..bcb995f8 100644 --- a/spring-pulsar-boot-autoconfigure/src/main/java/org/springframework/pulsar/autoconfigure/PulsarAnnotationDrivenConfiguration.java +++ b/spring-pulsar-boot-autoconfigure/src/main/java/org/springframework/pulsar/autoconfigure/PulsarAnnotationDrivenConfiguration.java @@ -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 { } diff --git a/spring-pulsar-boot-autoconfigure/src/main/java/org/springframework/pulsar/autoconfigure/PulsarAutoConfiguration.java b/spring-pulsar-boot-autoconfigure/src/main/java/org/springframework/pulsar/autoconfigure/PulsarAutoConfiguration.java index 83a41807..7a552be1 100644 --- a/spring-pulsar-boot-autoconfigure/src/main/java/org/springframework/pulsar/autoconfigure/PulsarAutoConfiguration.java +++ b/spring-pulsar-boot-autoconfigure/src/main/java/org/springframework/pulsar/autoconfigure/PulsarAutoConfiguration.java @@ -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 factory = new DefaultPulsarConsumerFactory<>(pulsarClient, - this.properties.buildConsumerProperties()); - return factory; + return new DefaultPulsarConsumerFactory<>(pulsarClient, this.properties.buildConsumerProperties()); } - } diff --git a/spring-pulsar-boot-autoconfigure/src/main/java/org/springframework/pulsar/autoconfigure/PulsarProperties.java b/spring-pulsar-boot-autoconfigure/src/main/java/org/springframework/pulsar/autoconfigure/PulsarProperties.java index fac31d83..d42229bf 100644 --- a/spring-pulsar-boot-autoconfigure/src/main/java/org/springframework/pulsar/autoconfigure/PulsarProperties.java +++ b/spring-pulsar-boot-autoconfigure/src/main/java/org/springframework/pulsar/autoconfigure/PulsarProperties.java @@ -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 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 { diff --git a/spring-pulsar-boot-autoconfigure/src/test/java/org/springframework/pulsar/autoconfigure/PulsarAutoConfigurationTests.java b/spring-pulsar-boot-autoconfigure/src/test/java/org/springframework/pulsar/autoconfigure/PulsarAutoConfigurationTests.java new file mode 100644 index 00000000..fc8fed8e --- /dev/null +++ b/spring-pulsar-boot-autoconfigure/src/test/java/org/springframework/pulsar/autoconfigure/PulsarAutoConfigurationTests.java @@ -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 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 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 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> 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 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); + } + } + +} diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/config/PulsarListenerBeanNames.java b/spring-pulsar/src/main/java/org/springframework/pulsar/config/PulsarListenerBeanNames.java index 534c2823..479ab906 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/config/PulsarListenerBeanNames.java +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/config/PulsarListenerBeanNames.java @@ -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"; } diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/core/CachingPulsarProducerFactory.java b/spring-pulsar/src/main/java/org/springframework/pulsar/core/CachingPulsarProducerFactory.java new file mode 100644 index 00000000..788b0038 --- /dev/null +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/core/CachingPulsarProducerFactory.java @@ -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. + *

+ * 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. + *

+ * The proxied producer is cached in an LRU fashion and evicted when it has not been used within a configured time + * period. + * + * @param producer type. + * + * @author Chris Bono + */ +public class CachingPulsarProducerFactory extends DefaultPulsarProducerFactory implements DisposableBean { + + private final LogAccessor logger = new LogAccessor(LogFactory.getLog(this.getClass())); + + private final Cache, Producer> 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 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, Producer>) (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 createProducer(String topic, Schema schema, MessageRouter messageRouter) { + final String topicName = ProducerUtils.resolveTopicName(topic, this); + SchemaTopic 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 doCreateProducer(String topic, Schema schema, MessageRouter messageRouter) throws PulsarClientException { + Producer 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 wrapProducerWithCloseCallback(Producer producer, Consumer> 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) invocation.getThis()); + return null; + } + if (invocation.getMethod().getName().equals("closeAsync")) { + closeCallback.accept((Producer) invocation.getThis()); + return CompletableFuture.completedFuture(null); + } + return invocation.proceed(); + } + }); + return (Producer) factory.getProxy(); + } + + @Override + public void destroy() { + this.producerCache.asMap().forEach((schemaTopic, producer) -> { + this.producerCache.invalidate(schemaTopic); + closeProducer(producer); + }); + } + + @SuppressWarnings("unchecked") + private void closeProducer(Producer producer) { + Producer actualProducer = (Producer) 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 type of the schema + */ + record SchemaTopic(Schema schema, String topicName, MessageRouter messageRouter) { + public SchemaTopic { + Assert.notNull(schema, () -> "'schema' must be non-null"); + Assert.notNull(topicName, () -> "'topicName' must be non-null"); + } + } +} diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/core/DefaultPulsarProducerFactory.java b/spring-pulsar/src/main/java/org/springframework/pulsar/core/DefaultPulsarProducerFactory.java index dbfc047d..4090aaf3 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/core/DefaultPulsarProducerFactory.java +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/core/DefaultPulsarProducerFactory.java @@ -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 implements PulsarProducerFactory, DisposableBean { +public class DefaultPulsarProducerFactory implements PulsarProducerFactory { private final LogAccessor logger = new LogAccessor(LogFactory.getLog(this.getClass())); - // TODO add caching of producers per schema/topic w/ ttl - private final Map producerConfig = new HashMap<>(); private final PulsarClient pulsarClient; @@ -63,17 +60,20 @@ public class DefaultPulsarProducerFactory implements PulsarProducerFactory @Override public Producer createProducer(String topic, Schema schema, MessageRouter messageRouter) throws PulsarClientException { - this.logger.trace(() -> String.format("Creating producer for '%s' topic", topic)); + return doCreateProducer(topic, schema, messageRouter); + } + + protected Producer doCreateProducer(String topic, Schema 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 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 implements PulsarProducerFactory public Map getProducerConfig() { return this.producerConfig; } - - @Override - public void destroy() { - } } diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/core/ProducerUtils.java b/spring-pulsar/src/main/java/org/springframework/pulsar/core/ProducerUtils.java new file mode 100644 index 00000000..f32fb516 --- /dev/null +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/core/ProducerUtils.java @@ -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 String formatProducer(Producer producer) { + return String.format("(%s:%s)", producer.getProducerName(), producer.getTopic()); + } + + static String resolveTopicName(String userSpecifiedTopic, PulsarProducerFactory 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 void closeProducerAsync(Producer producer, LogAccessor logger) { + producer.closeAsync().exceptionally(e -> { + logger.warn(e, () -> String.format("Failed to close producer %s", ProducerUtils.formatProducer(producer))); + return null; + }); + } +} diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/core/PulsarProducerFactory.java b/spring-pulsar/src/main/java/org/springframework/pulsar/core/PulsarProducerFactory.java index a4a4704f..584c0e35 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/core/PulsarProducerFactory.java +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/core/PulsarProducerFactory.java @@ -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 producer payload type * diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/core/PulsarTemplate.java b/spring-pulsar/src/main/java/org/springframework/pulsar/core/PulsarTemplate.java index 84fe14c7..231220f7 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/core/PulsarTemplate.java +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/core/PulsarTemplate.java @@ -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 implements PulsarOperations { @Override public CompletableFuture 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 producer = prepareProducerForSend(topic, message, messageRouter); return producer.sendAsync(message) @@ -76,28 +74,12 @@ public class PulsarTemplate implements PulsarOperations { 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 prepareProducerForSend(String topic, T message, MessageRouter messageRouter) throws PulsarClientException { Schema schema = SchemaUtils.getSchema(message); return this.producerFactory.createProducer(topic, schema, messageRouter); } - - private void closeProducerAsync(Producer producer) { - producer.closeAsync().exceptionally(e -> { - this.logger.warn(e, () -> String.format("Failed to close producer %s:%s", producer.getProducerName(), producer.getTopic())); - return null; - }); - } } diff --git a/spring-pulsar/src/test/java/org/springframework/pulsar/core/CachingPulsarProducerFactoryTests.java b/spring-pulsar/src/test/java/org/springframework/pulsar/core/CachingPulsarProducerFactoryTests.java new file mode 100644 index 00000000..c2832cd7 --- /dev/null +++ b/spring-pulsar/src/test/java/org/springframework/pulsar/core/CachingPulsarProducerFactoryTests.java @@ -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> producerFactories; + + @BeforeEach + void prepareForTests() { + producerFactories = new ArrayList<>(); + } + + @AfterEach + void cleanupFromTests() { + producerFactories.forEach(CachingPulsarProducerFactory::destroy); + } + + @Test + void createProducerMultipleCalls() throws PulsarClientException { + PulsarProducerFactory producerFactory = producerFactory(pulsarClient, Collections.emptyMap()); + SchemaTopic cacheKey = new SchemaTopic<>(schema, "topic1", null); + + Producer producer1 = producerFactory.createProducer("topic1", schema); + Producer producer2 = producerFactory.createProducer("topic1", schema); + Producer producer3 = producerFactory.createProducer("topic1", schema); + assertThat(producer1).isSameAs(producer2).isSameAs(producer3); + + Cache, Producer> producerCache = getAssertedProducerCache(producerFactory, Collections.singletonList(cacheKey)); + Producer cachedProducerProxy = producerCache.asMap().get(cacheKey); + assertThat(cachedProducerProxy).isSameAs(producer1); + } + + @Test + void cachedProducerIsCloseSafeProxy() throws PulsarClientException { + PulsarProducerFactory producerFactory = producerFactory(pulsarClient, Collections.emptyMap()); + + Producer proxyProducer = producerFactory.createProducer("topic1", schema); + Producer 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 schema1 = new StringSchema(); + Schema schema2 = new StringSchema(); + MessageRouter router1 = Mockito.mock(MessageRouter.class); + MessageRouter router2 = Mockito.mock(MessageRouter.class); + + PulsarProducerFactory 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> 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 producerFactory = producerFactory(pulsarClient, Collections.emptyMap()); + SchemaTopic cacheKey1 = new SchemaTopic<>(schema, "topic1", null); + SchemaTopic cacheKey2 = new SchemaTopic<>(schema, "topic2", null); + + Producer actualProducer1 = actualProducerFrom(producerFactory.createProducer("topic1", schema)); + Producer actualProducer2 = actualProducerFrom(producerFactory.createProducer("topic2", schema)); + + Cache, Producer> 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 producerFactory = new CachingPulsarProducerFactory<>(pulsarClient, + Collections.emptyMap(), Duration.ofSeconds(3L), 10L, 2); + SchemaTopic cacheKey = new SchemaTopic<>(schema, "topic1", null); + + Producer actualProducer = actualProducerFrom(producerFactory.createProducer("topic1", schema)); + + Cache, Producer> 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 producerFactory = producerFactory(pulsarClient, Collections.emptyMap()); + assertThatThrownBy(() -> producerFactory.createProducer("topic1", schema)) + .isInstanceOf(RuntimeException.class) + .hasMessage("5150"); + getAssertedProducerCache(producerFactory, Collections.emptyList()); + } + + @Override + protected void assertProducerHasTopicSchemaAndRouter(Producer producer, String topic, Schema schema, MessageRouter router) { + super.assertProducerHasTopicSchemaAndRouter(actualProducerFrom(producer), topic, schema, router); + } + + @SuppressWarnings("unchecked") + private Producer actualProducerFrom(Producer proxyProducer) { + Producer actualProducer = (Producer) AopProxyUtils.getSingletonTarget(proxyProducer); + assertThat(actualProducer).isNotNull(); + return actualProducer; + } + + @SuppressWarnings("unchecked") + private Cache, Producer> getAssertedProducerCache(PulsarProducerFactory producerFactory, + List> expectedCacheKeys) { + Cache, Producer> producerCache = (Cache, Producer>) + 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 producerFactory(PulsarClient pulsarClient, Map producerConfig) { + CachingPulsarProducerFactory producerFactory = new CachingPulsarProducerFactory<>(pulsarClient, + producerConfig, Duration.ofMinutes(5L), 10L, 2); + producerFactories.add(producerFactory); + return producerFactory; + } +} diff --git a/spring-pulsar/src/test/java/org/springframework/pulsar/core/DefaultPulsarProducerFactoryTests.java b/spring-pulsar/src/test/java/org/springframework/pulsar/core/DefaultPulsarProducerFactoryTests.java new file mode 100644 index 00000000..c5db9040 --- /dev/null +++ b/spring-pulsar/src/test/java/org/springframework/pulsar/core/DefaultPulsarProducerFactoryTests.java @@ -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 producerConfig = Collections.emptyMap(); + PulsarProducerFactory producerFactory = producerFactory(pulsarClient, producerConfig); + try (Producer producer1 = producerFactory.createProducer("topic1", schema)) { + try (Producer producer2 = producerFactory.createProducer("topic1", schema)) { + try (Producer producer3 = producerFactory.createProducer("topic1", schema)) { + assertThat(producer1).isNotSameAs(producer2).isNotSameAs(producer3); + } + } + } + } + + @Override + protected PulsarProducerFactory producerFactory(PulsarClient pulsarClient, Map producerConfig) { + return new DefaultPulsarProducerFactory<>(pulsarClient, producerConfig); + } +} diff --git a/spring-pulsar/src/test/java/org/springframework/pulsar/core/PulsarProducerFactoryTests.java b/spring-pulsar/src/test/java/org/springframework/pulsar/core/PulsarProducerFactoryTests.java new file mode 100644 index 00000000..847bd752 --- /dev/null +++ b/spring-pulsar/src/test/java/org/springframework/pulsar/core/PulsarProducerFactoryTests.java @@ -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 schema = Schema.STRING; + + protected PulsarClient pulsarClient; + + @BeforeEach + void setup() throws PulsarClientException { + pulsarClient = PulsarClient.builder() + .serviceUrl(getPulsarBrokerUrl()) + .build(); + } + + @Test + void createProducerWithSpecificTopic() throws PulsarClientException { + PulsarProducerFactory producerFactory = producerFactory(pulsarClient, Collections.emptyMap()); + try (Producer producer = producerFactory.createProducer("topic1", schema)) { + assertProducerHasTopicSchemaAndRouter(producer, "topic1", schema, null); + } + } + + @Test + void createProducerWithSpecificTopicAndMessageRouter() throws PulsarClientException { + PulsarProducerFactory producerFactory = producerFactory(pulsarClient, Collections.emptyMap()); + MessageRouter router = mock(MessageRouter.class); + try (Producer producer = producerFactory.createProducer("topic1", schema, router)) { + assertProducerHasTopicSchemaAndRouter(producer, "topic1", schema, router); + } + } + + @Test + void createProducerWithDefaultTopic() throws PulsarClientException { + PulsarProducerFactory producerFactory = producerFactory(pulsarClient, Collections.singletonMap("topicName", "topic0")); + try (Producer producer = producerFactory.createProducer(null, schema)) { + assertProducerHasTopicSchemaAndRouter(producer, "topic0", schema, null); + } + } + + @Test + void createProducerWithDefaultTopicAndMessageRouter() throws PulsarClientException { + PulsarProducerFactory producerFactory = producerFactory(pulsarClient, Collections.singletonMap("topicName", "topic0")); + MessageRouter router = mock(MessageRouter.class); + try (Producer producer = producerFactory.createProducer(null, schema, router)) { + assertProducerHasTopicSchemaAndRouter(producer, "topic0", schema, router); + } + } + + @Test + void createProducerWithNoTopic() { + PulsarProducerFactory 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 producer, String topic, Schema 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 producerFactory(PulsarClient pulsarClient, Map producerConfig); + +} diff --git a/src/checkstyle/checkstyle.xml b/src/checkstyle/checkstyle.xml index 313da7c2..742b02de 100644 --- a/src/checkstyle/checkstyle.xml +++ b/src/checkstyle/checkstyle.xml @@ -148,13 +148,6 @@ value="Line has leading space characters; indentation should be performed with tabs only."/> - - - - - -