Added testDeviceReachedMaximumAllowedOnCloud

application/src/main/java/org/thingsboard/server/service/edge/DefaultEdgeNotificationService.java

@@ -122,7 +122,7 @@ public class DefaultEdgeNotificationService implements EdgeNotificationService {
 
     @Override
     public void pushNotificationToEdge(TransportProtos.EdgeNotificationMsgProto edgeNotificationMsg, TbCallback callback) {
-        log.trace("Pushing notification to edge {}", edgeNotificationMsg);
+        log.debug("Pushing notification to edge {}", edgeNotificationMsg);
         try {
             TenantId tenantId = TenantId.fromUUID(new UUID(edgeNotificationMsg.getTenantIdMSB(), edgeNotificationMsg.getTenantIdLSB()));
             EdgeEventType type = EdgeEventType.valueOf(edgeNotificationMsg.getType());
@@ -153,11 +153,12 @@ public class DefaultEdgeNotificationService implements EdgeNotificationService {
                     relationProcessor.processRelationNotification(tenantId, edgeNotificationMsg);
                     break;
                 default:
-                    log.debug("Edge event type [{}] is not designed to be pushed to edge", type);
+                    log.warn("Edge event type [{}] is not designed to be pushed to edge", type);
             }
         } catch (Exception e) {
             callback.onFailure(e);
-            log.error("Can't push to edge updates, edgeNotificationMsg [{}]", edgeNotificationMsg, e);
+            String errMsg = String.format("Can't push to edge updates, edgeNotificationMsg [%s]", edgeNotificationMsg);
+            log.error(errMsg, e);
         } finally {
             callback.onSuccess();
         }

application/src/main/java/org/thingsboard/server/service/edge/rpc/EdgeGrpcSession.java

@@ -644,6 +644,7 @@ public final class EdgeGrpcSession implements Closeable {
             }
         } catch (Exception e) {
             log.error("[{}] Can't process uplink msg [{}]", this.sessionId, uplinkMsg, e);
+            return Futures.immediateFailedFuture(e);
         }
         return Futures.allAsList(result);
     }

application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/DeviceEdgeProcessor.java

@@ -51,6 +51,7 @@ import org.thingsboard.server.common.msg.TbMsg;
 import org.thingsboard.server.common.msg.TbMsgDataType;
 import org.thingsboard.server.common.msg.TbMsgMetaData;
 import org.thingsboard.server.common.msg.rpc.FromDeviceRpcResponse;
+import org.thingsboard.server.dao.exception.DataValidationException;
 import org.thingsboard.server.dao.model.ModelConstants;
 import org.thingsboard.server.gen.edge.v1.DeviceCredentialsUpdateMsg;
 import org.thingsboard.server.gen.edge.v1.DeviceRpcCallMsg;
@@ -79,38 +80,37 @@ public class DeviceEdgeProcessor extends BaseEdgeProcessor {
                 String deviceName = deviceUpdateMsg.getName();
                 Device device = deviceService.findDeviceByTenantIdAndName(tenantId, deviceName);
                 if (device != null) {
-                    PageLink pageLink = new PageLink(DEFAULT_PAGE_SIZE);
+                    boolean deviceAlreadyExistsForThisEdge = isDeviceAlreadyExistsOnCloudForThisEdge(tenantId, edge, device);
-                    PageData<EdgeId> pageData;
+                    if (deviceAlreadyExistsForThisEdge) {
-                    do {
+                        log.info("[{}] Device with name '{}' already exists on the cloud, and related to this edge [{}]. " +
-                        pageData = edgeService.findRelatedEdgeIdsByEntityId(tenantId, device.getId(), pageLink);
+                                "deviceUpdateMsg [{}], Updating device", tenantId, deviceName, edge.getId(), deviceUpdateMsg);
-                        boolean update = false;
+                        updateDevice(tenantId, edge, deviceUpdateMsg);
-                        if (pageData != null && pageData.getData() != null && !pageData.getData().isEmpty()) {
+                    } else {
-                            if (pageData.getData().contains(edge.getId())) {
+                        log.info("[{}] Device with name '{}' already exists on the cloud, but not related to this edge [{}]. deviceUpdateMsg [{}]." +
-                                update = true;
+                                "Creating a new device with random prefix and relate to this edge", tenantId, deviceName, edge.getId(), deviceUpdateMsg);
-                            }
+                        String newDeviceName = deviceUpdateMsg.getName() + "_" + RandomStringUtils.randomAlphabetic(15);
-                            if (pageData.hasNext()) {
+                        Device newDevice;
-                                pageLink = pageLink.nextPageLink();
+                        try {
-                            }
+                            newDevice = createDevice(tenantId, edge, deviceUpdateMsg, newDeviceName);
+                        } catch (DataValidationException e) {
+                            String errMsg = String.format("[%s] Device update msg can't be processed due to data validation [%s]", tenantId, deviceUpdateMsg);
+                            log.error(errMsg, e);
+                            return Futures.immediateFuture(null);
                         }
-
+                        ObjectNode body = mapper.createObjectNode();
-                        if (update) {
+                        body.put("conflictName", deviceName);
-                            log.info("[{}] Device with name '{}' already exists on the cloud, and related to this edge [{}]. " +
+                        saveEdgeEvent(tenantId, edge.getId(), EdgeEventType.DEVICE, EdgeEventActionType.ENTITY_MERGE_REQUEST, newDevice.getId(), body);
-                                    "deviceUpdateMsg [{}], Updating device", tenantId, deviceName, edge.getId(), deviceUpdateMsg);
+                        saveEdgeEvent(tenantId, edge.getId(), EdgeEventType.DEVICE, EdgeEventActionType.CREDENTIALS_REQUEST, newDevice.getId(), null);
-                            updateDevice(tenantId, edge, deviceUpdateMsg);
+                    }
-                        } else {
-                            log.info("[{}] Device with name '{}' already exists on the cloud, but not related to this edge [{}]. deviceUpdateMsg [{}]." +
-                                    "Creating a new device with random prefix and relate to this edge", tenantId, deviceName, edge.getId(), deviceUpdateMsg);
-                            String newDeviceName = deviceUpdateMsg.getName() + "_" + RandomStringUtils.randomAlphabetic(15);
-                            Device newDevice = createDevice(tenantId, edge, deviceUpdateMsg, newDeviceName);
-                            ObjectNode body = mapper.createObjectNode();
-                            body.put("conflictName", deviceName);
-                            saveEdgeEvent(tenantId, edge.getId(), EdgeEventType.DEVICE, EdgeEventActionType.ENTITY_MERGE_REQUEST, newDevice.getId(), body);
-                            saveEdgeEvent(tenantId, edge.getId(), EdgeEventType.DEVICE, EdgeEventActionType.CREDENTIALS_REQUEST, newDevice.getId(), null);
-                        }
-                    } while (pageData != null && pageData.hasNext());
                 } else {
                     log.info("[{}] Creating new device and replacing device entity on the edge [{}]", tenantId, deviceUpdateMsg);
-                    device = createDevice(tenantId, edge, deviceUpdateMsg, deviceUpdateMsg.getName());
+                    try {
+                        device = createDevice(tenantId, edge, deviceUpdateMsg, deviceUpdateMsg.getName());
+                    } catch (DataValidationException e) {
+                        String errMsg = String.format("[%s] Device update msg can't be processed due to data validation [%s]", tenantId, deviceUpdateMsg);
+                        log.error(errMsg, e);
+                        return Futures.immediateFuture(null);
+                    }
                     saveEdgeEvent(tenantId, edge.getId(), EdgeEventType.DEVICE, EdgeEventActionType.CREDENTIALS_REQUEST, device.getId(), null);
                 }
                 break;
@@ -131,6 +131,23 @@ public class DeviceEdgeProcessor extends BaseEdgeProcessor {
         return Futures.immediateFuture(null);
     }
 
+    private boolean isDeviceAlreadyExistsOnCloudForThisEdge(TenantId tenantId, Edge edge, Device device) {
+        PageLink pageLink = new PageLink(DEFAULT_PAGE_SIZE);
+        PageData<EdgeId> pageData;
+        do {
+            pageData = edgeService.findRelatedEdgeIdsByEntityId(tenantId, device.getId(), pageLink);
+            if (pageData != null && pageData.getData() != null && !pageData.getData().isEmpty()) {
+                if (pageData.getData().contains(edge.getId())) {
+                    return true;
+                }
+                if (pageData.hasNext()) {
+                    pageLink = pageLink.nextPageLink();
+                }
+            }
+        } while (pageData != null && pageData.hasNext());
+        return false;
+    }
+
     public ListenableFuture<Void> processDeviceCredentialsFromEdge(TenantId tenantId, DeviceCredentialsUpdateMsg deviceCredentialsUpdateMsg) {
         log.debug("Executing onDeviceCredentialsUpdate, deviceCredentialsUpdateMsg [{}]", deviceCredentialsUpdateMsg);
         DeviceId deviceId = new DeviceId(new UUID(deviceCredentialsUpdateMsg.getDeviceIdMSB(), deviceCredentialsUpdateMsg.getDeviceIdLSB()));

application/src/test/java/org/thingsboard/server/edge/BaseEdgeTest.java

@@ -42,6 +42,7 @@ import org.thingsboard.server.common.data.DeviceProfile;
 import org.thingsboard.server.common.data.EntityType;
 import org.thingsboard.server.common.data.EntityView;
 import org.thingsboard.server.common.data.Tenant;
+import org.thingsboard.server.common.data.TenantProfile;
 import org.thingsboard.server.common.data.User;
 import org.thingsboard.server.common.data.alarm.Alarm;
 import org.thingsboard.server.common.data.alarm.AlarmInfo;
@@ -80,6 +81,7 @@ import org.thingsboard.server.common.data.rule.RuleNode;
 import org.thingsboard.server.common.data.security.Authority;
 import org.thingsboard.server.common.data.security.DeviceCredentials;
 import org.thingsboard.server.common.data.security.DeviceCredentialsType;
+import org.thingsboard.server.common.data.tenant.profile.DefaultTenantProfileConfiguration;
 import org.thingsboard.server.common.data.widget.WidgetType;
 import org.thingsboard.server.common.data.widget.WidgetsBundle;
 import org.thingsboard.server.common.transport.adaptor.JsonConverter;
@@ -110,6 +112,7 @@ import org.thingsboard.server.gen.edge.v1.RuleChainMetadataUpdateMsg;
 import org.thingsboard.server.gen.edge.v1.RuleChainUpdateMsg;
 import org.thingsboard.server.gen.edge.v1.UpdateMsgType;
 import org.thingsboard.server.gen.edge.v1.UplinkMsg;
+import org.thingsboard.server.gen.edge.v1.UplinkResponseMsg;
 import org.thingsboard.server.gen.edge.v1.UserCredentialsRequestMsg;
 import org.thingsboard.server.gen.edge.v1.UserCredentialsUpdateMsg;
 import org.thingsboard.server.gen.edge.v1.UserUpdateMsg;
@@ -385,6 +388,40 @@ abstract public class BaseEdgeTest extends AbstractControllerTest {
 
     }
 
+    @Test
+    public void testDeviceReachedMaximumAllowedOnCloud() throws Exception {
+        // update tenant profile configuration
+        loginSysAdmin();
+        TenantProfile tenantProfile = doGet("/api/tenantProfile/" + savedTenant.getTenantProfileId().getId(), TenantProfile.class);
+        DefaultTenantProfileConfiguration profileConfiguration =
+                (DefaultTenantProfileConfiguration) tenantProfile.getProfileData().getConfiguration();
+        profileConfiguration.setMaxDevices(1);
+        tenantProfile.getProfileData().setConfiguration(profileConfiguration);
+        doPost("/api/tenantProfile/", tenantProfile, TenantProfile.class);
+
+        loginTenantAdmin();
+
+        UUID uuid = Uuids.timeBased();
+
+        UplinkMsg.Builder uplinkMsgBuilder = UplinkMsg.newBuilder();
+        DeviceUpdateMsg.Builder deviceUpdateMsgBuilder = DeviceUpdateMsg.newBuilder();
+        deviceUpdateMsgBuilder.setIdMSB(uuid.getMostSignificantBits());
+        deviceUpdateMsgBuilder.setIdLSB(uuid.getLeastSignificantBits());
+        deviceUpdateMsgBuilder.setName("Edge Device");
+        deviceUpdateMsgBuilder.setType("default");
+        deviceUpdateMsgBuilder.setMsgType(UpdateMsgType.ENTITY_CREATED_RPC_MESSAGE);
+        uplinkMsgBuilder.addDeviceUpdateMsg(deviceUpdateMsgBuilder.build());
+
+        edgeImitator.expectResponsesAmount(1);
+
+        edgeImitator.sendUplinkMsg(uplinkMsgBuilder.build());
+
+        Assert.assertTrue(edgeImitator.waitForResponses());
+
+        UplinkResponseMsg latestResponseMsg = edgeImitator.getLatestResponseMsg();
+        Assert.assertTrue(latestResponseMsg.getSuccess());
+    }
+
     @Test
     public void testAssets() throws Exception {
         // 1

application/src/test/java/org/thingsboard/server/edge/imitator/EdgeImitator.java

@@ -88,6 +88,9 @@ public class EdgeImitator {
     @Getter
     private List<AbstractMessage> downlinkMsgs;
 
+    @Getter
+    private UplinkResponseMsg latestResponseMsg;
+
     private boolean connected = false;
 
     public EdgeImitator(String host, int port, String routingKey, String routingSecret) throws NoSuchFieldException, IllegalAccessException {
@@ -132,6 +135,7 @@ public class EdgeImitator {
 
     private void onUplinkResponse(UplinkResponseMsg msg) {
         log.info("onUplinkResponse: {}", msg);
+        latestResponseMsg = msg;
         responsesLatch.countDown();
     }