SpringCloud/spring-cloud-stream
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

GH-2939: Rely on custom AmqpHeaders.RETRY_COUNT for server retries

Browse Source 
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
This commit is contained in:
Artem Bilan
2024-10-07 17:34:46 -04:00
committed by Oleg Zhurakousky
parent bf9ea9e626
commit c96a62f163
2 changed files with 18 additions and 36 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
binders/rabbit-binder/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitMessageChannelBinder.java
View File

@@ -815,6 +815,7 @@ public class RabbitMessageChannelBinder extends
messageProperties.setDeliveryMode(
properties.getExtension().getRepublishDeliveyMode());
}
messageProperties.incrementRetryCount();
return messageProperties;
}

53
docs/modules/ROOT/pages/rabbit/rabbit_dlq.adoc
View File

@@ -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<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);
}
@@ -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<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");
@@ -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<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);