ARTEMIS-5597 initial distribution broken when sending to FQQN

Ensure the broker considers remote queue bindings as well when routing
messages to an FQQN. Otherwise consumers might starve.

Author: jbertram

artemis-commons/src/main/java/org/apache/activemq/artemis/utils/UUID.java

@@ -16,6 +16,8 @@
  */
 package org.apache.activemq.artemis.utils;
 
+import org.apache.activemq.artemis.api.core.SimpleString;
+
 /**
  * UUID represents Universally Unique Identifiers (aka Global UID in Windows world). UUIDs are usually generated via
  * UUIDGenerator (or in case of 'Null UUID', 16 zero bytes, via static method getNullUUID()), or received from external
@@ -63,6 +65,8 @@ public final class UUID {
 
    public static final char HYPHEN = '-';
 
+   public static final int UUID_STRING_LENGTH = 36;
+
    // 'Standard' namespaces defined (suggested) by UUID specs:
    public static final String NAMESPACE_DNS = "6ba7b810-9dad-11d1-80b4-00c04fd430c8";
 
@@ -174,7 +178,7 @@ public String toString() {
        */
 
       if (mDesc == null) {
-         StringBuilder b = new StringBuilder(36);
+         StringBuilder b = new StringBuilder(UUID_STRING_LENGTH);
 
          for (int i = 0; i < 16; ++i) {
             // Need to bypass hyphens:
@@ -262,7 +266,7 @@ public boolean equals(final Object obj) {
     * @return {@code true} if the string is a valid UUID format; {@code false} otherwise
     */
    public static boolean isUUID(CharSequence str) {
-      if (str == null || str.length() != 36) {
+      if (str == null || str.length() != UUID_STRING_LENGTH) {
          return false;
       }
 
@@ -285,4 +289,21 @@ public static boolean isUUID(CharSequence str) {
    private static boolean isHex(char c) {
       return (c >= '0' && c <= '9') || (c >= 'a' && c <= 'f');
    }
+
+   /**
+    * Removes a trailing UUID from the given input, if present. The method checks whether the last characters of the
+    * input adhere to a valid UUID format and removes them if true.
+    *
+    * @param input the input {@code SimpleString} which may contain a trailing UUID to be stripped
+    * @return a {@code SimpleString} with the trailing UUID removed if present; otherwise, returns the original input
+    *         unchanged
+    */
+   public static SimpleString stripTrailingUUID(SimpleString input) {
+      int length = input.length();
+      if (length >= UUID_STRING_LENGTH && UUID.isUUID(input.subSeq(length - UUID_STRING_LENGTH, length))) {
+         return input.subSeq(0, length - UUID_STRING_LENGTH);
+      } else {
+         return input;
+      }
+   }
 }

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

@@ -49,6 +49,7 @@
 import org.apache.activemq.artemis.core.server.group.impl.Response;
 import org.apache.activemq.artemis.core.transaction.impl.TransactionImpl;
 import org.apache.activemq.artemis.utils.CompositeAddress;
+import org.apache.activemq.artemis.utils.UUID;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import java.lang.invoke.MethodHandles;
@@ -343,12 +344,6 @@ private void route(final Message message,
          } else if (groupRouting && groupingHandler != null && (groupId = message.getGroupID()) != null) {
             context.clear().setReusable(false);
             routeUsingStrictOrdering(message, context, groupingHandler, groupId, 0);
-         } else if (CompositeAddress.isFullyQualified(message.getAddress())) {
-            context.clear().setReusable(false);
-            final Binding theBinding = bindingsNameMap.get(String.valueOf(CompositeAddress.extractQueueName(message.getAddressSimpleString())));
-            if (theBinding != null && (theBinding.getFilter() == null || theBinding.getFilter().match(message))) {
-               theBinding.route(message, context);
-            }
          } else {
             // in a optimization, we are reusing the previous context if everything is right for it
             // so the simpleRouting will only happen if needed
@@ -494,8 +489,11 @@ private static boolean matchBinding(final Message message,
          return false;
       }
 
-      final Filter filter = binding.getFilter();
+      if (CompositeAddress.isFullyQualified(message.getAddress()) && binding instanceof QueueBinding && !UUID.stripTrailingUUID(binding.getClusterName()).equals(CompositeAddress.extractQueueName(message.getAddressSimpleString()))) {
+         return false;
+      }
 
+      final Filter filter = binding.getFilter();
       if (filter == null || filter.match(message)) {
          return true;
       }

tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/cluster/distribution/SimpleSymmetricClusterTest.java

@@ -16,11 +16,6 @@
  */
 package org.apache.activemq.artemis.tests.integration.cluster.distribution;
 
-import static org.junit.jupiter.api.Assertions.assertEquals;
-import static org.junit.jupiter.api.Assertions.assertFalse;
-import static org.junit.jupiter.api.Assertions.assertNotNull;
-import static org.junit.jupiter.api.Assertions.assertTrue;
-
 import java.lang.invoke.MethodHandles;
 import java.util.ArrayList;
 import java.util.Arrays;
@@ -37,8 +32,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.core.config.DivertConfiguration;
 import org.apache.activemq.artemis.api.core.management.QueueControl;
+import org.apache.activemq.artemis.core.config.DivertConfiguration;
 import org.apache.activemq.artemis.core.postoffice.Binding;
 import org.apache.activemq.artemis.core.server.ComponentConfigurationRoutingType;
 import org.apache.activemq.artemis.core.server.cluster.ClusterConnection;
@@ -49,11 +44,17 @@
 import org.apache.activemq.artemis.core.settings.impl.AddressSettings;
 import org.apache.activemq.artemis.tests.integration.management.ManagementControlHelper;
 import org.apache.activemq.artemis.tests.util.Wait;
+import org.apache.activemq.artemis.utils.CompositeAddress;
 import org.junit.jupiter.api.Disabled;
 import org.junit.jupiter.api.Test;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
 public class SimpleSymmetricClusterTest extends ClusterTestBase {
 
    private static final Logger logger = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
@@ -370,6 +371,64 @@ public void testSimple_TwoNodes() throws Exception {
 
    }
 
+   @Test
+   public void testInitialDistributionSendToFQQNConsumeFromFQQN() throws Exception {
+      testInitialDistribution(true, true);
+   }
+
+   @Test
+   public void testInitialDistributionSendToFQQNConsumeFromQueue() throws Exception {
+      testInitialDistribution(true, false);
+   }
+
+   @Test
+   public void testInitialDistributionSendToAddressConsumeFromFQQN() throws Exception {
+      testInitialDistribution(false, true);
+   }
+
+   @Test
+   public void testInitialDistributionSendToAddressConsumeFromQueue() throws Exception {
+      testInitialDistribution(false, false);
+   }
+
+   private void testInitialDistribution(boolean sendToFQQN, boolean consumeFromFQQN) throws Exception {
+      CountDownLatch countDownLatch = new CountDownLatch(1);
+      final String address = "myAddress";
+      final String queue = "myQueue";
+      final String fqqn = CompositeAddress.toFullyQualified(address, queue);
+
+      setupServer(0, false, isNetty());
+      setupServer(1, false, isNetty());
+
+      setupClusterConnection("cluster0", address, MessageLoadBalancingType.ON_DEMAND, 1, isNetty(), 0, 1);
+      setupClusterConnection("cluster1", address, MessageLoadBalancingType.ON_DEMAND, 1, isNetty(), 1, 0);
+
+      startServers(0, 1);
+
+      setupSessionFactory(0, isNetty());
+      setupSessionFactory(1, isNetty());
+
+      createQueue(0, address, queue, null, false);
+      createQueue(1, address, queue, null, false);
+
+      waitForBindings(0, address, 1, 0, true);
+      waitForBindings(0, address, 1, 0, false);
+      waitForBindings(1, address, 1, 0, true);
+      waitForBindings(1, address, 1, 0, false);
+
+      addConsumer(0, 0, consumeFromFQQN ? fqqn : queue, null);
+      consumers[0].consumer.setMessageHandler((m) -> countDownLatch.countDown());
+
+      waitForBindings(0, address, 1, 1, true);
+      waitForBindings(1, address, 1, 1, false);
+
+      send(1, sendToFQQN ? fqqn : address, 1, true, null);
+
+      assertTrue(countDownLatch.await(5, TimeUnit.SECONDS));
+
+      closeAllConsumers();
+   }
+
    @Test
    public void testSimpleSnFManagement() throws Exception {
       final String address = "queues.testaddress";

tests/unit-tests/src/test/java/org/apache/activemq/artemis/tests/unit/util/UUIDTest.java

@@ -25,6 +25,7 @@
 
 import org.apache.activemq.artemis.api.core.SimpleString;
 import org.apache.activemq.artemis.tests.util.ActiveMQTestBase;
+import org.apache.activemq.artemis.utils.RandomUtil;
 import org.apache.activemq.artemis.utils.UUID;
 import org.apache.activemq.artemis.utils.UUIDGenerator;
 import org.junit.jupiter.api.Test;
@@ -101,4 +102,13 @@ public void testIsUUID() {
       // null
       assertFalse(UUID.isUUID(null));
    }
+
+   @Test
+   public void testStringTrailingUUID() {
+      SimpleString uuid = RandomUtil.randomUUIDSimpleString();
+      assertEquals(uuid, UUID.stripTrailingUUID(uuid.concat(RandomUtil.randomUUIDSimpleString())));
+
+      SimpleString foo = SimpleString.of("foo");
+      assertEquals(foo, UUID.stripTrailingUUID(foo));
+   }
 }