You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@activemq.apache.org by cl...@apache.org on 2020/12/08 19:43:24 UTC
[activemq-artemis] branch master updated: ARTEMIS-3024 Expiry or
DLQ Retry not working with AMQP
This is an automated email from the ASF dual-hosted git repository.
clebertsuconic pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/activemq-artemis.git
The following commit(s) were added to refs/heads/master by this push:
new 8253417 ARTEMIS-3024 Expiry or DLQ Retry not working with AMQP
new caf314e This closes #3374
8253417 is described below
commit 825341734a84807634f44c6bd14c40b7379182f2
Author: Clebert Suconic <cl...@apache.org>
AuthorDate: Tue Dec 8 14:02:19 2020 -0500
ARTEMIS-3024 Expiry or DLQ Retry not working with AMQP
---
.../artemis/protocol/amqp/broker/AMQPMessage.java | 6 ++-
.../apache/activemq/artemis/core/server/Queue.java | 4 ++
.../artemis/core/server/impl/QueueImpl.java | 18 +++++++++
.../integration/amqp/AmqpExpiredMessageTest.java | 46 ++++++++++++++++++++++
4 files changed, 73 insertions(+), 1 deletion(-)
diff --git a/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/broker/AMQPMessage.java b/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/broker/AMQPMessage.java
index 0bcf572..6c9e6fb 100644
--- a/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/broker/AMQPMessage.java
+++ b/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/broker/AMQPMessage.java
@@ -524,7 +524,11 @@ public abstract class AMQPMessage extends RefCountMessage implements org.apache.
}
protected Object getMessageAnnotation(String annotation) {
- return getMessageAnnotation(Symbol.getSymbol(annotation));
+ Object messageAnnotation = getMessageAnnotation(Symbol.getSymbol(AMQPMessageSupport.toAnnotationName(annotation)));
+ if (messageAnnotation == null) {
+ messageAnnotation = getMessageAnnotation(Symbol.getSymbol(annotation));
+ }
+ return messageAnnotation;
}
protected Object getMessageAnnotation(Symbol annotation) {
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/Queue.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/Queue.java
index eb04963..628e43e 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/Queue.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/Queue.java
@@ -361,6 +361,10 @@ public interface Queue extends Bindable,CriticalComponent {
int retryMessages(Filter filter) throws Exception;
+ default int retryMessages(Filter filter, Integer expectedHits) throws Exception {
+ return retryMessages(filter);
+ }
+
void addRedistributor(long delay);
void cancelRedistributor() throws Exception;
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/QueueImpl.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/QueueImpl.java
index ef0abf2..2379d5e 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/QueueImpl.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/QueueImpl.java
@@ -2140,6 +2140,7 @@ public class QueueImpl extends CriticalComponentImpl implements Queue {
QueueIterateAction messageAction) throws Exception {
int count = 0;
int txCount = 0;
+ Integer expectedHits = messageAction.expectedHits();
// This is to avoid scheduling depaging while iterQueue is happening
// this should minimize the use of the paged executor.
depagePending = true;
@@ -2170,6 +2171,9 @@ public class QueueImpl extends CriticalComponentImpl implements Queue {
}
txCount++;
count++;
+ if (expectedHits != null && count >= expectedHits.intValue()) {
+ break;
+ }
}
}
@@ -2611,11 +2615,21 @@ public class QueueImpl extends CriticalComponentImpl implements Queue {
@Override
public int retryMessages(Filter filter) throws Exception {
+ return retryMessages(filter, null);
+ }
+
+ @Override
+ public int retryMessages(Filter filter, Integer expectedHits) throws Exception {
final HashMap<String, Long> queues = new HashMap<>();
return iterQueue(DEFAULT_FLUSH_LIMIT, filter, new QueueIterateAction() {
@Override
+ public Integer expectedHits() {
+ return expectedHits;
+ }
+
+ @Override
public boolean actMessage(Transaction tx, MessageReference ref) throws Exception {
String originalMessageAddress = ref.getMessage().getAnnotationString(Message.HDR_ORIGINAL_ADDRESS);
@@ -4164,6 +4178,10 @@ public class QueueImpl extends CriticalComponentImpl implements Queue {
*/
abstract class QueueIterateAction {
+ public Integer expectedHits() {
+ return null;
+ }
+
/**
*
* @param tx the transaction which the message action should participate in
diff --git a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/amqp/AmqpExpiredMessageTest.java b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/amqp/AmqpExpiredMessageTest.java
index 30b3f7a..38d324f 100644
--- a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/amqp/AmqpExpiredMessageTest.java
+++ b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/amqp/AmqpExpiredMessageTest.java
@@ -128,6 +128,52 @@ public class AmqpExpiredMessageTest extends AmqpClientTestSupport {
connection.close();
}
+ @Test(timeout = 60000)
+ public void testRetryExpiry() throws Exception {
+ AmqpClient client = createAmqpClient();
+ AmqpConnection connection = addConnection(client.connect());
+ AmqpSession session = connection.createSession();
+
+ AmqpSender sender = session.createSender(getQueueName());
+
+ // Get the Queue View early to avoid racing the delivery.
+ final Queue queueView = getProxyToQueue(getQueueName());
+ assertNotNull(queueView);
+
+ AmqpMessage message = new AmqpMessage();
+ message.setTimeToLive(1);
+ message.setText("Test-Message");
+ message.setDurable(true);
+ message.setApplicationProperty("key1", "Value1");
+ sender.send(message);
+
+ message = new AmqpMessage();
+ message.setTimeToLive(1);
+ message.setBytes(new byte[500 * 1024]);
+ sender.send(message);
+ sender.close();
+
+ final Queue dlqView = getProxyToQueue(getDeadLetterAddress());
+
+ Wait.assertEquals(2, dlqView::getMessageCount);
+ Assert.assertEquals(2, dlqView.retryMessages(null));
+ Wait.assertEquals(0, dlqView::getMessageCount);
+ Wait.assertEquals(2, queueView::getMessageCount);
+
+
+ AmqpReceiver receiver = session.createReceiver(getQueueName());
+ // Now try and get the message
+ receiver.flow(2);
+ for (int i = 0; i < 2; i++) {
+ AmqpMessage received = receiver.receive(5, TimeUnit.SECONDS);
+ assertNotNull(received);
+ received.accept();
+ }
+ connection.close();
+ Wait.assertEquals(0, queueView::getMessageCount);
+ Wait.assertEquals(0, dlqView::getMessageCount);
+ }
+
/** This test is validating a broker feature where the message copy through the DLQ will receive an annotation.
* It is also testing filter on that annotation. */
@Test(timeout = 60000)