Skip to content

Commit

Permalink
GH-2939: Rely on custom AmqpHeaders.RETRY_COUNT for server retries
Browse files Browse the repository at this point in the history 
Fixes: #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
  • Loading branch information
@artembilan @olegz
artembilan authored and olegz committed Oct 17, 2024
1 parent bf9ea9e commit c96a62f
Show file tree
Hide file tree
Showing 2 changed files with 18 additions and 36 deletions.
1 change: 1 addition & 0 deletions .../main/java/org/springframework/cloud/stream/binder/rabbit/RabbitMessageChannelBinder.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -815,6 +815,7 @@ private MessageProperties adjustMessagePropertiesHeader(Throwable cause, String
messageProperties.setDeliveryMode(
properties.getExtension().getRepublishDeliveyMode());
}
messageProperties.incrementRetryCount();
return messageProperties;
}

Expand Down
53 changes: 17 additions & 36 deletions docs/modules/ROOT/pages/rabbit/rabbit_dlq.adoc
View file
Original file line number Diff line number Diff line change
Expand Up @@ -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");
Expand All @@ -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 {
Expand All @@ -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 {
Expand All @@ -90,13 +83,10 @@ public class ReRouteDlqApplication {
@RabbitListener(queues = DLQ)
public void rePublish(Message failedMessage) {
Map<String, Object> 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<String, Object> headers = failedMessage.getMessageProperties().getHeaders();
headers.put("x-delay", 5000 * retriesHeader);
this.rabbitTemplate.send(DELAY_EXCHANGE, ORIGINAL_QUEUE, failedMessage);
}
Expand Down Expand Up @@ -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:
Expand All @@ -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");
Expand All @@ -164,12 +153,9 @@ public class ReRouteDlqApplication {
@RabbitListener(queues = DLQ)
public void rePublish(Message failedMessage) {
Map<String, Object> 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<Map<String, ?>> xDeath = (List<Map<String, ?>>) headers.get(X_DEATH_HEADER);
String exchange = (String) xDeath.get(0).get("exchange");
List<String> routingKeys = (List<String>) xDeath.get(0).get("routing-keys");
Expand All @@ -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:
Expand All @@ -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;
Expand All @@ -223,12 +207,9 @@ public class ReRouteDlqApplication {
@RabbitListener(queues = DLQ)
public void rePublish(Message failedMessage) {
Map<String, Object> 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);
Expand Down

0 comments on commit c96a62f

Please sign in to comment.