From 0d925f8aed6cbcdab58d22804d9d712cadae7990 Mon Sep 17 00:00:00 2001 From: dshvaika Date: Wed, 1 Jul 2026 12:55:56 +0300 Subject: [PATCH 1/7] feat(transport): emit DELIVERED for gateway persistent RPC on PUBACK Gateway sub-device persistent RPC previously went QUEUED -> SENT -> SUCCESSFUL: the gateway send path never registered the outbound publish packetId, so the gateway PUBACK produced no DELIVERED (unlike direct devices). AbstractGatewaySessionHandler now keeps a packetId -> {device sessionInfo, rpc} awaiting-ack map; the sub-device session registers the packetId BEFORE the write (mirroring MqttTransportHandler#sendToDeviceRpcRequest so the loopback PUBACK cannot race ahead of registration and drop DELIVERED). The connection PUBACK handler delegates to onPubAck on a miss to emit DELIVERED against the device session; a scheduled task emits TIMEOUT if no PUBACK arrives; the map is cleared silently on gateway disconnect. Registration is guarded only by isAckExpected (every QoS1 RPC, not just persisted), matching the direct-device path: DELIVERED/TIMEOUT also drive one-way completion, the sequential delivered flag, and retry/re-queue for non-persistent RPCs in DeviceActorMessageProcessor; only the DB status write is persistence-gated. --- ...SideRpcDeliveryTimeoutIntegrationTest.java | 107 ++++++++++++++++++ .../transport/mqtt/MqttTransportHandler.java | 2 + .../AbstractGatewayDeviceSessionContext.java | 24 ++-- .../AbstractGatewaySessionHandler.java | 32 ++++++ 4 files changed, 154 insertions(+), 11 deletions(-) create mode 100644 application/src/test/java/org/thingsboard/server/transport/mqtt/mqttv3/rpc/MqttGatewayServerSideRpcDeliveryTimeoutIntegrationTest.java diff --git a/application/src/test/java/org/thingsboard/server/transport/mqtt/mqttv3/rpc/MqttGatewayServerSideRpcDeliveryTimeoutIntegrationTest.java b/application/src/test/java/org/thingsboard/server/transport/mqtt/mqttv3/rpc/MqttGatewayServerSideRpcDeliveryTimeoutIntegrationTest.java new file mode 100644 index 0000000000..b79d472e4b --- /dev/null +++ b/application/src/test/java/org/thingsboard/server/transport/mqtt/mqttv3/rpc/MqttGatewayServerSideRpcDeliveryTimeoutIntegrationTest.java @@ -0,0 +1,107 @@ +/** + * Copyright © 2016-2026 The Thingsboard Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.thingsboard.server.transport.mqtt.mqttv3.rpc; + +import io.netty.handler.codec.mqtt.MqttQoS; +import lombok.extern.slf4j.Slf4j; +import org.junit.Before; +import org.junit.Test; +import org.springframework.test.context.TestPropertySource; +import org.thingsboard.common.util.JacksonUtil; +import org.thingsboard.server.common.data.Device; +import org.thingsboard.server.common.data.TransportPayloadType; +import org.thingsboard.server.common.data.rpc.Rpc; +import org.thingsboard.server.common.data.rpc.RpcStatus; +import org.thingsboard.server.common.msg.session.FeatureType; +import org.thingsboard.server.dao.service.DaoSqlTest; +import org.thingsboard.server.transport.mqtt.MqttTestConfigProperties; +import org.thingsboard.server.transport.mqtt.mqttv3.MqttTestCallback; +import org.thingsboard.server.transport.mqtt.mqttv3.MqttTestClient; +import org.thingsboard.server.transport.mqtt.mqttv3.MqttTestSubscribeOnTopicCallback; + +import java.util.concurrent.TimeUnit; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotNull; +import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.status; +import static org.thingsboard.server.common.data.device.profile.MqttTopics.GATEWAY_CONNECT_TOPIC; +import static org.thingsboard.server.common.data.device.profile.MqttTopics.GATEWAY_RPC_TOPIC; + +@Slf4j +@DaoSqlTest +@TestPropertySource(properties = { + "actors.rpc.close_session_on_rpc_delivery_timeout=true", + "transport.mqtt.timeout=100", +}) +public class MqttGatewayServerSideRpcDeliveryTimeoutIntegrationTest extends AbstractMqttServerSideRpcIntegrationTest { + + @Before + public void beforeTest() throws Exception { + MqttTestConfigProperties configProperties = MqttTestConfigProperties.builder() + .deviceName("RPC timeout test device") + .gatewayName("RPC timeout test gateway") + .transportPayloadType(TransportPayloadType.JSON) + .build(); + processBeforeTest(configProperties); + } + + @Test + public void testGatewayPersistentRpcTimeoutWhenNoPuback() throws Exception { + MqttTestClient client = new MqttTestClient(); + client.enableManualAcks(); + client.connectAndWait(gatewayAccessToken); + + String deviceName = "Gateway Device Persistent RPC Timeout Json"; + String connectPayload = "{\"device\": \"" + deviceName + "\", \"type\": \"" + TransportPayloadType.JSON.name() + "\"}"; + client.publish(GATEWAY_CONNECT_TOPIC, connectPayload.getBytes()); + + Device device = doExecuteWithRetriesAndInterval(() -> getDeviceByName(deviceName), 20, 100); + assertNotNull(device); + + MqttTestCallback callback = new MqttTestSubscribeOnTopicCallback(GATEWAY_RPC_TOPIC); + client.setCallback(callback); + // QoS1 downlink so the gateway would need to PUBACK — but manual acks withhold it. + // Use the two-step form (client.subscribeAndWait + awaitForDeviceActorToReceiveSubscription) + // rather than subscribeAndWait(client, topic, deviceId, featureType, qos) because the gateway + // device actor already has one RPC subscription entry created when the CONNECT message was + // processed above, so subscribeAndWait (which waits for count+1) would wait for 2 forever. + client.subscribeAndWait(GATEWAY_RPC_TOPIC, MqttQoS.AT_LEAST_ONCE); + awaitForDeviceActorToReceiveSubscription(device.getId(), FeatureType.RPC, 1); + + long expirationTime = System.currentTimeMillis() + TimeUnit.DAYS.toMillis(1); + String rpcRequest = "{\"method\":\"toggle_gpio\",\"params\":{\"pin\":1},\"persistent\":true,\"retries\":0,\"expirationTime\":" + expirationTime + "}"; + String response = doPostAsync("/api/rpc/twoway/" + device.getId().getId().toString(), rpcRequest, String.class, status().isOk()); + String rpcId = JacksonUtil.toJsonNode(response).get("rpcId").asText(); + + callback.getSubscribeLatch().await(DEFAULT_WAIT_TIMEOUT_SECONDS, TimeUnit.SECONDS); + + // No PUBACK is ever sent. The gateway awaiting-ack scheduler emits TIMEOUT after + // transport.mqtt.timeout (100 ms); the server re-queues the RPC to QUEUED. + Rpc rpc = doExecuteWithRetriesAndInterval(() -> { + Rpc current = doGet("/api/rpc/persistent/" + rpcId, Rpc.class); + return RpcStatus.QUEUED.equals(current.getStatus()) ? current : null; + }, 50, 100); + assertNotNull(rpc); + assertEquals(RpcStatus.QUEUED, rpc.getStatus()); + + client.disconnect(); + } + + private Device getDeviceByName(String deviceName) throws Exception { + return doGet("/api/tenant/devices?deviceName=" + deviceName, Device.class); + } + +} 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 f753579292..414af93f40 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 @@ -385,6 +385,8 @@ public class MqttTransportHandler extends ChannelInboundHandlerAdapter implement TransportProtos.ToDeviceRpcRequestMsg rpcRequest = rpcAwaitingAck.remove(msgId); if (rpcRequest != null) { transportService.process(deviceSessionCtx.getSessionInfo(), rpcRequest, RpcStatus.DELIVERED, true, TransportServiceCallback.EMPTY); + } else if (gatewaySessionHandler != null) { + gatewaySessionHandler.onPubAck(msgId); } break; default: diff --git a/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/AbstractGatewayDeviceSessionContext.java b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/AbstractGatewayDeviceSessionContext.java index a1162f1990..8fa8b862e3 100644 --- a/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/AbstractGatewayDeviceSessionContext.java +++ b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/AbstractGatewayDeviceSessionContext.java @@ -17,6 +17,7 @@ package org.thingsboard.server.transport.mqtt.session; import io.netty.channel.ChannelFuture; import io.netty.handler.codec.mqtt.MqttMessage; +import io.netty.handler.codec.mqtt.MqttPublishMessage; import lombok.Getter; import lombok.ToString; import lombok.extern.slf4j.Slf4j; @@ -106,19 +107,20 @@ public abstract class AbstractGatewayDeviceSessionContext { + if (isAckExpected(payload)) { + int msgId = ((MqttPublishMessage) payload).variableHeader().packetId(); + parent.registerRpcAwaitingAck(msgId, getSessionInfo(), request); + } ChannelFuture channelFuture = parent.writeAndFlush(payload); - if (request.getPersisted()) { - channelFuture.addListener(result -> { - if (result.cause() == null) { - if (!isAckExpected(payload)) { - transportService.process(getSessionInfo(), request, RpcStatus.DELIVERED, TransportServiceCallback.EMPTY); - } else if (request.getPersisted()) { - transportService.process(getSessionInfo(), request, RpcStatus.SENT, TransportServiceCallback.EMPTY); - - } + channelFuture.addListener(result -> { + if (result.cause() == null) { + if (!isAckExpected(payload)) { + transportService.process(getSessionInfo(), request, RpcStatus.DELIVERED, TransportServiceCallback.EMPTY); + } else if (request.getPersisted()) { + transportService.process(getSessionInfo(), request, RpcStatus.SENT, TransportServiceCallback.EMPTY); } - }); - } + } + }); } ); } catch (Exception e) { 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 8ea6480142..ba570ba4a0 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 @@ -52,6 +52,7 @@ import org.thingsboard.server.common.data.StringUtils; import org.thingsboard.server.common.data.id.DeviceId; import org.thingsboard.server.common.data.util.TbPair; import org.thingsboard.server.common.msg.gateway.metrics.GatewayMetadata; +import org.thingsboard.server.common.data.rpc.RpcStatus; import org.thingsboard.server.common.msg.tools.TbRateLimitsException; import org.thingsboard.server.common.transport.TransportService; import org.thingsboard.server.common.transport.TransportServiceCallback; @@ -80,6 +81,7 @@ import java.util.Set; import java.util.UUID; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentMap; +import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.locks.Lock; @@ -123,6 +125,10 @@ public abstract class AbstractGatewaySessionHandler rpcAwaitingAck = new ConcurrentHashMap<>(); + + private record RpcAwaitingAck(SessionInfoProto deviceSessionInfo, TransportProtos.ToDeviceRpcRequestMsg rpc) {} + @Getter @Setter private boolean overwriteDevicesActivity = false; @@ -201,6 +207,10 @@ public abstract class AbstractGatewaySessionHandler { Futures.addCallback(future, new FutureCallback() { @@ -252,6 +262,28 @@ public abstract class AbstractGatewaySessionHandler { + RpcAwaitingAck pending = rpcAwaitingAck.remove(msgId); + if (pending != null) { + log.trace("[{}] Gateway RPC [{}] PUBACK timeout, sending TIMEOUT status", sessionId, rpc.getRequestId()); + transportService.process(pending.deviceSessionInfo(), pending.rpc(), RpcStatus.TIMEOUT, TransportServiceCallback.EMPTY); + } + }, delay, TimeUnit.MILLISECONDS); + } + + public void onPubAck(int msgId) { + RpcAwaitingAck pending = rpcAwaitingAck.remove(msgId); + if (pending == null) { + return; + } + log.trace("[{}] Gateway RPC [{}] PUBACK received, sending DELIVERED status", sessionId, pending.rpc().getRequestId()); + transportService.process(pending.deviceSessionInfo(), pending.rpc(), RpcStatus.DELIVERED, true, TransportServiceCallback.EMPTY); + } + protected boolean isJsonPayloadType() { return deviceSessionCtx.isJsonPayloadType(); } From a0fce168994ecc5dc21c690a52e2dce7d5670c0e Mon Sep 17 00:00:00 2001 From: dshvaika Date: Wed, 1 Jul 2026 12:55:57 +0300 Subject: [PATCH 2/7] test(transport): gateway persistent RPC reaches DELIVERED on PUBACK Adds a QoS1 gateway persistent-RPC integration test that drives a real PUBACK and asserts the persisted RPC status reaches DELIVERED via the REST API. Makes getDeviceByName protected for reuse. --- ...tractMqttServerSideRpcIntegrationTest.java | 43 ++++++++++++++++++- .../MqttServerSideRpcJsonIntegrationTest.java | 5 +++ 2 files changed, 47 insertions(+), 1 deletion(-) diff --git a/application/src/test/java/org/thingsboard/server/transport/mqtt/mqttv3/rpc/AbstractMqttServerSideRpcIntegrationTest.java b/application/src/test/java/org/thingsboard/server/transport/mqtt/mqttv3/rpc/AbstractMqttServerSideRpcIntegrationTest.java index 75d529d4b1..f515611d10 100644 --- a/application/src/test/java/org/thingsboard/server/transport/mqtt/mqttv3/rpc/AbstractMqttServerSideRpcIntegrationTest.java +++ b/application/src/test/java/org/thingsboard/server/transport/mqtt/mqttv3/rpc/AbstractMqttServerSideRpcIntegrationTest.java @@ -38,6 +38,7 @@ import org.thingsboard.server.common.data.device.profile.MqttDeviceProfileTransp import org.thingsboard.server.common.data.device.profile.ProtoTransportPayloadConfiguration; import org.thingsboard.server.common.data.device.profile.TransportPayloadTypeConfiguration; import org.thingsboard.server.common.data.rpc.Rpc; +import org.thingsboard.server.common.data.rpc.RpcStatus; import org.thingsboard.server.common.msg.session.FeatureType; import org.thingsboard.server.gen.transport.TransportApiProtos; import org.thingsboard.server.transport.mqtt.AbstractMqttIntegrationTest; @@ -331,6 +332,46 @@ public abstract class AbstractMqttServerSideRpcIntegrationTest extends AbstractM assertEquals(MqttQoS.AT_MOST_ONCE.value(), callback.getMessageArrivedQoS()); } + protected void validateGatewayPersistentRpcDelivered(String deviceName) throws Exception { + MqttTestClient client = new MqttTestClient(); + client.connectAndWait(gatewayAccessToken); + + String connectPayload = "{\"device\": \"" + deviceName + "\", \"type\": \"" + TransportPayloadType.JSON.name() + "\"}"; + client.publish(GATEWAY_CONNECT_TOPIC, connectPayload.getBytes()); + + Device device = doExecuteWithRetriesAndInterval(() -> getDeviceByName(deviceName), 20, 100); + assertNotNull(device); + + MqttTestCallback callback = new MqttTestSubscribeOnTopicCallback(GATEWAY_RPC_TOPIC); + client.setCallback(callback); + // Subscribe at QoS1 so the downlink publish is QoS1 and Paho auto-PUBACKs (-> DELIVERED). + // Two-step subscribe (client.subscribeAndWait + awaitForDeviceActorToReceiveSubscription, 1) + // rather than subscribeAndWait(client, topic, deviceId, featureType, qos): the latter waits for + // count+1, but the gateway CONNECT already registered one RPC subscription, so it would hang. + client.subscribeAndWait(GATEWAY_RPC_TOPIC, MqttQoS.AT_LEAST_ONCE); + awaitForDeviceActorToReceiveSubscription(device.getId(), FeatureType.RPC, 1); + + long expirationTime = System.currentTimeMillis() + TimeUnit.DAYS.toMillis(1); + String rpcRequest = "{\"method\":\"toggle_gpio\",\"params\":{\"pin\":1},\"persistent\":true,\"expirationTime\":" + expirationTime + "}"; + String response = doPostAsync("/api/rpc/twoway/" + device.getId().getId().toString(), rpcRequest, String.class, status().isOk()); + String rpcId = JacksonUtil.toJsonNode(response).get("rpcId").asText(); + + callback.getSubscribeLatch().await(DEFAULT_WAIT_TIMEOUT_SECONDS, TimeUnit.SECONDS); + // Confirm the downlink arrived at QoS1 — otherwise no PUBACK is produced and the test could + // pass for the wrong reason if a regression downgraded the gateway RPC to QoS0. + assertEquals(MqttQoS.AT_LEAST_ONCE.value(), callback.getMessageArrivedQoS()); + + // Poll the persisted status until DELIVERED (200 x 100ms = 20s window). + Rpc rpc = doExecuteWithRetriesAndInterval(() -> { + Rpc current = doGet("/api/rpc/persistent/" + rpcId, Rpc.class); + return RpcStatus.DELIVERED.equals(current.getStatus()) ? current : null; + }, 200, 100); + assertNotNull(rpc); + assertEquals(RpcStatus.DELIVERED, rpc.getStatus()); + + client.disconnect(); + } + protected void validateProtoTwoWayRpcGatewayResponse(String deviceName, MqttTestClient client, byte[] connectPayloadBytes) throws Exception { client.publish(GATEWAY_CONNECT_TOPIC, connectPayloadBytes); @@ -353,7 +394,7 @@ public abstract class AbstractMqttServerSideRpcIntegrationTest extends AbstractM assertEquals(MqttQoS.AT_MOST_ONCE.value(), callback.getMessageArrivedQoS()); } - private Device getDeviceByName(String deviceName) throws Exception { + protected Device getDeviceByName(String deviceName) throws Exception { return doGet("/api/tenant/devices?deviceName=" + deviceName, Device.class); } diff --git a/application/src/test/java/org/thingsboard/server/transport/mqtt/mqttv3/rpc/MqttServerSideRpcJsonIntegrationTest.java b/application/src/test/java/org/thingsboard/server/transport/mqtt/mqttv3/rpc/MqttServerSideRpcJsonIntegrationTest.java index 10ef549c87..842904baef 100644 --- a/application/src/test/java/org/thingsboard/server/transport/mqtt/mqttv3/rpc/MqttServerSideRpcJsonIntegrationTest.java +++ b/application/src/test/java/org/thingsboard/server/transport/mqtt/mqttv3/rpc/MqttServerSideRpcJsonIntegrationTest.java @@ -81,6 +81,11 @@ public class MqttServerSideRpcJsonIntegrationTest extends AbstractMqttServerSide processJsonTwoWayRpcTestGateway("Gateway Device TwoWay RPC Json"); } + @Test + public void testGatewayServerMqttPersistentRpcDeliveredOnPuback() throws Exception { + validateGatewayPersistentRpcDelivered("Gateway Device Persistent RPC Json"); + } + protected void processJsonOneWayRpcTestGateway(String deviceName) throws Exception { MqttTestClient client = new MqttTestClient(); client.connectAndWait(gatewayAccessToken); From 6510a3d5eff821f1edfbfde3aaa1f8e60f507205 Mon Sep 17 00:00:00 2001 From: dshvaika Date: Wed, 1 Jul 2026 12:55:57 +0300 Subject: [PATCH 3/7] test(transport): gateway persistent RPC times out when gateway never PUBACKs With manual acks (PUBACK withheld) and transport.mqtt.timeout=100 + close_session_on_rpc_delivery_timeout=true, asserts the persisted RPC is re-queued to QUEUED after the delivery timeout. --- ...GatewayServerSideRpcDeliveryTimeoutIntegrationTest.java | 7 +++---- 1 file changed, 3 insertions(+), 4 deletions(-) diff --git a/application/src/test/java/org/thingsboard/server/transport/mqtt/mqttv3/rpc/MqttGatewayServerSideRpcDeliveryTimeoutIntegrationTest.java b/application/src/test/java/org/thingsboard/server/transport/mqtt/mqttv3/rpc/MqttGatewayServerSideRpcDeliveryTimeoutIntegrationTest.java index b79d472e4b..18d1ad8105 100644 --- a/application/src/test/java/org/thingsboard/server/transport/mqtt/mqttv3/rpc/MqttGatewayServerSideRpcDeliveryTimeoutIntegrationTest.java +++ b/application/src/test/java/org/thingsboard/server/transport/mqtt/mqttv3/rpc/MqttGatewayServerSideRpcDeliveryTimeoutIntegrationTest.java @@ -87,6 +87,9 @@ public class MqttGatewayServerSideRpcDeliveryTimeoutIntegrationTest extends Abst String rpcId = JacksonUtil.toJsonNode(response).get("rpcId").asText(); callback.getSubscribeLatch().await(DEFAULT_WAIT_TIMEOUT_SECONDS, TimeUnit.SECONDS); + // Confirm the downlink arrived at QoS1 — a QoS0 downlink would produce no PUBACK path at all, + // so the timeout would fire for the wrong reason and mask a regression. + assertEquals(MqttQoS.AT_LEAST_ONCE.value(), callback.getMessageArrivedQoS()); // No PUBACK is ever sent. The gateway awaiting-ack scheduler emits TIMEOUT after // transport.mqtt.timeout (100 ms); the server re-queues the RPC to QUEUED. @@ -100,8 +103,4 @@ public class MqttGatewayServerSideRpcDeliveryTimeoutIntegrationTest extends Abst client.disconnect(); } - private Device getDeviceByName(String deviceName) throws Exception { - return doGet("/api/tenant/devices?deviceName=" + deviceName, Device.class); - } - } From c59762a5b8a981e2a7d8bbbc805fdf89e9f5a57a Mon Sep 17 00:00:00 2001 From: dshvaika Date: Wed, 1 Jul 2026 13:32:19 +0300 Subject: [PATCH 4/7] fix(transport): skip redundant DELIVERED for non-persistent one-way RPCs Non-persistent one-way RPCs self-complete on send (DeviceActorMessageProcessor) and are never added to the device actor's pending map. Emitting a transport delivery status (DELIVERED on PUBACK / QoS0) for them therefore only produced a benign 'RPC has already been removed from pending map' WARN and no effect. Guard both the gateway path (AbstractGatewayDeviceSessionContext.onToDeviceRpcRequest) and the direct-device path (MqttTransportHandler.sendToDeviceRpcRequest) with 'oneway && !persisted': skip awaiting-ack registration and the delivery-status emission for those. DELIVERED is preserved for non-persistent two-way RPCs (sets the delivered flag used by SEQUENTIAL_ON_ACK/RESPONSE strategies) and for all persistent RPCs; SENT stays persisted-only. The device-API path had this pre-existing warning; it is now suppressed there too. The residual 'already removed from pending map' WARN is intentionally kept for the rare persistent loopback race (response beats SENT/DELIVERED) so it stays a meaningful signal. --- .../server/transport/mqtt/MqttTransportHandler.java | 12 +++++++++--- .../session/AbstractGatewayDeviceSessionContext.java | 10 ++++++++-- 2 files changed, 17 insertions(+), 5 deletions(-) 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 414af93f40..8c5c650e6a 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 @@ -1525,7 +1525,11 @@ public class MqttTransportHandler extends ChannelInboundHandlerAdapter implement public void sendToDeviceRpcRequest(MqttMessage payload, TransportProtos.ToDeviceRpcRequestMsg rpcRequest, TransportProtos.SessionInfoProto sessionInfo) { int msgId = ((MqttPublishMessage) payload).variableHeader().packetId(); int requestId = rpcRequest.getRequestId(); - if (isAckExpected(payload)) { + // Non-persistent one-way RPCs self-complete on send (DeviceActorMessageProcessor) and are never + // in the pending map, so a delivery status for them only yields a benign "already removed from + // pending map" WARN. Track/emit delivery status only for the rest. + boolean oneWayNonPersisted = rpcRequest.getOneway() && !rpcRequest.getPersisted(); + if (!oneWayNonPersisted && isAckExpected(payload)) { rpcAwaitingAck.put(msgId, rpcRequest); context.getScheduler().schedule(() -> { TransportProtos.ToDeviceRpcRequestMsg msg = rpcAwaitingAck.remove(msgId); @@ -1545,8 +1549,10 @@ public class MqttTransportHandler extends ChannelInboundHandlerAdapter implement return; } if (!isAckExpected(payload)) { - log.trace("[{}][{}][{}] Going to send to device actor RPC request DELIVERED status update ...", deviceSessionCtx.getDeviceId(), sessionId, requestId); - transportService.process(sessionInfo, rpcRequest, RpcStatus.DELIVERED, TransportServiceCallback.EMPTY); + if (!oneWayNonPersisted) { + log.trace("[{}][{}][{}] Going to send to device actor RPC request DELIVERED status update ...", deviceSessionCtx.getDeviceId(), sessionId, requestId); + transportService.process(sessionInfo, rpcRequest, RpcStatus.DELIVERED, TransportServiceCallback.EMPTY); + } } else if (rpcRequest.getPersisted()) { log.trace("[{}][{}][{}] Going to send to device actor RPC request SENT status update ...", deviceSessionCtx.getDeviceId(), sessionId, requestId); transportService.process(sessionInfo, rpcRequest, RpcStatus.SENT, TransportServiceCallback.EMPTY); diff --git a/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/AbstractGatewayDeviceSessionContext.java b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/AbstractGatewayDeviceSessionContext.java index 8fa8b862e3..ea34a2492e 100644 --- a/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/AbstractGatewayDeviceSessionContext.java +++ b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/AbstractGatewayDeviceSessionContext.java @@ -107,7 +107,11 @@ public abstract class AbstractGatewayDeviceSessionContext { - if (isAckExpected(payload)) { + // Non-persistent one-way RPCs self-complete on send; a delivery status for them + // only yields a benign "already removed from pending map" WARN. Skip those + // (mirrors MqttTransportHandler#sendToDeviceRpcRequest). + boolean oneWayNonPersisted = request.getOneway() && !request.getPersisted(); + if (!oneWayNonPersisted && isAckExpected(payload)) { int msgId = ((MqttPublishMessage) payload).variableHeader().packetId(); parent.registerRpcAwaitingAck(msgId, getSessionInfo(), request); } @@ -115,7 +119,9 @@ public abstract class AbstractGatewayDeviceSessionContext { if (result.cause() == null) { if (!isAckExpected(payload)) { - transportService.process(getSessionInfo(), request, RpcStatus.DELIVERED, TransportServiceCallback.EMPTY); + if (!oneWayNonPersisted) { + transportService.process(getSessionInfo(), request, RpcStatus.DELIVERED, TransportServiceCallback.EMPTY); + } } else if (request.getPersisted()) { transportService.process(getSessionInfo(), request, RpcStatus.SENT, TransportServiceCallback.EMPTY); } From 50210966deb0e5a70694886c4f29a07039fac5e4 Mon Sep 17 00:00:00 2001 From: dshvaika Date: Wed, 1 Jul 2026 14:48:00 +0300 Subject: [PATCH 5/7] refactor(transport): address review feedback for gateway RPC DELIVERED MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Extract the delivery-tracking predicate (oneway && !persisted) into MqttRpcStatusUtil.requireDeliveryTracking, shared by the gateway path and MqttTransportHandler.sendToDeviceRpcRequest (was duplicated). AbstractGatewayDeviceSessionContext casts the payload to a typed MqttPublishMessage local once. AbstractGatewaySessionHandler: move the RpcStatus import into the alphabetical common.data.* block and make registerRpcAwaitingAck package-private (onPubAck stays public — cross-package call from MqttTransportHandler). Tests: extract shared helpers connectGatewayAndSubscribeForRpc + awaitRpcStatus (and a GatewayRpcSession record) into AbstractMqttServerSideRpcIntegrationTest, used by both the DELIVERED and TIMEOUT tests. --- ...tractMqttServerSideRpcIntegrationTest.java | 56 ++++++++++++------- ...SideRpcDeliveryTimeoutIntegrationTest.java | 49 +++------------- .../transport/mqtt/MqttTransportHandler.java | 10 ++-- .../AbstractGatewayDeviceSessionContext.java | 18 +++--- .../AbstractGatewaySessionHandler.java | 4 +- .../mqtt/util/MqttRpcStatusUtil.java | 35 ++++++++++++ 6 files changed, 94 insertions(+), 78 deletions(-) create mode 100644 common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/util/MqttRpcStatusUtil.java diff --git a/application/src/test/java/org/thingsboard/server/transport/mqtt/mqttv3/rpc/AbstractMqttServerSideRpcIntegrationTest.java b/application/src/test/java/org/thingsboard/server/transport/mqtt/mqttv3/rpc/AbstractMqttServerSideRpcIntegrationTest.java index f515611d10..bb1037f481 100644 --- a/application/src/test/java/org/thingsboard/server/transport/mqtt/mqttv3/rpc/AbstractMqttServerSideRpcIntegrationTest.java +++ b/application/src/test/java/org/thingsboard/server/transport/mqtt/mqttv3/rpc/AbstractMqttServerSideRpcIntegrationTest.java @@ -333,7 +333,34 @@ public abstract class AbstractMqttServerSideRpcIntegrationTest extends AbstractM } protected void validateGatewayPersistentRpcDelivered(String deviceName) throws Exception { + GatewayRpcSession session = connectGatewayAndSubscribeForRpc(deviceName, false); + + long expirationTime = System.currentTimeMillis() + TimeUnit.DAYS.toMillis(1); + String rpcRequest = "{\"method\":\"toggle_gpio\",\"params\":{\"pin\":1},\"persistent\":true,\"expirationTime\":" + expirationTime + "}"; + String response = doPostAsync("/api/rpc/twoway/" + session.device().getId().getId().toString(), rpcRequest, String.class, status().isOk()); + String rpcId = JacksonUtil.toJsonNode(response).get("rpcId").asText(); + + session.callback().getSubscribeLatch().await(DEFAULT_WAIT_TIMEOUT_SECONDS, TimeUnit.SECONDS); + // Confirm the downlink arrived at QoS1 — otherwise no PUBACK is produced and the test could + // pass for the wrong reason if a regression downgraded the gateway RPC to QoS0. + assertEquals(MqttQoS.AT_LEAST_ONCE.value(), session.callback().getMessageArrivedQoS()); + + awaitRpcStatus(rpcId, RpcStatus.DELIVERED, 200, 100); + + session.client().disconnect(); + } + + /** + * Connects a gateway, provisions the sub-device via CONNECT, and subscribes to the gateway RPC topic + * at QoS1. Two-step subscribe (client.subscribeAndWait + awaitForDeviceActorToReceiveSubscription, 1) + * rather than the 5-arg subscribeAndWait, which waits for subscription count+1 — the gateway CONNECT + * already registered one RPC subscription, so it would hang. + */ + protected GatewayRpcSession connectGatewayAndSubscribeForRpc(String deviceName, boolean manualAcks) throws Exception { MqttTestClient client = new MqttTestClient(); + if (manualAcks) { + client.enableManualAcks(); + } client.connectAndWait(gatewayAccessToken); String connectPayload = "{\"device\": \"" + deviceName + "\", \"type\": \"" + TransportPayloadType.JSON.name() + "\"}"; @@ -344,34 +371,23 @@ public abstract class AbstractMqttServerSideRpcIntegrationTest extends AbstractM MqttTestCallback callback = new MqttTestSubscribeOnTopicCallback(GATEWAY_RPC_TOPIC); client.setCallback(callback); - // Subscribe at QoS1 so the downlink publish is QoS1 and Paho auto-PUBACKs (-> DELIVERED). - // Two-step subscribe (client.subscribeAndWait + awaitForDeviceActorToReceiveSubscription, 1) - // rather than subscribeAndWait(client, topic, deviceId, featureType, qos): the latter waits for - // count+1, but the gateway CONNECT already registered one RPC subscription, so it would hang. client.subscribeAndWait(GATEWAY_RPC_TOPIC, MqttQoS.AT_LEAST_ONCE); awaitForDeviceActorToReceiveSubscription(device.getId(), FeatureType.RPC, 1); + return new GatewayRpcSession(client, device, callback); + } - long expirationTime = System.currentTimeMillis() + TimeUnit.DAYS.toMillis(1); - String rpcRequest = "{\"method\":\"toggle_gpio\",\"params\":{\"pin\":1},\"persistent\":true,\"expirationTime\":" + expirationTime + "}"; - String response = doPostAsync("/api/rpc/twoway/" + device.getId().getId().toString(), rpcRequest, String.class, status().isOk()); - String rpcId = JacksonUtil.toJsonNode(response).get("rpcId").asText(); - - callback.getSubscribeLatch().await(DEFAULT_WAIT_TIMEOUT_SECONDS, TimeUnit.SECONDS); - // Confirm the downlink arrived at QoS1 — otherwise no PUBACK is produced and the test could - // pass for the wrong reason if a regression downgraded the gateway RPC to QoS0. - assertEquals(MqttQoS.AT_LEAST_ONCE.value(), callback.getMessageArrivedQoS()); - - // Poll the persisted status until DELIVERED (200 x 100ms = 20s window). + protected Rpc awaitRpcStatus(String rpcId, RpcStatus expectedStatus, int retries, int intervalMs) throws Exception { Rpc rpc = doExecuteWithRetriesAndInterval(() -> { Rpc current = doGet("/api/rpc/persistent/" + rpcId, Rpc.class); - return RpcStatus.DELIVERED.equals(current.getStatus()) ? current : null; - }, 200, 100); + return expectedStatus.equals(current.getStatus()) ? current : null; + }, retries, intervalMs); assertNotNull(rpc); - assertEquals(RpcStatus.DELIVERED, rpc.getStatus()); - - client.disconnect(); + assertEquals(expectedStatus, rpc.getStatus()); + return rpc; } + protected record GatewayRpcSession(MqttTestClient client, Device device, MqttTestCallback callback) {} + protected void validateProtoTwoWayRpcGatewayResponse(String deviceName, MqttTestClient client, byte[] connectPayloadBytes) throws Exception { client.publish(GATEWAY_CONNECT_TOPIC, connectPayloadBytes); diff --git a/application/src/test/java/org/thingsboard/server/transport/mqtt/mqttv3/rpc/MqttGatewayServerSideRpcDeliveryTimeoutIntegrationTest.java b/application/src/test/java/org/thingsboard/server/transport/mqtt/mqttv3/rpc/MqttGatewayServerSideRpcDeliveryTimeoutIntegrationTest.java index 18d1ad8105..9c314339e5 100644 --- a/application/src/test/java/org/thingsboard/server/transport/mqtt/mqttv3/rpc/MqttGatewayServerSideRpcDeliveryTimeoutIntegrationTest.java +++ b/application/src/test/java/org/thingsboard/server/transport/mqtt/mqttv3/rpc/MqttGatewayServerSideRpcDeliveryTimeoutIntegrationTest.java @@ -21,24 +21,15 @@ import org.junit.Before; import org.junit.Test; import org.springframework.test.context.TestPropertySource; import org.thingsboard.common.util.JacksonUtil; -import org.thingsboard.server.common.data.Device; import org.thingsboard.server.common.data.TransportPayloadType; -import org.thingsboard.server.common.data.rpc.Rpc; import org.thingsboard.server.common.data.rpc.RpcStatus; -import org.thingsboard.server.common.msg.session.FeatureType; import org.thingsboard.server.dao.service.DaoSqlTest; import org.thingsboard.server.transport.mqtt.MqttTestConfigProperties; -import org.thingsboard.server.transport.mqtt.mqttv3.MqttTestCallback; -import org.thingsboard.server.transport.mqtt.mqttv3.MqttTestClient; -import org.thingsboard.server.transport.mqtt.mqttv3.MqttTestSubscribeOnTopicCallback; import java.util.concurrent.TimeUnit; import static org.junit.Assert.assertEquals; -import static org.junit.Assert.assertNotNull; import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.status; -import static org.thingsboard.server.common.data.device.profile.MqttTopics.GATEWAY_CONNECT_TOPIC; -import static org.thingsboard.server.common.data.device.profile.MqttTopics.GATEWAY_RPC_TOPIC; @Slf4j @DaoSqlTest @@ -60,47 +51,25 @@ public class MqttGatewayServerSideRpcDeliveryTimeoutIntegrationTest extends Abst @Test public void testGatewayPersistentRpcTimeoutWhenNoPuback() throws Exception { - MqttTestClient client = new MqttTestClient(); - client.enableManualAcks(); - client.connectAndWait(gatewayAccessToken); - - String deviceName = "Gateway Device Persistent RPC Timeout Json"; - String connectPayload = "{\"device\": \"" + deviceName + "\", \"type\": \"" + TransportPayloadType.JSON.name() + "\"}"; - client.publish(GATEWAY_CONNECT_TOPIC, connectPayload.getBytes()); - - Device device = doExecuteWithRetriesAndInterval(() -> getDeviceByName(deviceName), 20, 100); - assertNotNull(device); - - MqttTestCallback callback = new MqttTestSubscribeOnTopicCallback(GATEWAY_RPC_TOPIC); - client.setCallback(callback); - // QoS1 downlink so the gateway would need to PUBACK — but manual acks withhold it. - // Use the two-step form (client.subscribeAndWait + awaitForDeviceActorToReceiveSubscription) - // rather than subscribeAndWait(client, topic, deviceId, featureType, qos) because the gateway - // device actor already has one RPC subscription entry created when the CONNECT message was - // processed above, so subscribeAndWait (which waits for count+1) would wait for 2 forever. - client.subscribeAndWait(GATEWAY_RPC_TOPIC, MqttQoS.AT_LEAST_ONCE); - awaitForDeviceActorToReceiveSubscription(device.getId(), FeatureType.RPC, 1); + // manualAcks = true → the gateway client never PUBACKs the QoS1 downlink. + GatewayRpcSession session = connectGatewayAndSubscribeForRpc("Gateway Device Persistent RPC Timeout Json", true); long expirationTime = System.currentTimeMillis() + TimeUnit.DAYS.toMillis(1); String rpcRequest = "{\"method\":\"toggle_gpio\",\"params\":{\"pin\":1},\"persistent\":true,\"retries\":0,\"expirationTime\":" + expirationTime + "}"; - String response = doPostAsync("/api/rpc/twoway/" + device.getId().getId().toString(), rpcRequest, String.class, status().isOk()); + String response = doPostAsync("/api/rpc/twoway/" + session.device().getId().getId().toString(), rpcRequest, String.class, status().isOk()); String rpcId = JacksonUtil.toJsonNode(response).get("rpcId").asText(); - callback.getSubscribeLatch().await(DEFAULT_WAIT_TIMEOUT_SECONDS, TimeUnit.SECONDS); + session.callback().getSubscribeLatch().await(DEFAULT_WAIT_TIMEOUT_SECONDS, TimeUnit.SECONDS); // Confirm the downlink arrived at QoS1 — a QoS0 downlink would produce no PUBACK path at all, // so the timeout would fire for the wrong reason and mask a regression. - assertEquals(MqttQoS.AT_LEAST_ONCE.value(), callback.getMessageArrivedQoS()); + assertEquals(MqttQoS.AT_LEAST_ONCE.value(), session.callback().getMessageArrivedQoS()); // No PUBACK is ever sent. The gateway awaiting-ack scheduler emits TIMEOUT after - // transport.mqtt.timeout (100 ms); the server re-queues the RPC to QUEUED. - Rpc rpc = doExecuteWithRetriesAndInterval(() -> { - Rpc current = doGet("/api/rpc/persistent/" + rpcId, Rpc.class); - return RpcStatus.QUEUED.equals(current.getStatus()) ? current : null; - }, 50, 100); - assertNotNull(rpc); - assertEquals(RpcStatus.QUEUED, rpc.getStatus()); + // transport.mqtt.timeout (100 ms); the RPC status reverts to QUEUED (re-queued via the + // close_session_on_rpc_delivery_timeout path). + awaitRpcStatus(rpcId, RpcStatus.QUEUED, 50, 100); - client.disconnect(); + session.client().disconnect(); } } 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 8c5c650e6a..7b4ec3f057 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 @@ -88,6 +88,7 @@ import org.thingsboard.server.transport.mqtt.session.GatewaySessionHandler; import org.thingsboard.server.transport.mqtt.session.MqttTopicMatcher; import org.thingsboard.server.transport.mqtt.session.SparkplugDeviceSessionContext; import org.thingsboard.server.transport.mqtt.session.SparkplugNodeSessionHandler; +import org.thingsboard.server.transport.mqtt.util.MqttRpcStatusUtil; import org.thingsboard.server.transport.mqtt.util.ReturnCodeResolver; import org.thingsboard.server.transport.mqtt.util.sparkplug.SparkplugMessageType; import org.thingsboard.server.transport.mqtt.util.sparkplug.SparkplugRpcRequestHeader; @@ -1525,11 +1526,8 @@ public class MqttTransportHandler extends ChannelInboundHandlerAdapter implement public void sendToDeviceRpcRequest(MqttMessage payload, TransportProtos.ToDeviceRpcRequestMsg rpcRequest, TransportProtos.SessionInfoProto sessionInfo) { int msgId = ((MqttPublishMessage) payload).variableHeader().packetId(); int requestId = rpcRequest.getRequestId(); - // Non-persistent one-way RPCs self-complete on send (DeviceActorMessageProcessor) and are never - // in the pending map, so a delivery status for them only yields a benign "already removed from - // pending map" WARN. Track/emit delivery status only for the rest. - boolean oneWayNonPersisted = rpcRequest.getOneway() && !rpcRequest.getPersisted(); - if (!oneWayNonPersisted && isAckExpected(payload)) { + boolean requireDeliveryTracking = MqttRpcStatusUtil.requireDeliveryTracking(rpcRequest); + if (requireDeliveryTracking && isAckExpected(payload)) { rpcAwaitingAck.put(msgId, rpcRequest); context.getScheduler().schedule(() -> { TransportProtos.ToDeviceRpcRequestMsg msg = rpcAwaitingAck.remove(msgId); @@ -1549,7 +1547,7 @@ public class MqttTransportHandler extends ChannelInboundHandlerAdapter implement return; } if (!isAckExpected(payload)) { - if (!oneWayNonPersisted) { + if (requireDeliveryTracking) { log.trace("[{}][{}][{}] Going to send to device actor RPC request DELIVERED status update ...", deviceSessionCtx.getDeviceId(), sessionId, requestId); transportService.process(sessionInfo, rpcRequest, RpcStatus.DELIVERED, TransportServiceCallback.EMPTY); } diff --git a/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/AbstractGatewayDeviceSessionContext.java b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/AbstractGatewayDeviceSessionContext.java index ea34a2492e..5b0f8c15be 100644 --- a/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/AbstractGatewayDeviceSessionContext.java +++ b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/AbstractGatewayDeviceSessionContext.java @@ -30,6 +30,7 @@ import org.thingsboard.server.common.transport.TransportServiceCallback; import org.thingsboard.server.common.transport.auth.TransportDeviceInfo; import org.thingsboard.server.gen.transport.TransportProtos; import org.thingsboard.server.gen.transport.TransportProtos.SessionInfoProto; +import org.thingsboard.server.transport.mqtt.util.MqttRpcStatusUtil; import java.util.UUID; import java.util.concurrent.ConcurrentMap; @@ -107,19 +108,16 @@ public abstract class AbstractGatewayDeviceSessionContext { - // Non-persistent one-way RPCs self-complete on send; a delivery status for them - // only yields a benign "already removed from pending map" WARN. Skip those - // (mirrors MqttTransportHandler#sendToDeviceRpcRequest). - boolean oneWayNonPersisted = request.getOneway() && !request.getPersisted(); - if (!oneWayNonPersisted && isAckExpected(payload)) { - int msgId = ((MqttPublishMessage) payload).variableHeader().packetId(); - parent.registerRpcAwaitingAck(msgId, getSessionInfo(), request); + MqttPublishMessage publishMsg = (MqttPublishMessage) payload; + boolean requireDeliveryTracking = MqttRpcStatusUtil.requireDeliveryTracking(request); + if (requireDeliveryTracking && isAckExpected(publishMsg)) { + parent.registerRpcAwaitingAck(publishMsg.variableHeader().packetId(), getSessionInfo(), request); } - ChannelFuture channelFuture = parent.writeAndFlush(payload); + ChannelFuture channelFuture = parent.writeAndFlush(publishMsg); channelFuture.addListener(result -> { if (result.cause() == null) { - if (!isAckExpected(payload)) { - if (!oneWayNonPersisted) { + if (!isAckExpected(publishMsg)) { + if (requireDeliveryTracking) { transportService.process(getSessionInfo(), request, RpcStatus.DELIVERED, TransportServiceCallback.EMPTY); } } else if (request.getPersisted()) { 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 ba570ba4a0..0a15d5d255 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 @@ -50,9 +50,9 @@ import org.thingsboard.server.common.data.Device; import org.thingsboard.server.common.data.DeviceProfile; import org.thingsboard.server.common.data.StringUtils; import org.thingsboard.server.common.data.id.DeviceId; +import org.thingsboard.server.common.data.rpc.RpcStatus; import org.thingsboard.server.common.data.util.TbPair; import org.thingsboard.server.common.msg.gateway.metrics.GatewayMetadata; -import org.thingsboard.server.common.data.rpc.RpcStatus; import org.thingsboard.server.common.msg.tools.TbRateLimitsException; import org.thingsboard.server.common.transport.TransportService; import org.thingsboard.server.common.transport.TransportServiceCallback; @@ -262,7 +262,7 @@ public abstract class AbstractGatewaySessionHandler Date: Wed, 1 Jul 2026 15:03:18 +0300 Subject: [PATCH 6/7] refactor(transport): use typed publish message local in sendToDeviceRpcRequest Mirror AbstractGatewayDeviceSessionContext: cast the payload to a typed MqttPublishMessage local once instead of casting inline for packetId, so both RPC send paths read consistently. --- .../server/transport/mqtt/MqttTransportHandler.java | 9 +++++---- 1 file changed, 5 insertions(+), 4 deletions(-) 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 7b4ec3f057..7b321dc5f1 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 @@ -1524,10 +1524,11 @@ public class MqttTransportHandler extends ChannelInboundHandlerAdapter implement } public void sendToDeviceRpcRequest(MqttMessage payload, TransportProtos.ToDeviceRpcRequestMsg rpcRequest, TransportProtos.SessionInfoProto sessionInfo) { - int msgId = ((MqttPublishMessage) payload).variableHeader().packetId(); + MqttPublishMessage publishMsg = (MqttPublishMessage) payload; + int msgId = publishMsg.variableHeader().packetId(); int requestId = rpcRequest.getRequestId(); boolean requireDeliveryTracking = MqttRpcStatusUtil.requireDeliveryTracking(rpcRequest); - if (requireDeliveryTracking && isAckExpected(payload)) { + if (requireDeliveryTracking && isAckExpected(publishMsg)) { rpcAwaitingAck.put(msgId, rpcRequest); context.getScheduler().schedule(() -> { TransportProtos.ToDeviceRpcRequestMsg msg = rpcAwaitingAck.remove(msgId); @@ -1537,7 +1538,7 @@ public class MqttTransportHandler extends ChannelInboundHandlerAdapter implement } }, Math.max(0, Math.min(deviceSessionCtx.getContext().getTimeout(), rpcRequest.getExpirationTime() - System.currentTimeMillis())), TimeUnit.MILLISECONDS); } - var cf = publish(payload, deviceSessionCtx); + var cf = publish(publishMsg, deviceSessionCtx); cf.addListener(result -> { Throwable throwable = result.cause(); if (throwable != null) { @@ -1546,7 +1547,7 @@ public class MqttTransportHandler extends ChannelInboundHandlerAdapter implement ThingsboardErrorCode.INVALID_ARGUMENTS, " Failed send To Device Rpc Request: " + rpcRequest.getMethodName()); return; } - if (!isAckExpected(payload)) { + if (!isAckExpected(publishMsg)) { if (requireDeliveryTracking) { log.trace("[{}][{}][{}] Going to send to device actor RPC request DELIVERED status update ...", deviceSessionCtx.getDeviceId(), sessionId, requestId); transportService.process(sessionInfo, rpcRequest, RpcStatus.DELIVERED, TransportServiceCallback.EMPTY); From 4c0bdd32b91c32bacb6a626625cbad2cf39d6e85 Mon Sep 17 00:00:00 2001 From: dshvaika Date: Tue, 7 Jul 2026 17:12:18 +0300 Subject: [PATCH 7/7] docs(rpc): clarify close-session-on-delivery-timeout for MQTT gateway sub-devices --- application/src/main/resources/thingsboard.yml | 7 +++++++ 1 file changed, 7 insertions(+) diff --git a/application/src/main/resources/thingsboard.yml b/application/src/main/resources/thingsboard.yml index 003cee2eb7..64b9e4e58e 100644 --- a/application/src/main/resources/thingsboard.yml +++ b/application/src/main/resources/thingsboard.yml @@ -614,6 +614,13 @@ actors: #
  • QoS level 0: This feature does not apply, as no acknowledgment is expected, and therefore no timeout is triggered.
    • #
  • QoS level 1: This feature applies, as an acknowledgment is expected.
    • #
  • QoS level 2: Unsupported.
+ #
  • For MQTT Gateway (connected sub-devices): the same QoS rules as MQTT transport apply + # (only a QoS level 1 PUBACK triggers a delivery timeout). However, the gateway's own transport + # session is never closed, so the gateway connection and its other sub-devices stay unaffected. + # Only the sub-device session is deregistered and the RPC is reverted to the queued state. It is + # redelivered automatically once the sub-device session is re-registered, which happens on the next + # gateway message for that device (connect, telemetry, or attribute publish): the RPC subscription + # is reapplied and the queued RPC is resent. Until then the RPC remains queued.
    • #
  • For CoAP transport: #
    • Confirmable requests: This feature applies, as delivery confirmation is expected.
      • #
    • Non-confirmable requests: This feature does not apply, as no delivery acknowledgment is expected.