From 4f1d975c109089e9530febf5792d437ad37bedaf Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Tue, 6 Jun 2017 14:41:16 -0400 Subject: [PATCH] GH-77: Send to DLQ Via ErrorChannel 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 (`.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 --- .../RabbitExchangeQueueProvisioner.java | 4 +- .../rabbit/RabbitMessageChannelBinder.java | 155 ++++++++++++++---- ...bbitMessageChannelBinderConfiguration.java | 2 +- .../binder/rabbit/RabbitBinderTests.java | 78 +++++++-- .../binder/rabbit/RabbitTestBinder.java | 4 + 5 files changed, 188 insertions(+), 55 deletions(-) diff --git a/spring-cloud-stream-binder-rabbit-core/src/main/java/org/springframework/cloud/stream/binder/rabbit/provisioning/RabbitExchangeQueueProvisioner.java b/spring-cloud-stream-binder-rabbit-core/src/main/java/org/springframework/cloud/stream/binder/rabbit/provisioning/RabbitExchangeQueueProvisioner.java index 5d22aa7f5..6788f6cdd 100644 --- a/spring-cloud-stream-binder-rabbit-core/src/main/java/org/springframework/cloud/stream/binder/rabbit/provisioning/RabbitExchangeQueueProvisioner.java +++ b/spring-cloud-stream-binder-rabbit-core/src/main/java/org/springframework/cloud/stream/binder/rabbit/provisioning/RabbitExchangeQueueProvisioner.java @@ -391,9 +391,7 @@ public class RabbitExchangeQueueProvisioner implements ProvisioningProvider, RabbitExchangeQueueProvisioner> implements ExtendedPropertiesBinder { + 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 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 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 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(); + } + } diff --git a/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/config/RabbitMessageChannelBinderConfiguration.java b/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/config/RabbitMessageChannelBinderConfiguration.java index d2b4f543e..b61f9b49a 100644 --- a/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/config/RabbitMessageChannelBinderConfiguration.java +++ b/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/config/RabbitMessageChannelBinderConfiguration.java @@ -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; diff --git a/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitBinderTests.java b/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitBinderTests.java index 6ef71be3b..e0885e33f 100644 --- a/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitBinderTests.java +++ b/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitBinderTests.java @@ -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, ExtendedProducerProperties> { @@ -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 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 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> boundErrorChannelMessage = new AtomicReference<>(); + final AtomicReference> 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); diff --git a/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitTestBinder.java b/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitTestBinder.java index 82a952f11..22af545b7 100644 --- a/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitTestBinder.java +++ b/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitTestBinder.java @@ -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