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 56baf86be1..ecd7e1ec5d 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(new RetryCallback() { @@ -222,9 +227,6 @@ public class AmqpInboundChannelAdapter extends MessageProducerSupport implements throw e; } } - finally { - attributesHolder.remove(); - } } private void processMessage(Message message, Channel channel) { @@ -246,14 +248,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 be06eb6c6e..e9a0b42c84 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 @@ -255,7 +255,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(new RetryCallback() { @@ -354,14 +359,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