From c96a62f163b99cdc4ca92d684e14363e712f236d Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Mon, 7 Oct 2024 17:34:46 -0400 Subject: [PATCH] GH-2939: Rely on custom `AmqpHeaders.RETRY_COUNT` for server retries Fixes: https://github.com/spring-cloud/spring-cloud-stream/issues/2939 The RabbitMQ 4.0 does not deal with client side `x-*` headers. Therefore, an `x-death.count` is not incremented anymore when message is re-published from client back to the broker. * Spring AMQP 3.2 has introduced an `AmqpHeaders.RETRY_COUNT` custom header. Use `messageProperties.incrementRetryCount()` in the `RabbitMessageChannelBinder` when we re-published message back to the broker for server-side retries * Fix docs respectively Resolves #3019 --- .../rabbit/RabbitMessageChannelBinder.java | 1 + .../modules/ROOT/pages/rabbit/rabbit_dlq.adoc | 53 ++++++------------- 2 files changed, 18 insertions(+), 36 deletions(-) diff --git a/binders/rabbit-binder/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitMessageChannelBinder.java b/binders/rabbit-binder/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitMessageChannelBinder.java index 5f610b57e..e24b28032 100644 --- a/binders/rabbit-binder/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitMessageChannelBinder.java +++ b/binders/rabbit-binder/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitMessageChannelBinder.java @@ -815,6 +815,7 @@ public class RabbitMessageChannelBinder extends messageProperties.setDeliveryMode( properties.getExtension().getRepublishDeliveyMode()); } + messageProperties.incrementRetryCount(); return messageProperties; } diff --git a/docs/modules/ROOT/pages/rabbit/rabbit_dlq.adoc b/docs/modules/ROOT/pages/rabbit/rabbit_dlq.adoc index c20f1ce92..749ed3f1c 100644 --- a/docs/modules/ROOT/pages/rabbit/rabbit_dlq.adoc +++ b/docs/modules/ROOT/pages/rabbit/rabbit_dlq.adoc @@ -28,8 +28,6 @@ public class ReRouteDlqApplication { 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("Press enter to exit"); @@ -42,12 +40,9 @@ public class ReRouteDlqApplication { @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); + long retries = failedMessage.getMessageProperties().getRetryCount(); + if (retries < 3) { + failedMessage.getMessageProperties().incrementRetryCount(); this.rabbitTemplate.send(ORIGINAL_QUEUE, failedMessage); } else { @@ -74,8 +69,6 @@ public class ReRouteDlqApplication { 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 { @@ -90,13 +83,10 @@ public class ReRouteDlqApplication { @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); + long retries = failedMessage.getMessageProperties().getRetryCount(); + if (retries < 3) { + failedMessage.getMessageProperties().incrementRetryCount(); + Map headers = failedMessage.getMessageProperties().getHeaders(); headers.put("x-delay", 5000 * retriesHeader); this.rabbitTemplate.send(DELAY_EXCHANGE, ORIGINAL_QUEUE, failedMessage); } @@ -128,9 +118,10 @@ public class ReRouteDlqApplication { [[partitioned-destinations]] == Partitioned Destinations -With partitioned destinations, there is one DLQ for all partitions. We determine the original queue from the headers. +With partitioned destinations, there is one DLQ for all partitions. +We determine the original queue from the headers. -[[republishtodlq=false]] +[[republishtodlq-false]] === `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, as shown in the following example: @@ -148,8 +139,6 @@ public class ReRouteDlqApplication { 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("Press enter to exit"); @@ -164,12 +153,9 @@ public class ReRouteDlqApplication { @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); + long retries = failedMessage.getMessageProperties().getRetryCount(); + if (retries < 3) { + failedMessage.getMessageProperties().incrementRetryCount(); 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"); @@ -188,7 +174,7 @@ public class ReRouteDlqApplication { } ---- -[[republishtodlq=true]] +[[republishtodlq-true]] === `republishToDlq=true` When `republishToDlq` is `true`, the republishing recoverer adds the original exchange and routing key to headers, as shown in the following example: @@ -204,8 +190,6 @@ public class ReRouteDlqApplication { 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; @@ -223,12 +207,9 @@ public class ReRouteDlqApplication { @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); + long retries = failedMessage.getMessageProperties().getRetryCount(); + if (retries < 3) { + failedMessage.getMessageProperties().incrementRetryCount(); 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);