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

Fix JMS Inbound Endpoints for observation

Browse Source 
The `JmsMessageDrivenEndpoint` delegates all the hard work to the
`ChannelPublishingJmsMessageListener`, but missed to propagate an `ObservationRegistry`
and other related options.
The `JmsInboundGateway` is worse: it delegated to the `JmsMessageDrivenEndpoint`

* Add `IntegrationObservation.HANDLER` observation to the `MessagingGatewaySupport.send()`
operation: used by the delegate in the `ChannelPublishingJmsMessageListener`
* Expose and propagate observation-related options from `JmsInboundGateway`
and `JmsMessageDrivenEndpoint`
* Expose `observationConvention()` option on the `MessagingGatewaySpec`
and `MessageProducerSpec`
* Remove unused imports
* Do not start a new `RECEIVER` observation if there is already `SERVER` one
* Fix `MessagingGatewaySupport` for `Observation.NOOP` check.
The parent process may still use `ObservationRegistry.NOOP` which sets
`Observation.NOOP` instance into the current context and thread local.

**Cherry-pick to `6.1.x` & `6.0.x`**
This commit is contained in:
Artem Bilan
2023-09-19 12:02:49 -04:00
committed by GitHub
parent 459de03ea3
commit 9ad2b8a8e2
8 changed files with 208 additions and 35 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
build.gradle
View File

@@ -766,6 +766,7 @@ project('spring-integration-jms') {
testImplementation "org.apache.activemq:artemis-jakarta-client:$artemisVersion"
testImplementation 'org.springframework:spring-oxm'
testImplementation 'com.fasterxml.jackson.core:jackson-databind'
testImplementation 'io.micrometer:micrometer-observation-test'
}
}

13
spring-integration-core/src/main/java/org/springframework/integration/dsl/MessageProducerSpec.java
View File

@@ -18,6 +18,7 @@ package org.springframework.integration.dsl;
import org.springframework.integration.endpoint.MessageProducerSupport;
import org.springframework.integration.support.ErrorMessageStrategy;
import org.springframework.integration.support.management.observation.MessageReceiverObservationConvention;
import org.springframework.lang.Nullable;
import org.springframework.messaging.MessageChannel;
@@ -152,4 +153,16 @@ public abstract class MessageProducerSpec<S extends MessageProducerSpec<S, P>, P
return _this();
}
/**
* Provide a custom {@link MessageReceiverObservationConvention}.
* @param observationConvention the observation convention to use.
* @return the spec.
* @since 6.0.8
* @see MessageProducerSupport#setObservationConvention(MessageReceiverObservationConvention)
*/
public S observationConvention(MessageReceiverObservationConvention observationConvention) {
this.target.setObservationConvention(observationConvention);
return _this();
}
}

13
spring-integration-core/src/main/java/org/springframework/integration/dsl/MessagingGatewaySpec.java
View File

@@ -19,6 +19,7 @@ package org.springframework.integration.dsl;
import org.springframework.integration.gateway.MessagingGatewaySupport;
import org.springframework.integration.mapping.InboundMessageMapper;
import org.springframework.integration.mapping.OutboundMessageMapper;
import org.springframework.integration.support.management.observation.MessageRequestReplyReceiverObservationConvention;
import org.springframework.lang.Nullable;
import org.springframework.messaging.MessageChannel;
@@ -205,4 +206,16 @@ public abstract class MessagingGatewaySpec<S extends MessagingGatewaySpec<S, G>,
return _this();
}
/**
* Provide a custom {@link MessageRequestReplyReceiverObservationConvention}.
* @param observationConvention the observation convention to use.
* @return the spec.
* @since 6.0.8
* @see MessagingGatewaySupport#setObservationConvention(MessageRequestReplyReceiverObservationConvention)
*/
public S observationConvention(MessageRequestReplyReceiverObservationConvention observationConvention) {
this.target.setObservationConvention(observationConvention);
return _this();
}
}

87
spring-integration-core/src/main/java/org/springframework/integration/gateway/MessagingGatewaySupport.java
View File

@@ -22,6 +22,7 @@ import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
import io.micrometer.observation.Observation;
import io.micrometer.observation.ObservationRegistry;
import org.reactivestreams.Publisher;
import org.reactivestreams.Subscriber;
@@ -56,12 +57,16 @@ import org.springframework.integration.support.MutableMessageBuilder;
import org.springframework.integration.support.converter.SimpleMessageConverter;
import org.springframework.integration.support.management.IntegrationInboundManagement;
import org.springframework.integration.support.management.IntegrationManagedResource;
import org.springframework.integration.support.management.TrackableComponent;
import org.springframework.integration.support.management.metrics.MeterFacade;
import org.springframework.integration.support.management.metrics.MetricsCaptor;
import org.springframework.integration.support.management.metrics.SampleFacade;
import org.springframework.integration.support.management.metrics.TimerFacade;
import org.springframework.integration.support.management.observation.DefaultMessageReceiverObservationConvention;
import org.springframework.integration.support.management.observation.DefaultMessageRequestReplyReceiverObservationConvention;
import org.springframework.integration.support.management.observation.IntegrationObservation;
import org.springframework.integration.support.management.observation.MessageReceiverContext;
import org.springframework.integration.support.management.observation.MessageReceiverObservationConvention;
import org.springframework.integration.support.management.observation.MessageRequestReplyReceiverContext;
import org.springframework.integration.support.management.observation.MessageRequestReplyReceiverObservationConvention;
import org.springframework.lang.Nullable;
@@ -91,7 +96,7 @@ import org.springframework.util.Assert;
*/
@IntegrationManagedResource
public abstract class MessagingGatewaySupport extends AbstractEndpoint
implements org.springframework.integration.support.management.TrackableComponent,
implements TrackableComponent,
IntegrationInboundManagement, IntegrationPattern {
protected final ConvertingMessagingTemplate messagingTemplate; // NOSONAR
@@ -144,6 +149,8 @@ public abstract class MessagingGatewaySupport extends AbstractEndpoint
@Nullable
private MessageRequestReplyReceiverObservationConvention observationConvention;
private MessageReceiverObservationConvention receiverObservationConvention;
private volatile AbstractEndpoint replyMessageCorrelator;
private volatile boolean initialized;
@@ -384,6 +391,10 @@ public abstract class MessagingGatewaySupport extends AbstractEndpoint
this.observationConvention = observationConvention;
}
public void setReceiverObservationConvention(MessageReceiverObservationConvention receiverObservationConvention) {
this.receiverObservationConvention = receiverObservationConvention;
}
@Override
protected void onInit() {
Assert.state(!(this.requestChannelName != null && this.requestChannel != null),
@@ -468,27 +479,65 @@ public abstract class MessagingGatewaySupport extends AbstractEndpoint
MessageChannel channel = getRequestChannel();
Assert.state(channel != null,
"send is not supported, because no request channel has been configured");
SampleFacade sample = null;
if (this.metricsCaptor != null) {
sample = this.metricsCaptor.start();
Message<?> requestMessage = this.messagingTemplate.doConvert(object, null, this.historyWritingPostProcessor);
if (!ObservationRegistry.NOOP.equals(this.observationRegistry)
&& (this.observationRegistry.getCurrentObservation() == null
|| Observation.NOOP.equals(this.observationRegistry.getCurrentObservation()))) {
sendWithObservation(channel, requestMessage);
}
else if (this.metricsCaptor != null) {
sendWithMetrics(channel, requestMessage);
}
else {
doSend(channel, requestMessage);
}
}
private void sendWithObservation(MessageChannel channel, Message<?> message) {
try {
this.messagingTemplate.convertAndSend(channel, object, this.historyWritingPostProcessor);
if (sample != null) {
sample.stop(sendTimer());
}
IntegrationObservation.HANDLER.observation(
this.receiverObservationConvention,
DefaultMessageReceiverObservationConvention.INSTANCE,
() -> new MessageReceiverContext(message, getComponentName()),
this.observationRegistry)
.observe(() -> this.messagingTemplate.send(channel, message));
}
catch (Exception e) {
if (sample != null) {
sample.stop(buildSendTimer(false, e.getClass().getSimpleName()));
}
MessageChannel errorChan = getErrorChannel();
if (errorChan != null) {
this.messagingTemplate.send(errorChan, new ErrorMessage(e));
}
else {
rethrow(e, "failed to send message");
}
catch (Exception ex) {
sendErrorMessage(ex, message);
}
}
private void sendWithMetrics(MessageChannel channel, Message<?> message) {
SampleFacade sample = this.metricsCaptor.start();
try {
this.messagingTemplate.send(channel, message);
sample.stop(sendTimer());
}
catch (Exception ex) {
sample.stop(buildSendTimer(false, ex.getClass().getSimpleName()));
sendErrorMessage(ex, message);
}
}
private void doSend(MessageChannel channel, Message<?> message) {
try {
this.messagingTemplate.send(channel, message);
}
catch (Exception ex) {
sendErrorMessage(ex, message);
}
}
private void sendErrorMessage(Exception exception, Message<?> failedMessage) {
MessageChannel errorChan = getErrorChannel();
if (errorChan != null) {
this.messagingTemplate.send(errorChan, buildErrorMessage(failedMessage, exception));
}
else {
rethrow(exception, "failed to send message");
}
}

24
spring-integration-jms/src/main/java/org/springframework/integration/jms/ChannelPublishingJmsMessageListener.java
View File

@@ -18,6 +18,7 @@ package org.springframework.integration.jms;
import java.util.Map;
import io.micrometer.observation.ObservationRegistry;
import jakarta.jms.DeliveryMode;
import jakarta.jms.Destination;
import jakarta.jms.InvalidDestinationException;
@@ -38,6 +39,9 @@ import org.springframework.integration.gateway.MessagingGatewaySupport;
import org.springframework.integration.support.DefaultMessageBuilderFactory;
import org.springframework.integration.support.MessageBuilderFactory;
import org.springframework.integration.support.management.TrackableComponent;
import org.springframework.integration.support.management.metrics.MetricsCaptor;
import org.springframework.integration.support.management.observation.MessageReceiverObservationConvention;
import org.springframework.integration.support.management.observation.MessageRequestReplyReceiverObservationConvention;
import org.springframework.integration.support.utils.IntegrationUtils;
import org.springframework.jms.listener.SessionAwareMessageListener;
import org.springframework.jms.support.JmsUtils;
@@ -323,6 +327,26 @@ public class ChannelPublishingJmsMessageListener
this.extractReplyPayload = extractReplyPayload;
}
public void setMetricsCaptor(MetricsCaptor captor) {
this.gatewayDelegate.registerMetricsCaptor(captor);
}
public void setObservationRegistry(ObservationRegistry observationRegistry) {
this.gatewayDelegate.registerObservationRegistry(observationRegistry);
}
public void setRequestReplyObservationConvention(
@Nullable MessageRequestReplyReceiverObservationConvention observationConvention) {
this.gatewayDelegate.setObservationConvention(observationConvention);
}
public void setReceiverObservationConvention(
@Nullable MessageReceiverObservationConvention observationConvention) {
this.gatewayDelegate.setReceiverObservationConvention(observationConvention);
}
@Override
public void setBeanFactory(BeanFactory beanFactory) throws BeansException {
this.beanFactory = beanFactory;

32
spring-integration-jms/src/main/java/org/springframework/integration/jms/JmsInboundGateway.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2016-2022 the original author or authors.
* Copyright 2016-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.
@@ -16,10 +16,14 @@
package org.springframework.integration.jms;
import io.micrometer.observation.ObservationRegistry;
import org.springframework.beans.BeansException;
import org.springframework.context.ApplicationContext;
import org.springframework.integration.context.OrderlyShutdownCapable;
import org.springframework.integration.gateway.MessagingGatewaySupport;
import org.springframework.integration.support.management.metrics.MetricsCaptor;
import org.springframework.integration.support.management.observation.MessageRequestReplyReceiverObservationConvention;
import org.springframework.jms.listener.AbstractMessageListenerContainer;
import org.springframework.messaging.MessageChannel;
@@ -114,28 +118,38 @@ public class JmsInboundGateway extends MessagingGatewaySupport implements Orderl
this.endpoint.setShutdownContainerOnStop(shutdownContainerOnStop);
}
@Override
public void registerMetricsCaptor(MetricsCaptor metricsCaptorToRegister) {
super.registerMetricsCaptor(metricsCaptorToRegister);
this.endpoint.registerMetricsCaptor(metricsCaptorToRegister);
}
@Override
public void registerObservationRegistry(ObservationRegistry observationRegistry) {
super.registerObservationRegistry(observationRegistry);
this.endpoint.registerObservationRegistry(observationRegistry);
}
@Override
public void setObservationConvention(MessageRequestReplyReceiverObservationConvention observationConvention) {
super.setObservationConvention(observationConvention);
this.endpoint.getListener().setRequestReplyObservationConvention(observationConvention);
}
@Override
public String getComponentType() {
return this.endpoint.getComponentType();
}
@Override
public void setComponentName(String componentName) {
super.setComponentName(componentName);
this.endpoint.setComponentName(getComponentName());
}
@Override
public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
super.setApplicationContext(applicationContext);
this.endpoint.setApplicationContext(applicationContext);
this.endpoint.setBeanFactory(applicationContext);
this.endpoint.getListener().setBeanFactory(applicationContext);
}
@Override
protected void onInit() {
this.endpoint.setComponentName(getComponentName());
this.endpoint.afterPropertiesSet();
}

30
spring-integration-jms/src/main/java/org/springframework/integration/jms/JmsMessageDrivenEndpoint.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2021 the original author or authors.
* Copyright 2002-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.
@@ -16,11 +16,15 @@
package org.springframework.integration.jms;
import io.micrometer.observation.ObservationRegistry;
import org.springframework.beans.BeansException;
import org.springframework.context.ApplicationContext;
import org.springframework.integration.context.OrderlyShutdownCapable;
import org.springframework.integration.endpoint.MessageProducerSupport;
import org.springframework.integration.jms.util.JmsAdapterUtils;
import org.springframework.integration.support.management.metrics.MetricsCaptor;
import org.springframework.integration.support.management.observation.MessageReceiverObservationConvention;
import org.springframework.jms.listener.AbstractMessageListenerContainer;
import org.springframework.jms.listener.DefaultMessageListenerContainer;
import org.springframework.messaging.MessageChannel;
@@ -91,7 +95,7 @@ public class JmsMessageDrivenEndpoint extends MessageProducerSupport implements
* container setting even if an external container is provided. Defaults to null
* (won't change container) if an external container is provided or `transacted` when
* the framework creates an implicit {@link DefaultMessageListenerContainer}.
* @param sessionAcknowledgeMode the acknowledge mode.
* @param sessionAcknowledgeMode the acknowledgement mode.
*/
public void setSessionAcknowledgeMode(String sessionAcknowledgeMode) {
this.sessionAcknowledgeMode = sessionAcknowledgeMode;
@@ -134,9 +138,9 @@ public class JmsMessageDrivenEndpoint extends MessageProducerSupport implements
}
/**
* Set to false to prevent listener container shutdown when the endpoint is stopped.
* Set to {@code false} to prevent listener container shutdown when the endpoint is stopped.
* Then, if so configured, any cached consumer(s) in the container will remain.
* Otherwise the shared connection and will be closed and the listener invokers shut
* Otherwise, the shared connection and will be closed and the listener invokers shut
* down; this behavior is new starting with version 5.1. Default: true.
* @param shutdownContainerOnStop false to not shutdown.
* @since 5.1
@@ -149,6 +153,24 @@ public class JmsMessageDrivenEndpoint extends MessageProducerSupport implements
return this.listener;
}
@Override
public void registerMetricsCaptor(MetricsCaptor captor) {
super.registerMetricsCaptor(captor);
this.listener.setMetricsCaptor(captor);
}
@Override
public void registerObservationRegistry(ObservationRegistry observationRegistry) {
super.registerObservationRegistry(observationRegistry);
this.listener.setObservationRegistry(observationRegistry);
}
@Override
public void setObservationConvention(MessageReceiverObservationConvention observationConvention) {
super.setObservationConvention(observationConvention);
this.listener.setReceiverObservationConvention(observationConvention);
}
@Override
public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
super.setApplicationContext(applicationContext);

43
spring-integration-jms/src/test/java/org/springframework/integration/jms/dsl/JmsTests.java
View File

@@ -22,8 +22,11 @@ import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import io.micrometer.observation.tck.TestObservationRegistry;
import io.micrometer.observation.tck.TestObservationRegistryAssert;
import jakarta.jms.JMSException;
import jakarta.jms.TextMessage;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.ListableBeanFactory;
@@ -40,6 +43,7 @@ import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.channel.FixedSubscriberChannel;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.config.EnableIntegration;
import org.springframework.integration.config.EnableIntegrationManagement;
import org.springframework.integration.config.GlobalChannelInterceptor;
import org.springframework.integration.core.MessageSource;
import org.springframework.integration.dsl.IntegrationFlow;
@@ -55,6 +59,7 @@ import org.springframework.integration.jms.JmsDestinationPollingSource;
import org.springframework.integration.jms.SubscribableJmsChannel;
import org.springframework.integration.scheduling.PollerMetadata;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.integration.support.management.observation.IntegrationObservation;
import org.springframework.integration.test.util.TestUtils;
import org.springframework.jms.core.JmsTemplate;
import org.springframework.jms.listener.DefaultMessageListenerContainer;
@@ -144,6 +149,14 @@ public class JmsTests extends ActiveMQMultiContextTests {
@Autowired
JmsTemplate jmsTemplate;
@Autowired
TestObservationRegistry observationRegistry;
@BeforeEach
void setup() {
this.observationRegistry.clear();
}
@Test
public void testPollingFlow() {
this.controlBus.send("@'integerMessageSource.inboundChannelAdapter'.start()");
@@ -204,6 +217,14 @@ public class JmsTests extends ActiveMQMultiContextTests {
.isEqualTo("foo");
assertThat(this.jmsOutboundFlowTemplate).isNotNull();
TestObservationRegistryAssert.assertThat(this.observationRegistry)
.hasObservationWithNameEqualTo("spring.integration.handler")
.that()
.hasLowCardinalityKeyValue(IntegrationObservation.HandlerTags.COMPONENT_NAME.asString(),
"observedJmsMessageDrivenChannelAdapter")
.hasBeenStarted()
.hasBeenStopped();
}
@Test
@@ -239,6 +260,14 @@ public class JmsTests extends ActiveMQMultiContextTests {
.isNotNull()
.extracting(Message::getPayload)
.isEqualTo("error: junk is not convertible");
TestObservationRegistryAssert.assertThat(this.observationRegistry)
.hasObservationWithNameEqualTo("spring.integration.gateway")
.that()
.hasLowCardinalityKeyValue(IntegrationObservation.GatewayTags.COMPONENT_NAME.asString(),
"observedJmsInboundGateway")
.hasBeenStarted()
.hasBeenStopped();
}
@Test
@@ -294,8 +323,14 @@ public class JmsTests extends ActiveMQMultiContextTests {
@Configuration
@EnableIntegration
@IntegrationComponentScan
@EnableIntegrationManagement(observationPatterns = "observedJms*")
public static class ContextConfiguration {
@Bean
TestObservationRegistry observationRegistry() {
return TestObservationRegistry.create();
}
@Bean
public JmsTemplate jmsTemplate() {
JmsTemplate jmsTemplate = new JmsTemplate(connectionFactory);
@@ -407,9 +442,10 @@ public class JmsTests extends ActiveMQMultiContextTests {
public IntegrationFlow jmsMessageDrivenFlowWithContainer() {
return IntegrationFlow
.from(Jms.messageDrivenChannelAdapter(
Jms.container(amqFactory, "containerSpecDestination")
.pubSubDomain(false)
.taskExecutor(Executors.newCachedThreadPool())))
Jms.container(amqFactory, "containerSpecDestination")
.pubSubDomain(false)
.taskExecutor(Executors.newCachedThreadPool()))
.id("observedJmsMessageDrivenChannelAdapter"))
.transform(String::trim)
.channel(jmsOutboundInboundReplyChannel())
.get();
@@ -428,6 +464,7 @@ public class JmsTests extends ActiveMQMultiContextTests {
public IntegrationFlow jmsInboundGatewayFlow() {
return IntegrationFlow.from(
Jms.inboundGateway(amqFactory)
.id("observedJmsInboundGateway")
.requestChannel(jmsInboundGatewayInputChannel())
.replyTimeout(1)
.errorOnTimeout(true)