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

SCSt-913: Fix AMQP Inbound Retry logic

Browse Source 
Relates to spring-cloud/spring-cloud-stream#913
and 914094e51d

Also see https://github.com/spring-projects/spring-integration-kafka/pull/163

The `RetryTemplate` and `errorChannel` are mutually exclusive options:
```
Assert.state(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");
```
Therefore we don't have any error handling functionality if there is a retry but no `RecoveryCallback`.
So, don't add `RetryContext` to `ThreadLocal<AttributeAccessor>` if we don't have `RecoveryCallback` provided,
because that value is out of use then.

To make code more logical perform `attributesHolder.remove()` only for the non-retryable branch.
With that refactoring the missed `attributesHolder.remove()` has been observed in the `AmqpInboundGateway`
Add `attributesHolder.remove()` to the `RetryListener.close()` for retryable branch

**Cherry-pick to 4.3.x**
This commit is contained in:
Artem Bilan
2017-06-06 14:29:59 -04:00
committed by Gary Russell
parent edb9ea142d
commit 4f83a953bf
2 changed files with 20 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files

16
spring-integration-amqp/src/main/java/org/springframework/integration/amqp/inbound/AmqpInboundChannelAdapter.java
View File

@@ -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<Object, RuntimeException>() {
@@ -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 <T, E extends Throwable> boolean open(RetryContext context, RetryCallback<T, E> callback) {
attributesHolder.set(context);
if (AmqpInboundChannelAdapter.this.recoveryCallback != null) {
attributesHolder.set(context);
}
return true;
}
@Override
public <T, E extends Throwable> void close(RetryContext context, RetryCallback<T, E> callback,
Throwable throwable) {
// Empty
attributesHolder.remove();
}
@Override

13
spring-integration-amqp/src/main/java/org/springframework/integration/amqp/inbound/AmqpInboundGateway.java
View File

@@ -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<Object, RuntimeException>() {
@@ -354,14 +359,16 @@ public class AmqpInboundGateway extends MessagingGatewaySupport {
@Override
public <T, E extends Throwable> boolean open(RetryContext context, RetryCallback<T, E> callback) {
attributesHolder.set(context);
if (AmqpInboundGateway.this.recoveryCallback != null) {
attributesHolder.set(context);
}
return true;
}
@Override
public <T, E extends Throwable> void close(RetryContext context, RetryCallback<T, E> callback,
Throwable throwable) {
// Empty
attributesHolder.remove();
}
@Override