ARTEMIS-6076 Prevent auto delete for non-empty addresses

When an address has no direct bindings but the address size is > 0, it means
queues on other addresses have one or more message references pointing to this
address (e.g., queues bound to wildcard addresses).

Author: brusdev

artemis-server/src/main/java/org/apache/activemq/artemis/core/postoffice/impl/PostOfficeImpl.java

@@ -1068,7 +1068,12 @@ private void deleteDuplicateCache(SimpleString address) throws Exception {
    @Override
    public boolean isAddressBound(final SimpleString address) throws Exception {
       Collection<Binding> bindings = getDirectBindings(address);
-      return bindings != null && !bindings.isEmpty();
+      return (bindings != null && !bindings.isEmpty()) ||
+         // When an address has no direct bindings but the address size is > 0, it means queues on other addresses
+         // have one or more message references pointing to this address (e.g., queues bound to wildcard addresses).
+         // The address must be considered bound because these message references keep it in use, preventing
+         // operations like auto-deletion that should only occur when the address is truly unused.
+         pagingManager.getPageStore(address).getAddressSize() > 0;
    }
 
    @Override

tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/client/AutoDeleteAddressTest.java

@@ -18,18 +18,16 @@
 
 import static org.junit.jupiter.api.Assertions.assertNotNull;
 import static org.junit.jupiter.api.Assertions.assertNull;
-import static org.junit.jupiter.api.Assertions.assertTrue;
 
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.List;
-import java.util.concurrent.CountDownLatch;
-import java.util.concurrent.TimeUnit;
 
 import org.apache.activemq.artemis.api.core.QueueConfiguration;
 import org.apache.activemq.artemis.api.core.RoutingType;
 import org.apache.activemq.artemis.api.core.SimpleString;
 import org.apache.activemq.artemis.api.core.client.ClientConsumer;
+import org.apache.activemq.artemis.api.core.client.ClientMessage;
 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;
@@ -122,13 +120,11 @@ public void testAutoDeleteAddressWithWildcardAddress() throws Exception {
       String wildcardAddress = prefix + ".#";
       String queue = RandomUtil.randomUUIDString();
       final int MESSAGE_COUNT = 10;
-      final CountDownLatch latch = new CountDownLatch(MESSAGE_COUNT);
 
       server.createQueue(QueueConfiguration.of(queue).setAddress(wildcardAddress).setRoutingType(RoutingType.ANYCAST).setAutoCreated(true));
 
       ClientSession consumerSession = cf.createSession();
       ClientConsumer consumer = consumerSession.createConsumer(queue);
-      consumer.setMessageHandler(message -> latch.countDown());
       consumerSession.start();
 
       ClientSession producerSession = cf.createSession();
@@ -143,7 +139,19 @@ public void testAutoDeleteAddressWithWildcardAddress() throws Exception {
       }
       producerSession.close();
 
-      assertTrue(latch.await(2, TimeUnit.SECONDS));
+      PostOfficeTestAccessor.sweepAndReapAddresses((PostOfficeImpl) server.getPostOffice());
+
+      for (String address : addresses) {
+         assertNotNull(server.getAddressInfo(SimpleString.of(address)));
+         Wait.assertTrue(() -> Arrays.asList(server.getPagingManager().getStoreNames()).contains(SimpleString.of(address)), 2000, 100);
+      }
+
+      for (int i = 0; i < MESSAGE_COUNT; i++) {
+         ClientMessage message = consumer.receive(3000);
+         assertNotNull(message);
+         message.acknowledge();
+      }
+      consumerSession.commit();
 
       for (String address : addresses) {
          assertNotNull(server.getAddressInfo(SimpleString.of(address)));