(activemq-artemis) branch main updated: ARTEMIS-4664 - autoCreatedResource can get removed while receiving batch of messages

[cl...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

clebertsuconic pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/activemq-artemis.git


The following commit(s) were added to refs/heads/main by this push:
     new 45533c38da ARTEMIS-4664 - autoCreatedResource can get removed while receiving batch of messages
45533c38da is described below

commit 45533c38da7b160733f5c48e3f313b3175b14dd9
Author: AntonRoskvist <an...@volvo.com>
AuthorDate: Thu Mar 7 11:36:08 2024 +0100

    ARTEMIS-4664 - autoCreatedResource can get removed while receiving batch of messages
---
 .../core/server/impl/ActiveMQServerImpl.java       |  3 ++
 .../server/AutoCreateDeadLetterResourcesTest.java  | 37 ++++++++++++++++++++++
 2 files changed, 40 insertions(+)

diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/ActiveMQServerImpl.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/ActiveMQServerImpl.java
index 5886d5637e..20cace6b67 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/ActiveMQServerImpl.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/ActiveMQServerImpl.java
@@ -4090,6 +4090,9 @@ public class ActiveMQServerImpl implements ActiveMQServer {
             }
             final QueueBinding queueBinding = (QueueBinding) rawBinding;
             if (ignoreIfExists) {
+               //Reset potentially ongoing auto-delete status of queue
+               queueBinding.getQueue().setSwept(false);
+
                return queueBinding.getQueue();
             } else {
                throw ActiveMQMessageBundle.BUNDLE.queueAlreadyExists(queueConfiguration.getName(), queueBinding.getAddress());
diff --git a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/server/AutoCreateDeadLetterResourcesTest.java b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/server/AutoCreateDeadLetterResourcesTest.java
index 6a29677578..c02ba3d524 100644
--- a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/server/AutoCreateDeadLetterResourcesTest.java
+++ b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/server/AutoCreateDeadLetterResourcesTest.java
@@ -27,6 +27,8 @@ import org.apache.activemq.artemis.api.core.client.ClientProducer;
 import org.apache.activemq.artemis.api.core.client.ClientSession;
 import org.apache.activemq.artemis.api.core.client.ClientSessionFactory;
 import org.apache.activemq.artemis.api.core.client.ServerLocator;
+import org.apache.activemq.artemis.api.core.management.QueueControl;
+import org.apache.activemq.artemis.api.core.management.ResourceNames;
 import org.apache.activemq.artemis.core.config.DivertConfiguration;
 import org.apache.activemq.artemis.core.server.ActiveMQServer;
 import org.apache.activemq.artemis.core.server.Queue;
@@ -256,6 +258,41 @@ public class AutoCreateDeadLetterResourcesTest extends ActiveMQTestBase {
       message.acknowledge();
    }
 
+   @Test
+   public void testOngingSendToDLAPreventAutoDelete() throws Exception {
+      final int messageCount = 100;
+      SimpleString dlqName = AddressSettings.DEFAULT_DEAD_LETTER_QUEUE_PREFIX.concat(addressA).concat(AddressSettings.DEFAULT_DEAD_LETTER_QUEUE_SUFFIX);
+
+      server.createQueue(new QueueConfiguration(queueA).setAddress(addressA).setRoutingType(RoutingType.ANYCAST));
+
+      ServerLocator locator = createInVMNonHALocator();
+      ClientSessionFactory sessionFactory = createSessionFactory(locator);
+      ClientSession session = addClientSession(sessionFactory.createSession(true, true));
+      ClientProducer producer = addClientProducer(session.createProducer(addressA));
+
+      ClientMessage message = session.createMessage(true);
+      message.getBodyBuffer().writeBytes(createFakeLargeStream(1024 * 1024).readAllBytes());
+
+      for (int i = 0; i < messageCount; i++) {
+         producer.send(message);
+      }
+
+      QueueControl queueControl = (QueueControl)server.getManagementService().getResource(ResourceNames.QUEUE + queueA);
+      queueControl.sendMessagesToDeadLetterAddress(null);
+
+      QueueControl dlqControl = (QueueControl)server.getManagementService().getResource(ResourceNames.QUEUE + dlqName);
+      dlqControl.retryMessages();
+
+      for (int i = 0; i < 10; i++) {
+         queueControl.sendMessagesToDeadLetterAddress(null);
+         dlqControl.retryMessages();
+      }
+
+      Wait.assertTrue(() -> queueControl.getMessageCount() == messageCount, 2000);
+      Wait.assertTrue(() -> server.locateQueue(dlqName) == null, 2000);
+
+   }
+
    private void triggerDlaDelivery() throws Exception {
       try {
          server.createQueue(new QueueConfiguration(queueA).setAddress(addressA).setRoutingType(RoutingType.ANYCAST));