Skip to content

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

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
wants to merge 2 commits into from
Closed

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

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
9466b91
GH-2939: Rely on custom `AmqpHeaders.RETRY_COUNT` for server retries
artembilan Oct 7, 2024
30c2b3d
* Add `RabbitDlqTests` to cover DLX functionality
artembilan Oct 8, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 5 additions & 0 deletions binders/rabbit-binder/spring-cloud-stream-binder-rabbit/pom.xml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -98,6 +98,11 @@
<artifactId>spring-cloud-stream-binder-rabbit-test-support</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-testcontainers</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.testcontainers</groupId>
<artifactId>rabbitmq</artifactId>
Expand Down
1 change: 1 addition & 0 deletions .../main/java/org/springframework/cloud/stream/binder/rabbit/RabbitMessageChannelBinder.java
View file
Open in desktop
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
113 changes: 113 additions & 0 deletions ...bbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/dlq/RabbitDlqTests.java
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,113 @@
/*
* Copyright 2024-2024 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* https://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/

package org.springframework.cloud.stream.binder.rabbit.dlq;

import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.function.Consumer;

import org.junit.jupiter.api.Nested;
import org.junit.jupiter.api.Test;
import org.testcontainers.containers.RabbitMQContainer;
import org.testcontainers.utility.DockerImageName;

import org.springframework.amqp.AmqpRejectAndDontRequeueException;
import org.springframework.amqp.ImmediateAcknowledgeAmqpException;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.SpringBootConfiguration;
import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.boot.testcontainers.service.connection.ServiceConnection;
import org.springframework.context.annotation.Bean;
import org.springframework.messaging.Message;
import org.springframework.test.annotation.DirtiesContext;
import org.springframework.test.context.TestPropertySource;

import static org.assertj.core.api.Assertions.assertThat;

/**
* @author Artem Bilan
*
* @since 4.2
*/
@SpringBootTest(properties = { "spring.cloud.stream.function.bindings.listener-in-0=input",
"spring.cloud.stream.bindings.input.destination=myDestination",
"spring.cloud.stream.bindings.input.group=consumerGroup",
"spring.cloud.stream.bindings.input.consumer.max-attempts=1",
"spring.cloud.stream.rabbit.bindings.input.consumer.auto-bind-dlq=true",
"spring.cloud.stream.rabbit.bindings.input.consumer.dlq-ttl=1000",
"spring.cloud.stream.rabbit.bindings.input.consumer.dlq-dead-letter-exchange=" })
@DirtiesContext
public class RabbitDlqTests {

@Test
void verifyServerSideRetry(@Autowired RabbitTemplate rabbitTemplate, @Autowired CountDownLatch dlqRetryExhausted)
throws InterruptedException {

rabbitTemplate.convertAndSend("myDestination.consumerGroup", "test data");

assertThat(dlqRetryExhausted.await(10, TimeUnit.SECONDS)).isTrue();
}

@EnableAutoConfiguration
@SpringBootConfiguration
public static class RabbitDlqConfiguration {

@Bean
@ServiceConnection
RabbitMQContainer rabbitContainer() {
return new RabbitMQContainer(DockerImageName.parse("rabbitmq:4"));
}

@Bean
CountDownLatch dlqRetryExhausted() {
return new CountDownLatch(1);
}

@Bean
public Consumer<Message<String>> listener(CountDownLatch dlqRetryExhausted) {
return message -> {
Long retryCount = message.getHeaders().get(AmqpHeaders.RETRY_COUNT, Long.class);
if (retryCount != null && retryCount.equals(3L)) {
dlqRetryExhausted.countDown();
// giving up - don't send to DLX
throw new ImmediateAcknowledgeAmqpException("Failed after 4 attempts");
}
throw new AmqpRejectAndDontRequeueException("failed");
};
}

}

@Nested
@TestPropertySource(properties = "spring.cloud.stream.rabbit.bindings.input.consumer.republish-to-dlq=false")
class NoRepublishToDlx {

@Test
void verifyServerSideRetry(@Autowired RabbitTemplate rabbitTemplate,
@Autowired CountDownLatch dlqRetryExhausted) throws InterruptedException {

rabbitTemplate.convertAndSend("myDestination.consumerGroup", "test data");

assertThat(dlqRetryExhausted.await(10, TimeUnit.SECONDS)).isTrue();
}

}

}
53 changes: 17 additions & 36 deletions docs/modules/ROOT/pages/rabbit/rabbit_dlq.adoc
View file
Open in desktop
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
Loading