diff --git a/spring-cloud-stream-binder-rabbit-docs/src/main/asciidoc/contributing.adoc b/spring-cloud-stream-binder-rabbit-docs/src/main/asciidoc/contributing.adoc index 14505d112..4b83f04c7 100644 --- a/spring-cloud-stream-binder-rabbit-docs/src/main/asciidoc/contributing.adoc +++ b/spring-cloud-stream-binder-rabbit-docs/src/main/asciidoc/contributing.adoc @@ -1,4 +1,4 @@ -[[contributing] +[[contributing]] == Contributing Spring Cloud is released under the non-restrictive Apache 2.0 license, @@ -39,4 +39,4 @@ added after the original pull request but before a merge. other target branch in the main project). * When writing a commit message please follow http://tbaggery.com/2008/04/19/a-note-about-git-commit-messages.html[these conventions], if you are fixing an existing issue please add `Fixes gh-XXXX` at the end of the commit - message (where XXXX is the issue number). \ No newline at end of file + message (where XXXX is the issue number). diff --git a/spring-cloud-stream-binder-rabbit-docs/src/main/asciidoc/dlq.adoc b/spring-cloud-stream-binder-rabbit-docs/src/main/asciidoc/dlq.adoc new file mode 100644 index 000000000..3a318014d --- /dev/null +++ b/spring-cloud-stream-binder-rabbit-docs/src/main/asciidoc/dlq.adoc @@ -0,0 +1,242 @@ +[[rabbit-dlq-processing]] +== Dead-Letter Queue Processing + +Because it can't be anticipated how users would want to dispose of dead-lettered messages, the framework does not provide any standard mechanism to handle them. +If the reason for the dead-lettering is transient, you may wish to route the messages back to the original queue. +However, if the problem is a permanent issue, that could cause an infinite loop. +The following `spring-boot` application is an example of how to route those messages back to the original queue, but moves them to a third "parking lot" queue after three attempts. +The second example utilizes the https://www.rabbitmq.com/blog/2015/04/16/scheduling-messages-with-rabbitmq/[RabbitMQ Delayed Message Exchange] to introduce a delay to the requeued message. +In this example, the delay increases for each attempt. +These examples use a `@RabbitListener` to receive messages from the DLQ, you could also use `RabbitTemplate.receive()` in a batch process. + +The examples assume the original destination is `so8400in` and the consumer group is `so8400`. + +=== Non-Partioned Destinations + +The first two examples are when the destination is **not** partitioned. + +[source, java] +---- +@SpringBootApplication +public class ReRouteDlqApplication { + + private static final String ORIGINAL_QUEUE = "so8400in.so8400"; + + private static final String DLQ = ORIGINAL_QUEUE + ".dlq"; + + private static final String PARKING_LOT = ORIGINAL_QUEUE + ".parkingLot"; + + private static final String X_RETRIES_HEADER = "x-retries"; + + public static void main(String[] args) throws Exception { + ConfigurableApplicationContext context = SpringApplication.run(ReRouteDlqApplication.class, args); + System.out.println("Hit enter to terminate"); + System.in.read(); + context.close(); + } + + @Autowired + private RabbitTemplate rabbitTemplate; + + @RabbitListener(queues = DLQ) + public void rePublish(Message failedMessage) { + Integer retriesHeader = (Integer) failedMessage.getMessageProperties().getHeaders().get(X_RETRIES_HEADER); + if (retriesHeader == null) { + retriesHeader = Integer.valueOf(0); + } + if (retriesHeader < 3) { + failedMessage.getMessageProperties().getHeaders().put(X_RETRIES_HEADER, retriesHeader + 1); + this.rabbitTemplate.send(ORIGINAL_QUEUE, failedMessage); + } + else { + this.rabbitTemplate.send(PARKING_LOT, failedMessage); + } + } + + @Bean + public Queue parkingLot() { + return new Queue(PARKING_LOT); + } + +} +---- + +[source, java] +---- +@SpringBootApplication +public class ReRouteDlqApplication { + + private static final String ORIGINAL_QUEUE = "so8400in.so8400"; + + private static final String DLQ = ORIGINAL_QUEUE + ".dlq"; + + private static final String PARKING_LOT = ORIGINAL_QUEUE + ".parkingLot"; + + private static final String X_RETRIES_HEADER = "x-retries"; + + private static final String DELAY_EXCHANGE = "dlqReRouter"; + + public static void main(String[] args) throws Exception { + ConfigurableApplicationContext context = SpringApplication.run(ReRouteDlqApplication.class, args); + System.out.println("Hit enter to terminate"); + System.in.read(); + context.close(); + } + + @Autowired + private RabbitTemplate rabbitTemplate; + + @RabbitListener(queues = DLQ) + public void rePublish(Message failedMessage) { + Map headers = failedMessage.getMessageProperties().getHeaders(); + Integer retriesHeader = (Integer) headers.get(X_RETRIES_HEADER); + if (retriesHeader == null) { + retriesHeader = Integer.valueOf(0); + } + if (retriesHeader < 3) { + headers.put(X_RETRIES_HEADER, retriesHeader + 1); + headers.put("x-delay", 5000 * retriesHeader); + this.rabbitTemplate.send(DELAY_EXCHANGE, ORIGINAL_QUEUE, failedMessage); + } + else { + this.rabbitTemplate.send(PARKING_LOT, failedMessage); + } + } + + @Bean + public DirectExchange delayExchange() { + DirectExchange exchange = new DirectExchange(DELAY_EXCHANGE); + exchange.setDelayed(true); + return exchange; + } + + @Bean + public Binding bindOriginalToDelay() { + return BindingBuilder.bind(new Queue(ORIGINAL_QUEUE)).to(delayExchange()).with(ORIGINAL_QUEUE); + } + + @Bean + public Queue parkingLot() { + return new Queue(PARKING_LOT); + } + +} +---- + +=== Partitioned Destinations + +With partitioned destinations, there is one DLQ for all partitions and we determine the original queue from the headers. + +==== republishToDlq=false + +When `republishToDlq` is `false`, RabbitMQ publishes the message to the DLX/DLQ with an `x-death` header containing information about the original destination. + +[source, java] +---- +@SpringBootApplication +public class ReRouteDlqApplication { + + private static final String ORIGINAL_QUEUE = "so8400in.so8400"; + + private static final String DLQ = ORIGINAL_QUEUE + ".dlq"; + + private static final String PARKING_LOT = ORIGINAL_QUEUE + ".parkingLot"; + + private static final String X_DEATH_HEADER = "x-death"; + + private static final String X_RETRIES_HEADER = "x-retries"; + + public static void main(String[] args) throws Exception { + ConfigurableApplicationContext context = SpringApplication.run(ReRouteDlqApplication.class, args); + System.out.println("Hit enter to terminate"); + System.in.read(); + context.close(); + } + + @Autowired + private RabbitTemplate rabbitTemplate; + + @SuppressWarnings("unchecked") + @RabbitListener(queues = DLQ) + public void rePublish(Message failedMessage) { + Map headers = failedMessage.getMessageProperties().getHeaders(); + Integer retriesHeader = (Integer) headers.get(X_RETRIES_HEADER); + if (retriesHeader == null) { + retriesHeader = Integer.valueOf(0); + } + if (retriesHeader < 3) { + headers.put(X_RETRIES_HEADER, retriesHeader + 1); + List> xDeath = (List>) headers.get(X_DEATH_HEADER); + String exchange = (String) xDeath.get(0).get("exchange"); + List routingKeys = (List) xDeath.get(0).get("routing-keys"); + this.rabbitTemplate.send(exchange, routingKeys.get(0), failedMessage); + } + else { + this.rabbitTemplate.send(PARKING_LOT, failedMessage); + } + } + + @Bean + public Queue parkingLot() { + return new Queue(PARKING_LOT); + } + +} +---- + +==== republishToDlq=true + +When `republishToDlq` is `true`, the republishing recoverer adds the original exchange and routing key to headers. + +[source, java] +---- +@SpringBootApplication +public class ReRouteDlqApplication { + + private static final String ORIGINAL_QUEUE = "so8400in.so8400"; + + private static final String DLQ = ORIGINAL_QUEUE + ".dlq"; + + private static final String PARKING_LOT = ORIGINAL_QUEUE + ".parkingLot"; + + private static final String X_RETRIES_HEADER = "x-retries"; + + private static final String X_ORIGINAL_EXCHANGE_HEADER = RepublishMessageRecoverer.X_ORIGINAL_EXCHANGE; + + private static final String X_ORIGINAL_ROUTING_KEY_HEADER = RepublishMessageRecoverer.X_ORIGINAL_ROUTING_KEY; + + public static void main(String[] args) throws Exception { + ConfigurableApplicationContext context = SpringApplication.run(ReRouteDlqApplication.class, args); + System.out.println("Hit enter to terminate"); + System.in.read(); + context.close(); + } + + @Autowired + private RabbitTemplate rabbitTemplate; + + @RabbitListener(queues = DLQ) + public void rePublish(Message failedMessage) { + Map headers = failedMessage.getMessageProperties().getHeaders(); + Integer retriesHeader = (Integer) headers.get(X_RETRIES_HEADER); + if (retriesHeader == null) { + retriesHeader = Integer.valueOf(0); + } + if (retriesHeader < 3) { + headers.put(X_RETRIES_HEADER, retriesHeader + 1); + String exchange = (String) headers.get(X_ORIGINAL_EXCHANGE_HEADER); + String originalRoutingKey = (String) headers.get(X_ORIGINAL_ROUTING_KEY_HEADER); + this.rabbitTemplate.send(exchange, originalRoutingKey, failedMessage); + } + else { + this.rabbitTemplate.send(PARKING_LOT, failedMessage); + } + } + + @Bean + public Queue parkingLot() { + return new Queue(PARKING_LOT); + } + +} +---- diff --git a/spring-cloud-stream-binder-rabbit-docs/src/main/asciidoc/index.adoc b/spring-cloud-stream-binder-rabbit-docs/src/main/asciidoc/index.adoc index 38e551a3d..76626b66b 100644 --- a/spring-cloud-stream-binder-rabbit-docs/src/main/asciidoc/index.adoc +++ b/spring-cloud-stream-binder-rabbit-docs/src/main/asciidoc/index.adoc @@ -1,6 +1,6 @@ [[spring-cloud-stream-binder-rabbit-reference]] = Spring Cloud Stream RabbitMQ Binder Reference Guide -Sabby Anandan, Marius Bogoevici, Eric Bottard, Mark Fisher, Ilayaperumal Gopinathan, Gunnar Hillert, Mark Pollack, Patrick Peralta, Glenn Renfro, Thomas Risberg, Dave Syer, David Turanski, Janne Valkealahti, Benjamin Klein +Sabby Anandan, Marius Bogoevici, Eric Bottard, Mark Fisher, Ilayaperumal Gopinathan, Gunnar Hillert, Mark Pollack, Patrick Peralta, Glenn Renfro, Thomas Risberg, Dave Syer, David Turanski, Janne Valkealahti, Benjamin Klein, Gary Russell :doctype: book :toc: :toclevels: 4 @@ -23,11 +23,13 @@ Sabby Anandan, Marius Bogoevici, Eric Bottard, Mark Fisher, Ilayaperumal Gopinat = Reference Guide include::overview.adoc[] +include::dlq.adoc[] = Appendices [appendix] include::building.adoc[] +[appendix] include::contributing.adoc[] // ====================================================================================== diff --git a/spring-cloud-stream-binder-rabbit-docs/src/main/asciidoc/overview.adoc b/spring-cloud-stream-binder-rabbit-docs/src/main/asciidoc/overview.adoc index 567f05d7a..fb9d5d7f1 100644 --- a/spring-cloud-stream-binder-rabbit-docs/src/main/asciidoc/overview.adoc +++ b/spring-cloud-stream-binder-rabbit-docs/src/main/asciidoc/overview.adoc @@ -38,6 +38,17 @@ For each consumer group, a `Queue` will be bound to that `TopicExchange`. Each consumer instance have a corresponding RabbitMQ `Consumer` instance for its group's `Queue`. For partitioned producers/consumers the queues are suffixed with the partition index and use the partition index as routing key. +Using the `autoBindDlq` option, you can optionally configure the binder to create and configure dead-letter queues (DLQs) (and a dead-letter exchange `DLX`). +The dead letter queue has the name of the destination, appended with `.dlq`. +If retry is enabled (`maxAttempts > 1`) failed messages will be delivered to the DLQ. +If retry is disabled (`maxAttempts = 1`), you should set `requeueRejected` to false so the failed message will be routed to the DLQ, instead of being requeued. +In addition, `republishToDlq` causes the binder to publish a failed message to the DLQ (instead of rejecting it); this enables additional information to be added to the message in headers, such as the stack trace in the `x-exception-stacktrace` header. +This option does not need retry enabled or the `requeueRejected` property set to `true`. +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). +Some options are described in <>. + == Configuration Options This section contains settings specific to the RabbitMQ Binder and bound channels. @@ -45,6 +56,7 @@ This section contains settings specific to the RabbitMQ Binder and bound channel For general binding configuration options and properties, please refer to the https://github.com/spring-cloud/spring-cloud-stream/blob/master/spring-cloud-stream-docs/src/main/asciidoc/spring-cloud-stream-overview.adoc#configuration-options[Spring Cloud Stream core documentation]. +[[rabbit-binder-properties]] === RabbitMQ Binder Properties By default, the RabbitMQ binder uses Spring Boot's `ConnectionFactory`, and it therefore supports all Spring Boot configuration options for RabbitMQ. @@ -174,4 +186,4 @@ Default: `[STANDARD_REPLY_HEADERS,'*']`. ==== In the case of RabbitMQ, content type headers can be set by external applications. Spring Cloud Stream supports them as part of an extended internal protocol used for any type of transport (including transports, such as Kafka, that do not normally support headers). -==== \ No newline at end of file +====