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

GH-77: Send to DLQ Via ErrorChannel

Browse Source 
Resolves spring-cloud/spring-cloud-stream-binder-rabbit#77

Relates to spring-cloud/spring-cloud-stream#913

This PR allows user code to subscribe to a specific error channel (`<destination>.errors`) or the
global Spring Integration `errorChannel` to receive a copy of messages that fail, whether or
not a DLQ is configured.

Polishing - PR Comments; also add EM Strategy to adapter.

Override error naming - destination already contains group.

* Polishing according PR comments
* Fix `RabbitBinderTests` to use bean names without extra `group`
* Fix deprecation warnings
This commit is contained in:
Gary Russell
2017-06-06 14:41:16 -04:00
committed by Artem Bilan
parent df33bd0126
commit 4f1d975c10
5 changed files with 188 additions and 55 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
spring-cloud-stream-binder-rabbit-core/src/main/java/org/springframework/cloud/stream/binder/rabbit/provisioning/RabbitExchangeQueueProvisioner.java
View File

@@ -391,9 +391,7 @@ public class RabbitExchangeQueueProvisioner implements ProvisioningProvider<Exte
private Exchange buildExchange(RabbitCommonProperties properties, String exchangeName) {
try {
ExchangeBuilder builder = new ExchangeBuilder(exchangeName, properties.getExchangeType());
if (properties.isExchangeDurable()) {
builder.durable();
}
builder.durable(properties.isExchangeDurable());
if (properties.isExchangeAutoDelete()) {
builder.autoDelete();
}

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

@@ -16,14 +16,17 @@
package org.springframework.cloud.stream.binder.rabbit;
import java.io.PrintWriter;
import java.io.StringWriter;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.Map;
import org.springframework.amqp.AmqpRejectAndDontRequeueException;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessagePostProcessor;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.rabbit.config.RetryInterceptorBuilder;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.connection.LocalizedQueueConnectionFactory;
import org.springframework.amqp.rabbit.core.BatchingRabbitTemplate;
@@ -31,7 +34,7 @@ import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.core.support.BatchingStrategy;
import org.springframework.amqp.rabbit.core.support.SimpleBatchingStrategy;
import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
import org.springframework.amqp.rabbit.retry.MessageRecoverer;
import org.springframework.amqp.rabbit.listener.exception.ListenerExecutionFailedException;
import org.springframework.amqp.rabbit.retry.RejectAndDontRequeueRecoverer;
import org.springframework.amqp.rabbit.retry.RepublishMessageRecoverer;
import org.springframework.amqp.rabbit.support.DefaultMessagePropertiesConverter;
@@ -54,12 +57,15 @@ import org.springframework.cloud.stream.provisioning.ProducerDestination;
import org.springframework.core.task.SimpleAsyncTaskExecutor;
import org.springframework.integration.amqp.inbound.AmqpInboundChannelAdapter;
import org.springframework.integration.amqp.outbound.AmqpOutboundEndpoint;
import org.springframework.integration.amqp.support.AmqpMessageHeaderErrorMessageStrategy;
import org.springframework.integration.amqp.support.DefaultAmqpHeaderMapper;
import org.springframework.integration.context.IntegrationContextUtils;
import org.springframework.integration.core.MessageProducer;
import org.springframework.integration.support.ErrorMessageStrategy;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHandler;
import org.springframework.retry.interceptor.RetryOperationsInterceptor;
import org.springframework.messaging.MessagingException;
import org.springframework.messaging.support.ErrorMessage;
import org.springframework.scheduling.TaskScheduler;
import org.springframework.util.Assert;
import org.springframework.util.StringUtils;
@@ -82,6 +88,9 @@ public class RabbitMessageChannelBinder
ExtendedProducerProperties<RabbitProducerProperties>, RabbitExchangeQueueProvisioner>
implements ExtendedPropertiesBinder<MessageChannel, RabbitConsumerProperties, RabbitProducerProperties> {
private static final AmqpMessageHeaderErrorMessageStrategy errorMessageStrategy =
new AmqpMessageHeaderErrorMessageStrategy();
private static final MessagePropertiesConverter inboundMessagePropertiesConverter =
new DefaultMessagePropertiesConverter() {
@@ -246,13 +255,6 @@ public class RabbitMessageChannelBinder
listenerContainer.setTxSize(properties.getExtension().getTxSize());
listenerContainer.setTaskExecutor(new SimpleAsyncTaskExecutor(consumerDestination.getName() + "-"));
listenerContainer.setQueueNames(consumerDestination.getName());
if (properties.getMaxAttempts() > 1 || properties.getExtension().isRepublishToDlq()) {
RetryOperationsInterceptor retryInterceptor = RetryInterceptorBuilder.stateless()
.retryOperations(buildRetryTemplate(properties))
.recoverer(determineRecoverer(destination, properties.getExtension()))
.build();
listenerContainer.setAdviceChain(retryInterceptor);
}
listenerContainer.setAfterReceivePostProcessors(this.decompressingPostProcessor);
listenerContainer.setMessagePropertiesConverter(RabbitMessageChannelBinder.inboundMessagePropertiesConverter);
listenerContainer.afterPropertiesSet();
@@ -263,13 +265,112 @@ public class RabbitMessageChannelBinder
DefaultAmqpHeaderMapper mapper = DefaultAmqpHeaderMapper.inboundMapper();
mapper.setRequestHeaderNames(properties.getExtension().getHeaderPatterns());
adapter.setHeaderMapper(mapper);
adapter.afterPropertiesSet();
ErrorInfrastructure errorInfrastructure = registerErrorInfrastructure(consumerDestination, group, properties);
if (properties.getMaxAttempts() > 1) {
adapter.setRetryTemplate(buildRetryTemplate(properties));
if (properties.getExtension().isRepublishToDlq()) {
adapter.setRecoveryCallback(errorInfrastructure.getRecoverer());
}
}
else {
adapter.setErrorMessageStrategy(errorMessageStrategy);
adapter.setErrorChannel(errorInfrastructure.getErrorChannel());
}
return adapter;
}
@Override
protected ErrorMessageStrategy getErrorMessageStrategy() {
return errorMessageStrategy;
}
@Override
protected MessageHandler getErrorMessageHandler(ConsumerDestination destination, String group,
final ExtendedConsumerProperties<RabbitConsumerProperties> properties) {
if (properties.getExtension().isRepublishToDlq()) {
return new MessageHandler() {
private final RabbitTemplate template = new RabbitTemplate(
RabbitMessageChannelBinder.this.connectionFactory);
private final String exchange = deadLetterExchangeName(properties.getExtension());
private final String routingKey = properties.getExtension().getDeadLetterRoutingKey();
@Override
public void handleMessage(org.springframework.messaging.Message<?> message) throws MessagingException {
Message amqpMessage = (Message) message.getHeaders()
.get(AmqpMessageHeaderErrorMessageStrategy.AMQP_RAW_MESSAGE);
if (!(message instanceof ErrorMessage)) {
logger.error("Expected an ErrorMessage, not a " + message.getClass().toString() + " for: "
+ message);
}
else if (amqpMessage == null) {
logger.error("No raw message header in " + message);
}
else {
Throwable cause = (Throwable) message.getPayload();
MessageProperties messageProperties = amqpMessage.getMessageProperties();
Map<String, Object> headers = messageProperties.getHeaders();
headers.put(RepublishMessageRecoverer.X_EXCEPTION_STACKTRACE, getStackTraceAsString(cause));
headers.put(RepublishMessageRecoverer.X_EXCEPTION_MESSAGE,
cause.getCause() != null ? cause.getCause().getMessage() : cause.getMessage());
headers.put(RepublishMessageRecoverer.X_ORIGINAL_EXCHANGE,
messageProperties.getReceivedExchange());
headers.put(RepublishMessageRecoverer.X_ORIGINAL_ROUTING_KEY,
messageProperties.getReceivedRoutingKey());
if (properties.getExtension().getRepublishDeliveyMode() != null) {
messageProperties.setDeliveryMode(properties.getExtension().getRepublishDeliveyMode());
}
template.send(this.exchange,
this.routingKey != null ? this.routingKey : messageProperties.getConsumerQueue(),
amqpMessage);
}
}
};
}
else if (properties.getMaxAttempts() > 1) {
return new MessageHandler() {
private final RejectAndDontRequeueRecoverer recoverer = new RejectAndDontRequeueRecoverer();
@Override
public void handleMessage(org.springframework.messaging.Message<?> message) throws MessagingException {
Message amqpMessage = (Message) message.getHeaders()
.get(AmqpMessageHeaderErrorMessageStrategy.AMQP_RAW_MESSAGE);
if (!(message instanceof ErrorMessage)) {
logger.error("Expected an ErrorMessage, not a " + message.getClass().toString() + " for: "
+ message);
throw new ListenerExecutionFailedException("Unexpected error message " + message,
new AmqpRejectAndDontRequeueException(""), null);
}
else if (amqpMessage == null) {
logger.error("No raw message header in " + message);
throw new ListenerExecutionFailedException("Unexpected error message " + message,
new AmqpRejectAndDontRequeueException(""), amqpMessage);
}
else {
this.recoverer.recover(amqpMessage, (Throwable) message.getPayload());
}
}
};
}
else {
return super.getErrorMessageHandler(destination, group, properties);
}
}
@Override
protected String errorsBaseName(ConsumerDestination destination, String group,
ExtendedConsumerProperties<RabbitConsumerProperties> consumerProperties) {
return destination.getName() + ".errors";
}
private String deadLetterExchangeName(RabbitCommonProperties properties) {
if (properties.getDeadLetterExchange() == null) {
return properties.getPrefix() + RabbitCommonProperties.DEAD_LETTER_EXCHANGE;
return applyPrefix(properties.getPrefix(), RabbitCommonProperties.DEAD_LETTER_EXCHANGE);
}
else {
return properties.getDeadLetterExchange();
@@ -282,29 +383,6 @@ public class RabbitMessageChannelBinder
provisioningProvider.cleanAutoDeclareContext(consumerDestination.getName());
}
private MessageRecoverer determineRecoverer(String name, final RabbitConsumerProperties properties) {
if (properties.isRepublishToDlq()) {
RabbitTemplate errorTemplate = new RabbitTemplate(this.connectionFactory);
if (properties.getRepublishDeliveyMode() != null) {
return new RepublishMessageRecoverer(errorTemplate, deadLetterExchangeName(properties), name) {
@Override
public void recover(Message message, Throwable cause) {
message.getMessageProperties().setDeliveryMode(properties.getRepublishDeliveyMode());
super.recover(message, cause);
}
};
}
else {
return new RepublishMessageRecoverer(errorTemplate, deadLetterExchangeName(properties), name);
}
}
else {
return new RejectAndDontRequeueRecoverer();
}
}
private RabbitTemplate buildRabbitTemplate(RabbitProducerProperties properties) {
RabbitTemplate rabbitTemplate;
if (properties.isBatchingEnabled()) {
@@ -328,4 +406,11 @@ public class RabbitMessageChannelBinder
return rabbitTemplate;
}
private String getStackTraceAsString(Throwable cause) {
StringWriter stringWriter = new StringWriter();
PrintWriter printWriter = new PrintWriter(stringWriter, true);
cause.printStackTrace(printWriter);
return stringWriter.getBuffer().toString();
}
}

2
spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/config/RabbitMessageChannelBinderConfiguration.java
View File

@@ -21,8 +21,8 @@ import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.support.postprocessor.DelegatingDecompressingPostProcessor;
import org.springframework.amqp.support.postprocessor.GZipPostProcessor;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.PropertyPlaceholderAutoConfiguration;
import org.springframework.boot.autoconfigure.amqp.RabbitProperties;
import org.springframework.boot.autoconfigure.context.PropertyPlaceholderAutoConfiguration;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.cloud.stream.binder.rabbit.RabbitMessageChannelBinder;
import org.springframework.cloud.stream.binder.rabbit.properties.RabbitBinderConfigurationProperties;

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

@@ -21,14 +21,16 @@ import static org.mockito.Mockito.spy;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicReference;
import java.util.zip.Deflater;
import org.aopalliance.aop.Advice;
import org.apache.commons.logging.Log;
import org.junit.Rule;
import org.junit.Test;
@@ -77,7 +79,9 @@ import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHandler;
import org.springframework.messaging.MessageHeaders;
import org.springframework.messaging.MessagingException;
import org.springframework.messaging.SubscribableChannel;
import org.springframework.messaging.support.GenericMessage;
import org.springframework.retry.support.RetryTemplate;
import com.rabbitmq.http.client.domain.QueueInfo;
@@ -85,6 +89,7 @@ import com.rabbitmq.http.client.domain.QueueInfo;
* @author Mark Fisher
* @author Gary Russell
* @author David Turanski
* @author Artem Bilan
*/
public class RabbitBinderTests extends
PartitionCapableBinderTests<RabbitTestBinder, ExtendedConsumerProperties<RabbitConsumerProperties>, ExtendedProducerProperties<RabbitProducerProperties>> {
@@ -172,11 +177,11 @@ public class RabbitBinderTests extends
assertThat(TestUtils.getPropertyValue(container, "defaultRequeueRejected", Boolean.class)).isTrue();
assertThat(TestUtils.getPropertyValue(container, "prefetchCount")).isEqualTo(1);
assertThat(TestUtils.getPropertyValue(container, "txSize")).isEqualTo(1);
Advice retry = TestUtils.getPropertyValue(container, "adviceChain", Advice[].class)[0];
assertThat(TestUtils.getPropertyValue(retry, "retryOperations.retryPolicy.maxAttempts")).isEqualTo(3);
assertThat(TestUtils.getPropertyValue(retry, "retryOperations.backOffPolicy.initialInterval")).isEqualTo(1000L);
assertThat(TestUtils.getPropertyValue(retry, "retryOperations.backOffPolicy.maxInterval")).isEqualTo(10000L);
assertThat(TestUtils.getPropertyValue(retry, "retryOperations.backOffPolicy.multiplier")).isEqualTo(2.0);
RetryTemplate retry = TestUtils.getPropertyValue(endpoint, "retryTemplate", RetryTemplate.class);
assertThat(TestUtils.getPropertyValue(retry, "retryPolicy.maxAttempts")).isEqualTo(3);
assertThat(TestUtils.getPropertyValue(retry, "backOffPolicy.initialInterval")).isEqualTo(1000L);
assertThat(TestUtils.getPropertyValue(retry, "backOffPolicy.maxInterval")).isEqualTo(10000L);
assertThat(TestUtils.getPropertyValue(retry, "backOffPolicy.multiplier")).isEqualTo(2.0);
consumerBinding.unbind();
assertThat(endpoint.isRunning()).isFalse();
@@ -190,7 +195,7 @@ public class RabbitBinderTests extends
properties.getExtension().setMaxConcurrency(3);
properties.getExtension().setPrefix("foo.");
properties.getExtension().setPrefetch(20);
properties.getExtension().setRequestHeaderPatterns(new String[] { "foo" });
properties.getExtension().setHeaderPatterns(new String[] { "foo" });
properties.getExtension().setTxSize(10);
properties.setInstanceIndex(0);
consumerBinding = binder.bindConsumer("props.0", "test", createBindableChannel("input", new BindingProperties()),
@@ -388,7 +393,7 @@ public class RabbitBinderTests extends
ExtendedProducerProperties<RabbitProducerProperties> producerProperties = createProducerProperties();
producerProperties.getExtension().setPrefix("foo.");
producerProperties.getExtension().setDeliveryMode(MessageDeliveryMode.NON_PERSISTENT);
producerProperties.getExtension().setRequestHeaderPatterns(new String[] { "foo" });
producerProperties.getExtension().setHeaderPatterns(new String[] { "foo" });
producerProperties.setPartitionKeyExpression(spelExpressionParser.parseExpression("'foo'"));
producerProperties.setPartitionKeyExtractorClass(TestPartitionKeyExtractorClass.class);
producerProperties.setPartitionSelectorExpression(spelExpressionParser.parseExpression("0"));
@@ -630,14 +635,23 @@ public class RabbitBinderTests extends
}
@Test
public void testAutoBindDLQPartionedConsumerFirstWithRepublish() throws Exception {
public void testAutoBindDLQPartionedConsumerFirstWithRepublishNoRetry() throws Exception {
testAutoBindDLQPartionedConsumerFirstWithRepublishGuts(false);
}
@Test
public void testAutoBindDLQPartionedConsumerFirstWithRepublishWithRetry() throws Exception {
testAutoBindDLQPartionedConsumerFirstWithRepublishGuts(true);
}
private void testAutoBindDLQPartionedConsumerFirstWithRepublishGuts(final boolean withRetry) throws Exception {
RabbitTestBinder binder = getBinder();
ExtendedConsumerProperties<RabbitConsumerProperties> properties = createConsumerProperties();
properties.getExtension().setPrefix("bindertest.");
properties.getExtension().setAutoBindDlq(true);
properties.getExtension().setRepublishToDlq(true);
properties.getExtension().setRepublishDeliveyMode(MessageDeliveryMode.NON_PERSISTENT);
properties.setMaxAttempts(1); // disable retry
properties.setMaxAttempts(withRetry ? 2 : 1);
properties.setPartitioned(true);
properties.setInstanceIndex(0);
DirectChannel input0 = createBindableChannel("input", createConsumerBindingProperties(properties));
@@ -689,6 +703,33 @@ public class RabbitBinderTests extends
});
ApplicationContext context = TestUtils.getPropertyValue(binder.getBinder(), "applicationContext",
ApplicationContext.class);
SubscribableChannel boundErrorChannel = context
.getBean("bindertest.partPubDLQ.0.dlqPartGrp-0.errors", SubscribableChannel.class);
SubscribableChannel globalErrorChannel = context.getBean("errorChannel", SubscribableChannel.class);
final AtomicReference<Message<?>> boundErrorChannelMessage = new AtomicReference<>();
final AtomicReference<Message<?>> globalErrorChannelMessage = new AtomicReference<>();
final AtomicBoolean hasRecovererInCallStack = new AtomicBoolean(!withRetry);
boundErrorChannel.subscribe(new MessageHandler() {
@Override
public void handleMessage(Message<?> message) throws MessagingException {
boundErrorChannelMessage.set(message);
String stackTrace = Arrays.toString(new RuntimeException().getStackTrace());
hasRecovererInCallStack.set(stackTrace.contains("ErrorMessageSendingRecoverer"));
}
});
globalErrorChannel.subscribe(new MessageHandler() {
@Override
public void handleMessage(Message<?> message) throws MessagingException {
globalErrorChannelMessage.set(message);
}
});
output.send(new GenericMessage<>(1));
assertThat(latch1.await(10, TimeUnit.SECONDS)).isTrue();
@@ -717,6 +758,11 @@ public class RabbitBinderTests extends
.isEqualTo("partPubDLQ.0-0");
assertThat(received.getMessageProperties().getHeaders()).doesNotContainKey(BinderHeaders.PARTITION_HEADER);
// verify we got a message on the dedicated error channel and the global (via bridge)
assertThat(boundErrorChannelMessage.get()).isNotNull();
assertThat(globalErrorChannelMessage.get()).isNotNull();
assertThat(hasRecovererInCallStack.get()).isEqualTo(withRetry);
input0Binding.unbind();
input1Binding.unbind();
defaultConsumerBinding1.unbind();
@@ -1045,7 +1091,7 @@ public class RabbitBinderTests extends
private SimpleMessageListenerContainer verifyContainer(Lifecycle endpoint) {
SimpleMessageListenerContainer container;
Advice retry;
RetryTemplate retry;
container = TestUtils.getPropertyValue(endpoint, "messageListenerContainer",
SimpleMessageListenerContainer.class);
assertThat(container.getAcknowledgeMode()).isEqualTo(AcknowledgeMode.NONE);
@@ -1056,11 +1102,11 @@ public class RabbitBinderTests extends
assertThat(TestUtils.getPropertyValue(container, "defaultRequeueRejected", Boolean.class)).isFalse();
assertThat(TestUtils.getPropertyValue(container, "prefetchCount")).isEqualTo(20);
assertThat(TestUtils.getPropertyValue(container, "txSize")).isEqualTo(10);
retry = TestUtils.getPropertyValue(container, "adviceChain", Advice[].class)[0];
assertThat(TestUtils.getPropertyValue(retry, "retryOperations.retryPolicy.maxAttempts")).isEqualTo(23);
assertThat(TestUtils.getPropertyValue(retry, "retryOperations.backOffPolicy.initialInterval")).isEqualTo(2000L);
assertThat(TestUtils.getPropertyValue(retry, "retryOperations.backOffPolicy.maxInterval")).isEqualTo(20000L);
assertThat(TestUtils.getPropertyValue(retry, "retryOperations.backOffPolicy.multiplier")).isEqualTo(5.0);
retry = TestUtils.getPropertyValue(endpoint, "retryTemplate", RetryTemplate.class);
assertThat(TestUtils.getPropertyValue(retry, "retryPolicy.maxAttempts")).isEqualTo(23);
assertThat(TestUtils.getPropertyValue(retry, "backOffPolicy.initialInterval")).isEqualTo(2000L);
assertThat(TestUtils.getPropertyValue(retry, "backOffPolicy.maxInterval")).isEqualTo(20000L);
assertThat(TestUtils.getPropertyValue(retry, "backOffPolicy.multiplier")).isEqualTo(5.0);
List<?> requestMatchers = TestUtils.getPropertyValue(endpoint, "headerMapper.requestHeaderMatcher.matchers",
List.class);

4
spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitTestBinder.java
View File

@@ -31,6 +31,7 @@ import org.springframework.cloud.stream.binder.rabbit.properties.RabbitConsumerP
import org.springframework.cloud.stream.binder.rabbit.properties.RabbitProducerProperties;
import org.springframework.cloud.stream.binder.rabbit.provisioning.RabbitExchangeQueueProvisioner;
import org.springframework.context.support.GenericApplicationContext;
import org.springframework.integration.channel.PublishSubscribeChannel;
import org.springframework.integration.codec.kryo.PojoCodec;
import org.springframework.integration.context.IntegrationContextUtils;
import org.springframework.messaging.MessageChannel;
@@ -65,6 +66,9 @@ public class RabbitTestBinder extends AbstractTestBinder<RabbitMessageChannelBin
scheduler.setPoolSize(1);
scheduler.afterPropertiesSet();
context.getBeanFactory().registerSingleton(IntegrationContextUtils.TASK_SCHEDULER_BEAN_NAME, scheduler);
PublishSubscribeChannel errorChannel = new PublishSubscribeChannel();
context.getBeanFactory().initializeBean(errorChannel, IntegrationContextUtils.ERROR_CHANNEL_BEAN_NAME);
context.getBeanFactory().registerSingleton(IntegrationContextUtils.ERROR_CHANNEL_BEAN_NAME, errorChannel);
context.refresh();
binder.setApplicationContext(context);
binder.setCodec(new PojoCodec());