QoS2 duplicate PUBLISH stuff

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

@@ -62,8 +62,8 @@ public interface MQTTLogger {
    @LogMessage(id = 834008, value = "Failed to remove session state for client with ID: {}", level = LogMessage.Level.ERROR)
    void failedToRemoveSessionState(String clientID, Exception e);
 
-   @LogMessage(id = 834009, value = "Ignoring duplicate MQTT QoS2 PUBLISH; packet ID: {}; client ID: {}.", level = LogMessage.Level.WARN)
-   void ignoringQoS2Publish(long packetId, String clientId);
+   @LogMessage(id = 834009, value = "Ignoring duplicate MQTT QoS2 PUBLISH (DUP flag not set); packet ID: {}; client ID: {}. Unexpected reuse of an in-flight packet ID.", level = LogMessage.Level.WARN)
+   void ignoringUnexpectedDuplicatePacketId(long packetId, String clientId);
 
    @LogMessage(id = 834010, value = "Unable to scan MQTT sessions", level = LogMessage.Level.ERROR)
    void unableToScanSessions(Exception e);
@@ -85,4 +85,7 @@ public interface MQTTLogger {
 
    @LogMessage(id = 834016, value = "Storage operation failed. Error code: {}; message: {}", level = LogMessage.Level.ERROR)
    void storageOperationError(int errorCode, String errorMessage);
+
+   @LogMessage(id = 834017, value = "Ignoring duplicate MQTT QoS2 PUBLISH (DUP flag set); packet ID: {}; client ID: {}. Expected reuse of an in-flight packet ID.", level = LogMessage.Level.INFO)
+   void ignoringExpectedDuplicatePacketId(long packetId, String clientId);
 }

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

@@ -341,8 +341,16 @@ void handlePublish(MqttPublishMessage message) throws Exception {
       }
 
       if (message.fixedHeader().qosLevel().value() == 2 && session.getState().getPublishCache().contains(message.variableHeader().packetId())) {
-         MQTTLogger.LOGGER.ignoringQoS2Publish(message.variableHeader().packetId(), session.getState().getClientId());
-         sendPubRec(message.variableHeader().packetId(), MQTTReasonCodes.SUCCESS);
+         byte reasonCode = MQTTReasonCodes.SUCCESS;
+         if (message.fixedHeader().isDup()) {
+            MQTTLogger.LOGGER.ignoringExpectedDuplicatePacketId(message.variableHeader().packetId(), session.getState().getClientId());
+         } else {
+            MQTTLogger.LOGGER.ignoringUnexpectedDuplicatePacketId(message.variableHeader().packetId(), session.getState().getClientId());
+            if (session.getVersion() == MQTTVersion.MQTT_5 && session.getProtocolManager().isRejectUnexpectedDuplicatePacketId()) {
+               reasonCode = MQTTReasonCodes.PACKET_IDENTIFIER_IN_USE;
+            }
+         }
+         sendPubRec(message.variableHeader().packetId(), reasonCode);
          return;
       }
 

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

@@ -73,6 +73,8 @@ public class MQTTProtocolManager extends AbstractProtocolManager<MqttMessage, MQ
 
    private boolean allowLinkStealing = true;
 
+   private boolean rejectUnexpectedDuplicatePacketId = false;
+
    private int defaultMaximumInFlightPublishMessages = MQTTUtil.DEFAULT_MAXIMUM_IN_FLIGHT_PUBLISH_MESSAGES;
 
    private final MQTTRoutingHandler routingHandler;
@@ -199,6 +201,15 @@ public void setAllowLinkStealing(boolean allowLinkStealing) {
       this.allowLinkStealing = allowLinkStealing;
    }
 
+   public boolean isRejectUnexpectedDuplicatePacketId() {
+      return rejectUnexpectedDuplicatePacketId;
+   }
+
+   public MQTTProtocolManager setRejectUnexpectedDuplicatePacketId(boolean rejectUnexpectedDuplicatePacketId) {
+      this.rejectUnexpectedDuplicatePacketId = rejectUnexpectedDuplicatePacketId;
+      return this;
+   }
+
    public int getDefaultMaximumInFlightPublishMessages() {
       return defaultMaximumInFlightPublishMessages;
    }

docs/user-manual/mqtt.adoc

@@ -307,3 +307,35 @@ However if you'd rather have the broker make a positive acknowledgement then set
 ----
 <acceptor name="mqtt">tcp://0.0.0:1883?protocols=MQTT;closeMqttConnectionOnPublishAuthorizationFailure=false</acceptor>
 ----
+
+== QoS 2 Packet Identifier Reuse
+
+A QoS 2 `PUBLISH` from a client uses a packet identifier which the MQTT specification considers "in use" from the moment the `PUBLISH` is sent until the corresponding `PUBCOMP` completes the exactly-once flow.
+A well-behaved client will never send a _new_ `PUBLISH` with a packet identifier that is still in use.
+If a client loses its outgoing session state but resumes an existing session it may reuse a packet identifier that the broker still considers in use.
+
+When the broker receives a QoS 2 `PUBLISH` whose packet identifier is already in use it inspects the `DUP` flag to distinguish two cases:
+
+* `DUP=1`: an expected retransmission of the original in-flight message. The broker does not deliver the message to subscribers and logs `AMQ834017` at `INFO`.
+* `DUP=0`: an unexpected reuse of an in-flight packet identifier. The broker does not deliver the message to subscribers and logs `AMQ834009` at `WARN`.
+
+By default the broker responds to the client in both cases with a `PUBREC` reason code of `0x00` (i.e. "Success") in order to allow the QoS 2 flow to complete.
+For MQTT 5 clients you can instead have the broker reject the `DUP=0` case with a `PUBREC` reason code of https://docs.oasis-open.org/mqtt/mqtt/v5.0/os/mqtt-v5.0-os.html#_Toc3901147[`0x91` (i.e. "Packet Identifier in use")] by setting the URL parameter `rejectUnexpectedDuplicatePacketId` to `true` on the relevant MQTT `acceptor` in `broker.xml`, e.g.:
+
+[,xml]
+----
+<acceptor name="mqtt">tcp://0.0.0:1883?protocols=MQTT;rejectUnexpectedDuplicatePacketId=true</acceptor>
+----
+
+By default `rejectUnexpectedDuplicatePacketId` is `false`.
+This setting has no effect on MQTT 3.x clients as those versions don't support `PUBREC` reason codes.
+The MQTT 5 specification doesn't require this behavior, but it allows it.
+It is provided as an option to detect and alert non-compliant clients.
+A `PUBREC` reason code >= `0x80` terminates the QoS 2 flow, so the affected client's publish will fail.
+
+Regarding the `0x91` reason code, https://docs.oasis-open.org/mqtt/mqtt/v5.0/os/mqtt-v5.0-os.html#_Toc3901031[section 2.4 of the MQTT 5 specification] states:
+
+[quote,]
+____
+For Reason Code 0x91 (Packet identifier in use), the response to this is either to try to fix the state, or to reset the Session state by connecting using Clean Start set to 1, or to decide if the Client or Server implementations are defective.
+____

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

@@ -28,15 +28,18 @@
 import java.util.concurrent.Executors;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicReference;
 
 import io.netty.handler.codec.mqtt.MqttMessageType;
+import io.netty.handler.codec.mqtt.MqttPubReplyMessageVariableHeader;
 import org.apache.activemq.artemis.api.core.ActiveMQException;
 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.core.paging.impl.PagingManagerImpl;
 import org.apache.activemq.artemis.core.paging.impl.PagingManagerImplAccessor;
+import org.apache.activemq.artemis.core.postoffice.DuplicateIDCache;
 import org.apache.activemq.artemis.core.postoffice.impl.PostOfficeImpl;
 import org.apache.activemq.artemis.core.postoffice.impl.PostOfficeTestAccessor;
 import org.apache.activemq.artemis.core.protocol.mqtt.MQTTInterceptor;
@@ -45,11 +48,13 @@
 import org.apache.activemq.artemis.core.protocol.mqtt.MQTTSessionAccessor;
 import org.apache.activemq.artemis.core.protocol.mqtt.MQTTSessionState;
 import org.apache.activemq.artemis.core.protocol.mqtt.MQTTUtil;
+import org.apache.activemq.artemis.core.protocol.mqtt.PacketIdCache;
 import org.apache.activemq.artemis.core.server.Queue;
 import org.apache.activemq.artemis.core.server.ServerSession;
 import org.apache.activemq.artemis.core.server.plugin.ActiveMQServerSessionPlugin;
 import org.apache.activemq.artemis.core.settings.impl.AddressSettings;
 import org.apache.activemq.artemis.logs.AssertionLoggerHandler;
+import org.apache.activemq.artemis.utils.ByteUtil;
 import org.apache.activemq.artemis.utils.RandomUtil;
 import org.apache.activemq.artemis.utils.ReusableLatch;
 import org.apache.activemq.artemis.utils.Wait;
@@ -1110,4 +1115,114 @@ public void testPublishWithDelimiterInTopicNameAndWildcardSubscription() throws
 
       assertTrue(latch.await(500, TimeUnit.MILLISECONDS));
    }
+
+   /**
+    * A spec-compliant client can never reuse a packet ID whose QoS 2 handshake is still in-flight, so the "reused packet
+    * ID" path (DUP flag not set) can't be reproduced with a normal client. Instead, seed the broker's PUBLISH cache with
+    * the packet ID the client is about to use to simulate a non-compliant client that resumed its session without
+    * preserving its outgoing QoS 2 state. The broker must treat the fresh (DUP=0) PUBLISH as a duplicate and log a WARN
+    * because the new message is silently dropped.
+    */
+   @Test
+   @Timeout(DEFAULT_TIMEOUT_SEC)
+   public void testDuplicateQoS2PublishWithReusedPacketIdLogsWarning() throws Exception {
+      final String TOPIC = RandomUtil.randomUUIDString();
+      final String CLIENT_ID = "publisher";
+      // Paho assigns packet ID 1 to the first QoS > 0 message sent on a fresh connection
+      final int PACKET_ID = 1;
+
+      server.createQueue(QueueConfiguration.of(TOPIC)
+                            .setAddress(TOPIC)
+                            .setRoutingType(RoutingType.MULTICAST)
+                            .setDurable(true));
+
+      MqttClient publisher = createPahoClient(CLIENT_ID);
+      publisher.connect(new MqttConnectionOptionsBuilder().cleanStart(false).sessionExpiryInterval(300L).build());
+
+      // Seed the broker's PUBLISH cache with the packet ID the client is about to use.
+      SimpleString cacheName = PacketIdCache.getCacheName(server.getInternalNamingPrefix(), CLIENT_ID, PacketIdCache.TYPE.PUBLISH);
+      DuplicateIDCache pubCache = server.getPostOffice().getDuplicateIDCache(cacheName, MQTTUtil.TWO_BYTE_INT_MAX);
+      pubCache.addToCache(ByteUtil.intToBytes(PACKET_ID), null);
+
+      try (AssertionLoggerHandler loggerHandler = new AssertionLoggerHandler()) {
+         // Fresh (DUP=0) PUBLISH reusing the cached packet ID; blocks until the QoS 2 handshake completes
+         publisher.publish(TOPIC, RandomUtil.randomBytes(), EXACTLY_ONCE, false);
+
+         assertTrue(loggerHandler.findText("AMQ834009"), "expected WARN for reused packet ID");
+         assertFalse(loggerHandler.findText("AMQ834017"), "did not expect the retransmit INFO message");
+      }
+
+      // The "new" message was silently dropped as a duplicate; nothing was delivered to the queue
+      assertEquals(0L, server.locateQueue(TOPIC).getMessageCount());
+
+      publisher.disconnect();
+      publisher.close();
+   }
+
+   /**
+    * Companion to {@link #testDuplicateQoS2PublishWithReusedPacketIdLogsWarning()}. With the
+    * {@code rejectQoS2PublishWithReusedPacketId} setting enabled the broker must respond to the DUP=0 reused-packet-ID
+    * case with a {@code PUBREC} reason code of {@code 0x91} (i.e. "Packet Identifier in use") instead of {@code 0x00}
+    * (i.e. "Success").
+    */
+   @Test
+   @Timeout(DEFAULT_TIMEOUT_SEC)
+   public void testDuplicateQoS2PublishWithReusedPacketIdRejected() throws Exception {
+      setAcceptorProperty("rejectQoS2PublishWithReusedPacketId=true");
+
+      final String TOPIC = RandomUtil.randomUUIDString();
+      final String CLIENT_ID = "publisher";
+      // Paho assigns packet ID 1 to the first QoS > 0 message sent on a fresh connection
+      final int PACKET_ID = 1;
+
+      server.createQueue(QueueConfiguration.of(TOPIC)
+                            .setAddress(TOPIC)
+                            .setRoutingType(RoutingType.MULTICAST)
+                            .setDurable(true));
+
+      // capture the reason code of the outgoing PUBREC
+      AtomicInteger pubRecReasonCode = new AtomicInteger(-1);
+      CountDownLatch pubRecLatch = new CountDownLatch(1);
+      MQTTInterceptor outgoingInterceptor = (packet, connection) -> {
+         if (packet.fixedHeader().messageType() == MqttMessageType.PUBREC && packet.variableHeader() instanceof MqttPubReplyMessageVariableHeader header) {
+            pubRecReasonCode.set(header.reasonCode() & 0xFF);
+            pubRecLatch.countDown();
+         }
+         return true;
+      };
+      server.getRemotingService().addOutgoingInterceptor(outgoingInterceptor);
+
+      MqttClient publisher = createPahoClient(CLIENT_ID);
+      publisher.connect(new MqttConnectionOptionsBuilder().cleanStart(false).sessionExpiryInterval(300L).build());
+
+      // Seed the broker's PUBLISH cache with the packet ID the client is about to use.
+      SimpleString cacheName = PacketIdCache.getCacheName(server.getInternalNamingPrefix(), CLIENT_ID, PacketIdCache.TYPE.PUBLISH);
+      DuplicateIDCache pubCache = server.getPostOffice().getDuplicateIDCache(cacheName, MQTTUtil.TWO_BYTE_INT_MAX);
+      pubCache.addToCache(ByteUtil.intToBytes(PACKET_ID), null);
+
+      try (AssertionLoggerHandler loggerHandler = new AssertionLoggerHandler()) {
+         // Fresh (DUP=0) PUBLISH reusing the cached packet ID; the broker rejects it so Paho reports the reason code
+         try {
+            publisher.publish(TOPIC, RandomUtil.randomBytes(), EXACTLY_ONCE, false);
+            fail("expected the publish to fail with a 'packet identifier in use' reason code");
+         } catch (MqttException e) {
+            assertEquals(MQTTReasonCodes.PACKET_IDENTIFIER_IN_USE & 0xFF, e.getReasonCode(), "expected reason code 0x91");
+         }
+
+         assertTrue(loggerHandler.findText("AMQ834009"), "expected WARN for reused packet ID");
+      }
+
+      assertTrue(pubRecLatch.await(2, TimeUnit.SECONDS), "expected a PUBREC to be sent");
+      assertEquals(MQTTReasonCodes.PACKET_IDENTIFIER_IN_USE & 0xFF, pubRecReasonCode.get(), "expected PUBREC reason code 0x91");
+
+      // The "new" message was silently dropped as a duplicate; nothing was delivered to the queue
+      assertEquals(0L, server.locateQueue(TOPIC).getMessageCount());
+
+      try {
+         publisher.disconnect();
+      } catch (MqttException e) {
+         // the client may already be disconnected as a result of the rejected publish
+      }
+      publisher.close();
+   }
 }

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

@@ -24,6 +24,7 @@
 import org.apache.activemq.artemis.api.core.QueueConfiguration;
 import org.apache.activemq.artemis.api.core.RoutingType;
 import org.apache.activemq.artemis.core.protocol.mqtt.MQTTInterceptor;
+import org.apache.activemq.artemis.logs.AssertionLoggerHandler;
 import org.apache.activemq.artemis.utils.RandomUtil;
 import org.apache.activemq.artemis.utils.ReusableLatch;
 import org.apache.activemq.artemis.utils.Wait;
@@ -36,6 +37,7 @@
 import org.junit.jupiter.api.Timeout;
 
 import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
 import static org.junit.jupiter.api.Assertions.assertNull;
 import static org.junit.jupiter.api.Assertions.assertTrue;
 
@@ -201,6 +203,7 @@ public void testQoS2ClientDisconnectAfterPubRecSent() throws Exception {
    }
 
    public void testQoS2FailureAfterPubRecSent(boolean restart) throws Exception {
+      final String MQTT_LOGGER_NAME = "org.apache.activemq.artemis.core.protocol.mqtt";
       final String TOPIC = RandomUtil.randomUUIDString();
       final String CLIENTID = "publisher";
       final CountDownLatch pubRecLatch = new CountDownLatch(1);
@@ -272,11 +275,20 @@ public void disconnected(MqttDisconnectResponse disconnectResponse) {
 
       assertEquals(1, getPubCacheSize(CLIENTID));
 
-      // The client will automatically re-initiate the QoS2 protocol after reconnecting since it never got a PUBREC
-      reconnectSafely(publisher);
+      AssertionLoggerHandler.LogLevel previousLevel = AssertionLoggerHandler.setLevel(MQTT_LOGGER_NAME, AssertionLoggerHandler.LogLevel.INFO);
+      try (AssertionLoggerHandler loggerHandler = new AssertionLoggerHandler()) {
+         // The client will automatically re-initiate the QoS2 protocol after reconnecting since it never got a PUBREC.
+         // The retransmitted PUBLISH has the DUP flag set, so the broker recognizes the expected duplicate and logs at INFO.
+         reconnectSafely(publisher);
 
-      // Wait for the PUBCOMP to confirm QoS2 protocol is done
-      assertTrue(pubCompLatch.await(5, TimeUnit.SECONDS));
+         // Wait for the PUBCOMP to confirm QoS2 protocol is done
+         assertTrue(pubCompLatch.await(5, TimeUnit.SECONDS));
+
+         assertTrue(loggerHandler.findText("AMQ834017"), "expected INFO log for the QoS2 PUBLISH retransmission");
+         assertFalse(loggerHandler.findText("AMQ834009"), "did not expect WARN for reused packet ID");
+      } finally {
+         AssertionLoggerHandler.setLevel(MQTT_LOGGER_NAME, previousLevel);
+      }
 
       // Verify only one message is in the queue despite the QoS2 interruption
       Wait.assertEquals(1L, () -> server.locateQueue(TOPIC).getMessageCount(), 500, 25);