From 9e00c2be7cea33a58f8894bc791b6ae85e272811 Mon Sep 17 00:00:00 2001 From: Chris Bono Date: Mon, 10 Apr 2023 20:35:04 -0500 Subject: [PATCH] Hide observation API from public API on template and listener (#389) * Remove observation registry and convention from constructors on PulsarTemplate and CPLCF * Lookup observation registry and convention from app context at start time --- .../PulsarAnnotationDrivenConfiguration.java | 5 +- .../PulsarAutoConfiguration.java | 10 +- .../PulsarAutoConfigurationTests.java | 66 ++++------ ...bstractPulsarListenerContainerFactory.java | 14 +-- ...currentPulsarListenerContainerFactory.java | 42 +++---- .../pulsar/core/PulsarTemplate.java | 65 +++++++--- ...bstractPulsarMessageListenerContainer.java | 12 +- ...currentPulsarMessageListenerContainer.java | 23 +++- ...DefaultPulsarMessageListenerContainer.java | 23 +--- .../listener/PulsarContainerProperties.java | 24 +++- ...TemplateObservationConfigurationTests.java | 115 ++++++++++++++++++ .../pulsar/core/PulsarTemplateTests.java | 4 +- ...ntPulsarMessageListenerContainerTests.java | 112 +++++++++++++++-- .../pulsar/listener/PulsarListenerTests.java | 6 +- .../ObservationIntegrationTests.java | 12 +- .../pulsar/observation/ObservationTests.java | 32 +++-- 16 files changed, 397 insertions(+), 168 deletions(-) create mode 100644 spring-pulsar/src/test/java/org/springframework/pulsar/core/PulsarTemplateObservationConfigurationTests.java diff --git a/spring-pulsar-spring-boot-autoconfigure/src/main/java/org/springframework/pulsar/autoconfigure/PulsarAnnotationDrivenConfiguration.java b/spring-pulsar-spring-boot-autoconfigure/src/main/java/org/springframework/pulsar/autoconfigure/PulsarAnnotationDrivenConfiguration.java index 41dcf927..a278f7f2 100644 --- a/spring-pulsar-spring-boot-autoconfigure/src/main/java/org/springframework/pulsar/autoconfigure/PulsarAnnotationDrivenConfiguration.java +++ b/spring-pulsar-spring-boot-autoconfigure/src/main/java/org/springframework/pulsar/autoconfigure/PulsarAnnotationDrivenConfiguration.java @@ -67,7 +67,7 @@ public class PulsarAnnotationDrivenConfiguration { containerProperties.setSchemaResolver(schemaResolver); containerProperties.setTopicResolver(topicResolver); containerProperties.setSubscriptionType(this.pulsarProperties.getConsumer().getSubscriptionType()); - containerProperties.setObservationConvention(observationConventionProvider.getIfUnique()); + containerProperties.setObservationEnabled(this.pulsarProperties.getListener().isObservationsEnabled()); PropertyMapper map = PropertyMapper.get().alwaysApplyingWhenNonNull(); PulsarProperties.Listener listenerProperties = this.pulsarProperties.getListener(); @@ -79,8 +79,7 @@ public class PulsarAnnotationDrivenConfiguration { map.from(listenerProperties::getMaxNumMessages).to(containerProperties::setMaxNumMessages); return new ConcurrentPulsarListenerContainerFactory<>(consumerFactoryProvider.getIfAvailable(), - containerProperties, this.pulsarProperties.getListener().isObservationsEnabled() - ? observationRegistryProvider.getIfUnique() : null); + containerProperties); } @Bean diff --git a/spring-pulsar-spring-boot-autoconfigure/src/main/java/org/springframework/pulsar/autoconfigure/PulsarAutoConfiguration.java b/spring-pulsar-spring-boot-autoconfigure/src/main/java/org/springframework/pulsar/autoconfigure/PulsarAutoConfiguration.java index 2bf071ba..1a5af4f9 100644 --- a/spring-pulsar-spring-boot-autoconfigure/src/main/java/org/springframework/pulsar/autoconfigure/PulsarAutoConfiguration.java +++ b/spring-pulsar-spring-boot-autoconfigure/src/main/java/org/springframework/pulsar/autoconfigure/PulsarAutoConfiguration.java @@ -49,9 +49,6 @@ import org.springframework.pulsar.function.PulsarFunction; import org.springframework.pulsar.function.PulsarFunctionAdministration; import org.springframework.pulsar.function.PulsarSink; import org.springframework.pulsar.function.PulsarSource; -import org.springframework.pulsar.observation.PulsarTemplateObservationConvention; - -import io.micrometer.observation.ObservationRegistry; /** * {@link EnableAutoConfiguration Auto-configuration} for Apache Pulsar. @@ -101,12 +98,9 @@ public class PulsarAutoConfiguration { @ConditionalOnMissingBean public PulsarTemplate pulsarTemplate(PulsarProducerFactory pulsarProducerFactory, ObjectProvider interceptorsProvider, SchemaResolver schemaResolver, - TopicResolver topicResolver, ObjectProvider observationRegistryProvider, - ObjectProvider observationConventionProvider) { + TopicResolver topicResolver) { return new PulsarTemplate<>(pulsarProducerFactory, interceptorsProvider.orderedStream().toList(), - schemaResolver, topicResolver, this.properties.getTemplate().isObservationsEnabled() - ? observationRegistryProvider.getIfUnique() : null, - observationConventionProvider.getIfUnique()); + schemaResolver, topicResolver, this.properties.getTemplate().isObservationsEnabled()); } @Bean diff --git a/spring-pulsar-spring-boot-autoconfigure/src/test/java/org/springframework/pulsar/autoconfigure/PulsarAutoConfigurationTests.java b/spring-pulsar-spring-boot-autoconfigure/src/test/java/org/springframework/pulsar/autoconfigure/PulsarAutoConfigurationTests.java index ba5bdaa0..1951e298 100644 --- a/spring-pulsar-spring-boot-autoconfigure/src/test/java/org/springframework/pulsar/autoconfigure/PulsarAutoConfigurationTests.java +++ b/spring-pulsar-spring-boot-autoconfigure/src/test/java/org/springframework/pulsar/autoconfigure/PulsarAutoConfigurationTests.java @@ -65,11 +65,8 @@ import org.springframework.pulsar.core.TopicResolver; import org.springframework.pulsar.function.PulsarFunctionAdministration; import org.springframework.pulsar.listener.AckMode; import org.springframework.pulsar.listener.PulsarContainerProperties; -import org.springframework.pulsar.observation.PulsarListenerObservationConvention; -import org.springframework.pulsar.observation.PulsarTemplateObservationConvention; import com.github.benmanes.caffeine.cache.Caffeine; -import io.micrometer.observation.ObservationRegistry; /** * Autoconfiguration tests for {@link PulsarAutoConfiguration}. @@ -420,61 +417,42 @@ class PulsarAutoConfigurationTests { @Test void templateObservationsEnabledByDefault() { - ObservationRegistry observationRegistry = mock(ObservationRegistry.class); - contextRunner.withBean("observationRegistry", ObservationRegistry.class, () -> observationRegistry) - .run((context -> assertThat(context).hasNotFailed().getBean(PulsarTemplate.class) - .extracting("observationRegistry").isSameAs(observationRegistry))); + contextRunner.run((context -> assertThat(context).getBean(PulsarTemplate.class) + .hasFieldOrPropertyWithValue("observationEnabled", true))); + } + + @Test + void templateObservationsEnabledExplicitly() { + contextRunner.withPropertyValues("spring.pulsar.template.observations-enabled=true") + .run((context -> assertThat(context).getBean(PulsarTemplate.class) + .hasFieldOrPropertyWithValue("observationEnabled", true))); } @Test void templateObservationsCanBeDisabled() { - ObservationRegistry observationRegistry = mock(ObservationRegistry.class); contextRunner.withPropertyValues("spring.pulsar.template.observations-enabled=false") - .withBean("observationRegistry", ObservationRegistry.class, () -> observationRegistry) - .run((context -> assertThat(context).hasNotFailed().getBean(PulsarTemplate.class) - .extracting("observationRegistry").isNull())); - } - - @Test - void templateObservationsWithCustomConvention() { - ObservationRegistry observationRegistry = mock(ObservationRegistry.class); - PulsarTemplateObservationConvention customConvention = mock(PulsarTemplateObservationConvention.class); - contextRunner.withBean("observationRegistry", ObservationRegistry.class, () -> observationRegistry) - .withBean("customConvention", PulsarTemplateObservationConvention.class, () -> customConvention) - .run((context -> assertThat(context).hasNotFailed().getBean(PulsarTemplate.class) - .extracting("observationConvention").isSameAs(customConvention))); + .run((context -> assertThat(context).getBean(PulsarTemplate.class) + .hasFieldOrPropertyWithValue("observationEnabled", false))); } @Test void listenerObservationsEnabledByDefault() { - ObservationRegistry observationRegistry = mock(ObservationRegistry.class); - contextRunner.withBean("observationRegistry", ObservationRegistry.class, () -> observationRegistry) - .run((context -> assertThat(context).hasNotFailed() - .getBean(ConcurrentPulsarListenerContainerFactory.class).extracting("observationRegistry") - .isSameAs(observationRegistry))); + contextRunner.run((context -> assertThat(context).getBean(ConcurrentPulsarListenerContainerFactory.class) + .hasFieldOrPropertyWithValue("containerProperties.observationEnabled", true))); + } + + @Test + void listenerObservationsEnabledExplicitly() { + contextRunner.withPropertyValues("spring.pulsar.listener.observations-enabled=true") + .run((context -> assertThat(context).getBean(ConcurrentPulsarListenerContainerFactory.class) + .hasFieldOrPropertyWithValue("containerProperties.observationEnabled", true))); } @Test void listenerObservationsCanBeDisabled() { - ObservationRegistry observationRegistry = mock(ObservationRegistry.class); contextRunner.withPropertyValues("spring.pulsar.listener.observations-enabled=false") - .withBean("observationRegistry", ObservationRegistry.class, () -> observationRegistry) - .run((context -> assertThat(context).hasNotFailed() - .getBean(ConcurrentPulsarListenerContainerFactory.class).extracting("observationRegistry") - .isNull())); - } - - @Test - void listenerObservationsWithCustomConvention() { - ObservationRegistry observationRegistry = mock(ObservationRegistry.class); - PulsarListenerObservationConvention customConvention = mock(PulsarListenerObservationConvention.class); - contextRunner.withBean("observationRegistry", ObservationRegistry.class, () -> observationRegistry) - .withBean("customConvention", PulsarListenerObservationConvention.class, () -> customConvention) - .run((context -> assertThat(context).hasNotFailed() - .getBean(ConcurrentPulsarListenerContainerFactory.class) - .extracting(ConcurrentPulsarListenerContainerFactory::getContainerProperties) - .extracting(PulsarContainerProperties::getObservationConvention) - .isSameAs(customConvention))); + .run((context -> assertThat(context).getBean(ConcurrentPulsarListenerContainerFactory.class) + .hasFieldOrPropertyWithValue("containerProperties.observationEnabled", false))); } } diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/config/AbstractPulsarListenerContainerFactory.java b/spring-pulsar/src/main/java/org/springframework/pulsar/config/AbstractPulsarListenerContainerFactory.java index 846cec3a..922ed742 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/config/AbstractPulsarListenerContainerFactory.java +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/config/AbstractPulsarListenerContainerFactory.java @@ -24,7 +24,6 @@ import org.springframework.context.ApplicationContextAware; import org.springframework.context.ApplicationEventPublisher; import org.springframework.context.ApplicationEventPublisherAware; import org.springframework.core.log.LogAccessor; -import org.springframework.lang.Nullable; import org.springframework.pulsar.core.PulsarConsumerFactory; import org.springframework.pulsar.listener.AbstractPulsarMessageListenerContainer; import org.springframework.pulsar.listener.AckMode; @@ -32,8 +31,6 @@ import org.springframework.pulsar.listener.PulsarContainerProperties; import org.springframework.pulsar.support.JavaUtils; import org.springframework.pulsar.support.MessageConverter; -import io.micrometer.observation.ObservationRegistry; - /** * Base {@link PulsarListenerContainerFactory} implementation. * @@ -51,8 +48,6 @@ public abstract class AbstractPulsarListenerContainerFactory consumerFactory, - PulsarContainerProperties containerProperties, @Nullable ObservationRegistry observationRegistry) { + PulsarContainerProperties containerProperties) { this.consumerFactory = consumerFactory; this.containerProperties = containerProperties; - this.observationRegistry = observationRegistry; } protected PulsarConsumerFactory getConsumerFactory() { return this.consumerFactory; } - protected ObservationRegistry getObservationRegistry() { - return this.observationRegistry; - } - public PulsarContainerProperties getContainerProperties() { return this.containerProperties; } @@ -171,7 +161,7 @@ public abstract class AbstractPulsarListenerContainerFactory private Integer concurrency; public ConcurrentPulsarListenerContainerFactory(PulsarConsumerFactory consumerFactory, - PulsarContainerProperties containerProperties, @Nullable ObservationRegistry observationRegistry) { - super(consumerFactory, containerProperties, observationRegistry); + PulsarContainerProperties containerProperties) { + super(consumerFactory, containerProperties); } /** @@ -55,6 +52,22 @@ public class ConcurrentPulsarListenerContainerFactory this.concurrency = concurrency; } + @Override + public ConcurrentPulsarMessageListenerContainer createContainer(String... topics) { + PulsarListenerEndpoint endpoint = new PulsarListenerEndpointAdapter() { + + @Override + public Collection getTopics() { + return Arrays.asList(topics); + } + + }; + ConcurrentPulsarMessageListenerContainer container = createContainerInstance(endpoint); + initializeContainer(container, endpoint); + // customizeContainer(container); + return container; + } + @Override protected ConcurrentPulsarMessageListenerContainer createContainerInstance(PulsarListenerEndpoint endpoint) { @@ -84,8 +97,7 @@ public class ConcurrentPulsarListenerContainerFactory properties.setSchemaType(endpoint.getSchemaType()); - return new ConcurrentPulsarMessageListenerContainer<>(this.getConsumerFactory(), properties, - this.getObservationRegistry()); + return new ConcurrentPulsarMessageListenerContainer<>(this.getConsumerFactory(), properties); } @Override @@ -100,20 +112,4 @@ public class ConcurrentPulsarListenerContainerFactory } } - @Override - public ConcurrentPulsarMessageListenerContainer createContainer(String... topics) { - PulsarListenerEndpoint endpoint = new PulsarListenerEndpointAdapter() { - - @Override - public Collection getTopics() { - return Arrays.asList(topics); - } - - }; - ConcurrentPulsarMessageListenerContainer container = createContainerInstance(endpoint); - initializeContainer(container, endpoint); - // customizeContainer(container); - return container; - } - } 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 82289d93..e005a7e7 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 @@ -31,6 +31,9 @@ import org.apache.pulsar.client.api.TypedMessageBuilder; import org.apache.pulsar.client.api.interceptor.ProducerInterceptor; import org.springframework.beans.factory.BeanNameAware; +import org.springframework.beans.factory.SmartInitializingSingleton; +import org.springframework.context.ApplicationContext; +import org.springframework.context.ApplicationContextAware; import org.springframework.core.log.LogAccessor; import org.springframework.lang.Nullable; import org.springframework.pulsar.observation.DefaultPulsarTemplateObservationConvention; @@ -51,7 +54,8 @@ import io.micrometer.observation.ObservationRegistry; * @author Alexander Preuß * @author Christophe Bornet */ -public class PulsarTemplate implements PulsarOperations, BeanNameAware { +public class PulsarTemplate + implements PulsarOperations, ApplicationContextAware, BeanNameAware, SmartInitializingSingleton { private final LogAccessor logger = new LogAccessor(this.getClass()); @@ -63,11 +67,25 @@ public class PulsarTemplate implements PulsarOperations, BeanNameAware { private final TopicResolver topicResolver; + /** + * Whether to record observations. + */ + private boolean observationEnabled; + + /** + * The registry to record observations with. + */ @Nullable - private final ObservationRegistry observationRegistry; + private ObservationRegistry observationRegistry; + + /** + * The optional custom observation convention to use when recording observations. + */ + @Nullable + private PulsarTemplateObservationConvention observationConvention; @Nullable - private final PulsarTemplateObservationConvention observationConvention; + private ApplicationContext applicationContext; private String beanName = ""; @@ -82,12 +100,12 @@ public class PulsarTemplate implements PulsarOperations, BeanNameAware { /** * Construct a template instance with interceptors that uses the default schema - * resolver and default topic resolver. + * resolver and default topic resolver and enables observation recording. * @param producerFactory the factory used to create the backing Pulsar producers. * @param interceptors the interceptors to add to the producer. */ public PulsarTemplate(PulsarProducerFactory producerFactory, List interceptors) { - this(producerFactory, interceptors, new DefaultSchemaResolver(), new DefaultTopicResolver(), null, null); + this(producerFactory, interceptors, new DefaultSchemaResolver(), new DefaultTopicResolver(), true); } /** @@ -96,21 +114,40 @@ public class PulsarTemplate implements PulsarOperations, BeanNameAware { * @param interceptors the list of interceptors to add to the producer * @param schemaResolver the schema resolver to use * @param topicResolver the topic resolver to use - * @param observationRegistry the registry to record observations with or {@code null} - * to not record observations - * @param observationConvention the optional custom observation convention to use when - * recording observations + * @param observationEnabled whether to record observations */ public PulsarTemplate(PulsarProducerFactory producerFactory, List interceptors, - SchemaResolver schemaResolver, TopicResolver topicResolver, - @Nullable ObservationRegistry observationRegistry, - @Nullable PulsarTemplateObservationConvention observationConvention) { + SchemaResolver schemaResolver, TopicResolver topicResolver, boolean observationEnabled) { this.producerFactory = producerFactory; this.interceptors = interceptors; this.schemaResolver = schemaResolver; this.topicResolver = topicResolver; - this.observationRegistry = observationRegistry; - this.observationConvention = observationConvention; + this.observationEnabled = observationEnabled; + } + + @Override + public void setApplicationContext(ApplicationContext applicationContext) { + this.applicationContext = applicationContext; + } + + /** + * If observations are enabled, attempt to obtain the Observation registry and + * convention. + */ + @Override + public void afterSingletonsInstantiated() { + if (!this.observationEnabled) { + this.logger.debug(() -> "Observations are not enabled - not recording"); + return; + } + if (this.applicationContext == null) { + this.logger.warn(() -> "Observations enabled but application context null - not recording"); + return; + } + this.observationRegistry = this.applicationContext.getBeanProvider(ObservationRegistry.class) + .getIfUnique(() -> this.observationRegistry); + this.observationConvention = this.applicationContext.getBeanProvider(PulsarTemplateObservationConvention.class) + .getIfUnique(() -> this.observationConvention); } @Override diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/listener/AbstractPulsarMessageListenerContainer.java b/spring-pulsar/src/main/java/org/springframework/pulsar/listener/AbstractPulsarMessageListenerContainer.java index 7f84ca6a..13cf2da6 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/listener/AbstractPulsarMessageListenerContainer.java +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/listener/AbstractPulsarMessageListenerContainer.java @@ -19,14 +19,11 @@ package org.springframework.pulsar.listener; import org.apache.pulsar.client.api.DeadLetterPolicy; import org.apache.pulsar.client.api.RedeliveryBackoff; -import org.springframework.lang.Nullable; import org.springframework.pulsar.core.AbstractPulsarMessageContainer; import org.springframework.pulsar.core.ConsumerBuilderCustomizer; import org.springframework.pulsar.core.PulsarConsumerFactory; import org.springframework.util.Assert; -import io.micrometer.observation.ObservationRegistry; - /** * Base implementation for the {@link PulsarMessageListenerContainer}. * @@ -41,8 +38,6 @@ public non-sealed abstract class AbstractPulsarMessageListenerContainer exten private final PulsarContainerProperties pulsarContainerProperties; - private final ObservationRegistry observationRegistry; - protected final Object lifecycleMonitor = new Object(); private volatile boolean paused; @@ -59,10 +54,9 @@ public non-sealed abstract class AbstractPulsarMessageListenerContainer exten @SuppressWarnings("unchecked") protected AbstractPulsarMessageListenerContainer(PulsarConsumerFactory pulsarConsumerFactory, - PulsarContainerProperties pulsarContainerProperties, @Nullable ObservationRegistry observationRegistry) { + PulsarContainerProperties pulsarContainerProperties) { this.pulsarConsumerFactory = (PulsarConsumerFactory) pulsarConsumerFactory; this.pulsarContainerProperties = pulsarContainerProperties; - this.observationRegistry = observationRegistry; } public PulsarConsumerFactory getPulsarConsumerFactory() { @@ -73,10 +67,6 @@ public non-sealed abstract class AbstractPulsarMessageListenerContainer exten return this.pulsarContainerProperties; } - public ObservationRegistry getObservationRegistry() { - return this.observationRegistry; - } - @Override public boolean isRunning() { return this.running; diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/listener/ConcurrentPulsarMessageListenerContainer.java b/spring-pulsar/src/main/java/org/springframework/pulsar/listener/ConcurrentPulsarMessageListenerContainer.java index c66ca38c..0465b5ba 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/listener/ConcurrentPulsarMessageListenerContainer.java +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/listener/ConcurrentPulsarMessageListenerContainer.java @@ -25,8 +25,8 @@ import org.springframework.context.ApplicationContext; import org.springframework.context.ApplicationEventPublisher; import org.springframework.core.task.AsyncTaskExecutor; import org.springframework.core.task.SimpleAsyncTaskExecutor; -import org.springframework.lang.Nullable; import org.springframework.pulsar.core.PulsarConsumerFactory; +import org.springframework.pulsar.observation.PulsarListenerObservationConvention; import org.springframework.util.Assert; import io.micrometer.observation.ObservationRegistry; @@ -50,8 +50,8 @@ public class ConcurrentPulsarMessageListenerContainer extends AbstractPulsarM private final List executors = new ArrayList<>(); public ConcurrentPulsarMessageListenerContainer(PulsarConsumerFactory pulsarConsumerFactory, - PulsarContainerProperties pulsarContainerProperties, @Nullable ObservationRegistry observationRegistry) { - super(pulsarConsumerFactory, pulsarContainerProperties, observationRegistry); + PulsarContainerProperties pulsarContainerProperties) { + super(pulsarConsumerFactory, pulsarContainerProperties); } public int getConcurrency() { @@ -80,6 +80,20 @@ public class ConcurrentPulsarMessageListenerContainer extends AbstractPulsarM setRunning(true); + // Set observation registry accordingly + if (containerProperties.isObservationEnabled()) { + ApplicationContext applicationContext = getApplicationContext(); + if (applicationContext == null) { + this.logger.warn(() -> "Observations enabled but application context null - not recording"); + } + else { + applicationContext.getBeanProvider(ObservationRegistry.class) + .ifUnique(containerProperties::setObservationRegistry); + applicationContext.getBeanProvider(PulsarListenerObservationConvention.class) + .ifUnique(containerProperties::setObservationConvention); + } + } + for (int i = 0; i < this.concurrency; i++) { DefaultPulsarMessageListenerContainer container = constructContainer(containerProperties); configureChildContainer(i, container); @@ -90,8 +104,7 @@ public class ConcurrentPulsarMessageListenerContainer extends AbstractPulsarM } private DefaultPulsarMessageListenerContainer constructContainer(PulsarContainerProperties containerProperties) { - return new DefaultPulsarMessageListenerContainer<>(this.getPulsarConsumerFactory(), containerProperties, - this.getObservationRegistry()); + return new DefaultPulsarMessageListenerContainer<>(this.getPulsarConsumerFactory(), containerProperties); } private void configureChildContainer(int index, DefaultPulsarMessageListenerContainer container) { diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/listener/DefaultPulsarMessageListenerContainer.java b/spring-pulsar/src/main/java/org/springframework/pulsar/listener/DefaultPulsarMessageListenerContainer.java index 9a6ad280..8c6b1e93 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/listener/DefaultPulsarMessageListenerContainer.java +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/listener/DefaultPulsarMessageListenerContainer.java @@ -51,7 +51,6 @@ import org.springframework.context.ApplicationEventPublisher; import org.springframework.core.log.LogAccessor; import org.springframework.core.task.AsyncTaskExecutor; import org.springframework.core.task.SimpleAsyncTaskExecutor; -import org.springframework.lang.Nullable; import org.springframework.pulsar.core.ConsumerBuilderConfigurationUtil; import org.springframework.pulsar.core.ConsumerBuilderCustomizer; import org.springframework.pulsar.core.PulsarConsumerFactory; @@ -66,7 +65,6 @@ import org.springframework.util.Assert; import org.springframework.util.StringUtils; import io.micrometer.observation.Observation; -import io.micrometer.observation.ObservationRegistry; /** * Default implementation for {@link PulsarMessageListenerContainer}. @@ -97,12 +95,7 @@ public class DefaultPulsarMessageListenerContainer extends AbstractPulsarMess public DefaultPulsarMessageListenerContainer(PulsarConsumerFactory pulsarConsumerFactory, PulsarContainerProperties pulsarContainerProperties) { - this(pulsarConsumerFactory, pulsarContainerProperties, null); - } - - public DefaultPulsarMessageListenerContainer(PulsarConsumerFactory pulsarConsumerFactory, - PulsarContainerProperties pulsarContainerProperties, @Nullable ObservationRegistry observationRegistry) { - super(pulsarConsumerFactory, pulsarContainerProperties, observationRegistry); + super(pulsarConsumerFactory, pulsarContainerProperties); this.thisOrParentContainer = this; } @@ -122,8 +115,7 @@ public class DefaultPulsarMessageListenerContainer extends AbstractPulsarMess containerProperties.setConsumerTaskExecutor(consumerExecutor); } - this.listenerConsumer = new Listener(messageListener, this.getContainerProperties(), - this.getObservationRegistry()); + this.listenerConsumer = new Listener(messageListener, this.getContainerProperties()); setRunning(true); this.startLatch = new CountDownLatch(1); this.listenerConsumerFuture = consumerExecutor.submitCompletable(this.listenerConsumer); @@ -229,8 +221,6 @@ public class DefaultPulsarMessageListenerContainer extends AbstractPulsarMess private final PulsarContainerProperties containerProperties; - private final ObservationRegistry observationRegistry; - private Consumer consumer; private final Set nackableMessages = new HashSet<>(); @@ -246,8 +236,7 @@ public class DefaultPulsarMessageListenerContainer extends AbstractPulsarMess private final SubscriptionType subscriptionType; @SuppressWarnings({ "unchecked", "rawtypes" }) - Listener(MessageListener messageListener, PulsarContainerProperties containerProperties, - @Nullable ObservationRegistry observationRegistry) { + Listener(MessageListener messageListener, PulsarContainerProperties containerProperties) { this.containerProperties = containerProperties; this.isBatchListener = this.containerProperties.isBatchListener(); @@ -266,7 +255,6 @@ public class DefaultPulsarMessageListenerContainer extends AbstractPulsarMess this.listener = null; this.batchMessageListener = null; } - this.observationRegistry = observationRegistry; this.pulsarConsumerErrorHandler = getPulsarConsumerErrorHandler(); this.consumerBuilderCustomizer = getConsumerBuilderCustomizer(); try { @@ -480,13 +468,14 @@ public class DefaultPulsarMessageListenerContainer extends AbstractPulsarMess } private Observation newObservation(Message message) { - if (this.observationRegistry == null) { + if (this.containerProperties.getObservationRegistry() == null) { return Observation.NOOP; } return PulsarListenerObservation.LISTENER_OBSERVATION.observation( this.containerProperties.getObservationConvention(), DefaultPulsarListenerObservationConvention.INSTANCE, - () -> new PulsarMessageReceiverContext(message, getBeanName()), this.observationRegistry); + () -> new PulsarMessageReceiverContext(message, getBeanName()), + this.containerProperties.getObservationRegistry()); } private void dispatchMessageToListener(Message message, AtomicBoolean inRetryMode) { diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/listener/PulsarContainerProperties.java b/spring-pulsar/src/main/java/org/springframework/pulsar/listener/PulsarContainerProperties.java index d60a9a4d..cc810ec1 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/listener/PulsarContainerProperties.java +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/listener/PulsarContainerProperties.java @@ -33,6 +33,8 @@ import org.springframework.pulsar.core.TopicResolver; import org.springframework.pulsar.observation.PulsarListenerObservationConvention; import org.springframework.util.Assert; +import io.micrometer.observation.ObservationRegistry; + /** * Contains runtime properties for a listener container. * @@ -80,6 +82,10 @@ public class PulsarContainerProperties { private AckMode ackMode = AckMode.BATCH; + private boolean observationEnabled; + + private ObservationRegistry observationRegistry; + private PulsarListenerObservationConvention observationConvention; private Properties pulsarConsumerProperties = new Properties(); @@ -162,6 +168,22 @@ public class PulsarContainerProperties { this.ackMode = ackMode; } + public boolean isObservationEnabled() { + return this.observationEnabled; + } + + public void setObservationEnabled(boolean observationEnabled) { + this.observationEnabled = observationEnabled; + } + + public ObservationRegistry getObservationRegistry() { + return this.observationRegistry; + } + + void setObservationRegistry(ObservationRegistry observationRegistry) { + this.observationRegistry = observationRegistry; + } + public PulsarListenerObservationConvention getObservationConvention() { return this.observationConvention; } @@ -170,7 +192,7 @@ public class PulsarContainerProperties { * Set a custom observation convention. * @param observationConvention the convention. */ - public void setObservationConvention(PulsarListenerObservationConvention observationConvention) { + void setObservationConvention(PulsarListenerObservationConvention observationConvention) { this.observationConvention = observationConvention; } diff --git a/spring-pulsar/src/test/java/org/springframework/pulsar/core/PulsarTemplateObservationConfigurationTests.java b/spring-pulsar/src/test/java/org/springframework/pulsar/core/PulsarTemplateObservationConfigurationTests.java new file mode 100644 index 00000000..84f05034 --- /dev/null +++ b/spring-pulsar/src/test/java/org/springframework/pulsar/core/PulsarTemplateObservationConfigurationTests.java @@ -0,0 +1,115 @@ +/* + * Copyright 2022-2023 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.pulsar.core; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.mockito.Mockito.mock; + +import java.util.Collections; + +import org.junit.jupiter.api.Test; + +import org.springframework.context.support.GenericApplicationContext; +import org.springframework.pulsar.observation.PulsarTemplateObservationConvention; + +import io.micrometer.observation.ObservationRegistry; + +/** + * Tests for the observation configuration aspect of {@link PulsarTemplate}. + * + * @author Chris Bono + */ +@SuppressWarnings({ "rawtypes", "unchecked" }) +class PulsarTemplateObservationConfigurationTests { + + @Test + void uniqueRegistryAndConventionBeansAvailable() { + var template = new PulsarTemplate(mock(PulsarProducerFactory.class), Collections.emptyList(), + new DefaultSchemaResolver(), new DefaultTopicResolver(), true); + var appContext = new GenericApplicationContext(); + var observationRegistry = mock(ObservationRegistry.class); + var observationConvention = mock(PulsarTemplateObservationConvention.class); + appContext.registerBean("obsReg", ObservationRegistry.class, () -> observationRegistry); + appContext.registerBean("obsConv", PulsarTemplateObservationConvention.class, () -> observationConvention); + appContext.refresh(); + template.setApplicationContext(appContext); + template.afterSingletonsInstantiated(); + assertThat(template).hasFieldOrPropertyWithValue("observationEnabled", true); + assertThat(template).hasFieldOrPropertyWithValue("observationRegistry", observationRegistry); + assertThat(template).hasFieldOrPropertyWithValue("observationConvention", observationConvention); + } + + @Test + void enabledPropertySetToFalse() { + var template = new PulsarTemplate(mock(PulsarProducerFactory.class), Collections.emptyList(), + new DefaultSchemaResolver(), new DefaultTopicResolver(), false); + var appContext = new GenericApplicationContext(); + var observationRegistry = mock(ObservationRegistry.class); + var observationConvention = mock(PulsarTemplateObservationConvention.class); + appContext.registerBean("obsReg", ObservationRegistry.class, () -> observationRegistry); + appContext.registerBean("obsConv", PulsarTemplateObservationConvention.class, () -> observationConvention); + appContext.refresh(); + template.setApplicationContext(appContext); + template.afterSingletonsInstantiated(); + assertThat(template).hasFieldOrPropertyWithValue("observationEnabled", false); + assertThat(template).hasFieldOrPropertyWithValue("observationRegistry", null); + assertThat(template).hasFieldOrPropertyWithValue("observationConvention", null); + } + + @Test + void noAppContextAvailable() { + var template = new PulsarTemplate(mock(PulsarProducerFactory.class), Collections.emptyList(), + new DefaultSchemaResolver(), new DefaultTopicResolver(), true); + template.afterSingletonsInstantiated(); + assertThat(template).hasFieldOrPropertyWithValue("observationEnabled", true); + assertThat(template).hasFieldOrPropertyWithValue("observationRegistry", null); + assertThat(template).hasFieldOrPropertyWithValue("observationConvention", null); + } + + @Test + void noRegistryOrConventionBeansAvailable() { + var template = new PulsarTemplate(mock(PulsarProducerFactory.class), Collections.emptyList(), + new DefaultSchemaResolver(), new DefaultTopicResolver(), true); + var appContext = new GenericApplicationContext(); + appContext.refresh(); + template.setApplicationContext(appContext); + template.afterSingletonsInstantiated(); + assertThat(template).hasFieldOrPropertyWithValue("observationEnabled", true); + assertThat(template).hasFieldOrPropertyWithValue("observationRegistry", null); + assertThat(template).hasFieldOrPropertyWithValue("observationConvention", null); + } + + @Test + void noUniqueRegistryOrConventionBeansAvailable() { + var template = new PulsarTemplate(mock(PulsarProducerFactory.class), Collections.emptyList(), + new DefaultSchemaResolver(), new DefaultTopicResolver(), true); + var appContext = new GenericApplicationContext(); + var observationRegistry = mock(ObservationRegistry.class); + var observationConvention = mock(PulsarTemplateObservationConvention.class); + appContext.registerBean("obsReg1", ObservationRegistry.class, () -> observationRegistry); + appContext.registerBean("obsReg2", ObservationRegistry.class, () -> observationRegistry); + appContext.registerBean("obsConv1", PulsarTemplateObservationConvention.class, () -> observationConvention); + appContext.registerBean("obsConv2", PulsarTemplateObservationConvention.class, () -> observationConvention); + appContext.refresh(); + template.setApplicationContext(appContext); + template.afterSingletonsInstantiated(); + assertThat(template).hasFieldOrPropertyWithValue("observationEnabled", true); + assertThat(template).hasFieldOrPropertyWithValue("observationRegistry", null); + assertThat(template).hasFieldOrPropertyWithValue("observationConvention", null); + } + +} diff --git a/spring-pulsar/src/test/java/org/springframework/pulsar/core/PulsarTemplateTests.java b/spring-pulsar/src/test/java/org/springframework/pulsar/core/PulsarTemplateTests.java index e30fdfcf..bb4fb27d 100644 --- a/spring-pulsar/src/test/java/org/springframework/pulsar/core/PulsarTemplateTests.java +++ b/spring-pulsar/src/test/java/org/springframework/pulsar/core/PulsarTemplateTests.java @@ -221,7 +221,7 @@ class PulsarTemplateTests implements PulsarTestContainerSupport { DefaultTopicResolver topicResolver = new DefaultTopicResolver(); topicResolver.addCustomTopicMapping(Foo.class, topic); PulsarTemplate pulsarTemplate = new PulsarTemplate<>(producerFactory, Collections.emptyList(), - new DefaultSchemaResolver(), topicResolver, null, null); + new DefaultSchemaResolver(), topicResolver, false); Foo foo = new Foo("Foo-" + UUID.randomUUID(), "Bar-" + UUID.randomUUID()); ThrowingConsumer> sendFunction = (template) -> template.send(foo, Schema.JSON(Foo.class)); sendAndConsume(pulsarTemplate, sendFunction, topic, Schema.JSON(Foo.class), foo); @@ -288,7 +288,7 @@ class PulsarTemplateTests implements PulsarTestContainerSupport { DefaultSchemaResolver schemaResolver = new DefaultSchemaResolver(); schemaResolver.addCustomSchemaMapping(Foo.class, Schema.JSON(Foo.class)); PulsarTemplate pulsarTemplate = new PulsarTemplate<>(producerFactory, Collections.emptyList(), - schemaResolver, new DefaultTopicResolver(), null, null); + schemaResolver, new DefaultTopicResolver(), false); Foo foo = new Foo("Foo-" + UUID.randomUUID(), "Bar-" + UUID.randomUUID()); ThrowingConsumer> sendFunction = (template) -> template.newMessage(foo).send(); sendAndConsume(pulsarTemplate, sendFunction, topic, Schema.JSON(Foo.class), foo); diff --git a/spring-pulsar/src/test/java/org/springframework/pulsar/listener/ConcurrentPulsarMessageListenerContainerTests.java b/spring-pulsar/src/test/java/org/springframework/pulsar/listener/ConcurrentPulsarMessageListenerContainerTests.java index 2cecce09..d9525fb2 100644 --- a/spring-pulsar/src/test/java/org/springframework/pulsar/listener/ConcurrentPulsarMessageListenerContainerTests.java +++ b/spring-pulsar/src/test/java/org/springframework/pulsar/listener/ConcurrentPulsarMessageListenerContainerTests.java @@ -36,14 +36,18 @@ import org.apache.pulsar.client.api.RedeliveryBackoff; import org.apache.pulsar.client.api.Schema; import org.apache.pulsar.client.api.SubscriptionType; import org.apache.pulsar.client.impl.MultiplierRedeliveryBackoff; +import org.junit.jupiter.api.Nested; import org.junit.jupiter.api.Test; +import org.springframework.context.support.GenericApplicationContext; import org.springframework.pulsar.config.ConcurrentPulsarListenerContainerFactory; import org.springframework.pulsar.config.PulsarListenerEndpoint; import org.springframework.pulsar.core.PulsarConsumerFactory; import org.springframework.pulsar.observation.PulsarListenerObservationConvention; import org.springframework.util.backoff.BackOff; +import io.micrometer.observation.ObservationRegistry; + /** * @author Soby Chacko * @author Alexander Preuß @@ -60,7 +64,7 @@ public class ConcurrentPulsarMessageListenerContainerTests { containerProperties.setMaxNumMessages(120); containerProperties.setMaxNumBytes(32000); ConcurrentPulsarListenerContainerFactory containerFactory = new ConcurrentPulsarListenerContainerFactory<>( - consumerFactory, containerProperties, null); + consumerFactory, containerProperties); containerFactory.setConcurrency(1); PulsarListenerEndpoint pulsarListenerEndpoint = mock(PulsarListenerEndpoint.class); when(pulsarListenerEndpoint.getConcurrency()).thenReturn(1); @@ -135,14 +139,17 @@ public class ConcurrentPulsarMessageListenerContainerTests { void observationConfigAppliedOnChildContainer() throws Exception { PulsarListenerMockComponents env = setupListenerMockComponents(SubscriptionType.Shared); ConcurrentPulsarMessageListenerContainer concurrentContainer = env.concurrentContainer(); - PulsarListenerObservationConvention customObservationConvention = mock( - PulsarListenerObservationConvention.class); - concurrentContainer.getContainerProperties().setObservationConvention(customObservationConvention); + ObservationRegistry observationRegistry = mock(ObservationRegistry.class); + PulsarListenerObservationConvention observationConvention = mock(PulsarListenerObservationConvention.class); + concurrentContainer.getContainerProperties().setObservationEnabled(true); + concurrentContainer.getContainerProperties().setObservationRegistry(observationRegistry); + concurrentContainer.getContainerProperties().setObservationConvention(observationConvention); concurrentContainer.start(); DefaultPulsarMessageListenerContainer childContainer = concurrentContainer.getContainers().get(0); - assertThat(childContainer.getContainerProperties().getObservationConvention()) - .isSameAs(customObservationConvention); + assertThat(childContainer.getContainerProperties().isObservationEnabled()).isTrue(); + assertThat(childContainer.getContainerProperties().getObservationRegistry()).isSameAs(observationRegistry); + assertThat(childContainer.getContainerProperties().getObservationConvention()).isSameAs(observationConvention); } @Test @@ -188,7 +195,7 @@ public class ConcurrentPulsarMessageListenerContainerTests { }); ConcurrentPulsarMessageListenerContainer concurrentContainer = new ConcurrentPulsarMessageListenerContainer<>( - consumerFactory, pulsarContainerProperties, null); + consumerFactory, pulsarContainerProperties); return new PulsarListenerMockComponents(consumerFactory, consumer, concurrentContainer); } @@ -197,4 +204,95 @@ public class ConcurrentPulsarMessageListenerContainerTests { Consumer consumer, ConcurrentPulsarMessageListenerContainer concurrentContainer) { } + @Nested + class ObservationConfigurationTests { + + @Test + void uniqueRegistryAndConventionBeansAvailable() throws Exception { + PulsarListenerMockComponents env = setupListenerMockComponents(SubscriptionType.Shared); + ConcurrentPulsarMessageListenerContainer concurrentContainer = env.concurrentContainer(); + GenericApplicationContext appContext = new GenericApplicationContext(); + ObservationRegistry observationRegistry = mock(ObservationRegistry.class); + PulsarListenerObservationConvention observationConvention = mock(PulsarListenerObservationConvention.class); + appContext.registerBean("obsReg", ObservationRegistry.class, () -> observationRegistry); + appContext.registerBean("obsConv", PulsarListenerObservationConvention.class, () -> observationConvention); + appContext.refresh(); + concurrentContainer.setApplicationContext(appContext); + concurrentContainer.getContainerProperties().setObservationEnabled(true); + concurrentContainer.start(); + PulsarContainerProperties containerProps = concurrentContainer.getContainerProperties(); + assertThat(containerProps.isObservationEnabled()).isTrue(); + assertThat(containerProps.getObservationRegistry()).isSameAs(observationRegistry); + assertThat(containerProps.getObservationConvention()).isSameAs(observationConvention); + } + + @Test + void enabledPropertySetToFalse() throws Exception { + PulsarListenerMockComponents env = setupListenerMockComponents(SubscriptionType.Shared); + ConcurrentPulsarMessageListenerContainer concurrentContainer = env.concurrentContainer(); + GenericApplicationContext appContext = new GenericApplicationContext(); + ObservationRegistry observationRegistry = mock(ObservationRegistry.class); + PulsarListenerObservationConvention observationConvention = mock(PulsarListenerObservationConvention.class); + appContext.registerBean("obsReg", ObservationRegistry.class, () -> observationRegistry); + appContext.registerBean("obsConv", PulsarListenerObservationConvention.class, () -> observationConvention); + appContext.refresh(); + concurrentContainer.setApplicationContext(appContext); + concurrentContainer.getContainerProperties().setObservationEnabled(false); + concurrentContainer.start(); + PulsarContainerProperties containerProps = concurrentContainer.getContainerProperties(); + assertThat(containerProps.isObservationEnabled()).isFalse(); + assertThat(containerProps.getObservationConvention()).isNull(); + assertThat(containerProps.getObservationRegistry()).isNull(); + } + + @Test + void noAppContextAvailable() throws Exception { + PulsarListenerMockComponents env = setupListenerMockComponents(SubscriptionType.Shared); + ConcurrentPulsarMessageListenerContainer concurrentContainer = env.concurrentContainer(); + concurrentContainer.getContainerProperties().setObservationEnabled(true); + concurrentContainer.start(); + PulsarContainerProperties containerProps = concurrentContainer.getContainerProperties(); + assertThat(containerProps.isObservationEnabled()).isTrue(); + assertThat(containerProps.getObservationConvention()).isNull(); + assertThat(containerProps.getObservationRegistry()).isNull(); + } + + @Test + void noRegistryOrConventionBeansAvailable() throws Exception { + PulsarListenerMockComponents env = setupListenerMockComponents(SubscriptionType.Shared); + ConcurrentPulsarMessageListenerContainer concurrentContainer = env.concurrentContainer(); + GenericApplicationContext appContext = new GenericApplicationContext(); + appContext.refresh(); + concurrentContainer.setApplicationContext(appContext); + concurrentContainer.getContainerProperties().setObservationEnabled(true); + concurrentContainer.start(); + PulsarContainerProperties containerProps = concurrentContainer.getContainerProperties(); + assertThat(containerProps.isObservationEnabled()).isTrue(); + assertThat(containerProps.getObservationConvention()).isNull(); + assertThat(containerProps.getObservationRegistry()).isNull(); + } + + @Test + void noUniqueRegistryOrConventionBeansAvailable() throws Exception { + PulsarListenerMockComponents env = setupListenerMockComponents(SubscriptionType.Shared); + ConcurrentPulsarMessageListenerContainer concurrentContainer = env.concurrentContainer(); + GenericApplicationContext appContext = new GenericApplicationContext(); + appContext.registerBean("obsReg1", ObservationRegistry.class, () -> mock(ObservationRegistry.class)); + appContext.registerBean("obsReg2", ObservationRegistry.class, () -> mock(ObservationRegistry.class)); + appContext.registerBean("obsConv1", PulsarListenerObservationConvention.class, + () -> mock(PulsarListenerObservationConvention.class)); + appContext.registerBean("obsConv2", PulsarListenerObservationConvention.class, + () -> mock(PulsarListenerObservationConvention.class)); + appContext.refresh(); + concurrentContainer.setApplicationContext(appContext); + concurrentContainer.getContainerProperties().setObservationEnabled(true); + concurrentContainer.start(); + PulsarContainerProperties containerProps = concurrentContainer.getContainerProperties(); + assertThat(containerProps.isObservationEnabled()).isTrue(); + assertThat(containerProps.getObservationRegistry()).isNull(); + assertThat(containerProps.getObservationConvention()).isNull(); + } + + } + } diff --git a/spring-pulsar/src/test/java/org/springframework/pulsar/listener/PulsarListenerTests.java b/spring-pulsar/src/test/java/org/springframework/pulsar/listener/PulsarListenerTests.java index 68136839..01cf3011 100644 --- a/spring-pulsar/src/test/java/org/springframework/pulsar/listener/PulsarListenerTests.java +++ b/spring-pulsar/src/test/java/org/springframework/pulsar/listener/PulsarListenerTests.java @@ -128,7 +128,7 @@ public class PulsarListenerTests implements PulsarTestContainerSupport { PulsarListenerContainerFactory pulsarListenerContainerFactory( PulsarConsumerFactory pulsarConsumerFactory) { ConcurrentPulsarListenerContainerFactory pulsarListenerContainerFactory = new ConcurrentPulsarListenerContainerFactory<>( - pulsarConsumerFactory, new PulsarContainerProperties(), null); + pulsarConsumerFactory, new PulsarContainerProperties()); return pulsarListenerContainerFactory; } @@ -707,7 +707,7 @@ public class PulsarListenerTests implements PulsarTestContainerSupport { PulsarContainerProperties containerProps = new PulsarContainerProperties(); containerProps.setSchemaResolver(schemaResolver); ConcurrentPulsarListenerContainerFactory pulsarListenerContainerFactory = new ConcurrentPulsarListenerContainerFactory<>( - pulsarConsumerFactory, containerProps, null); + pulsarConsumerFactory, containerProps); return pulsarListenerContainerFactory; } @@ -787,7 +787,7 @@ public class PulsarListenerTests implements PulsarTestContainerSupport { PulsarContainerProperties containerProps = new PulsarContainerProperties(); containerProps.setTopicResolver(topicResolver); ConcurrentPulsarListenerContainerFactory pulsarListenerContainerFactory = new ConcurrentPulsarListenerContainerFactory<>( - pulsarConsumerFactory, containerProps, null); + pulsarConsumerFactory, containerProps); return pulsarListenerContainerFactory; } diff --git a/spring-pulsar/src/test/java/org/springframework/pulsar/observation/ObservationIntegrationTests.java b/spring-pulsar/src/test/java/org/springframework/pulsar/observation/ObservationIntegrationTests.java index ab5f7238..4806e3e5 100644 --- a/spring-pulsar/src/test/java/org/springframework/pulsar/observation/ObservationIntegrationTests.java +++ b/spring-pulsar/src/test/java/org/springframework/pulsar/observation/ObservationIntegrationTests.java @@ -140,10 +140,9 @@ public class ObservationIntegrationTests extends SampleTestRunner implements Pul } @Bean - public PulsarTemplate pulsarTemplate(PulsarProducerFactory pulsarProducerFactory, - ObservationRegistry observationRegistry) { + public PulsarTemplate pulsarTemplate(PulsarProducerFactory pulsarProducerFactory) { return new PulsarTemplate<>(pulsarProducerFactory, null, new DefaultSchemaResolver(), - new DefaultTopicResolver(), observationRegistry, null); + new DefaultTopicResolver(), true); } @Bean @@ -153,9 +152,10 @@ public class ObservationIntegrationTests extends SampleTestRunner implements Pul @Bean PulsarListenerContainerFactory pulsarListenerContainerFactory( - PulsarConsumerFactory pulsarConsumerFactory, ObservationRegistry observationRegistry) { - return new ConcurrentPulsarListenerContainerFactory<>(pulsarConsumerFactory, - new PulsarContainerProperties(), observationRegistry); + PulsarConsumerFactory pulsarConsumerFactory) { + PulsarContainerProperties containerProps = new PulsarContainerProperties(); + containerProps.setObservationEnabled(true); + return new ConcurrentPulsarListenerContainerFactory<>(pulsarConsumerFactory, containerProps); } @Bean diff --git a/spring-pulsar/src/test/java/org/springframework/pulsar/observation/ObservationTests.java b/spring-pulsar/src/test/java/org/springframework/pulsar/observation/ObservationTests.java index 33dbfc28..6a442389 100644 --- a/spring-pulsar/src/test/java/org/springframework/pulsar/observation/ObservationTests.java +++ b/spring-pulsar/src/test/java/org/springframework/pulsar/observation/ObservationTests.java @@ -183,13 +183,17 @@ public class ObservationTests implements PulsarTestContainerSupport { PulsarTemplate pulsarTemplate(PulsarProducerFactory pulsarProducerFactory, ObservationRegistry observationRegistry) { return new PulsarTemplate<>(pulsarProducerFactory, null, new DefaultSchemaResolver(), - new DefaultTopicResolver(), observationRegistry, new DefaultPulsarTemplateObservationConvention() { - @Override - public KeyValues getLowCardinalityKeyValues(PulsarMessageSenderContext context) { - return super.getLowCardinalityKeyValues(context).and(SENDER_EXTRA_TAG, - context.getBeanName()); - } - }); + new DefaultTopicResolver(), true); + } + + @Bean + PulsarTemplateObservationConvention pulsarTemplateConvention() { + return new DefaultPulsarTemplateObservationConvention() { + @Override + public KeyValues getLowCardinalityKeyValues(PulsarMessageSenderContext context) { + return super.getLowCardinalityKeyValues(context).and(SENDER_EXTRA_TAG, context.getBeanName()); + } + }; } @Bean @@ -199,9 +203,15 @@ public class ObservationTests implements PulsarTestContainerSupport { @Bean PulsarListenerContainerFactory pulsarListenerContainerFactory( - PulsarConsumerFactory pulsarConsumerFactory, ObservationRegistry observationRegistry) { + PulsarConsumerFactory pulsarConsumerFactory) { PulsarContainerProperties containerProperties = new PulsarContainerProperties(); - containerProperties.setObservationConvention(new DefaultPulsarListenerObservationConvention() { + containerProperties.setObservationEnabled(true); + return new ConcurrentPulsarListenerContainerFactory<>(pulsarConsumerFactory, containerProperties); + } + + @Bean + PulsarListenerObservationConvention pulsarListenerConvention() { + return new DefaultPulsarListenerObservationConvention() { @Override public KeyValues getLowCardinalityKeyValues(PulsarMessageReceiverContext context) { // Only add the extra tag for the 1st listener @@ -210,9 +220,7 @@ public class ObservationTests implements PulsarTestContainerSupport { } return super.getLowCardinalityKeyValues(context).and(RECEIVER_EXTRA_TAG, context.getListenerId()); } - }); - return new ConcurrentPulsarListenerContainerFactory<>(pulsarConsumerFactory, containerProperties, - observationRegistry); + }; } @Bean