Merge pull request #13777 from ShvaykaD/bugfix/rpc-additional-info

Fixed handling of RPC with missing additional info in cluster mode

application/src/main/java/org/thingsboard/server/actors/device/DeviceActorMessageProcessor.java

@@ -270,10 +270,19 @@ public class DeviceActorMessageProcessor extends AbstractContextAwareMsgProcesso
         rpc.setExpirationTime(request.getExpirationTime());
         rpc.setRequest(JacksonUtil.valueToTree(request));
         rpc.setStatus(status);
-        rpc.setAdditionalInfo(JacksonUtil.toJsonNode(request.getAdditionalInfo()));
+        rpc.setAdditionalInfo(getAdditionalInfo(request));
         systemContext.getTbRpcService().save(tenantId, rpc);
     }
 
+    private JsonNode getAdditionalInfo(ToDeviceRpcRequest request) {
+        try {
+            return JacksonUtil.toJsonNode(request.getAdditionalInfo());
+        } catch (IllegalArgumentException e) {
+            log.debug("Failed to parse additional info [{}]", request.getAdditionalInfo());
+            return JacksonUtil.valueToTree(request.getAdditionalInfo());
+        }
+    }
+
     private ToDeviceRpcRequestMsg createToDeviceRpcRequestMsg(ToDeviceRpcRequest request) {
         ToDeviceRpcRequestBody body = request.getBody();
         return ToDeviceRpcRequestMsg.newBuilder()

common/proto/src/main/java/org/thingsboard/server/common/util/ProtoUtils.java

@@ -529,8 +529,10 @@ public class ProtoUtils {
                 .setRequestIdMSB(msg.getMsg().getId().getMostSignificantBits())
                 .setRequestIdLSB(msg.getMsg().getId().getLeastSignificantBits())
                 .setOneway(msg.getMsg().isOneway())
-                .setPersisted(msg.getMsg().isPersisted())
+                .setPersisted(msg.getMsg().isPersisted());
-                .setAdditionalInfo(msg.getMsg().getAdditionalInfo());
+        if (msg.getMsg().getAdditionalInfo() != null) {
+            builder.setAdditionalInfo(msg.getMsg().getAdditionalInfo());
+        }
         if (msg.getMsg().getRetries() != null) {
             builder.setRetries(msg.getMsg().getRetries());
         }
@@ -555,7 +557,9 @@ public class ProtoUtils {
                 toDeviceRpcRequestMsg.getOneway(),
                 toDeviceRpcRequestMsg.getExpirationTime(),
                 new ToDeviceRpcRequestBody(toDeviceRpcRequestMsg.getMethodName(), toDeviceRpcRequestMsg.getParams()),
-                toDeviceRpcRequestMsg.getPersisted(), toDeviceRpcRequestMsg.hasRetries() ? toDeviceRpcRequestMsg.getRetries() : null, toDeviceRpcRequestMsg.getAdditionalInfo());
+                toDeviceRpcRequestMsg.getPersisted(),
+                toDeviceRpcRequestMsg.hasRetries() ? toDeviceRpcRequestMsg.getRetries() : null,
+                toDeviceRpcRequestMsg.hasAdditionalInfo() ? toDeviceRpcRequestMsg.getAdditionalInfo() : null);
         return new ToDeviceRpcRequestActorMsg(proto.getServiceId(), toDeviceRpcRequest);
     }
 

common/proto/src/main/proto/queue.proto

@@ -699,7 +699,7 @@ message ToDeviceRpcRequestMsg {
   bool oneway = 7;
   bool persisted = 8;
   optional int32 retries = 9;
-  string additionalInfo = 10;
+  optional string additionalInfo = 10;
 }
 
 message ToDeviceRpcResponseMsg {

common/proto/src/test/java/org/thingsboard/server/common/util/ProtoUtilsTest.java

@@ -21,6 +21,9 @@ import org.jeasy.random.EasyRandomParameters;
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.BeforeAll;
 import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.NullAndEmptySource;
+import org.junit.jupiter.params.provider.ValueSource;
 import org.thingsboard.common.util.JacksonUtil;
 import org.thingsboard.server.common.data.ApiUsageState;
 import org.thingsboard.server.common.data.Device;
@@ -56,6 +59,7 @@ import org.thingsboard.server.common.data.security.DeviceCredentialsType;
 import org.thingsboard.server.common.data.sync.vc.RepositorySettings;
 import org.thingsboard.server.common.data.tenant.profile.DefaultTenantProfileConfiguration;
 import org.thingsboard.server.common.data.tenant.profile.TenantProfileConfiguration;
+import org.thingsboard.server.common.msg.ToDeviceActorNotificationMsg;
 import org.thingsboard.server.common.msg.edge.EdgeEventUpdateMsg;
 import org.thingsboard.server.common.msg.edge.EdgeHighPriorityMsg;
 import org.thingsboard.server.common.msg.edge.FromEdgeSyncResponse;
@@ -279,4 +283,65 @@ class ProtoUtilsTest {
         assertThat(actual).as(String.format(description, entityName, entityName)).isEqualTo(expected);
     }
 
+    @ParameterizedTest
+    @NullAndEmptySource
+    @ValueSource(strings = {"{\"key\":\"value\"}"})
+    void testRpcWithVariousAdditionalInfoToProtoAndBack(String additionalInfo) {
+        UUID requestId = UUID.fromString("93405c57-5787-46ff-806e-670bb60f49b6");
+        String methodName = "reboot";
+        String params = "";
+        String serviceId = "serviceId";
+        long expirationTime = System.currentTimeMillis();
+        int retries = 3;
+
+        ToDeviceRpcRequest request = new ToDeviceRpcRequest(
+                requestId,
+                tenantId,
+                deviceId,
+                false,
+                expirationTime,
+                new ToDeviceRpcRequestBody(methodName, params),
+                true,
+                retries,
+                additionalInfo
+        );
+        ToDeviceRpcRequestActorMsg msg = new ToDeviceRpcRequestActorMsg(serviceId, request);
+
+        // Serialize
+        TransportProtos.ToDeviceActorNotificationMsgProto toProto = ProtoUtils.toProto(msg);
+        assertThat(toProto).isNotNull();
+        assertThat(toProto.hasToDeviceRpcRequestMsg()).isTrue();
+
+        TransportProtos.ToDeviceRpcRequestActorMsgProto toDeviceRpcRequestActorMsgProto = toProto.getToDeviceRpcRequestMsg();
+        assertThat(toDeviceRpcRequestActorMsgProto.hasToDeviceRpcRequestMsg()).isTrue();
+
+        TransportProtos.ToDeviceRpcRequestMsg toDeviceRpcRequestMsg = toDeviceRpcRequestActorMsgProto.getToDeviceRpcRequestMsg();
+        assertThat(toDeviceRpcRequestMsg.getRequestIdMSB()).isEqualTo(requestId.getMostSignificantBits());
+        assertThat(toDeviceRpcRequestMsg.getRequestIdLSB()).isEqualTo(requestId.getLeastSignificantBits());
+        assertThat(toDeviceRpcRequestMsg.getMethodName()).isEqualTo(methodName);
+        assertThat(toDeviceRpcRequestMsg.getParams()).isEqualTo(params);
+        assertThat(toDeviceRpcRequestMsg.getExpirationTime()).isEqualTo(expirationTime);
+        assertThat(toDeviceRpcRequestMsg.getOneway()).isFalse();
+        assertThat(toDeviceRpcRequestMsg.getPersisted()).isTrue();
+        assertThat(toDeviceRpcRequestMsg.getRetries()).isEqualTo(retries);
+
+        if (additionalInfo != null) {
+            assertThat(toDeviceRpcRequestMsg.hasAdditionalInfo()).isTrue();
+            assertThat(toDeviceRpcRequestMsg.getAdditionalInfo()).isEqualTo(additionalInfo);
+        } else {
+            assertThat(toDeviceRpcRequestMsg.hasAdditionalInfo()).isFalse();
+        }
+
+        // Deserialize
+        ToDeviceActorNotificationMsg fromProto = ProtoUtils.fromProto(toProto);
+        assertThat(fromProto).isNotNull();
+        assertThat(fromProto).isInstanceOf(ToDeviceRpcRequestActorMsg.class);
+        ToDeviceRpcRequestActorMsg toDeviceRpcRequestActorMsg = (ToDeviceRpcRequestActorMsg) fromProto;
+
+        assertThat(toDeviceRpcRequestActorMsg.getDeviceId()).isEqualTo(deviceId);
+        assertThat(toDeviceRpcRequestActorMsg.getTenantId()).isEqualTo(tenantId);
+        assertThat(toDeviceRpcRequestActorMsg.getServiceId()).isEqualTo(serviceId);
+        assertThat(toDeviceRpcRequestActorMsg.getMsg()).isEqualTo(request);
+    }
+
 }