diff --git a/docs/src/main/asciidoc/overview.adoc b/docs/src/main/asciidoc/overview.adoc index 946d40db6..1ff4d4d2e 100644 --- a/docs/src/main/asciidoc/overview.adoc +++ b/docs/src/main/asciidoc/overview.adoc @@ -57,6 +57,11 @@ Starting with version 2.1, this is true regardless of the setting of `republishT 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`. +Starting with version 3.1.2, if the consumer is marked as `transacted`, publishing to the DLQ will participate in the transaction. +This allows the transaction to roll back if the publishing fails for some reason (for example, if the user is not authorized to publish to the dead letter exchange). +In addition, if the connection factory is configured for publisher confirms or returns, the publication to the DLQ will wait for the confirmation and check for a returned message. +If a negative acknowledgment or returned message is received, the binder will throw an `AmqpRejectAndDontRequeueException`, allowing the broker to take care of publishing to the DLQ as if the `republishToDlq` property is `false`. + See <> for more information about these properties. The framework does not provide any standard mechanism to consume dead-letter messages (or to re-route them back to the primary queue). diff --git a/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitMessageChannelBinder.java b/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitMessageChannelBinder.java index 5f39495b3..206646921 100644 --- a/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitMessageChannelBinder.java +++ b/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitMessageChannelBinder.java @@ -23,6 +23,9 @@ import java.util.ArrayList; import java.util.Arrays; import java.util.List; import java.util.Map; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; import java.util.concurrent.atomic.AtomicInteger; import java.util.regex.Pattern; @@ -39,7 +42,10 @@ import org.springframework.amqp.core.MessageProperties; import org.springframework.amqp.rabbit.batch.BatchingStrategy; import org.springframework.amqp.rabbit.batch.SimpleBatchingStrategy; import org.springframework.amqp.rabbit.connection.CachingConnectionFactory; +import org.springframework.amqp.rabbit.connection.CachingConnectionFactory.ConfirmType; import org.springframework.amqp.rabbit.connection.ConnectionFactory; +import org.springframework.amqp.rabbit.connection.CorrelationData; +import org.springframework.amqp.rabbit.connection.CorrelationData.Confirm; import org.springframework.amqp.rabbit.connection.LocalizedQueueConnectionFactory; import org.springframework.amqp.rabbit.connection.RabbitUtils; import org.springframework.amqp.rabbit.core.BatchingRabbitTemplate; @@ -618,11 +624,27 @@ public class RabbitMessageChannelBinder extends if (properties.getExtension().isRepublishToDlq()) { return new MessageHandler() { + private static final long ACK_TIMEOUT = 10_000; + private final RabbitTemplate template = new RabbitTemplate( RabbitMessageChannelBinder.this.connectionFactory); + private final ConfirmType confirmType; + + { this.template.setUsePublisherConnection(true); + this.template.setChannelTransacted(properties.getExtension().isTransacted()); + this.template.setMandatory(RabbitMessageChannelBinder.this.connectionFactory.isPublisherReturns()); + if (RabbitMessageChannelBinder.this.connectionFactory.isSimplePublisherConfirms()) { + this.confirmType = ConfirmType.SIMPLE; + } + else if (RabbitMessageChannelBinder.this.connectionFactory.isPublisherConfirms()) { + this.confirmType = ConfirmType.CORRELATED; + } + else { + this.confirmType = ConfirmType.NONE; + } } private final String exchange = deadLetterExchangeName(properties.getExtension()); @@ -694,7 +716,7 @@ public class RabbitMessageChannelBinder extends messageProperties.setDeliveryMode( properties.getExtension().getRepublishDeliveyMode()); } - this.template.send(this.exchange, + doSend(this.exchange, this.routingKey != null ? this.routingKey : messageProperties.getConsumerQueue(), amqpMessage); @@ -716,6 +738,45 @@ public class RabbitMessageChannelBinder extends } } + private void doSend(String exchange, String routingKey, Message amqpMessage) { + if (ConfirmType.SIMPLE.equals(this.confirmType)) { + this.template.invoke(temp -> { + temp.send(exchange, routingKey, amqpMessage); + if (!temp.waitForConfirms(ACK_TIMEOUT)) { + throw new AmqpRejectAndDontRequeueException("Negative ack for DLQ message received"); + } + return null; + }); + } + else if (ConfirmType.CORRELATED.equals(this.confirmType)) { + CorrelationData corr = new CorrelationData(); + this.template.send(exchange, routingKey, amqpMessage, corr); + try { + Confirm confirm = corr.getFuture().get(ACK_TIMEOUT, TimeUnit.MILLISECONDS); + if (!confirm.isAck()) { + throw new AmqpRejectAndDontRequeueException("Negative ack for DLQ message received"); + } + } + catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new AmqpRejectAndDontRequeueException(e); + } + catch (ExecutionException e) { + throw new AmqpRejectAndDontRequeueException(e.getCause()); + } + catch (TimeoutException e) { + throw new AmqpRejectAndDontRequeueException(e); + } + if (corr.getReturned() != null) { + RabbitMessageChannelBinder.this.logger.error("DLQ message was returned: " + amqpMessage); + throw new AmqpRejectAndDontRequeueException("DLQ message was returned"); + } + } + else { + this.template.send(exchange, routingKey, amqpMessage); + } + } + private boolean checkDlx() { if (this.dlxPresent == null) { if (properties.getExtension().isAutoBindDlq()) { 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 548c58b33..146e1a599 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 @@ -29,11 +29,13 @@ import java.util.Date; import java.util.HashMap; import java.util.List; import java.util.Map; +import java.util.Set; import java.util.UUID; 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.stream.Collectors; import java.util.zip.Deflater; import com.rabbitmq.client.LongString; @@ -115,6 +117,7 @@ import org.springframework.integration.channel.DirectChannel; import org.springframework.integration.channel.QueueChannel; import org.springframework.integration.context.IntegrationContextUtils; import org.springframework.integration.expression.ValueExpression; +import org.springframework.integration.handler.BridgeHandler; import org.springframework.integration.support.MessageBuilder; import org.springframework.messaging.Message; import org.springframework.messaging.MessageChannel; @@ -167,10 +170,7 @@ public class RabbitBinderTests extends protected RabbitTestBinder getBinder() { if (this.testBinder == null) { RabbitProperties rabbitProperties = new RabbitProperties(); - rabbitProperties.setPublisherConfirmType(ConfirmType.SIMPLE); - rabbitProperties.setPublisherReturns(true); - this.testBinder = new RabbitTestBinder(rabbitAvailableRule.getResource(), - rabbitProperties); + this.testBinder = new RabbitTestBinder(this.rabbitAvailableRule.getResource(), rabbitProperties); } return this.testBinder; } @@ -1520,6 +1520,148 @@ public class RabbitBinderTests extends consumerBinding.unbind(); } + @SuppressWarnings("unchecked") + @Test + public void testAutoBindDLQwithRepublishTx() throws Exception { + RabbitTestBinder binder = getBinder(); + ExtendedConsumerProperties consumerProperties = createConsumerProperties(); + consumerProperties.getExtension().setPrefix(TEST_PREFIX); + consumerProperties.getExtension().setAutoBindDlq(true); + consumerProperties.getExtension().setRepublishToDlq(true); + consumerProperties.setMaxAttempts(1); // disable retry + consumerProperties.getExtension().setDurableSubscription(true); + consumerProperties.getExtension().setTransacted(true); + DirectChannel moduleInputChannel = createBindableChannel("input", + createConsumerBindingProperties(consumerProperties)); + moduleInputChannel.setBeanName("dlqPubTestTx"); + moduleInputChannel.subscribe(new MessageHandler() { + + @Override + public void handleMessage(Message message) throws MessagingException { + throw new RuntimeException("test"); + } + + }); + Binding consumerBinding = binder.bindConsumer( + "foo.dlqpubtestTx", "foo", moduleInputChannel, consumerProperties); + + RabbitTemplate template = new RabbitTemplate(this.rabbitAvailableRule.getResource()); + template.convertAndSend("", TEST_PREFIX + "foo.dlqpubtestTx.foo", "foo"); + + template.setReceiveTimeout(10_000); + org.springframework.amqp.core.Message deadLetter = template + .receive(TEST_PREFIX + "foo.dlqpubtestTx.foo.dlq"); + assertThat(deadLetter).isNotNull(); + assertThat(deadLetter.getBody()).isEqualTo("foo".getBytes()); + assertThat(deadLetter.getMessageProperties().getHeaders()) + .containsKey((RepublishMessageRecoverer.X_EXCEPTION_STACKTRACE)); + List errorHandler = (List) TestUtils.getPropertyValue(consumerBinding, + "lifecycle.errorChannel.dispatcher.handlers", Set.class).stream() + .filter(handler -> !handler.getClass().equals(BridgeHandler.class)) + .collect(Collectors.toList()); + assertThat(errorHandler).hasSize(1); + assertThat(TestUtils.getPropertyValue(errorHandler.get(0), "template.transactional", Boolean.class)).isTrue(); + assertThat(TestUtils.getPropertyValue(errorHandler.get(0), "confirmType", ConfirmType.class)) + .isEqualTo(ConfirmType.NONE); + consumerBinding.unbind(); + } + + @SuppressWarnings("unchecked") + @Test + public void testAutoBindDLQwithRepublishSimpleConfirms() throws Exception { + CachingConnectionFactory ccf = this.rabbitAvailableRule.getResource(); + ccf.setPublisherReturns(true); + ccf.setPublisherConfirmType(ConfirmType.SIMPLE); + ccf.resetConnection(); + RabbitTestBinder binder = getBinder(); + ExtendedConsumerProperties consumerProperties = createConsumerProperties(); + consumerProperties.getExtension().setPrefix(TEST_PREFIX); + consumerProperties.getExtension().setAutoBindDlq(true); + consumerProperties.getExtension().setRepublishToDlq(true); + consumerProperties.setMaxAttempts(1); // disable retry + consumerProperties.getExtension().setDurableSubscription(true); + DirectChannel moduleInputChannel = createBindableChannel("input", + createConsumerBindingProperties(consumerProperties)); + moduleInputChannel.setBeanName("dlqPubtestSimple"); + moduleInputChannel.subscribe(new MessageHandler() { + + @Override + public void handleMessage(Message message) throws MessagingException { + throw new RuntimeException("test"); + } + + }); + Binding consumerBinding = binder.bindConsumer( + "foo.dlqpubtestSimple", "foo", moduleInputChannel, consumerProperties); + + RabbitTemplate template = new RabbitTemplate(this.rabbitAvailableRule.getResource()); + template.convertAndSend("", TEST_PREFIX + "foo.dlqpubtestSimple.foo", "foo"); + + template.setReceiveTimeout(10_000); + org.springframework.amqp.core.Message deadLetter = template + .receive(TEST_PREFIX + "foo.dlqpubtestSimple.foo.dlq"); + assertThat(deadLetter).isNotNull(); + assertThat(deadLetter.getBody()).isEqualTo("foo".getBytes()); + assertThat(deadLetter.getMessageProperties().getHeaders()) + .containsKey((RepublishMessageRecoverer.X_EXCEPTION_STACKTRACE)); + List errorHandler = (List) TestUtils.getPropertyValue(consumerBinding, + "lifecycle.errorChannel.dispatcher.handlers", Set.class).stream() + .filter(handler -> !handler.getClass().equals(BridgeHandler.class)) + .collect(Collectors.toList()); + assertThat(errorHandler).hasSize(1); + assertThat(TestUtils.getPropertyValue(errorHandler.get(0), "confirmType", ConfirmType.class)) + .isEqualTo(ConfirmType.SIMPLE); + consumerBinding.unbind(); + } + + @SuppressWarnings("unchecked") + @Test + public void testAutoBindDLQwithRepublishCorrelatedConfirms() throws Exception { + CachingConnectionFactory ccf = this.rabbitAvailableRule.getResource(); + ccf.setPublisherReturns(true); + ccf.setPublisherConfirmType(ConfirmType.CORRELATED); + ccf.resetConnection(); + RabbitTestBinder binder = getBinder(); + ExtendedConsumerProperties consumerProperties = createConsumerProperties(); + consumerProperties.getExtension().setPrefix(TEST_PREFIX); + consumerProperties.getExtension().setAutoBindDlq(true); + consumerProperties.getExtension().setRepublishToDlq(true); + consumerProperties.setMaxAttempts(1); // disable retry + consumerProperties.getExtension().setDurableSubscription(true); + DirectChannel moduleInputChannel = createBindableChannel("input", + createConsumerBindingProperties(consumerProperties)); + moduleInputChannel.setBeanName("dlqPubtestCorrelated"); + moduleInputChannel.subscribe(new MessageHandler() { + + @Override + public void handleMessage(Message message) throws MessagingException { + throw new RuntimeException("test"); + } + + }); + Binding consumerBinding = binder.bindConsumer( + "foo.dlqpubtestCorrelated", "foo", moduleInputChannel, consumerProperties); + + RabbitTemplate template = new RabbitTemplate(this.rabbitAvailableRule.getResource()); + template.convertAndSend("", TEST_PREFIX + "foo.dlqpubtestCorrelated.foo", "foo"); + + template.setReceiveTimeout(10_000); + org.springframework.amqp.core.Message deadLetter = template + .receive(TEST_PREFIX + "foo.dlqpubtestCorrelated.foo.dlq"); + assertThat(deadLetter).isNotNull(); + assertThat(deadLetter.getBody()).isEqualTo("foo".getBytes()); + assertThat(deadLetter.getMessageProperties().getHeaders()) + .containsKey((RepublishMessageRecoverer.X_EXCEPTION_STACKTRACE)); + List errorHandler = (List) TestUtils.getPropertyValue(consumerBinding, + "lifecycle.errorChannel.dispatcher.handlers", Set.class).stream() + .filter(handler -> !handler.getClass().equals(BridgeHandler.class)) + .collect(Collectors.toList()); + assertThat(errorHandler).hasSize(1); + assertThat(TestUtils.getPropertyValue(errorHandler.get(0), "confirmType", ConfirmType.class)) + .isEqualTo(ConfirmType.CORRELATED); + consumerBinding.unbind(); + } + @SuppressWarnings("unchecked") @Test public void testBatchingAndCompression() throws Exception {