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

Hide observation API from public API on template and listener (#389)

Browse Source 
* Remove observation registry and convention from
  constructors on PulsarTemplate and CPLCF

* Lookup observation registry and convention from
  app context at start time
This commit is contained in:
Chris Bono
2023-04-10 20:35:04 -05:00
committed by GitHub
parent 86c95f883e
commit 9e00c2be7c
16 changed files with 397 additions and 168 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -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

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

@@ -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<ProducerInterceptor> interceptorsProvider, SchemaResolver schemaResolver,
TopicResolver topicResolver, ObjectProvider<ObservationRegistry> observationRegistryProvider,
ObjectProvider<PulsarTemplateObservationConvention> 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

66
spring-pulsar-spring-boot-autoconfigure/src/test/java/org/springframework/pulsar/autoconfigure/PulsarAutoConfigurationTests.java
View File

@@ -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<Object>::getContainerProperties)
.extracting(PulsarContainerProperties::getObservationConvention)
.isSameAs(customConvention)));
.run((context -> assertThat(context).getBean(ConcurrentPulsarListenerContainerFactory.class)
.hasFieldOrPropertyWithValue("containerProperties.observationEnabled", false)));
}
}

14
spring-pulsar/src/main/java/org/springframework/pulsar/config/AbstractPulsarListenerContainerFactory.java
View File

@@ -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<C extends AbstractP
private final PulsarContainerProperties containerProperties;
private final ObservationRegistry observationRegistry;
private Boolean autoStartup;
private Integer phase;
@@ -66,20 +61,15 @@ public abstract class AbstractPulsarListenerContainerFactory<C extends AbstractP
private ApplicationContext applicationContext;
protected AbstractPulsarListenerContainerFactory(PulsarConsumerFactory<? super T> consumerFactory,
PulsarContainerProperties containerProperties, @Nullable ObservationRegistry observationRegistry) {
PulsarContainerProperties containerProperties) {
this.consumerFactory = consumerFactory;
this.containerProperties = containerProperties;
this.observationRegistry = observationRegistry;
}
protected PulsarConsumerFactory<? super T> getConsumerFactory() {
return this.consumerFactory;
}
protected ObservationRegistry getObservationRegistry() {
return this.observationRegistry;
}
public PulsarContainerProperties getContainerProperties() {
return this.containerProperties;
}
@@ -171,7 +161,7 @@ public abstract class AbstractPulsarListenerContainerFactory<C extends AbstractP
instanceProperties.setMaxNumMessages(this.containerProperties.getMaxNumMessages());
instanceProperties.setMaxNumBytes(this.containerProperties.getMaxNumBytes());
instanceProperties.setBatchTimeoutMillis(this.containerProperties.getBatchTimeoutMillis());
instanceProperties.setObservationConvention(this.containerProperties.getObservationConvention());
instanceProperties.setObservationEnabled(this.containerProperties.isObservationEnabled());
JavaUtils.INSTANCE.acceptIfNotNull(this.phase, instance::setPhase)
.acceptIfNotNull(this.applicationContext, instance::setApplicationContext)

42
spring-pulsar/src/main/java/org/springframework/pulsar/config/ConcurrentPulsarListenerContainerFactory.java
View File

@@ -20,15 +20,12 @@ import java.util.Arrays;
import java.util.Collection;
import java.util.HashSet;
import org.springframework.lang.Nullable;
import org.springframework.pulsar.core.PulsarConsumerFactory;
import org.springframework.pulsar.listener.ConcurrentPulsarMessageListenerContainer;
import org.springframework.pulsar.listener.PulsarContainerProperties;
import org.springframework.util.CollectionUtils;
import org.springframework.util.StringUtils;
import io.micrometer.observation.ObservationRegistry;
/**
* Concrete implementation for {@link PulsarListenerContainerFactory}.
*
@@ -43,8 +40,8 @@ public class ConcurrentPulsarListenerContainerFactory<T>
private Integer concurrency;
public ConcurrentPulsarListenerContainerFactory(PulsarConsumerFactory<? super T> consumerFactory,
PulsarContainerProperties containerProperties, @Nullable ObservationRegistry observationRegistry) {
super(consumerFactory, containerProperties, observationRegistry);
PulsarContainerProperties containerProperties) {
super(consumerFactory, containerProperties);
}
/**
@@ -55,6 +52,22 @@ public class ConcurrentPulsarListenerContainerFactory<T>
this.concurrency = concurrency;
}
@Override
public ConcurrentPulsarMessageListenerContainer<T> createContainer(String... topics) {
PulsarListenerEndpoint endpoint = new PulsarListenerEndpointAdapter() {
@Override
public Collection<String> getTopics() {
return Arrays.asList(topics);
}
};
ConcurrentPulsarMessageListenerContainer<T> container = createContainerInstance(endpoint);
initializeContainer(container, endpoint);
// customizeContainer(container);
return container;
}
@Override
protected ConcurrentPulsarMessageListenerContainer<T> createContainerInstance(PulsarListenerEndpoint endpoint) {
@@ -84,8 +97,7 @@ public class ConcurrentPulsarListenerContainerFactory<T>
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<T>
}
}
@Override
public ConcurrentPulsarMessageListenerContainer<T> createContainer(String... topics) {
PulsarListenerEndpoint endpoint = new PulsarListenerEndpointAdapter() {
@Override
public Collection<String> getTopics() {
return Arrays.asList(topics);
}
};
ConcurrentPulsarMessageListenerContainer<T> container = createContainerInstance(endpoint);
initializeContainer(container, endpoint);
// customizeContainer(container);
return container;
}
}

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

@@ -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<T> implements PulsarOperations<T>, BeanNameAware {
public class PulsarTemplate<T>
implements PulsarOperations<T>, ApplicationContextAware, BeanNameAware, SmartInitializingSingleton {
private final LogAccessor logger = new LogAccessor(this.getClass());
@@ -63,11 +67,25 @@ public class PulsarTemplate<T> implements PulsarOperations<T>, 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<T> implements PulsarOperations<T>, 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<T> producerFactory, List<ProducerInterceptor> 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<T> implements PulsarOperations<T>, 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<T> producerFactory, List<ProducerInterceptor> 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

12
spring-pulsar/src/main/java/org/springframework/pulsar/listener/AbstractPulsarMessageListenerContainer.java
View File

@@ -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<T> 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<T> exten
@SuppressWarnings("unchecked")
protected AbstractPulsarMessageListenerContainer(PulsarConsumerFactory<? super T> pulsarConsumerFactory,
PulsarContainerProperties pulsarContainerProperties, @Nullable ObservationRegistry observationRegistry) {
PulsarContainerProperties pulsarContainerProperties) {
this.pulsarConsumerFactory = (PulsarConsumerFactory<T>) pulsarConsumerFactory;
this.pulsarContainerProperties = pulsarContainerProperties;
this.observationRegistry = observationRegistry;
}
public PulsarConsumerFactory<T> getPulsarConsumerFactory() {
@@ -73,10 +67,6 @@ public non-sealed abstract class AbstractPulsarMessageListenerContainer<T> exten
return this.pulsarContainerProperties;
}
public ObservationRegistry getObservationRegistry() {
return this.observationRegistry;
}
@Override
public boolean isRunning() {
return this.running;

23
spring-pulsar/src/main/java/org/springframework/pulsar/listener/ConcurrentPulsarMessageListenerContainer.java
View File

@@ -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<T> extends AbstractPulsarM
private final List<AsyncTaskExecutor> executors = new ArrayList<>();
public ConcurrentPulsarMessageListenerContainer(PulsarConsumerFactory<? super T> 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<T> 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<T> container = constructContainer(containerProperties);
configureChildContainer(i, container);
@@ -90,8 +104,7 @@ public class ConcurrentPulsarMessageListenerContainer<T> extends AbstractPulsarM
}
private DefaultPulsarMessageListenerContainer<T> constructContainer(PulsarContainerProperties containerProperties) {
return new DefaultPulsarMessageListenerContainer<>(this.getPulsarConsumerFactory(), containerProperties,
this.getObservationRegistry());
return new DefaultPulsarMessageListenerContainer<>(this.getPulsarConsumerFactory(), containerProperties);
}
private void configureChildContainer(int index, DefaultPulsarMessageListenerContainer<T> container) {

23
spring-pulsar/src/main/java/org/springframework/pulsar/listener/DefaultPulsarMessageListenerContainer.java
View File

@@ -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<T> extends AbstractPulsarMess
public DefaultPulsarMessageListenerContainer(PulsarConsumerFactory<? super T> pulsarConsumerFactory,
PulsarContainerProperties pulsarContainerProperties) {
this(pulsarConsumerFactory, pulsarContainerProperties, null);
}
public DefaultPulsarMessageListenerContainer(PulsarConsumerFactory<? super T> pulsarConsumerFactory,
PulsarContainerProperties pulsarContainerProperties, @Nullable ObservationRegistry observationRegistry) {
super(pulsarConsumerFactory, pulsarContainerProperties, observationRegistry);
super(pulsarConsumerFactory, pulsarContainerProperties);
this.thisOrParentContainer = this;
}
@@ -122,8 +115,7 @@ public class DefaultPulsarMessageListenerContainer<T> 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<T> extends AbstractPulsarMess
private final PulsarContainerProperties containerProperties;
private final ObservationRegistry observationRegistry;
private Consumer<T> consumer;
private final Set<MessageId> nackableMessages = new HashSet<>();
@@ -246,8 +236,7 @@ public class DefaultPulsarMessageListenerContainer<T> 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<T> 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<T> extends AbstractPulsarMess
}
private Observation newObservation(Message<T> 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<T> message, AtomicBoolean inRetryMode) {

24
spring-pulsar/src/main/java/org/springframework/pulsar/listener/PulsarContainerProperties.java
View File

@@ -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;
}

115
spring-pulsar/src/test/java/org/springframework/pulsar/core/PulsarTemplateObservationConfigurationTests.java Normal file
View File

@@ -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);
}
}

4
spring-pulsar/src/test/java/org/springframework/pulsar/core/PulsarTemplateTests.java
View File

@@ -221,7 +221,7 @@ class PulsarTemplateTests implements PulsarTestContainerSupport {
DefaultTopicResolver topicResolver = new DefaultTopicResolver();
topicResolver.addCustomTopicMapping(Foo.class, topic);
PulsarTemplate<Foo> 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<PulsarTemplate<Foo>> 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<Foo> 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<PulsarTemplate<Foo>> sendFunction = (template) -> template.newMessage(foo).send();
sendAndConsume(pulsarTemplate, sendFunction, topic, Schema.JSON(Foo.class), foo);

112
spring-pulsar/src/test/java/org/springframework/pulsar/listener/ConcurrentPulsarMessageListenerContainerTests.java
View File

@@ -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<String> 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<String> 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<String> 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<String> concurrentContainer = new ConcurrentPulsarMessageListenerContainer<>(
consumerFactory, pulsarContainerProperties, null);
consumerFactory, pulsarContainerProperties);
return new PulsarListenerMockComponents(consumerFactory, consumer, concurrentContainer);
}
@@ -197,4 +204,95 @@ public class ConcurrentPulsarMessageListenerContainerTests {
Consumer<String> consumer, ConcurrentPulsarMessageListenerContainer<String> concurrentContainer) {
}
@Nested
class ObservationConfigurationTests {
@Test
void uniqueRegistryAndConventionBeansAvailable() throws Exception {
PulsarListenerMockComponents env = setupListenerMockComponents(SubscriptionType.Shared);
ConcurrentPulsarMessageListenerContainer<String> 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<String> 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<String> 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<String> 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<String> 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();
}
}
}

6
spring-pulsar/src/test/java/org/springframework/pulsar/listener/PulsarListenerTests.java
View File

@@ -128,7 +128,7 @@ public class PulsarListenerTests implements PulsarTestContainerSupport {
PulsarListenerContainerFactory pulsarListenerContainerFactory(
PulsarConsumerFactory<Object> 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;
}

12
spring-pulsar/src/test/java/org/springframework/pulsar/observation/ObservationIntegrationTests.java
View File

@@ -140,10 +140,9 @@ public class ObservationIntegrationTests extends SampleTestRunner implements Pul
}
@Bean
public PulsarTemplate<String> pulsarTemplate(PulsarProducerFactory<String> pulsarProducerFactory,
ObservationRegistry observationRegistry) {
public PulsarTemplate<String> pulsarTemplate(PulsarProducerFactory<String> 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<Object> pulsarConsumerFactory, ObservationRegistry observationRegistry) {
return new ConcurrentPulsarListenerContainerFactory<>(pulsarConsumerFactory,
new PulsarContainerProperties(), observationRegistry);
PulsarConsumerFactory<Object> pulsarConsumerFactory) {
PulsarContainerProperties containerProps = new PulsarContainerProperties();
containerProps.setObservationEnabled(true);
return new ConcurrentPulsarListenerContainerFactory<>(pulsarConsumerFactory, containerProps);
}
@Bean

32
spring-pulsar/src/test/java/org/springframework/pulsar/observation/ObservationTests.java
View File

@@ -183,13 +183,17 @@ public class ObservationTests implements PulsarTestContainerSupport {
PulsarTemplate<String> pulsarTemplate(PulsarProducerFactory<String> 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<Object> pulsarConsumerFactory, ObservationRegistry observationRegistry) {
PulsarConsumerFactory<Object> 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