From 9d651a7252e21ef758815f32a45a48973fff6202 Mon Sep 17 00:00:00 2001
From: Andrii Shvaika <ashvayka@thingsboard.io>
Date: Thu, 5 Aug 2021 15:10:04 +0300
Subject: [PATCH] Fix reference count issue

---
 .../thingsboard/server/transport/mqtt/MqttTransportHandler.java | 2 ++
 1 file changed, 2 insertions(+)

diff --git a/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/MqttTransportHandler.java b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/MqttTransportHandler.java
index 3e3f2e74d9..3af471c207 100644
--- a/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/MqttTransportHandler.java
+++ b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/MqttTransportHandler.java
@@ -237,6 +237,7 @@ public class MqttTransportHandler extends ChannelInboundHandlerAdapter implement
         }
 
         deviceSessionCtx.getMsgQueue().add(msg);
+        ReferenceCountUtil.retain(msg);
         processMsgQueue(ctx); //Under the normal conditions the msg queue will contain 0 messages. Many messages will be processed on device connect event in separate thread pool
     }
 
@@ -252,6 +253,7 @@ public class MqttTransportHandler extends ChannelInboundHandlerAdapter implement
                     while ((msg = deviceSessionCtx.getMsgQueue().poll()) != null) {
                         deviceSessionCtx.getMsgQueueSize().decrementAndGet();
                         processRegularSessionMsg(ctx, msg);
+                        ReferenceCountUtil.safeRelease(msg);
                     }
                 } finally {
                     deviceSessionCtx.getMsgQueueProcessorLock().unlock();