Fix a bug Session#resendInflightNotAcked() fails due to handling a freed ByteBuf

broker/src/main/java/io/moquette/broker/PostOffice.java

@@ -218,6 +218,8 @@ private void publish2Subscribers(ByteBuf origPayload, Topic topic, MqttQoS publi
                           sub.getClientId(), sub.getTopicFilter(), qos);
                 // we need to retain because duplicate only copy r/w indexes and don't retain() causing refCnt = 0
                 ByteBuf payload = origPayload.retainedDuplicate();
+                // without this retain(), Session.resendInflightNotAcked() fails because it tries to handle a freed ByteBuf
+                payload.retain();
                 targetSession.sendPublishOnSessionAtQos(topic, qos, payload);
             } else {
                 // If we are, the subscriber disconnected after the subscriptions tree selected that session as a

broker/src/main/java/io/moquette/broker/Session.java

@@ -276,7 +276,10 @@ private boolean inflighHasSlotsAndConnectionIsUp() {
 
     void pubAckReceived(int ackPacketId) {
         // TODO remain to invoke in somehow m_interceptor.notifyMessageAcknowledged
-        inflightWindow.remove(ackPacketId);
+        SessionRegistry.EnqueuedMessage message = inflightWindow.remove(ackPacketId);
+        if (message instanceof SessionRegistry.PublishedMessage) {
+            ((SessionRegistry.PublishedMessage) message).payload.release();
+        }
         inflightSlots.incrementAndGet();
         drainQueueToConnection();
     }

broker/src/test/java/io/moquette/integration/ServerLowlevelMessagesIntegrationTests.java

@@ -31,7 +31,10 @@
 import org.awaitility.Awaitility;
 import java.io.IOException;
 import java.util.Properties;
+import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+
 import static io.netty.handler.codec.mqtt.MqttConnectReturnCode.*;
 import static java.nio.charset.StandardCharsets.UTF_8;
 import static org.junit.Assert.*;
@@ -159,4 +162,46 @@ public void testWillMessageIsPublishedOnClientBadDisconnection() throws Interrup
         m_willSubscriber.disconnect();
     }
 
+    @Test
+    public void testResendNotAckedPublishes()
+            throws MqttException, InterruptedException
+    {
+        LOG.info("*** testResendNotAckedPublishes ***");
+        String topic = "/test";
+
+        MqttClient subscriber = new MqttClient("tcp://localhost:1883", "Subscriber");
+        MqttClient publisher = new MqttClient("tcp://localhost:1883", "Publisher");
+
+        try {
+            subscriber.connect();
+            publisher.connect();
+
+            AtomicBoolean isFirst = new AtomicBoolean(true);
+            CountDownLatch countDownLatch = new CountDownLatch(2);
+            subscriber.subscribe(topic, 1, new IMqttMessageListener() {
+                @Override
+                public void messageArrived(String topic, org.eclipse.paho.client.mqttv3.MqttMessage message)
+                        throws Exception
+                {
+                    if (isFirst.getAndSet(false)) {
+                        // wait to trigger resending PUBLISH
+                        TimeUnit.SECONDS.sleep(12);
+                    }
+                    countDownLatch.countDown();
+                }
+            });
+
+            publisher.publish(topic, "hello".getBytes(), 1, false);
+            assertTrue(countDownLatch.await(30, TimeUnit.SECONDS));
+        }
+        finally {
+            try {
+                subscriber.disconnect();
+            }
+            finally {
+                publisher.disconnect();
+            }
+        }
+    }
+
 }