ARTEMIS-5688 set routing-type when creating MQTT subscription queue

Author: jbertram

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

@@ -184,7 +184,7 @@ private Queue findOrCreateQueue(BindingQueryResult bindingQueryResult, AddressIn
        */
       boolean durable = session.getVersion() == MQTTVersion.MQTT_5 || (session.getVersion() != MQTTVersion.MQTT_5 && !session.isClean());
       if (addressInfo.getRoutingTypes().contains(RoutingType.MULTICAST)) {
-         return session.getServerSession().createQueue(QueueConfiguration.of(queue).setAddress(addressInfo.getName()).setFilterString(getMessageFilter(addressInfo.getName())).setDurable(durable));
+         return session.getServerSession().createQueue(QueueConfiguration.of(queue).setAddress(addressInfo.getName()).setRoutingType(RoutingType.MULTICAST).setFilterString(getMessageFilter(addressInfo.getName())).setDurable(durable));
       }
 
       if (addressInfo.getRoutingTypes().contains(RoutingType.ANYCAST)) {

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

@@ -924,4 +924,18 @@ public void testSessionCount() throws Exception {
       producer.connect();
       assertEquals(2, server.getActiveMQServerControl().getSessionCount());
    }
+
+   @Test
+   @Timeout(DEFAULT_TIMEOUT_SEC)
+   public void testSubscriptionQueueRoutingType() throws Exception {
+      String topic = getName();
+      String clientID = "myClientId";
+      server.getAddressSettingsRepository().addMatch(topic, new AddressSettings().setDefaultQueueRoutingType(RoutingType.ANYCAST));
+
+      MqttClient subscriber = createPahoClient(clientID);
+      subscriber.connect();
+      subscriber.subscribe(topic, AT_LEAST_ONCE);
+      // ensure the subscription queue uses multicast even though the default is anycast
+      assertEquals(RoutingType.MULTICAST, getSubscriptionQueue(topic, clientID).getRoutingType());
+   }
 }