Repository: moquette-io/moquette (java)

Issue:
QoS 1/2 retained message publishing doesn't work correctly
### Expected behavior
Sending QoS 1/2 retained messages to a subscriber who is initially not connected should work without error.

### Actual behavior
QoS 1 and 2 retained messages do not setup the necessary state management in Moquette which leads to failures, most specifically for QoS 2. When sending retained high-QoS messages, the inflightWindow map isn't used to keep the publish state as it would be when there is an existing session. Instead if uses this code path: https://github.com/moquette-io/moquette/blob/9ed6ca9d373e5315a58053ace88107b005c47de7/broker/src/main/java/io/moquette/broker/MQTTConnection.java#L456 which will publish the message to the client, but it doesn't setup the session structures correctly to accept the PUBACK, PUBREC, etc. We therefore get the error `Received a PUBACK with not matching packetId` and `Received a PUBREC with not matching packetId`

### Steps to reproduce

https://github.com/eclipse/paho.mqtt.testing/blob/8f5ff2d47d6099f1624958299e782034632772d2/interoperability/client_test.py#L160-L164

### Minimal yet complete reproducer code (or URL to code) or complete log file

### Moquette MQTT version
Tip of main (8e2a0b830b9c4e5110caa859253a1d458ab3b1d0)

### JVM version (e.g. `java -version`)

### OS version (e.g. `uname -a`)


Pull request:
Fix 653 retained messages publishing
This PR adds a test for, and fixes #653.

Retained messages used a difference code-path that skipped the Session, and thus did not properly set up in-flight status. 
To fix this, this PR:
- Adds a test for retained messages at different QoS levels
- Makes the retained messages use the same code path as non-retained messages
- Removes the retained-specific code.
- Adds a `retained` flag to this code-path
- Adds a `retained` flag to `SessionRegistry.PublishedMessage` so the flag is stored for queued messages

Patch:
diff --git a/broker/src/main/java/io/moquette/broker/MQTTConnection.java b/broker/src/main/java/io/moquette/broker/MQTTConnection.java
index aa77e169..170990fa 100644
--- a/broker/src/main/java/io/moquette/broker/MQTTConnection.java
+++ b/broker/src/main/java/io/moquette/broker/MQTTConnection.java
@@ -527,41 +527,39 @@ final class MQTTConnection {
         return NettyUtils.userName(channel);
     }
 
-    public void sendPublishRetainedQos0(Topic topic, MqttQoS qos, ByteBuf payload) {
-        MqttPublishMessage publishMsg = retainedPublish(topic.toString(), qos, payload);
+    public void sendPublishWithPacketId(Topic topic, MqttQoS qos, ByteBuf payload, boolean retained) {
+        final int packetId = nextPacketId();
+        MqttPublishMessage publishMsg = createPublishMessage(topic.toString(), qos, payload, packetId, retained);
         sendPublish(publishMsg);
     }
 
-    public void sendPublishRetainedWithPacketId(Topic topic, MqttQoS qos, ByteBuf payload) {
-        final int packetId = nextPacketId();
-        MqttPublishMessage publishMsg = retainedPublishWithMessageId(topic.toString(), qos, payload, packetId);
+    // TODO move this method in Session
+    void sendPublishQos0(Topic topic, MqttQoS qos, ByteBuf payload, boolean retained) {
+        MqttPublishMessage publishMsg = createPublishMessage(topic.toString(), qos, payload, 0, retained);
         sendPublish(publishMsg);
     }
 
-    private static MqttPublishMessage retainedPublish(String topic, MqttQoS qos, ByteBuf message) {
-        return retainedPublishWithMessageId(topic, qos, message, 0);
+    static MqttPublishMessage createRetainedPublishMessage(String topic, MqttQoS qos, ByteBuf message) {
+        return createPublishMessage(topic, qos, message, 0, true);
     }
 
-    private static MqttPublishMessage retainedPublishWithMessageId(String topic, MqttQoS qos, ByteBuf message,
-                                                                   int messageId) {
-        MqttFixedHeader fixedHeader = new MqttFixedHeader(MqttMessageType.PUBLISH, false, qos, true, 0);
-        MqttPublishVariableHeader varHeader = new MqttPublishVariableHeader(topic, messageId);
-        return new MqttPublishMessage(fixedHeader, varHeader, message);
+    static MqttPublishMessage createNonRetainedPublishMessage(String topic, MqttQoS qos, ByteBuf message) {
+        return createPublishMessage(topic, qos, message, 0, false);
     }
 
-    // TODO move this method in Session
-    void sendPublishNotRetainedQos0(Topic topic, MqttQoS qos, ByteBuf payload) {
-        MqttPublishMessage publishMsg = notRetainedPublish(topic.toString(), qos, payload);
-        sendPublish(publishMsg);
+    static MqttPublishMessage createRetainedPublishMessage(String topic, MqttQoS qos, ByteBuf message,
+                                                           int messageId) {
+        return createPublishMessage(topic, qos, message, messageId, true);
     }
 
-    static MqttPublishMessage notRetainedPublish(String topic, MqttQoS qos, ByteBuf message) {
-        return notRetainedPublishWithMessageId(topic, qos, message, 0);
+    static MqttPublishMessage createNotRetainedPublishMessage(String topic, MqttQoS qos, ByteBuf message,
+                                                              int messageId) {
+        return createPublishMessage(topic, qos, message, messageId, false);
     }
 
-    static MqttPublishMessage notRetainedPublishWithMessageId(String topic, MqttQoS qos, ByteBuf message,
-                                                              int messageId) {
-        MqttFixedHeader fixedHeader = new MqttFixedHeader(MqttMessageType.PUBLISH, false, qos, false, 0);
+    private static MqttPublishMessage createPublishMessage(String topic, MqttQoS qos, ByteBuf message,
+                                                           int messageId, boolean retained) {
+        MqttFixedHeader fixedHeader = new MqttFixedHeader(MqttMessageType.PUBLISH, false, qos, retained, 0);
         MqttPublishVariableHeader varHeader = new MqttPublishVariableHeader(topic, messageId);
         return new MqttPublishMessage(fixedHeader, varHeader, message);
     }
diff --git a/broker/src/main/java/io/moquette/broker/NewNettyAcceptor.java b/broker/src/main/java/io/moquette/broker/NewNettyAcceptor.java
index f1e59d00..20293d66 100644
--- a/broker/src/main/java/io/moquette/broker/NewNettyAcceptor.java
+++ b/broker/src/main/java/io/moquette/broker/NewNettyAcceptor.java
@@ -223,7 +223,7 @@ class NewNettyAcceptor {
         } catch (Exception ex) {
             if (ex instanceof BindException) {
                LOG.error("Cannot bind to port: " + port, ex);
-               System.exit(1);
+               throw new RuntimeException("Cannot bind to port: " + port, ex);
             } else {
                 LOG.error("An interruptedException was caught while initializing integration. Protocol={}", protocol, ex);
                 throw new RuntimeException(ex);
diff --git a/broker/src/main/java/io/moquette/broker/PostOffice.java b/broker/src/main/java/io/moquette/broker/PostOffice.java
index 507d80ba..d4278852 100644
--- a/broker/src/main/java/io/moquette/broker/PostOffice.java
+++ b/broker/src/main/java/io/moquette/broker/PostOffice.java
@@ -475,7 +475,7 @@ class PostOffice {
         if (isSessionPresent) {
             LOG.debug("Sending PUBLISH message to active subscriber CId: {}, topicFilter: {}, qos: {}",
                       sub.getClientId(), sub.getTopicFilter(), qos);
-            targetSession.sendPublishOnSessionAtQos(topic, qos, payload);
+            targetSession.sendNotRetainedPublishOnSessionAtQos(topic, qos, payload);
         } else {
             // If we are, the subscriber disconnected after the subscriptions tree selected that session as a
             // destination.
diff --git a/broker/src/main/java/io/moquette/broker/Server.java b/broker/src/main/java/io/moquette/broker/Server.java
index 1fab4260..42b9aa3c 100644
--- a/broker/src/main/java/io/moquette/broker/Server.java
+++ b/broker/src/main/java/io/moquette/broker/Server.java
@@ -55,7 +55,11 @@ public class Server {
 
     public static void main(String[] args) throws IOException {
         final Server server = new Server();
-        server.startServer();
+        try {
+            server.startServer();
+        } catch (RuntimeException e) {
+            System.exit(1);
+        }
         System.out.println("Server started, version 0.16-SNAPSHOT");
         //Bind a shutdown hook
         Runtime.getRuntime().addShutdownHook(new Thread(server::stopServer));
diff --git a/broker/src/main/java/io/moquette/broker/Session.java b/broker/src/main/java/io/moquette/broker/Session.java
index e8a55a9b..57150cae 100644
--- a/broker/src/main/java/io/moquette/broker/Session.java
+++ b/broker/src/main/java/io/moquette/broker/Session.java
@@ -229,25 +229,33 @@ class Session {
 //                m_interceptor.notifyMessageAcknowledged(interceptAckMsg);
     }
 
-    public void sendPublishOnSessionAtQos(Topic topic, MqttQoS qos, ByteBuf payload) {
+    public void sendRetainedPublishOnSessionAtQos(Topic topic, MqttQoS qos, ByteBuf payload) {
+        sendPublishOnSessionAtQos(topic, qos, payload, true);
+    }
+
+    public void sendNotRetainedPublishOnSessionAtQos(Topic topic, MqttQoS qos, ByteBuf payload) {
+        sendPublishOnSessionAtQos(topic, qos, payload, false);
+    }
+
+    private void sendPublishOnSessionAtQos(Topic topic, MqttQoS qos, ByteBuf payload, boolean retained) {
         switch (qos) {
             case AT_MOST_ONCE:
                 if (connected()) {
-                    mqttConnection.sendPublishNotRetainedQos0(topic, qos, payload);
+                    mqttConnection.sendPublishQos0(topic, qos, payload, retained);
                 }
                 break;
             case AT_LEAST_ONCE:
-                sendPublishQos1(topic, qos, payload);
+                sendPublishQos1(topic, qos, payload, retained);
                 break;
             case EXACTLY_ONCE:
-                sendPublishQos2(topic, qos, payload);
+                sendPublishQos2(topic, qos, payload, retained);
                 break;
             case FAILURE:
                 LOG.error("Not admissible");
         }
     }
 
-    private void sendPublishQos1(Topic topic, MqttQoS qos, ByteBuf payload) {
+    private void sendPublishQos1(Topic topic, MqttQoS qos, ByteBuf payload, boolean retained) {
         if (!connected() && isClean()) {
             //pushing messages to disconnected not clean session
             return;
@@ -260,7 +268,7 @@ class Session {
 
             // Adding to a map, retain.
             payload.retain();
-            EnqueuedMessage old = inflightWindow.put(packetId, new PublishedMessage(topic, qos, payload));
+            EnqueuedMessage old = inflightWindow.put(packetId, new PublishedMessage(topic, qos, payload, retained));
             // If there already was something, release it.
             if (old != null) {
                 old.release();
@@ -268,7 +276,7 @@ class Session {
             }
             inflightTimeouts.add(new InFlightPacket(packetId, FLIGHT_BEFORE_RESEND_MS));
 
-            MqttPublishMessage publishMsg = MQTTConnection.notRetainedPublishWithMessageId(topic.toString(), qos,
+            MqttPublishMessage publishMsg = MQTTConnection.createNotRetainedPublishMessage(topic.toString(), qos,
                                                                                            payload, packetId);
             localMqttConnectionRef.sendPublish(publishMsg);
             LOG.debug("Write direct to the peer, inflight slots: {}", inflightSlots.get());
@@ -278,7 +286,7 @@ class Session {
 
             // TODO drainQueueToConnection();?
         } else {
-            final SessionRegistry.PublishedMessage msg = new SessionRegistry.PublishedMessage(topic, qos, payload);
+            final SessionRegistry.PublishedMessage msg = new SessionRegistry.PublishedMessage(topic, qos, payload, retained);
             // Adding to a queue, retain.
             msg.retain();
             sessionQueue.add(msg);
@@ -286,7 +294,7 @@ class Session {
         }
     }
 
-    private void sendPublishQos2(Topic topic, MqttQoS qos, ByteBuf payload) {
+    private void sendPublishQos2(Topic topic, MqttQoS qos, ByteBuf payload, boolean retained) {
         final MQTTConnection localMqttConnectionRef = mqttConnection;
         if (canSkipQueue(localMqttConnectionRef)) {
             inflightSlots.decrementAndGet();
@@ -294,7 +302,7 @@ class Session {
 
             // Retain before adding to map
             payload.retain();
-            EnqueuedMessage old = inflightWindow.put(packetId, new SessionRegistry.PublishedMessage(topic, qos, payload));
+            EnqueuedMessage old = inflightWindow.put(packetId, new SessionRegistry.PublishedMessage(topic, qos, payload, retained));
             // If there already was something, release it.
             if (old != null) {
                 old.release();
@@ -302,13 +310,13 @@ class Session {
             }
             inflightTimeouts.add(new InFlightPacket(packetId, FLIGHT_BEFORE_RESEND_MS));
 
-            MqttPublishMessage publishMsg = MQTTConnection.notRetainedPublishWithMessageId(topic.toString(), qos,
+            MqttPublishMessage publishMsg = MQTTConnection.createNotRetainedPublishMessage(topic.toString(), qos,
                                                                                            payload, packetId);
             localMqttConnectionRef.sendPublish(publishMsg);
 
             drainQueueToConnection();
         } else {
-            final SessionRegistry.PublishedMessage msg = new SessionRegistry.PublishedMessage(topic, qos, payload);
+            final SessionRegistry.PublishedMessage msg = new SessionRegistry.PublishedMessage(topic, qos, payload, retained);
             // Adding to a queue, retain.
             msg.retain();
             sessionQueue.add(msg);
@@ -413,7 +421,7 @@ class Session {
             }
             inflightTimeouts.add(new InFlightPacket(sendPacketId, FLIGHT_BEFORE_RESEND_MS));
             final SessionRegistry.PublishedMessage msgPub = (SessionRegistry.PublishedMessage) msg;
-            MqttPublishMessage publishMsg = MQTTConnection.notRetainedPublishWithMessageId(
+            MqttPublishMessage publishMsg = MQTTConnection.createNotRetainedPublishMessage(
                 msgPub.topic.toString(),
                 msgPub.publishingQos,
                 msgPub.payload, sendPacketId);
@@ -432,15 +440,6 @@ class Session {
         drainQueueToConnection();
     }
 
-    void sendRetainedPublishOnSessionAtQos(Topic topic, MqttQoS qos, ByteBuf payload) {
-        if (qos != MqttQoS.AT_MOST_ONCE) {
-            // QoS 1 or 2
-            mqttConnection.sendPublishRetainedWithPacketId(topic, qos, payload);
-        } else {
-            mqttConnection.sendPublishRetainedQos0(topic, qos, payload);
-        }
-    }
-
     public void receivedPublishQos2(int messageID, MqttPublishMessage msg) {
         // Retain before putting msg in map.
         ReferenceCountUtil.retain(msg);
diff --git a/broker/src/main/java/io/moquette/broker/SessionRegistry.java b/broker/src/main/java/io/moquette/broker/SessionRegistry.java
index 8c7af1bf..bd73608c 100644
--- a/broker/src/main/java/io/moquette/broker/SessionRegistry.java
+++ b/broker/src/main/java/io/moquette/broker/SessionRegistry.java
@@ -57,11 +57,13 @@ public class SessionRegistry {
         final Topic topic;
         final MqttQoS publishingQos;
         final ByteBuf payload;
+        final boolean retained;
 
-        public PublishedMessage(Topic topic, MqttQoS publishingQos, ByteBuf payload) {
+        public PublishedMessage(Topic topic, MqttQoS publishingQos, ByteBuf payload, boolean retained) {
             this.topic = topic;
             this.publishingQos = publishingQos;
             this.payload = payload;
+            this.retained = false;
         }
 
         public Topic getTopic() {
diff --git a/broker/src/main/java/io/moquette/persistence/EnqueuedMessageValueType.java b/broker/src/main/java/io/moquette/persistence/EnqueuedMessageValueType.java
index a78ac012..a277ebda 100644
--- a/broker/src/main/java/io/moquette/persistence/EnqueuedMessageValueType.java
+++ b/broker/src/main/java/io/moquette/persistence/EnqueuedMessageValueType.java
@@ -82,7 +82,7 @@ public final class EnqueuedMessageValueType implements org.h2.mvstore.type.DataT
             final MqttQoS qos = MqttQoS.valueOf(buff.get());
             final String topicStr = topicDataType.read(buff);
             final ByteBuf payload = payloadDataType.read(buff);
-            return new SessionRegistry.PublishedMessage(Topic.asTopic(topicStr), qos, payload);
+            return new SessionRegistry.PublishedMessage(Topic.asTopic(topicStr), qos, payload, false);
         } else {
             throw new IllegalArgumentException("Can't recognize record of type: " + messageType);
         }


Test patch:
diff --git a/broker/src/test/java/io/moquette/broker/SessionRegistryTest.java b/broker/src/test/java/io/moquette/broker/SessionRegistryTest.java
index 0f6eb5ba..d6884c42 100644
--- a/broker/src/test/java/io/moquette/broker/SessionRegistryTest.java
+++ b/broker/src/test/java/io/moquette/broker/SessionRegistryTest.java
@@ -154,7 +154,7 @@ public class SessionRegistryTest {
 
         final ByteBuf payload = Unpooled.wrappedBuffer("Hello World!".getBytes(StandardCharsets.UTF_8));
         SessionRegistry.PublishedMessage msg = new SessionRegistry.PublishedMessage(Topic.asTopic("/say"),
-            MqttQoS.AT_LEAST_ONCE, payload);
+            MqttQoS.AT_LEAST_ONCE, payload, false);
         try {
             // store a message in the MVStore
             final String mapName = "test_map";
diff --git a/broker/src/test/java/io/moquette/broker/SessionTest.java b/broker/src/test/java/io/moquette/broker/SessionTest.java
index 5bcef6c5..1c8b02bf 100644
--- a/broker/src/test/java/io/moquette/broker/SessionTest.java
+++ b/broker/src/test/java/io/moquette/broker/SessionTest.java
@@ -49,7 +49,7 @@ public class SessionTest {
 
         // Verify
         assertTrue(queuedMessages.isEmpty(), "Messages should be drained");
-        
+
         // release the rest, to avoid leaking buffers
         for (int i = 2; i <= 11; i++) {
             client.pubAckReceived(i);
@@ -60,7 +60,7 @@ public class SessionTest {
 
     private void sendQoS1To(Session client, Topic destinationTopic, String message) {
         final ByteBuf payload = ByteBufUtil.writeUtf8(UnpooledByteBufAllocator.DEFAULT, message);
-        client.sendPublishOnSessionAtQos(destinationTopic, MqttQoS.AT_LEAST_ONCE, payload);
+        client.sendNotRetainedPublishOnSessionAtQos(destinationTopic, MqttQoS.AT_LEAST_ONCE, payload);
     }
 
     @Test
diff --git a/broker/src/test/java/io/moquette/integration/ServerIntegrationRetainTest.java b/broker/src/test/java/io/moquette/integration/ServerIntegrationRetainTest.java
new file mode 100644
index 00000000..4355bf5a
--- /dev/null
+++ b/broker/src/test/java/io/moquette/integration/ServerIntegrationRetainTest.java
@@ -0,0 +1,233 @@
+/*
+ * Copyright (c) 2012-2022 The original author or authors
+ * ------------------------------------------------------
+ * All rights reserved. This program and the accompanying materials
+ * are made available under the terms of the Eclipse Public License v1.0
+ * and Apache License v2.0 which accompanies this distribution.
+ *
+ * The Eclipse Public License is available at
+ * http://www.eclipse.org/legal/epl-v10.html
+ *
+ * The Apache License v2.0 is available at
+ * http://www.opensource.org/licenses/apache2.0.php
+ *
+ * You may elect to redistribute this code under either of these licenses.
+ */
+package io.moquette.integration;
+
+import io.moquette.broker.Server;
+import io.moquette.broker.config.IConfig;
+import io.moquette.broker.config.MemoryConfig;
+import org.awaitility.Awaitility;
+import org.awaitility.Durations;
+import org.eclipse.paho.client.mqttv3.IMqttClient;
+import org.eclipse.paho.client.mqttv3.MqttClient;
+import org.eclipse.paho.client.mqttv3.MqttMessage;
+import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.io.TempDir;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.nio.file.Path;
+import java.util.Properties;
+
+import static java.nio.charset.StandardCharsets.UTF_8;
+import java.util.stream.Stream;
+import org.awaitility.core.ConditionTimeoutException;
+import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
+import org.eclipse.paho.client.mqttv3.MqttException;
+import org.junit.jupiter.api.AfterAll;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.Arguments;
+import static org.junit.jupiter.params.provider.Arguments.arguments;
+import org.junit.jupiter.params.provider.MethodSource;
+
+public class ServerIntegrationRetainTest {
+
+    private static final Logger LOG = LoggerFactory.getLogger(ServerIntegrationRetainTest.class);
+
+    private static IConfig serverConfig;
+    private static Server server;
+
+    private IMqttClient clientSubscriber;
+    private IMqttClient clientPublisher;
+    private MessageCollector callbackPublisher;
+    private MessageCollector callbackSubscriber;
+
+    @TempDir
+    static Path tempFolder;
+
+    private static void startServer(String dbPath) throws IOException {
+        server = new Server();
+        final Properties configProps = IntegrationUtils.prepareTestProperties(dbPath);
+        serverConfig = new MemoryConfig(configProps);
+        server.startServer(serverConfig);
+    }
+
+    @BeforeAll
+    public static void beforeTests() throws IOException {
+        Awaitility.setDefaultTimeout(Durations.FIVE_SECONDS);
+        String dbPath = IntegrationUtils.tempH2Path(tempFolder);
+        startServer(dbPath);
+    }
+
+    @AfterAll
+    public static void afterTests() {
+        server.stopServer();
+        server = null;
+        serverConfig = null;
+    }
+
+    @BeforeEach
+    public void setUp() throws MqttException {
+        final MqttConnectOptions mqttConnectOptions = new MqttConnectOptions();
+        mqttConnectOptions.setCleanSession(true);
+
+        clientSubscriber = new MqttClient("tcp://localhost:1883", "Subscriber", new MemoryPersistence());
+        callbackSubscriber = new MessageCollector();
+        clientSubscriber.setCallback(callbackSubscriber);
+        clientSubscriber.connect(mqttConnectOptions);
+
+        clientPublisher = new MqttClient("tcp://localhost:1883", "Publisher", new MemoryPersistence());
+        callbackPublisher = new MessageCollector();
+        clientPublisher.setCallback(callbackPublisher);
+        clientPublisher.connect(mqttConnectOptions);
+    }
+
+    @AfterEach
+    public void tearDown() throws Exception {
+        if (clientPublisher.isConnected()) {
+            clientPublisher.disconnect();
+        }
+
+        if (clientSubscriber.isConnected()) {
+            clientSubscriber.disconnect();
+        }
+    }
+
+    private String createMessage(int qosPub, int qosSub) {
+        return "Hello world MQTT " + qosPub + " " + qosSub;
+    }
+
+    private String createTopic(int qosPub, int qosSub) {
+        return "/topic" + qosPub + qosSub;
+    }
+
+    private void sendRetainedAndSubscribe(int qosPub, int qosSub) throws MqttException {
+        String topic = createTopic(qosPub, qosSub);
+        String messageString = createMessage(qosPub, qosSub);
+        clientPublisher.subscribe(topic);
+        callbackPublisher.reinit();
+        clientPublisher.publish(topic, messageString.getBytes(UTF_8), qosPub, true);
+        // Wait for the publish to finish
+        Awaitility.await().until(() -> callbackPublisher.isMessageReceived());
+        validateRetainedFlagNotSet(callbackPublisher.getMessageImmediate());
+
+        callbackSubscriber.reinit();
+        clientSubscriber.subscribe(topic, qosSub);
+        try {
+            Awaitility.await().until(() -> callbackSubscriber.isMessageReceived());
+        } catch (ConditionTimeoutException ex) {
+            // This may be fine.
+        }
+    }
+
+    private void unsubscribeSubscriber(int qosPub, int qosSub) throws MqttException {
+        String topic = createTopic(qosPub, qosSub);
+        clientSubscriber.unsubscribe(topic);
+    }
+
+    private void sendEmptyRetainedAndSubscribe(int qosPub, int qosSub) throws MqttException {
+        String topic = createTopic(qosPub, qosSub);
+        callbackPublisher.reinit();
+        clientPublisher.publish(topic, new byte[0], qosPub, true);
+        // Wait for the publish to finish
+        Awaitility.await().until(() -> callbackPublisher.isMessageReceived());
+        validateRetainedFlagNotSet(callbackPublisher.getMessageImmediate());
+
+        callbackSubscriber.reinit();
+        clientSubscriber.subscribe(topic, qosSub);
+        try {
+            Awaitility.await().until(() -> callbackSubscriber.isMessageReceived());
+        } catch (ConditionTimeoutException ex) {
+            // This may be fine.
+        }
+    }
+
+    private void validateRetainedFlagNotSet(MqttMessage message) {
+        assertFalse(message.isRetained(), "Directly published version of messages should not have the retained flag set");
+    }
+
+    private void validateMustReceive(int qosPub, int qosSub) {
+        final boolean messageReceived = callbackSubscriber.isMessageReceived();
+        assertTrue(messageReceived, "Expected a message retained at QoS " + qosPub + ".");
+        MqttMessage message = callbackSubscriber.getMessageImmediate();
+        String expectedMessage = createMessage(qosPub, qosSub);
+        assertEquals(expectedMessage, message.toString());
+        assertEquals(Math.min(qosPub, qosSub), message.getQos());
+    }
+
+    private void validateMustNotReceive(int qosPub) {
+        final boolean messageReceived = callbackSubscriber.isMessageReceived();
+        MqttMessage message = callbackSubscriber.getMessageImmediate();
+        assertFalse(messageReceived, "Received an unexpected message retained at QoS " + qosPub + ": " + message);
+    }
+
+    static Stream<Arguments> notRetainedProvider() {
+        return Stream.of(
+            arguments(0, 0),
+            arguments(0, 1),
+            arguments(0, 2)
+        );
+    }
+
+    static Stream<Arguments> retainedProvider() {
+        return Stream.of(
+            arguments(1, 0),
+            arguments(1, 1),
+            arguments(1, 2),
+            arguments(2, 0),
+            arguments(2, 1),
+            arguments(2, 2)
+        );
+    }
+
+    @ParameterizedTest
+    @MethodSource("notRetainedProvider")
+    public void checkShouldNotRetain(int qosPub, int qosSub) throws MqttException {
+        LOG.info("*** checkShouldNotRetain: qosPub {}, qosSub {} ***", qosPub, qosSub);
+        sendRetainedAndSubscribe(qosPub, qosSub);
+        validateMustNotReceive(qosPub);
+    }
+
+    @ParameterizedTest
+    @MethodSource("retainedProvider")
+    public void checkShouldRetain(int qosPub, int qosSub) throws MqttException {
+        LOG.info("*** checkShouldRetain: qosPub {}, qosSub {} ***", qosPub, qosSub);
+        sendRetainedAndSubscribe(qosPub, qosSub);
+        validateMustReceive(qosPub, qosSub);
+        unsubscribeSubscriber(qosPub, qosSub);
+        sendEmptyRetainedAndSubscribe(qosPub, qosSub);
+        validateMustNotReceive(qosPub);
+    }
+
+    @Test
+    public void checkQos0CancelsRetain() throws MqttException {
+        LOG.info("*** checkQos0CancelsRetain ***");
+        // First send a QoS 2 retain, and check it arrives.
+        sendRetainedAndSubscribe(2, 2);
+        validateMustReceive(2, 2);
+        unsubscribeSubscriber(2, 2);
+        // Then send a QoS 0 retain, and check it cancels the previous retain.
+        sendRetainedAndSubscribe(0, 2);
+        validateMustNotReceive(0);
+    }
+}
diff --git a/broker/src/test/java/io/moquette/persistence/H2PersistentQueueTest.java b/broker/src/test/java/io/moquette/persistence/H2PersistentQueueTest.java
index 546ec100..505160c3 100644
--- a/broker/src/test/java/io/moquette/persistence/H2PersistentQueueTest.java
+++ b/broker/src/test/java/io/moquette/persistence/H2PersistentQueueTest.java
@@ -67,7 +67,7 @@ public class H2PersistentQueueTest {
 
     private SessionRegistry.PublishedMessage createMessage(String name) {
         final ByteBuf payload = Unpooled.wrappedBuffer(name.getBytes(StandardCharsets.UTF_8));
-        return new SessionRegistry.PublishedMessage(Topic.asTopic(name), MqttQoS.AT_LEAST_ONCE, payload);
+        return new SessionRegistry.PublishedMessage(Topic.asTopic(name), MqttQoS.AT_LEAST_ONCE, payload, false);
     }
 
     @Test