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

GH-8631: Add retry options to JMS inbound components

Browse Source 
Fixes: #8631

The `ChannelPublishingJmsMessageListener` (and respective Java DSL specs)
can now be supplied with a `RetryTemplate` & `RecoveryCallback` for retying `send` & `send-and-receive` operations
in the internal gateway implementation.
The `JmsMessageHeaderErrorMessageStrategy` was introduced to have access into a raw JMS message from retry context.
The functionality is modeled after `AmqpInboundChannelAdapter`
This commit is contained in:
Artem Bilan
2024-03-13 13:42:07 -04:00
parent ffe3605c90
commit 2dea91b4c3
8 changed files with 276 additions and 17 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2023 the original author or authors.
* Copyright 2002-2024 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.
@@ -17,6 +17,7 @@
package org.springframework.integration.jms;
import java.util.Map;
import java.util.concurrent.atomic.AtomicInteger;
import io.micrometer.observation.ObservationRegistry;
import jakarta.jms.DeliveryMode;
@@ -30,13 +31,18 @@ import org.springframework.beans.BeansException;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.beans.factory.BeanFactoryAware;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.core.AttributeAccessor;
import org.springframework.core.log.LogAccessor;
import org.springframework.expression.Expression;
import org.springframework.expression.spel.support.StandardEvaluationContext;
import org.springframework.integration.IntegrationMessageHeaderAccessor;
import org.springframework.integration.StaticMessageHeaderAccessor;
import org.springframework.integration.core.MessagingTemplate;
import org.springframework.integration.expression.ExpressionUtils;
import org.springframework.integration.gateway.MessagingGatewaySupport;
import org.springframework.integration.jms.support.JmsMessageHeaderErrorMessageStrategy;
import org.springframework.integration.support.DefaultMessageBuilderFactory;
import org.springframework.integration.support.ErrorMessageUtils;
import org.springframework.integration.support.MessageBuilderFactory;
import org.springframework.integration.support.management.TrackableComponent;
import org.springframework.integration.support.management.metrics.MetricsCaptor;
@@ -54,6 +60,10 @@ import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessagingException;
import org.springframework.messaging.support.ErrorMessage;
import org.springframework.retry.RecoveryCallback;
import org.springframework.retry.RetryOperations;
import org.springframework.retry.support.RetrySynchronizationManager;
import org.springframework.retry.support.RetryTemplate;
import org.springframework.util.Assert;
/**
@@ -347,6 +357,30 @@ public class ChannelPublishingJmsMessageListener
this.gatewayDelegate.setReceiverObservationConvention(observationConvention);
}
/**
* Set a {@link RetryTemplate} to use for retrying a message delivery within the
* adapter. Unlike adding retry at the container level, this can be used with an
* {@code ErrorMessageSendingRecoverer} {@link RecoveryCallback} to publish to the
* error channel after retries are exhausted. You generally should not configure an
* error channel when using retry here, use a {@link RecoveryCallback} instead.
* @param retryTemplate the template.
* @since 6.3
* @see #setRecoveryCallback(RecoveryCallback)
*/
public void setRetryTemplate(RetryTemplate retryTemplate) {
this.gatewayDelegate.retryTemplate = retryTemplate;
}
/**
* Set a {@link RecoveryCallback} when using retry within the adapter.
* @param recoveryCallback the callback.
* @since 6.3
* @see #setRetryTemplate(RetryTemplate)
*/
public void setRecoveryCallback(RecoveryCallback<Message<?>> recoveryCallback) {
this.gatewayDelegate.recoveryCallback = recoveryCallback;
}
@Override
public void setBeanFactory(BeanFactory beanFactory) throws BeansException {
this.beanFactory = beanFactory;
@@ -367,6 +401,9 @@ public class ChannelPublishingJmsMessageListener
}
Map<String, Object> headers = this.headerMapper.toHeaders(jmsMessage);
if (this.gatewayDelegate.retryTemplate != null) {
headers.put(IntegrationMessageHeaderAccessor.DELIVERY_ATTEMPT, new AtomicInteger());
}
requestMessage =
(result instanceof Message<?>) ?
this.messageBuilderFactory.fromMessage((Message<?>) result).copyHeaders(headers).build() :
@@ -385,10 +422,10 @@ public class ChannelPublishingJmsMessageListener
}
if (!this.expectReply) {
this.gatewayDelegate.send(requestMessage);
this.gatewayDelegate.send(jmsMessage, requestMessage);
}
else {
Message<?> replyMessage = this.gatewayDelegate.sendAndReceiveMessage(requestMessage);
Message<?> replyMessage = this.gatewayDelegate.sendAndReceiveMessage(jmsMessage, requestMessage);
if (replyMessage != null) {
Destination destination = getReplyDestination(jmsMessage, session);
this.logger.debug(() -> "Reply destination: " + destination);
@@ -424,6 +461,12 @@ public class ChannelPublishingJmsMessageListener
this.gatewayDelegate.setBeanFactory(this.beanFactory);
}
this.gatewayDelegate.afterPropertiesSet();
if (this.gatewayDelegate.retryTemplate != null) {
Assert.state(this.gatewayDelegate.getErrorChannel() == null,
"Cannot have an 'errorChannel' property when a 'RetryTemplate' is "
+ "provided; use an 'ErrorMessageSendingRecoverer' in the 'recoveryCallback' property to "
+ "send an error message when retries are exhausted");
}
this.messageBuilderFactory = IntegrationUtils.getMessageBuilderFactory(this.beanFactory);
this.evaluationContext = ExpressionUtils.createStandardEvaluationContext(this.beanFactory);
}
@@ -551,21 +594,65 @@ public class ChannelPublishingJmsMessageListener
private class GatewayDelegate extends MessagingGatewaySupport {
private static final ThreadLocal<AttributeAccessor> ATTRIBUTES_HOLDER = new ThreadLocal<>();
@Nullable
private RetryOperations retryTemplate;
@Nullable
private RecoveryCallback<Message<?>> recoveryCallback;
GatewayDelegate() {
setErrorMessageStrategy(new JmsMessageHeaderErrorMessageStrategy());
}
@Override
protected void send(Object request) { // NOSONAR - not useless, increases visibility
super.send(request);
private void send(jakarta.jms.Message jmsMessage, Message<?> requestMessage) {
try {
if (this.retryTemplate == null) {
setAttributesIfNecessary(jmsMessage, requestMessage);
send(requestMessage);
}
else {
this.retryTemplate.execute(
context -> {
StaticMessageHeaderAccessor.getDeliveryAttempt(requestMessage).incrementAndGet();
setAttributesIfNecessary(jmsMessage, requestMessage);
send(requestMessage);
return null;
}, this.recoveryCallback);
}
}
finally {
if (this.retryTemplate == null) {
ATTRIBUTES_HOLDER.remove();
}
}
}
@Override
protected Message<?> sendAndReceiveMessage(Object request) { // NOSONAR - not useless, increases visibility
return super.sendAndReceiveMessage(request);
private Message<?> sendAndReceiveMessage(jakarta.jms.Message jmsMessage, Message<?> requestMessage) {
try {
if (this.retryTemplate == null) {
setAttributesIfNecessary(jmsMessage, requestMessage);
return sendAndReceiveMessage(requestMessage);
}
else {
return this.retryTemplate.execute(
context -> {
StaticMessageHeaderAccessor.getDeliveryAttempt(requestMessage).incrementAndGet();
setAttributesIfNecessary(jmsMessage, requestMessage);
return sendAndReceiveMessage(requestMessage);
}, this.recoveryCallback);
}
}
finally {
if (this.retryTemplate == null) {
ATTRIBUTES_HOLDER.remove();
}
}
}
protected ErrorMessage buildErrorMessage(Throwable throwable) {
return super.buildErrorMessage(null, throwable);
return buildErrorMessage(null, throwable);
}
protected MessagingTemplate getMessagingTemplate() {
@@ -582,6 +669,29 @@ public class ChannelPublishingJmsMessageListener
}
}
@Override
protected AttributeAccessor getErrorMessageAttributes(@Nullable Message<?> message) {
AttributeAccessor attributes = ATTRIBUTES_HOLDER.get();
return (attributes != null) ? attributes : super.getErrorMessageAttributes(message);
}
private void setAttributesIfNecessary(Object jmsMessage, Message<?> message) {
boolean needHolder = getErrorChannel() != null && this.retryTemplate == null;
boolean needAttributes = needHolder || this.retryTemplate != null;
if (needHolder) {
ATTRIBUTES_HOLDER.set(ErrorMessageUtils.getAttributeAccessor(null, null));
}
if (needAttributes) {
AttributeAccessor attributes =
this.retryTemplate != null
? RetrySynchronizationManager.getContext()
: ATTRIBUTES_HOLDER.get();
if (attributes != null) {
attributes.setAttribute(ErrorMessageUtils.INPUT_MESSAGE_CONTEXT_KEY, message);
attributes.setAttribute(JmsMessageHeaderErrorMessageStrategy.JMS_RAW_MESSAGE, jmsMessage);
}
}
}
}
}

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

@@ -32,6 +32,8 @@ import org.springframework.integration.util.CheckedFunction;
import org.springframework.jms.listener.AbstractMessageListenerContainer;
import org.springframework.jms.support.converter.MessageConverter;
import org.springframework.jms.support.destination.DestinationResolver;
import org.springframework.retry.RecoveryCallback;
import org.springframework.retry.support.RetryTemplate;
import org.springframework.util.Assert;
/**
@@ -221,9 +223,35 @@ public class JmsInboundGatewaySpec<S extends JmsInboundGatewaySpec<S>>
}
/**
* Set to false to prevent listener container shutdown when the endpoint is stopped.
* Set a {@link RetryTemplate} to use for retrying a message delivery within the
* adapter. Unlike adding retry at the container level, this can be used with an
* {@code ErrorMessageSendingRecoverer} {@link RecoveryCallback} to publish to the
* error channel after retries are exhausted. You generally should not configure an
* error channel when using retry here, use a {@link RecoveryCallback} instead.
* @param retryTemplate the template.
* @since 6.3
* @see #recoveryCallback(RecoveryCallback)
*/
public S retryTemplate(RetryTemplate retryTemplate) {
this.target.getListener().setRetryTemplate(retryTemplate);
return _this();
}
/**
* Set a {@link RecoveryCallback} when using retry within the adapter.
* @param recoveryCallback the callback.
* @since 6.3
* @see #retryTemplate(RetryTemplate)
*/
public S recoveryCallback(RecoveryCallback<org.springframework.messaging.Message<?>> recoveryCallback) {
this.target.getListener().setRecoveryCallback(recoveryCallback);
return _this();
}
/**
* 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 shutdown false to not shutdown.
* @return the spec.

31
spring-integration-jms/src/main/java/org/springframework/integration/jms/dsl/JmsMessageDrivenChannelAdapterSpec.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2016-2023 the original author or authors.
* Copyright 2016-2024 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.
@@ -29,6 +29,9 @@ import org.springframework.integration.jms.JmsHeaderMapper;
import org.springframework.integration.jms.JmsMessageDrivenEndpoint;
import org.springframework.jms.listener.AbstractMessageListenerContainer;
import org.springframework.jms.support.converter.MessageConverter;
import org.springframework.messaging.Message;
import org.springframework.retry.RecoveryCallback;
import org.springframework.retry.support.RetryTemplate;
import org.springframework.util.Assert;
/**
@@ -93,6 +96,32 @@ public class JmsMessageDrivenChannelAdapterSpec<S extends JmsMessageDrivenChanne
return _this();
}
/**
* Set a {@link RetryTemplate} to use for retrying a message delivery within the
* adapter. Unlike adding retry at the container level, this can be used with an
* {@code ErrorMessageSendingRecoverer} {@link RecoveryCallback} to publish to the
* error channel after retries are exhausted. You generally should not configure an
* error channel when using retry here, use a {@link RecoveryCallback} instead.
* @param retryTemplate the template.
* @since 6.3
* @see #recoveryCallback(RecoveryCallback)
*/
public S retryTemplate(RetryTemplate retryTemplate) {
this.target.getListener().setRetryTemplate(retryTemplate);
return _this();
}
/**
* Set a {@link RecoveryCallback} when using retry within the adapter.
* @param recoveryCallback the callback.
* @since 6.3
* @see #retryTemplate(RetryTemplate)
*/
public S recoveryCallback(RecoveryCallback<Message<?>> recoveryCallback) {
this.target.getListener().setRecoveryCallback(recoveryCallback);
return _this();
}
/**
*
* @param <S> the target {@link JmsListenerContainerSpec} implementation type.

63
spring-integration-jms/src/main/java/org/springframework/integration/jms/support/JmsMessageHeaderErrorMessageStrategy.java Normal file
View File

@@ -0,0 +1,63 @@
/*
* Copyright 2024 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.integration.jms.support;
import java.util.HashMap;
import java.util.Map;
import org.springframework.core.AttributeAccessor;
import org.springframework.integration.IntegrationMessageHeaderAccessor;
import org.springframework.integration.support.ErrorMessageStrategy;
import org.springframework.integration.support.ErrorMessageUtils;
import org.springframework.lang.Nullable;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.ErrorMessage;
/**
* An {@link ErrorMessageStrategy} extension that adds the raw JMS message as
* a header to the {@link ErrorMessage}.
*
* @author Artem Bilan
*
* @since 6.3
*
*/
public class JmsMessageHeaderErrorMessageStrategy implements ErrorMessageStrategy {
/**
* Header name/retry context variable for the raw received message.
*/
public static final String JMS_RAW_MESSAGE = "jms_raw_message";
@Override
public ErrorMessage buildErrorMessage(Throwable throwable, @Nullable AttributeAccessor context) {
Object inputMessage = context == null ? null
: context.getAttribute(ErrorMessageUtils.INPUT_MESSAGE_CONTEXT_KEY);
Map<String, Object> headers = new HashMap<>();
if (context != null) {
headers.put(JMS_RAW_MESSAGE, context.getAttribute(JMS_RAW_MESSAGE));
headers.put(IntegrationMessageHeaderAccessor.SOURCE_DATA, context.getAttribute(JMS_RAW_MESSAGE));
}
if (inputMessage instanceof Message) {
return new ErrorMessage(throwable, headers, (Message<?>) inputMessage);
}
else {
return new ErrorMessage(throwable, headers);
}
}
}

6
spring-integration-jms/src/main/java/org/springframework/integration/jms/support/package-info.java Normal file
View File

@@ -0,0 +1,6 @@
/**
* Provides JMS Components support classes.
*/
@org.springframework.lang.NonNullApi
@org.springframework.lang.NonNullFields
package org.springframework.integration.jms.support;

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

@@ -1,5 +1,5 @@
/*
* Copyright 2016-2023 the original author or authors.
* Copyright 2016-2024 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.
@@ -35,7 +35,9 @@ import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.integration.IntegrationMessageHeaderAccessor;
import org.springframework.integration.MessageTimeoutException;
import org.springframework.integration.StaticMessageHeaderAccessor;
import org.springframework.integration.annotation.InboundChannelAdapter;
import org.springframework.integration.annotation.IntegrationComponentScan;
import org.springframework.integration.annotation.MessagingGateway;
@@ -74,6 +76,7 @@ import org.springframework.messaging.PollableChannel;
import org.springframework.messaging.simp.SimpMessageHeaderAccessor;
import org.springframework.messaging.support.ChannelInterceptor;
import org.springframework.messaging.support.InterceptableChannel;
import org.springframework.retry.support.RetryTemplate;
import org.springframework.test.annotation.DirtiesContext;
import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
import org.springframework.transaction.PlatformTransactionManager;
@@ -224,6 +227,8 @@ public class JmsTests extends ActiveMQMultiContextTests {
.extracting(Message::getPayload)
.isEqualTo("foo");
assertThat(StaticMessageHeaderAccessor.getDeliveryAttempt(receive).get()).isEqualTo(3);
assertThat(this.jmsOutboundFlowTemplate).isNotNull();
TestObservationRegistryAssert.assertThat(this.observationRegistry)
@@ -456,7 +461,14 @@ public class JmsTests extends ActiveMQMultiContextTests {
Jms.container(amqFactory, "containerSpecDestination")
.pubSubDomain(false)
.taskExecutor(Executors.newCachedThreadPool()))
.id("observedJmsMessageDrivenChannelAdapter"))
.id("observedJmsMessageDrivenChannelAdapter")
.retryTemplate(new RetryTemplate()))
.handle((p, h) -> {
if (h.get(IntegrationMessageHeaderAccessor.DELIVERY_ATTEMPT, AtomicInteger.class).get() < 3) {
throw new RuntimeException("intentional for retry");
}
return p;
})
.transform(String::trim)
.channel(jmsOutboundInboundReplyChannel())
.get();

7
src/reference/antora/modules/ROOT/pages/jms.adoc
View File

@@ -150,7 +150,7 @@ If you want the entire flow to be transactional (for example, if there is a down
Alternatively, consider using a `jms-message-driven-channel-adapter` with `acknowledge` set to `transacted` (the default).
[[jms-message-driven-channel-adapter]]
== Message-driven Channel Adapter
== Message Driven Channel Adapter
The `message-driven-channel-adapter` requires a reference to either an instance of a Spring `MessageListener` container (any subclass of `AbstractMessageListenerContainer`) or both `ConnectionFactory` and `Destination` (a 'destinationName' can be provided in place of the 'destination' reference).
The following example defines a message-driven channel adapter with a `Destination` reference:
@@ -269,6 +269,9 @@ Starting with version 5.1, when the endpoint is stopped while the application re
Previously, the connection and consumers remained open.
To revert to the previous behavior, set the `shutdownContainerOnStop` on the `JmsMessageDrivenEndpoint` to `false`.
Starting with version 6.3, the `ChannelPublishingJmsMessageListener` can now be supplied with a `RetryTemplate` and `RecoveryCallback<Message<?>>` for retries on the downstream send and send-and-receive operations.
These options are also exposed into a `JmsMessageDrivenChannelAdapterSpec` for Java DSL.
[[jms-md-conversion-errors]]
=== Inbound Conversion Errors
@@ -469,6 +472,8 @@ public IntegrationFlow jmsInboundGatewayFlow(ConnectionFactory connectionFactory
}
----
Starting with version 6.3, the `Jms.inboundGateway()` API exposes a `retryTemplate()` and `recoveryCallback()` options for retrying internal send-and-receive operations.
[[jms-outbound-gateway]]
== Outbound Gateway

8
src/reference/antora/modules/ROOT/pages/whats-new.adoc
View File

@@ -42,4 +42,10 @@ See xref:mqtt.adoc[MQTT Support] for more information.
=== Testing Support Changes
The `MockIntegrationContext.substituteTriggerFor()` API has been introduced.
See xref:testing.adoc[Testing Support] for more information.
See xref:testing.adoc[Testing Support] for more information.
[[x6.3-jms]]
=== JMS Support Changes
The `ChannelPublishingJmsMessageListener` can now be supplied with a `RetryTemplate` and `RecoveryCallback<Message<?>>` for retries on the downstream send operations.
See xref:jms.adoc#jms-message-driven-channel-adapter[Message Driven Channel Adapter] for more information.