(artemis) branch main updated: ARTEMIS-6314 MQTT 5 retained PUBLISH not delivered to an established wildcard or shared subscription

[jb...@apache.org]

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

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


The following commit(s) were added to refs/heads/main by this push:
     new 3eeb4fe721 ARTEMIS-6314 MQTT 5 retained PUBLISH not delivered to an established wildcard or shared subscription
3eeb4fe721 is described below

commit 3eeb4fe721fe8f754eec08e795bfcc769d84e563
Author: grepjava <gr...@gmail.com>
AuthorDate: Wed Oct 21 21:21:24 2026 +0800

    ARTEMIS-6314 MQTT 5 retained PUBLISH not delivered to an established wildcard or shared subscription
---
 .../core/protocol/mqtt/MQTTPublishManager.java     | 12 ++---
 .../core/protocol/mqtt/MQTTSessionState.java       | 14 +++++
 .../mqtt5/spec/controlpackets/PublishTests.java    | 61 ++++++++++++++++++++++
 3 files changed, 81 insertions(+), 6 deletions(-)

diff --git a/artemis-protocols/artemis-mqtt-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/mqtt/MQTTPublishManager.java b/artemis-protocols/artemis-mqtt-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/mqtt/MQTTPublishManager.java
index 3865dc1723..4e2330f64e 100644
--- a/artemis-protocols/artemis-mqtt-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/mqtt/MQTTPublishManager.java
+++ b/artemis-protocols/artemis-mqtt-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/mqtt/MQTTPublishManager.java
@@ -31,7 +31,6 @@ import io.netty.handler.codec.mqtt.MqttProperties;
 import io.netty.handler.codec.mqtt.MqttPublishMessage;
 import io.netty.handler.codec.mqtt.MqttPublishVariableHeader;
 import io.netty.handler.codec.mqtt.MqttQoS;
-import io.netty.handler.codec.mqtt.MqttTopicSubscription;
 import org.apache.activemq.artemis.api.core.ActiveMQBuffer;
 import org.apache.activemq.artemis.api.core.ActiveMQSecurityException;
 import org.apache.activemq.artemis.api.core.ICoreMessage;
@@ -109,7 +108,7 @@ public class MQTTPublishManager {
       if (qos == 0) {
          // [MQTT-2.2.1-2] Hard-code the packet ID to 0 as QoS0 PUBLISH packets don't have a packet ID
          // [MQTT-3.3.1-2] The DUP flag MUST be set to 0 for all QoS 0 messages.
-         if (publishToClient(0, message, false, qos)) {
+         if (publishToClient(0, message, consumer.getID(), false, qos)) {
             consumer.individualAcknowledge(null, message.getMessageID());
          }
       } else if (qos == 1 || qos == 2) {
@@ -131,7 +130,7 @@ public class MQTTPublishManager {
             state.putCoreDeliveryInfo(packetIdToUse, CoreDeliveryInfo.of(consumer.getID(), correlationKey));
             state.incrementSendQuota();
          }
-         publishToClient(packetIdToUse, message, redelivery, qos);
+         publishToClient(packetIdToUse, message, consumer.getID(), redelivery, qos);
       } else {
          // Client must have disconnected and it's Subscription QoS cleared
          consumer.individualCancel(message.getMessageID(), false);
@@ -417,7 +416,7 @@ public class MQTTPublishManager {
       }
    }
 
-   private boolean publishToClient(int packetId, ICoreMessage coreMessage, boolean redelivery, int qos) throws Exception {
+   private boolean publishToClient(int packetId, ICoreMessage coreMessage, long consumerId, boolean redelivery, int qos) throws Exception {
       String topic = MQTTUtil.getMqttTopicFromCoreAddress(Objects.requireNonNullElse(coreMessage.getAddress(), ""), session.getWildcardConfiguration());
 
       ByteBuf payload;
@@ -442,8 +441,9 @@ public class MQTTPublishManager {
       if (session.getVersion() == MQTTVersion.MQTT_5) {
          mqttProperties = getPublishProperties(coreMessage);
          if (!isRetain && coreMessage.getBooleanProperty(MQTT_MESSAGE_RETAIN_KEY)) {
-            MqttTopicSubscription sub = session.getState().getSubscriptionItem(topic).getSubscription();
-            if (sub != null && sub.option().isRetainAsPublished()) {
+            // the subscription this consumer serves, whose topic filter need not be the message's topic
+            SubscriptionItem item = session.getState().getSubscriptionItemForConsumer(consumerId);
+            if (item != null && item.getSubscription().option().isRetainAsPublished()) {
                isRetain = true;
             }
          }
diff --git a/artemis-protocols/artemis-mqtt-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/mqtt/MQTTSessionState.java b/artemis-protocols/artemis-mqtt-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/mqtt/MQTTSessionState.java
index 7b74378bf4..4edc863bd8 100644
--- a/artemis-protocols/artemis-mqtt-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/mqtt/MQTTSessionState.java
+++ b/artemis-protocols/artemis-mqtt-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/mqtt/MQTTSessionState.java
@@ -40,6 +40,7 @@ import org.apache.activemq.artemis.api.core.SimpleString;
 import org.apache.activemq.artemis.core.message.impl.CoreMessage;
 import org.apache.activemq.artemis.core.postoffice.Address;
 import org.apache.activemq.artemis.core.postoffice.impl.AddressImpl;
+import org.apache.activemq.artemis.core.server.ServerConsumer;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -224,6 +225,19 @@ public class MQTTSessionState {
       return subscriptionItems.get(topicFilter);
    }
 
+   /**
+    * {@return the subscription whose consumer has the given ID, or {@code null} if none does}
+    */
+   public SubscriptionItem getSubscriptionItemForConsumer(long consumerId) {
+      for (SubscriptionItem item : subscriptionItems.values()) {
+         ServerConsumer consumer = item.getConsumer();
+         if (consumer != null && consumer.getID() == consumerId) {
+            return item;
+         }
+      }
+      return null;
+   }
+
    public List<Integer> getMatchingSubscriptionIdentifiers(String address) {
       String topic = MQTTUtil.getMqttTopicFromCoreAddress(address, session.getServer().getConfiguration().getWildcardConfiguration());
       Address topicToMatch = new AddressImpl(SimpleString.of(topic), MQTTUtil.MQTT_WILDCARD);
diff --git a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/mqtt5/spec/controlpackets/PublishTests.java b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/mqtt5/spec/controlpackets/PublishTests.java
index 0ae5c34004..a81a6d22c1 100644
--- a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/mqtt5/spec/controlpackets/PublishTests.java
+++ b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/mqtt5/spec/controlpackets/PublishTests.java
@@ -685,6 +685,67 @@ public class PublishTests extends MQTT5TestSupport {
       consumer.close();
    }
 
+   /**
+    * [MQTT-3.3.1-13] as above, for a subscription whose topic filter is not the topic name of the message.
+    */
+   @Test
+   @Timeout(DEFAULT_TIMEOUT_SEC)
+   public void testRetainAsPublishedOneOnEstablishedWildcardSubscription() throws Exception {
+      testRetainAsPublishedOnEstablishedSubscription(this.getTopicName() + "/#", true);
+   }
+
+   /**
+    * [MQTT-3.3.1-12] as above, for a subscription whose topic filter is not the topic name of the message.
+    */
+   @Test
+   @Timeout(DEFAULT_TIMEOUT_SEC)
+   public void testRetainAsPublishedZeroOnEstablishedWildcardSubscription() throws Exception {
+      testRetainAsPublishedOnEstablishedSubscription(this.getTopicName() + "/#", false);
+   }
+
+   /**
+    * [MQTT-3.3.1-13] as above, for a shared subscription.
+    */
+   @Test
+   @Timeout(DEFAULT_TIMEOUT_SEC)
+   public void testRetainAsPublishedOneOnEstablishedSharedSubscription() throws Exception {
+      testRetainAsPublishedOnEstablishedSubscription("$share/group/" + this.getTopicName() + "/a", true);
+   }
+
+   private void testRetainAsPublishedOnEstablishedSubscription(String topicFilter, boolean retainAsPublished) throws Exception {
+      final String CONSUMER_ID = RandomUtil.randomUUIDString();
+      final String TOPIC = this.getTopicName() + "/a";
+
+      // create consumer
+      final CountDownLatch latch = new CountDownLatch(1);
+      final AtomicBoolean retained = new AtomicBoolean(!retainAsPublished);
+      MqttClient consumer = createPahoClient(CONSUMER_ID);
+      consumer.setCallback(new DefaultMqttCallback() {
+         @Override
+         public void messageArrived(String topic, MqttMessage message) throws Exception {
+            assertEqualsByteArrays("retained".getBytes(StandardCharsets.UTF_8), message.getPayload());
+            retained.set(message.isRetained());
+            latch.countDown();
+         }
+      });
+      consumer.connect();
+      MqttSubscription subscription = new MqttSubscription(topicFilter, 1);
+      subscription.setRetainAsPublished(retainAsPublished);
+      consumer.subscribe(new MqttSubscription[]{subscription});
+
+      // send retained message
+      MqttClient producer = createPahoClient("producer");
+      producer.connect();
+      producer.publish(TOPIC, "retained".getBytes(), 1, true);
+      producer.disconnect();
+      producer.close();
+
+      assertTrue(latch.await(2, TimeUnit.SECONDS));
+      assertEquals(retainAsPublished, retained.get());
+      consumer.disconnect();
+      consumer.close();
+   }
+
    /**
     * [MQTT-3.3.2-3] The Topic Name in a PUBLISH packet sent by a Server to a subscribing Client MUST match the
     * Subscription’s Topic Filter.


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@artemis.apache.org
For additional commands, e-mail: commits-help@artemis.apache.org