From 6cfa3f1b0b6b4a699f1df94f4528322b8199d107 Mon Sep 17 00:00:00 2001
From: dashevchenko <dshevchenko@thingsboard.io>
Date: Tue, 7 Oct 2025 12:22:57 +0300
Subject: [PATCH] fixed device session closing: should close every device
 session if message id wrong

---
 .../AbstractGatewaySessionHandler.java        | 20 +++++++++++--------
 1 file changed, 12 insertions(+), 8 deletions(-)

diff --git a/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/AbstractGatewaySessionHandler.java b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/AbstractGatewaySessionHandler.java
index 168fef6c2e..fee2f04e24 100644
--- a/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/AbstractGatewaySessionHandler.java
+++ b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/AbstractGatewaySessionHandler.java
@@ -845,27 +845,31 @@ public abstract class AbstractGatewaySessionHandler<T extends AbstractGatewayDev
                         log.trace("[{}][{}][{}] Wrong msg id: [{}]", gateway.getTenantId(), gateway.getDeviceId(), sessionId, msgId);
                         ctx.writeAndFlush(MqttTransportHandler.createMqttPubAckMsg(
                                 deviceSessionCtx, msgId, MqttReasonCodes.PubAck.UNSPECIFIED_ERROR.byteValue()));
-                        closeDeviceSession(deviceName, MqttReasonCodes.Disconnect.MALFORMED_PACKET);
                     }
                 }
+                if (msgId <= 0) {
+                    closeDeviceSession(deviceName, MqttReasonCodes.Disconnect.MALFORMED_PACKET);
+                }
             }
 
             @Override
             public void onError(Throwable e) {
                 log.trace("[{}][{}][{}] Failed to publish msg: [{}] for device: [{}]", gateway.getTenantId(), gateway.getDeviceId(), sessionId, msg, deviceName, e);
-                if (ackSent.compareAndSet(false, true)) {
-                    if (e instanceof TbRateLimitsException) {
+                if (e instanceof TbRateLimitsException) {
+                    if (ackSent.compareAndSet(false, true)) {
                         ctx.writeAndFlush(MqttTransportHandler.createMqttPubAckMsg(
                                 deviceSessionCtx, msgId, MqttReasonCodes.PubAck.QUOTA_EXCEEDED.byteValue()));
-                        closeDeviceSession(deviceName, MqttReasonCodes.Disconnect.MESSAGE_RATE_TOO_HIGH);
-                    } else {
+                        ctx.close();
+                    }
+                    closeDeviceSession(deviceName, MqttReasonCodes.Disconnect.MESSAGE_RATE_TOO_HIGH);
+                } else {
+                    if (ackSent.compareAndSet(false, true)) {
                         ctx.writeAndFlush(MqttTransportHandler.createMqttPubAckMsg(
                                 deviceSessionCtx, msgId, MqttReasonCodes.PubAck.UNSPECIFIED_ERROR.byteValue()));
-                        closeDeviceSession(deviceName, MqttReasonCodes.Disconnect.UNSPECIFIED_ERROR);
+                        ctx.close();
                     }
-                    ctx.close();
+                    closeDeviceSession(deviceName, MqttReasonCodes.Disconnect.UNSPECIFIED_ERROR);
                 }
-                remaining.set(0);
             }
         };
     }