diff --git a/application/src/main/java/org/thingsboard/server/actors/device/DeviceActorMessageProcessor.java b/application/src/main/java/org/thingsboard/server/actors/device/DeviceActorMessageProcessor.java index 00af0461d1..14ff096747 100644 --- a/application/src/main/java/org/thingsboard/server/actors/device/DeviceActorMessageProcessor.java +++ b/application/src/main/java/org/thingsboard/server/actors/device/DeviceActorMessageProcessor.java @@ -164,7 +164,7 @@ public class DeviceActorMessageProcessor extends AbstractContextAwareMsgProcesso } else { logger.debug("[{}] No pending RPC messages for new async session [{}]", deviceId, sessionId); } - Set sentOneWayIds = new HashSet<>(); + Set sentOneWayIds = new HashSet<>(); if (type == SessionType.ASYNC) { rpcPendingMap.entrySet().forEach(processPendingRpc(context, sessionId, server, sentOneWayIds)); } else { @@ -174,12 +174,12 @@ public class DeviceActorMessageProcessor extends AbstractContextAwareMsgProcesso sentOneWayIds.forEach(rpcPendingMap::remove); } - private Consumer> processPendingRpc(ActorContext context, SessionId sessionId, Optional server, Set sentOneWayIds) { + private Consumer> processPendingRpc(ActorContext context, SessionId sessionId, Optional server, Set sentOneWayIds) { return entry -> { ToDeviceRpcRequest request = entry.getValue().getMsg().getMsg(); ToDeviceRpcRequestBody body = request.getBody(); if (request.isOneway()) { - sentOneWayIds.add(request.getId()); + sentOneWayIds.add(entry.getKey()); ToPluginRpcResponseDeviceMsg responsePluginMsg = toPluginRpcResponseMsg(entry.getValue().getMsg(), (String) null); context.parent().tell(responsePluginMsg, ActorRef.noSender()); }