ARTEMIS-6262 fix MQTT sub queue name collision

artemis-protocols/artemis-mqtt-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/mqtt/MQTTUtil.java

@@ -157,13 +157,17 @@ public static String getCoreQueueFromMqttTopic(String topicFilter, String client
 
       if (isSharedSubscription(topicFilter)) {
          Pair<String, String> decomposed = decomposeSharedSubscriptionTopicFilter(topicFilter);
-         return new StringBuilder().append(decomposed.getA()).append(".").append(getCoreAddressFromMqttTopic(decomposed.getB(), wildcardConfiguration)).toString();
+         return new StringBuilder().append(escapeDots(decomposed.getA())).append(".").append(getCoreAddressFromMqttTopic(decomposed.getB(), wildcardConfiguration)).toString();
       } else {
          Objects.requireNonNull(clientId, "MQTT client ID must not be null");
-         return new StringBuilder().append(clientId).append(".").append(getCoreAddressFromMqttTopic(topicFilter, wildcardConfiguration)).toString();
+         return new StringBuilder().append(escapeDots(clientId)).append(".").append(getCoreAddressFromMqttTopic(topicFilter, wildcardConfiguration)).toString();
       }
    }
 
+   private static String escapeDots(String input) {
+      return input.replace("\\", "\\\\").replace(".", "\\.");
+   }
+
    /**
     * This method takes the MQTT-related input and translates it into the proper name for a core address. The
     * {@code topicFilter} must be normal (i.e. non-shared). It should not be in the format

artemis-protocols/artemis-mqtt-protocol/src/test/java/org/apache/activemq/artemis/core/protocol/mqtt/MQTTUtilTest.java

@@ -69,6 +69,16 @@ public void testGetCoreQueueFromMqttTopic() {
       assertEquals(clientId + ".$", MQTTUtil.getCoreQueueFromMqttTopic("+", clientId, customWildCardConfig));
       assertEquals(clientId + ".|", MQTTUtil.getCoreQueueFromMqttTopic("/", clientId, customWildCardConfig));
       assertEquals(clientId + ".!", MQTTUtil.getCoreQueueFromMqttTopic("#", clientId, customWildCardConfig));
+
+      // ensure dots in the client ID are escaped
+      assertEquals("x\\.y.a.b.c", MQTTUtil.getCoreQueueFromMqttTopic("a/b/c", "x.y", defaultWildCardConfig));
+      assertEquals("x\\.y\\.z.a.b.c", MQTTUtil.getCoreQueueFromMqttTopic("a/b/c", "x.y.z", defaultWildCardConfig));
+      assertEquals("x\\.y.a|b|c", MQTTUtil.getCoreQueueFromMqttTopic("a/b/c", "x.y", customWildCardConfig));
+      assertEquals("x\\.y\\.z.a|b|c", MQTTUtil.getCoreQueueFromMqttTopic("a/b/c", "x.y.z", customWildCardConfig));
+
+      // ensure wildcard characters in the topic are escaped
+      assertEquals(clientId + ".\\.\\*", MQTTUtil.getCoreQueueFromMqttTopic(".*", clientId, defaultWildCardConfig));
+      assertEquals(clientId + ".\\|\\$\\!", MQTTUtil.getCoreQueueFromMqttTopic("|$!", clientId, customWildCardConfig));
    }
 
    @Test
@@ -81,6 +91,15 @@ public void testGetCoreQueueFromMqttTopicWithSharedSubscription() {
       WildcardConfiguration customWildCardConfig = new WildcardConfiguration().setDelimiter('|').setSingleWord('$').setAnyWords('!');
       assertEquals("shareName.a|b|c", MQTTUtil.getCoreQueueFromMqttTopic("$share/shareName/a/b/c", clientId, customWildCardConfig));
 
+      // ensure dots in the share name are escaped
+      assertEquals("x\\.y.a.b.c", MQTTUtil.getCoreQueueFromMqttTopic("$share/x.y/a/b/c", clientId, defaultWildCardConfig));
+      assertEquals("x\\.y\\.z.a.b.c", MQTTUtil.getCoreQueueFromMqttTopic("$share/x.y.z/a/b/c", clientId, defaultWildCardConfig));
+      assertEquals("x\\.y.a|b|c", MQTTUtil.getCoreQueueFromMqttTopic("$share/x.y/a/b/c", clientId, customWildCardConfig));
+      assertEquals("x\\.y\\.z.a|b|c", MQTTUtil.getCoreQueueFromMqttTopic("$share/x.y.z/a/b/c", clientId, customWildCardConfig));
+
+      // ensure wildcard characters in the topic are escaped
+      assertEquals("shareName.\\.\\*", MQTTUtil.getCoreQueueFromMqttTopic("$share/shareName/.*", clientId, defaultWildCardConfig));
+      assertEquals("shareName.\\|\\$\\!", MQTTUtil.getCoreQueueFromMqttTopic("$share/shareName/|$!", clientId, customWildCardConfig));
    }
 
    @Test

artemis-unit-test-support/src/main/java/org/apache/activemq/artemis/utils/Wait.java

@@ -205,11 +205,19 @@ public static void assertTrue(String failureMessage, Condition condition, final
       assertTrue(failureMessage, condition, duration, SLEEP_MILLIS);
    }
 
+   public static <T> T assertNotNull(Supplier<T> supplier) throws Exception {
+      return assertNotNull(supplier, MAX_WAIT_MILLIS, SLEEP_MILLIS);
+   }
+
    public static <T> T assertNotNull(Supplier<T> supplier, final long duration, final long sleep) throws Exception {
       Assertions.assertTrue(waitFor(() -> supplier.get() != null, duration, sleep));
       return supplier.get();
    }
 
+   public static void assertNull(ObjectCondition condition) throws Exception {
+      assertNull(condition, MAX_WAIT_MILLIS, SLEEP_MILLIS);
+   }
+
    public static void assertNull(ObjectCondition condition, final long duration, final long sleep) throws Exception {
       boolean result = waitFor(() -> (condition.getObject() == null), duration, sleep);
 

tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/mqtt5/SubscriptionQueueNameCollisionTest.java

@@ -0,0 +1,66 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.activemq.artemis.tests.integration.mqtt5;
+
+import org.apache.activemq.artemis.core.protocol.mqtt.MQTTUtil;
+import org.apache.activemq.artemis.utils.RandomUtil;
+import org.apache.activemq.artemis.utils.Wait;
+import org.eclipse.paho.mqttv5.client.MqttClient;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.Timeout;
+
+import static org.junit.jupiter.api.Assertions.assertNotEquals;
+
+public class SubscriptionQueueNameCollisionTest extends MQTT5TestSupport {
+
+   @Test
+   @Timeout(DEFAULT_TIMEOUT_SEC)
+   public void testCollision() throws Exception {
+      final String clientID1 = "a.b";
+      final String topic1 = "x/y";
+      createSub(clientID1, topic1);
+
+      final String clientID2 = "a";
+      final String topic2 = "b/x/y";
+      createSub(clientID2, topic2);
+
+      assertNotEquals(getSubscriptionQueue(topic1, clientID1).getName(), getSubscriptionQueue(topic2, clientID2).getName());
+   }
+
+   @Test
+   @Timeout(DEFAULT_TIMEOUT_SEC)
+   public void testShareCollision() throws Exception {
+      final String sharedSub1 = MQTTUtil.SHARED_SUBSCRIPTION_PREFIX + "a.b/x/y";
+      final String sharedSub2 = MQTTUtil.SHARED_SUBSCRIPTION_PREFIX + "a/b/x/y";
+
+      createSub(RandomUtil.randomUUIDString(), sharedSub1);
+      createSub(RandomUtil.randomUUIDString(), sharedSub2);
+
+      assertNotEquals(getSharedSubscriptionQueue(sharedSub1).getName(), getSharedSubscriptionQueue(sharedSub2).getName());
+   }
+
+   private void createSub(String clientId, String topic) throws Exception {
+      MqttClient subscriber = createPahoClient(clientId);
+      subscriber.connect();
+      runAfter(() -> {
+         disconnectSafely(subscriber);
+         subscriber.close();
+      });
+      subscriber.subscribe(topic, AT_LEAST_ONCE);
+      Wait.assertNotNull(() -> getSubscriptionQueue(topic, clientId));
+   }
+}