diff --git a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/inbound/AmqpInboundChannelAdapter.java b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/inbound/AmqpInboundChannelAdapter.java index fdf5cd44a7..cc33b21969 100644 --- a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/inbound/AmqpInboundChannelAdapter.java +++ b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/inbound/AmqpInboundChannelAdapter.java @@ -199,7 +199,12 @@ public class AmqpInboundChannelAdapter extends MessageProducerSupport implements public void onMessage(final Message message, final Channel channel) throws Exception { try { if (AmqpInboundChannelAdapter.this.retryTemplate == null) { - processMessage(message, channel); + try { + processMessage(message, channel); + } + finally { + attributesHolder.remove(); + } } else { AmqpInboundChannelAdapter.this.retryTemplate.execute(context -> { @@ -218,9 +223,6 @@ public class AmqpInboundChannelAdapter extends MessageProducerSupport implements throw e; } } - finally { - attributesHolder.remove(); - } } private void processMessage(Message message, Channel channel) { @@ -242,14 +244,16 @@ public class AmqpInboundChannelAdapter extends MessageProducerSupport implements @Override public boolean open(RetryContext context, RetryCallback callback) { - attributesHolder.set(context); + if (AmqpInboundChannelAdapter.this.recoveryCallback != null) { + attributesHolder.set(context); + } return true; } @Override public void close(RetryContext context, RetryCallback callback, Throwable throwable) { - // Empty + attributesHolder.remove(); } @Override diff --git a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/inbound/AmqpInboundGateway.java b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/inbound/AmqpInboundGateway.java index 112b4f3929..5b158fa445 100644 --- a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/inbound/AmqpInboundGateway.java +++ b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/inbound/AmqpInboundGateway.java @@ -254,7 +254,12 @@ public class AmqpInboundGateway extends MessagingGatewaySupport { @Override public void onMessage(final Message message, final Channel channel) throws Exception { if (AmqpInboundGateway.this.retryTemplate == null) { - doOnMessage(message, channel); + try { + doOnMessage(message, channel); + } + finally { + attributesHolder.remove(); + } } else { AmqpInboundGateway.this.retryTemplate.execute(context -> { @@ -346,14 +351,16 @@ public class AmqpInboundGateway extends MessagingGatewaySupport { @Override public boolean open(RetryContext context, RetryCallback callback) { - attributesHolder.set(context); + if (AmqpInboundGateway.this.recoveryCallback != null) { + attributesHolder.set(context); + } return true; } @Override public void close(RetryContext context, RetryCallback callback, Throwable throwable) { - // Empty + attributesHolder.remove(); } @Override