You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@camel.apache.org by da...@apache.org on 2020/02/09 11:45:18 UTC
[camel] branch master updated: Fix RabbitMQ requeue header handling
(#3555)
This is an automated email from the ASF dual-hosted git repository.
davsclaus pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/camel.git
The following commit(s) were added to refs/heads/master by this push:
new 87d2127 Fix RabbitMQ requeue header handling (#3555)
87d2127 is described below
commit 87d21278c39fa5b2169d40bc15336f967c35db35
Author: Denis Istomin <is...@gmail.com>
AuthorDate: Sun Feb 9 16:45:00 2020 +0500
Fix RabbitMQ requeue header handling (#3555)
---
.../camel/component/rabbitmq/RabbitConsumer.java | 9 ++-
.../component/rabbitmq/RabbitMQRequeueIntTest.java | 88 ++++++++++++++++++++--
2 files changed, 89 insertions(+), 8 deletions(-)
diff --git a/components/camel-rabbitmq/src/main/java/org/apache/camel/component/rabbitmq/RabbitConsumer.java b/components/camel-rabbitmq/src/main/java/org/apache/camel/component/rabbitmq/RabbitConsumer.java
index 93a259a..aaaa4a0 100644
--- a/components/camel-rabbitmq/src/main/java/org/apache/camel/component/rabbitmq/RabbitConsumer.java
+++ b/components/camel-rabbitmq/src/main/java/org/apache/camel/component/rabbitmq/RabbitConsumer.java
@@ -30,6 +30,7 @@ import org.apache.camel.Exchange;
import org.apache.camel.ExchangePattern;
import org.apache.camel.Message;
import org.apache.camel.RuntimeCamelException;
+import org.apache.camel.TypeConversionException;
import org.apache.camel.support.service.ServiceSupport;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -156,7 +157,13 @@ class RabbitConsumer extends ServiceSupport implements com.rabbitmq.client.Consu
channel.basicAck(deliveryTag, false);
}
} else {
- boolean isRequeueHeaderSet = msg.getHeader(RabbitMQConstants.REQUEUE, false, boolean.class);
+ boolean isRequeueHeaderSet = false;
+ try {
+ isRequeueHeaderSet = msg.getHeader(RabbitMQConstants.REQUEUE, false, boolean.class);
+ } catch (TypeConversionException e) {
+ LOG.warn("Header [{}] with value=[{}] evaluated as [false]", RabbitMQConstants.REQUEUE, e.getValue());
+ }
+
// processing failed, then reject and handle the exception
if (deliveryTag != 0 && !consumer.getEndpoint().isAutoAck()) {
LOG.trace("Rejecting receipt [delivery_tag={}] with requeue={}", deliveryTag, isRequeueHeaderSet);
diff --git a/components/camel-rabbitmq/src/test/java/org/apache/camel/component/rabbitmq/RabbitMQRequeueIntTest.java b/components/camel-rabbitmq/src/test/java/org/apache/camel/component/rabbitmq/RabbitMQRequeueIntTest.java
index b4a3e4b..3f9b804 100644
--- a/components/camel-rabbitmq/src/test/java/org/apache/camel/component/rabbitmq/RabbitMQRequeueIntTest.java
+++ b/components/camel-rabbitmq/src/test/java/org/apache/camel/component/rabbitmq/RabbitMQRequeueIntTest.java
@@ -22,19 +22,25 @@ import org.apache.camel.Produce;
import org.apache.camel.ProducerTemplate;
import org.apache.camel.builder.RouteBuilder;
import org.apache.camel.component.mock.MockEndpoint;
+import org.awaitility.Awaitility;
import org.junit.Test;
+import static java.util.concurrent.TimeUnit.SECONDS;
+import static org.awaitility.Awaitility.await;
+
/**
* Integration test to confirm REQUEUE header causes message to be re-queued instead of sent to DLQ.
*/
public class RabbitMQRequeueIntTest extends AbstractRabbitMQIntTest {
public static final String ROUTING_KEY = "rk4";
+ public static final String DEAD_LETTER_QUEUE_NAME = "dlq";
@Produce("direct:rabbitMQ")
protected ProducerTemplate directProducer;
@EndpointInject("rabbitmq:localhost:5672/ex4??username=cameltest&password=cameltest"
- + "&autoAck=false&queue=q4&routingKey=" + ROUTING_KEY)
+ + "&autoAck=false&queue=q4&deadLetterExchange=dlx&deadLetterExchangeType=fanout"
+ + "&deadLetterQueue=" + DEAD_LETTER_QUEUE_NAME + "&routingKey=" + ROUTING_KEY)
private Endpoint rabbitMQEndpoint;
@EndpointInject("mock:producing")
@@ -42,13 +48,35 @@ public class RabbitMQRequeueIntTest extends AbstractRabbitMQIntTest {
@EndpointInject("mock:consuming")
private MockEndpoint consumingMockEndpoint;
+ private com.rabbitmq.client.Connection connection;
+ private com.rabbitmq.client.Channel channel;
+ private com.rabbitmq.client.Channel deadLetterChannel;
+
+ @Override
+ public void setUp() throws Exception {
+ super.setUp();
+
+ connection = connection();
+ channel = connection.createChannel();
+ deadLetterChannel = connection.createChannel();
+ }
@Override
- protected RouteBuilder createRouteBuilder() throws Exception {
+ public void tearDown() throws Exception {
+ super.tearDown();
+
+ channel.abort();
+ deadLetterChannel.abort();
+ connection.abort();
+ }
+
+
+ @Override
+ protected RouteBuilder createRouteBuilder() {
return new RouteBuilder() {
@Override
- public void configure() throws Exception {
+ public void configure() {
from("direct:rabbitMQ")
.id("producingRoute")
.log("Sending message")
@@ -66,49 +94,95 @@ public class RabbitMQRequeueIntTest extends AbstractRabbitMQIntTest {
@Test
public void testNoRequeueHeaderCausesReject() throws Exception {
+ DeadLetterConsumer deadLetterConsumer = new DeadLetterConsumer();
+
producingMockEndpoint.expectedMessageCount(1);
consumingMockEndpoint.expectedMessageCount(1);
directProducer.sendBody("Hello, World!");
+ deadLetterChannel.basicConsume(DEAD_LETTER_QUEUE_NAME, false, deadLetterConsumer);
+
+ // If message was rejected and not requeued, it will be published in dead letter queue
+ await().atMost(5, SECONDS).until(() -> deadLetterConsumer.getReceivedSize() == 1);
- Thread.sleep(100);
producingMockEndpoint.assertIsSatisfied();
consumingMockEndpoint.assertIsSatisfied();
}
@Test
public void testNonBooleanRequeueHeaderCausesReject() throws Exception {
+ DeadLetterConsumer deadLetterConsumer = new DeadLetterConsumer();
+
producingMockEndpoint.expectedMessageCount(1);
consumingMockEndpoint.expectedMessageCount(1);
directProducer.sendBodyAndHeader("Hello, World!", RabbitMQConstants.REQUEUE, 4L);
+ deadLetterChannel.basicConsume(DEAD_LETTER_QUEUE_NAME, false, deadLetterConsumer);
+
+ // If message was rejected and not requeued, it will be published in dead letter queue
+ await().atMost(5, SECONDS).until(() -> deadLetterConsumer.getReceivedSize() == 1);
- Thread.sleep(100);
producingMockEndpoint.assertIsSatisfied();
consumingMockEndpoint.assertIsSatisfied();
}
@Test
public void testFalseRequeueHeaderCausesReject() throws Exception {
+ DeadLetterConsumer deadLetterConsumer = new DeadLetterConsumer();
+
producingMockEndpoint.expectedMessageCount(1);
consumingMockEndpoint.expectedMessageCount(1);
directProducer.sendBodyAndHeader("Hello, World!", RabbitMQConstants.REQUEUE, false);
+ deadLetterChannel.basicConsume(DEAD_LETTER_QUEUE_NAME, false, deadLetterConsumer);
+
+ // If message was rejected and not requeued, it will be published in dead letter queue
+ await().atMost(5, SECONDS).until(() -> deadLetterConsumer.getReceivedSize() == 1);
- Thread.sleep(100);
producingMockEndpoint.assertIsSatisfied();
consumingMockEndpoint.assertIsSatisfied();
}
@Test
public void testTrueRequeueHeaderCausesRequeue() throws Exception {
+ DeadLetterConsumer deadLetterConsumer = new DeadLetterConsumer();
+
producingMockEndpoint.expectedMessageCount(1);
consumingMockEndpoint.setMinimumExpectedMessageCount(2);
directProducer.sendBodyAndHeader("Hello, World!", RabbitMQConstants.REQUEUE, true);
+ deadLetterChannel.basicConsume(DEAD_LETTER_QUEUE_NAME, false, deadLetterConsumer);
+
+
+ Awaitility.await()
+ .during(1, SECONDS)
+ .atMost(2, SECONDS)
+ .until(() -> deadLetterConsumer.getReceivedSize() >= 0);
- Thread.sleep(100);
+ // If message was rejected and requeued it will not be published in dead letter queue
+ assertEquals(deadLetterConsumer.getReceivedSize(), 0);
producingMockEndpoint.assertIsSatisfied();
consumingMockEndpoint.assertIsSatisfied();
}
+
+ private class DeadLetterConsumer extends com.rabbitmq.client.DefaultConsumer {
+ private final java.util.List<String> received;
+
+ DeadLetterConsumer() {
+ super(deadLetterChannel);
+ received = new java.util.ArrayList<>();
+ }
+
+ public int getReceivedSize() {
+ return received.size();
+ }
+
+ @Override
+ public void handleDelivery(String consumerTag,
+ com.rabbitmq.client.Envelope envelope,
+ com.rabbitmq.client.AMQP.BasicProperties properties,
+ byte[] body) {
+ received.add(new String(body));
+ }
+ }
}