SpringCloud/spring-cloud-stream
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

GH-180: republishToDlq and ImmediateAckAmqpEx

Browse Source 
Resolves https://github.com/spring-cloud/spring-cloud-stream-binder-rabbit/issues/180
This commit is contained in:
Gary Russell
2018-10-09 13:08:37 -04:00
committed by Soby Chacko
parent d4fbad5672
commit a69d2309eb
3 changed files with 48 additions and 25 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
spring-cloud-stream-binder-rabbit-docs/src/main/asciidoc/overview.adoc
View File

@@ -51,6 +51,9 @@ You can republish a failed message after just one attempt.
Starting with version 1.2, you can configure the delivery mode of republished messages.
See the <<spring-cloud-stream-rabbit-republish-delivery-mode,`republishDeliveryMode` property>>.
If the stream listener throws an `ImmediateAcknowledgeAmqpException`, the DLQ is bypassed and the message simply discarded.
Starting with version 2.1, this is true regardless of the setting of `republishToDlq`; previously it was only the case when `republishToDlq` was `false`.
IMPORTANT: Setting `requeueRejected` to `true` (with `republishToDlq=false` ) causes the message to be re-queued and redelivered continually, which is likely not what you want unless the reason for the failure is transient.
In general, you should enable retry within the binder by setting `maxAttempts` to greater than one or by setting `republishToDlq` to `true`.

23
spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitMessageChannelBinder.java
View File

@@ -25,6 +25,7 @@ import java.util.Map;
import java.util.concurrent.atomic.AtomicInteger;
import org.springframework.amqp.AmqpRejectAndDontRequeueException;
import org.springframework.amqp.ImmediateAcknowledgeAmqpException;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessagePostProcessor;
import org.springframework.amqp.core.MessageProperties;
@@ -507,6 +508,12 @@ public class RabbitMessageChannelBinder
}
else {
Throwable cause = (Throwable) message.getPayload();
if (!shouldRepublish(cause)) {
if (logger.isDebugEnabled()) {
logger.debug("Skipping republish of: " + message);
}
return;
}
MessageProperties messageProperties = amqpMessage.getMessageProperties();
Map<String, Object> headers = messageProperties.getHeaders();
String stackTraceAsString = getStackTraceAsString(cause);
@@ -539,6 +546,22 @@ public class RabbitMessageChannelBinder
}
}
/**
* Traverse the cause tree, stopping at AmqpRejectAndDontRequeueException
* or ImmediateAcknowledgeAmqpException.
* @param throwable the throwable.
* @return true if neither found or AmqpRejectAndDontRequeueException is
* found first.
*/
private boolean shouldRepublish(Throwable throwable) {
Throwable cause = throwable;
while (cause != null && !(cause instanceof AmqpRejectAndDontRequeueException)
&& !(cause instanceof ImmediateAcknowledgeAmqpException)) {
cause = cause.getCause();
}
return !(cause instanceof ImmediateAcknowledgeAmqpException);
}
};
}
else if (properties.getMaxAttempts() > 1) {

47
spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitBinderTests.java
View File

@@ -36,6 +36,7 @@ import org.junit.rules.TestName;
import org.mockito.ArgumentCaptor;
import org.springframework.amqp.AmqpIOException;
import org.springframework.amqp.ImmediateAcknowledgeAmqpException;
import org.springframework.amqp.core.AcknowledgeMode;
import org.springframework.amqp.core.AmqpTemplate;
import org.springframework.amqp.core.AnonymousQueue;
@@ -1083,10 +1084,14 @@ public class RabbitBinderTests extends
moduleInputChannel.setBeanName("dlqPubTest");
RuntimeException exception = bigCause(new RuntimeException(BIG_EXCEPTION_MESSAGE));
assertThat(getStackTraceAsString(exception).length()).isGreaterThan(this.maxStackTraceSize);
AtomicBoolean dontRepublish = new AtomicBoolean();
moduleInputChannel.subscribe(new MessageHandler() {
@Override
public void handleMessage(Message<?> message) throws MessagingException {
if (dontRepublish.get()) {
throw new ImmediateAcknowledgeAmqpException("testDontRepublish");
}
throw exception;
}
@@ -1098,34 +1103,26 @@ public class RabbitBinderTests extends
RabbitTemplate template = new RabbitTemplate(this.rabbitAvailableRule.getResource());
template.convertAndSend("", TEST_PREFIX + "foo.dlqpubtest.foo", "foo");
int n = 0;
while (n++ < 100) {
org.springframework.amqp.core.Message deadLetter = template.receive(TEST_PREFIX + "foo.dlqpubtest.foo.dlq");
if (deadLetter != null) {
assertThat(new String(deadLetter.getBody())).isEqualTo("foo");
assertThat(deadLetter.getMessageProperties().getHeaders())
.containsKey((RepublishMessageRecoverer.X_EXCEPTION_STACKTRACE));
assertThat(((LongString) deadLetter.getMessageProperties().getHeaders()
.get(RepublishMessageRecoverer.X_EXCEPTION_STACKTRACE)).length()).isEqualTo(this.maxStackTraceSize);
break;
}
Thread.sleep(100);
}
assertThat(n).isLessThan(100);
template.setReceiveTimeout(10_000);
org.springframework.amqp.core.Message deadLetter = template.receive(TEST_PREFIX + "foo.dlqpubtest.foo.dlq");
assertThat(deadLetter).isNotNull();
assertThat(new String(deadLetter.getBody())).isEqualTo("foo");
assertThat(deadLetter.getMessageProperties().getHeaders())
.containsKey((RepublishMessageRecoverer.X_EXCEPTION_STACKTRACE));
assertThat(((LongString) deadLetter.getMessageProperties().getHeaders()
.get(RepublishMessageRecoverer.X_EXCEPTION_STACKTRACE)).length()).isEqualTo(this.maxStackTraceSize);
template.convertAndSend("", TEST_PREFIX + "foo.dlqpubtest2.foo", "bar");
n = 0;
while (n++ < 100) {
org.springframework.amqp.core.Message deadLetter = template.receive(TEST_PREFIX + "foo.dlqpubtest2.foo.dlq");
if (deadLetter != null) {
assertThat(new String(deadLetter.getBody())).isEqualTo("bar");
assertThat(deadLetter.getMessageProperties().getHeaders()).containsKey(("x-exception-stacktrace"));
break;
}
Thread.sleep(100);
}
assertThat(n).isLessThan(100);
deadLetter = template.receive(TEST_PREFIX + "foo.dlqpubtest2.foo.dlq");
assertThat(deadLetter).isNotNull();
assertThat(new String(deadLetter.getBody())).isEqualTo("bar");
assertThat(deadLetter.getMessageProperties().getHeaders()).containsKey(("x-exception-stacktrace"));
dontRepublish.set(true);
template.convertAndSend("", TEST_PREFIX + "foo.dlqpubtest2.foo", "baz");
template.setReceiveTimeout(500);
assertThat(template.receive(TEST_PREFIX + "foo.dlqpubtest2.foo.dlq")).isNull();
consumerBinding.unbind();
}