Merge branch 'master' of github.com:thingsboard/thingsboard

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

@@ -45,7 +45,6 @@ import javax.annotation.PostConstruct;
 
 import static org.thingsboard.server.service.entitiy.DefaultTbNotificationEntityService.edgeTypeByActionType;
 
-
 /**
  * This event listener does not support async event processing because relay on ThreadLocal
  * Another possible approach is to implement a special annotation and a bunch of classes similar to TransactionalApplicationListener
@@ -75,9 +74,6 @@ public class EdgeEventSourcingListener {
 
     @TransactionalEventListener(fallbackExecution = true)
     public void handleEvent(SaveEntityEvent<?> event) {
-        if (edgeSynchronizationManager.isSync()) {
-            return;
-        }
         try {
             if (!isValidSaveEntityEventForEdgeProcessing(event.getEntity(), event.getOldEntity())) {
                 return;
@@ -85,7 +81,7 @@ public class EdgeEventSourcingListener {
             log.trace("[{}] SaveEntityEvent called: {}", event.getTenantId(), event);
             EdgeEventActionType action = Boolean.TRUE.equals(event.getAdded()) ? EdgeEventActionType.ADDED : EdgeEventActionType.UPDATED;
             tbClusterService.sendNotificationMsgToEdge(event.getTenantId(), null, event.getEntityId(),
-                    null, null, action);
+                    null, null, action, edgeSynchronizationManager.getEdgeId().get());
         } catch (Exception e) {
             log.error("[{}] failed to process SaveEntityEvent: {}", event.getTenantId(), event, e);
         }
@@ -93,13 +89,11 @@ public class EdgeEventSourcingListener {
 
     @TransactionalEventListener(fallbackExecution = true)
     public void handleEvent(DeleteEntityEvent<?> event) {
-        if (edgeSynchronizationManager.isSync()) {
-            return;
-        }
         try {
             log.trace("[{}] DeleteEntityEvent called: {}", event.getTenantId(), event);
-            tbClusterService.sendNotificationMsgToEdge(event.getTenantId(), event.getEdgeId(), event.getEntityId(),
-                    JacksonUtil.toString(event.getEntity()), null, EdgeEventActionType.DELETED);
+            tbClusterService.sendNotificationMsgToEdge(event.getTenantId(), null, event.getEntityId(),
+                    JacksonUtil.toString(event.getEntity()), null, EdgeEventActionType.DELETED,
+                    edgeSynchronizationManager.getEdgeId().get());
         } catch (Exception e) {
             log.error("[{}] failed to process DeleteEntityEvent: {}", event.getTenantId(), event, e);
         }
@@ -107,13 +101,11 @@ public class EdgeEventSourcingListener {
 
     @TransactionalEventListener(fallbackExecution = true)
     public void handleEvent(ActionEntityEvent event) {
-        if (edgeSynchronizationManager.isSync()) {
-            return;
-        }
         try {
             log.trace("[{}] ActionEntityEvent called: {}", event.getTenantId(), event);
             tbClusterService.sendNotificationMsgToEdge(event.getTenantId(), event.getEdgeId(), event.getEntityId(),
-                    event.getBody(), null, edgeTypeByActionType(event.getActionType()));
+                    event.getBody(), null, edgeTypeByActionType(event.getActionType()),
+                    edgeSynchronizationManager.getEdgeId().get());
         } catch (Exception e) {
             log.error("[{}] failed to process ActionEntityEvent: {}", event.getTenantId(), event, e);
         }
@@ -121,9 +113,6 @@ public class EdgeEventSourcingListener {
 
     @TransactionalEventListener(fallbackExecution = true)
     public void handleEvent(RelationActionEvent event) {
-        if (edgeSynchronizationManager.isSync()) {
-            return;
-        }
         try {
             EntityRelation relation = event.getRelation();
             if (relation == null) {
@@ -136,7 +125,8 @@ public class EdgeEventSourcingListener {
             }
             log.trace("[{}] RelationActionEvent called: {}", event.getTenantId(), event);
             tbClusterService.sendNotificationMsgToEdge(event.getTenantId(), null, null,
-                    JacksonUtil.toString(relation), EdgeEventType.RELATION, edgeTypeByActionType(event.getActionType()));
+                    JacksonUtil.toString(relation), EdgeEventType.RELATION, edgeTypeByActionType(event.getActionType()),
+                    edgeSynchronizationManager.getEdgeId().get());
         } catch (Exception e) {
             log.error("[{}] failed to process RelationActionEvent: {}", event.getTenantId(), event, e);
         }

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

@@ -677,7 +677,7 @@ public final class EdgeGrpcSession implements Closeable {
             }
             if (uplinkMsg.getDeviceCredentialsUpdateMsgCount() > 0) {
                 for (DeviceCredentialsUpdateMsg deviceCredentialsUpdateMsg : uplinkMsg.getDeviceCredentialsUpdateMsgList()) {
-                    result.add(ctx.getDeviceProcessor().processDeviceCredentialsMsg(edge.getTenantId(), deviceCredentialsUpdateMsg));
+                    result.add(ctx.getDeviceProcessor().processDeviceCredentialsMsgFromEdge(edge.getTenantId(), edge.getId(), deviceCredentialsUpdateMsg));
                 }
             }
             if (uplinkMsg.getAssetProfileUpdateMsgCount() > 0) {
@@ -692,7 +692,7 @@ public final class EdgeGrpcSession implements Closeable {
             }
             if (uplinkMsg.getAlarmUpdateMsgCount() > 0) {
                 for (AlarmUpdateMsg alarmUpdateMsg : uplinkMsg.getAlarmUpdateMsgList()) {
-                    result.add(ctx.getAlarmProcessor().processAlarmMsg(edge.getTenantId(), alarmUpdateMsg));
+                    result.add(ctx.getAlarmProcessor().processAlarmMsgFromEdge(edge.getTenantId(), edge.getId(), alarmUpdateMsg));
                 }
             }
             if (uplinkMsg.getEntityViewUpdateMsgCount() > 0) {
@@ -702,7 +702,7 @@ public final class EdgeGrpcSession implements Closeable {
             }
             if (uplinkMsg.getRelationUpdateMsgCount() > 0) {
                 for (RelationUpdateMsg relationUpdateMsg : uplinkMsg.getRelationUpdateMsgList()) {
-                    result.add(ctx.getRelationProcessor().processRelationMsg(edge.getTenantId(), relationUpdateMsg));
+                    result.add(ctx.getRelationProcessor().processRelationMsgFromEdge(edge.getTenantId(), edge, relationUpdateMsg));
                 }
             }
             if (uplinkMsg.getDashboardUpdateMsgCount() > 0) {

application/src/main/java/org/thingsboard/server/service/edge/rpc/constructor/rule/AbstractRuleChainMetadataConstructor.java

@@ -15,7 +15,6 @@
  */
 package org.thingsboard.server.service.edge.rpc.constructor.rule;
 
-import com.fasterxml.jackson.core.JsonProcessingException;
 import com.fasterxml.jackson.databind.node.ObjectNode;
 import lombok.AllArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
@@ -50,7 +49,7 @@ public abstract class AbstractRuleChainMetadataConstructor implements RuleChainM
             constructRuleChainMetadataUpdatedMsg(tenantId, builder, ruleChainMetaData);
             builder.setMsgType(msgType);
             return builder.build();
-        } catch (JsonProcessingException ex) {
+        } catch (Exception ex) {
             log.error("[{}] Can't construct RuleChainMetadataUpdateMsg", tenantId, ex);
         }
         return null;
@@ -58,7 +57,7 @@ public abstract class AbstractRuleChainMetadataConstructor implements RuleChainM
 
     protected abstract void constructRuleChainMetadataUpdatedMsg(TenantId tenantId,
                                                                  RuleChainMetadataUpdateMsg.Builder builder,
-                                                                 RuleChainMetaData ruleChainMetaData) throws JsonProcessingException;
+                                                                 RuleChainMetaData ruleChainMetaData);
 
     protected List<NodeConnectionInfoProto> constructConnections(List<NodeConnectionInfo> connections) {
         List<NodeConnectionInfoProto> result = new ArrayList<>();
@@ -78,7 +77,7 @@ public abstract class AbstractRuleChainMetadataConstructor implements RuleChainM
                 .build();
     }
 
-    protected List<RuleNodeProto> constructNodes(List<RuleNode> nodes) throws JsonProcessingException {
+    protected List<RuleNodeProto> constructNodes(List<RuleNode> nodes) {
         List<RuleNodeProto> result = new ArrayList<>();
         if (nodes != null && !nodes.isEmpty()) {
             for (RuleNode node : nodes) {
@@ -88,22 +87,22 @@ public abstract class AbstractRuleChainMetadataConstructor implements RuleChainM
         return result;
     }
 
-    private RuleNodeProto constructNode(RuleNode node) throws JsonProcessingException {
+    private RuleNodeProto constructNode(RuleNode node) {
         return RuleNodeProto.newBuilder()
                 .setIdMSB(node.getId().getId().getMostSignificantBits())
                 .setIdLSB(node.getId().getId().getLeastSignificantBits())
                 .setType(node.getType())
                 .setName(node.getName())
                 .setDebugMode(node.isDebugMode())
-                .setConfiguration(JacksonUtil.OBJECT_MAPPER.writeValueAsString(node.getConfiguration()))
-                .setAdditionalInfo(JacksonUtil.OBJECT_MAPPER.writeValueAsString(node.getAdditionalInfo()))
+                .setConfiguration(JacksonUtil.toString(node.getConfiguration()))
+                .setAdditionalInfo(JacksonUtil.toString(node.getAdditionalInfo()))
                 .setSingletonMode(node.isSingletonMode())
                 .setConfigurationVersion(node.getConfigurationVersion())
                 .build();
     }
 
     protected List<RuleChainConnectionInfoProto> constructRuleChainConnections(List<RuleChainConnectionInfo> ruleChainConnections,
-                                                                               NavigableSet<Integer> removedNodeIndexes) throws JsonProcessingException {
+                                                                               NavigableSet<Integer> removedNodeIndexes) {
         List<RuleChainConnectionInfoProto> result = new ArrayList<>();
         if (ruleChainConnections != null && !ruleChainConnections.isEmpty()) {
             for (RuleChainConnectionInfo ruleChainConnectionInfo : ruleChainConnections) {
@@ -127,13 +126,13 @@ public abstract class AbstractRuleChainMetadataConstructor implements RuleChainM
         return result;
     }
 
-    private RuleChainConnectionInfoProto constructRuleChainConnection(RuleChainConnectionInfo ruleChainConnectionInfo) throws JsonProcessingException {
+    private RuleChainConnectionInfoProto constructRuleChainConnection(RuleChainConnectionInfo ruleChainConnectionInfo) {
         return RuleChainConnectionInfoProto.newBuilder()
                 .setFromIndex(ruleChainConnectionInfo.getFromIndex())
                 .setTargetRuleChainIdMSB(ruleChainConnectionInfo.getTargetRuleChainId().getId().getMostSignificantBits())
                 .setTargetRuleChainIdLSB(ruleChainConnectionInfo.getTargetRuleChainId().getId().getLeastSignificantBits())
                 .setType(ruleChainConnectionInfo.getType())
-                .setAdditionalInfo(JacksonUtil.OBJECT_MAPPER.writeValueAsString(ruleChainConnectionInfo.getAdditionalInfo()))
+                .setAdditionalInfo(JacksonUtil.toString(ruleChainConnectionInfo.getAdditionalInfo()))
                 .build();
     }
 }

application/src/main/java/org/thingsboard/server/service/edge/rpc/constructor/rule/RuleChainMetadataConstructorV330.java

@@ -15,7 +15,6 @@
  */
 package org.thingsboard.server.service.edge.rpc.constructor.rule;
 
-import com.fasterxml.jackson.core.JsonProcessingException;
 import lombok.extern.slf4j.Slf4j;
 import org.thingsboard.common.util.JacksonUtil;
 import org.thingsboard.rule.engine.flow.TbRuleChainInputNode;
@@ -45,7 +44,7 @@ public class RuleChainMetadataConstructorV330 extends AbstractRuleChainMetadataC
     @Override
     protected void constructRuleChainMetadataUpdatedMsg(TenantId tenantId,
                                                         RuleChainMetadataUpdateMsg.Builder builder,
-                                                        RuleChainMetaData ruleChainMetaData) throws JsonProcessingException {
+                                                        RuleChainMetaData ruleChainMetaData) {
         List<RuleNode> supportedNodes = filterNodes(ruleChainMetaData.getNodes());
 
         NavigableSet<Integer> removedNodeIndexes = getRemovedNodeIndexes(ruleChainMetaData.getNodes(), ruleChainMetaData.getConnections());

application/src/main/java/org/thingsboard/server/service/edge/rpc/constructor/rule/RuleChainMetadataConstructorV340.java

@@ -15,7 +15,6 @@
  */
 package org.thingsboard.server.service.edge.rpc.constructor.rule;
 
-import com.fasterxml.jackson.core.JsonProcessingException;
 import lombok.extern.slf4j.Slf4j;
 import org.thingsboard.server.common.data.id.TenantId;
 import org.thingsboard.server.common.data.rule.RuleChainMetaData;
@@ -29,7 +28,7 @@ public class RuleChainMetadataConstructorV340 extends AbstractRuleChainMetadataC
     @Override
     protected void constructRuleChainMetadataUpdatedMsg(TenantId tenantId,
                                                         RuleChainMetadataUpdateMsg.Builder builder,
-                                                        RuleChainMetaData ruleChainMetaData) throws JsonProcessingException {
+                                                        RuleChainMetaData ruleChainMetaData) {
         builder.addAllNodes(constructNodes(ruleChainMetaData.getNodes()))
                 .addAllConnections(constructConnections(ruleChainMetaData.getConnections()))
                 .addAllRuleChainConnections(constructRuleChainConnections(ruleChainMetaData.getRuleChainConnections(), new TreeSet<>()));

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

@@ -79,19 +79,19 @@ public class AdminSettingsEdgeEventFetcher implements EdgeEventFetcher {
 
         AdminSettings systemMailSettings = adminSettingsService.findAdminSettingsByKey(TenantId.SYS_TENANT_ID, "mail");
         result.add(EdgeUtils.constructEdgeEvent(tenantId, edge.getId(), EdgeEventType.ADMIN_SETTINGS,
-                EdgeEventActionType.UPDATED, null, JacksonUtil.OBJECT_MAPPER.valueToTree(systemMailSettings)));
+                EdgeEventActionType.UPDATED, null, JacksonUtil.valueToTree(systemMailSettings)));
 
         AdminSettings tenantMailSettings = convertToTenantAdminSettings(tenantId, systemMailSettings.getKey(), (ObjectNode) systemMailSettings.getJsonValue());
         result.add(EdgeUtils.constructEdgeEvent(tenantId, edge.getId(), EdgeEventType.ADMIN_SETTINGS,
-                EdgeEventActionType.UPDATED, null, JacksonUtil.OBJECT_MAPPER.valueToTree(tenantMailSettings)));
+                EdgeEventActionType.UPDATED, null, JacksonUtil.valueToTree(tenantMailSettings)));
 
         AdminSettings systemMailTemplates = loadMailTemplates(tenantId);
         result.add(EdgeUtils.constructEdgeEvent(tenantId, edge.getId(), EdgeEventType.ADMIN_SETTINGS,
-                EdgeEventActionType.UPDATED, null, JacksonUtil.OBJECT_MAPPER.valueToTree(systemMailTemplates)));
+                EdgeEventActionType.UPDATED, null, JacksonUtil.valueToTree(systemMailTemplates)));
 
         AdminSettings tenantMailTemplates = convertToTenantAdminSettings(tenantId, systemMailTemplates.getKey(), (ObjectNode) systemMailTemplates.getJsonValue());
         result.add(EdgeUtils.constructEdgeEvent(tenantId, edge.getId(), EdgeEventType.ADMIN_SETTINGS,
-                EdgeEventActionType.UPDATED, null, JacksonUtil.OBJECT_MAPPER.valueToTree(tenantMailTemplates)));
+                EdgeEventActionType.UPDATED, null, JacksonUtil.valueToTree(tenantMailTemplates)));
 
         // return PageData object to be in sync with other fetchers
         return new PageData<>(result, 1, result.size(), false);
@@ -114,7 +114,7 @@ public class AdminSettingsEdgeEventFetcher implements EdgeEventFetcher {
         AdminSettings adminSettings = new AdminSettings();
         adminSettings.setId(new AdminSettingsId(Uuids.timeBased()));
         adminSettings.setKey("mailTemplates");
-        adminSettings.setJsonValue(JacksonUtil.OBJECT_MAPPER.convertValue(mailTemplates, JsonNode.class));
+        adminSettings.setJsonValue(JacksonUtil.convertValue(mailTemplates, JsonNode.class));
         return adminSettings;
     }
 

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

@@ -376,7 +376,9 @@ public abstract class BaseEdgeProcessor {
         }, dbCallbackExecutorService);
     }
 
-    protected ListenableFuture<Void> processActionForAllEdges(TenantId tenantId, EdgeEventType type, EdgeEventActionType actionType, EntityId entityId) {
+    protected ListenableFuture<Void> processActionForAllEdges(TenantId tenantId, EdgeEventType type,
+                                                              EdgeEventActionType actionType, EntityId entityId,
+                                                              EdgeId sourceEdgeId) {
         List<ListenableFuture<Void>> futures = new ArrayList<>();
         if (TenantId.SYS_TENANT_ID.equals(tenantId)) {
             PageLink pageLink = new PageLink(DEFAULT_PAGE_SIZE);
@@ -384,21 +386,22 @@ public abstract class BaseEdgeProcessor {
             do {
                 tenantsIds = tenantService.findTenantsIds(pageLink);
                 for (TenantId tenantId1 : tenantsIds.getData()) {
-                    futures.addAll(processActionForAllEdgesByTenantId(tenantId1, type, actionType, entityId, null));
+                    futures.addAll(processActionForAllEdgesByTenantId(tenantId1, type, actionType, entityId, null, sourceEdgeId));
                 }
                 pageLink = pageLink.nextPageLink();
             } while (tenantsIds.hasNext());
         } else {
-            futures = processActionForAllEdgesByTenantId(tenantId, type, actionType, entityId, null);
+            futures = processActionForAllEdgesByTenantId(tenantId, type, actionType, entityId, null, sourceEdgeId);
         }
         return Futures.transform(Futures.allAsList(futures), voids -> null, dbCallbackExecutorService);
     }
 
     private List<ListenableFuture<Void>> processActionForAllEdgesByTenantId(TenantId tenantId,
-                                                                              EdgeEventType type,
-                                                                              EdgeEventActionType actionType,
-                                                                              EntityId entityId,
-                                                                              JsonNode body) {
+                                                                            EdgeEventType type,
+                                                                            EdgeEventActionType actionType,
+                                                                            EntityId entityId,
+                                                                            JsonNode body,
+                                                                            EdgeId sourceEdgeId) {
         PageLink pageLink = new PageLink(DEFAULT_PAGE_SIZE);
         PageData<Edge> pageData;
         List<ListenableFuture<Void>> futures = new ArrayList<>();
@@ -406,7 +409,9 @@ public abstract class BaseEdgeProcessor {
             pageData = edgeService.findEdgesByTenantId(tenantId, pageLink);
             if (pageData != null && pageData.getData() != null && !pageData.getData().isEmpty()) {
                 for (Edge edge : pageData.getData()) {
-                    futures.add(saveEdgeEvent(tenantId, edge.getId(), type, actionType, entityId, body));
+                    if (!edge.getId().equals(sourceEdgeId)) {
+                        futures.add(saveEdgeEvent(tenantId, edge.getId(), type, actionType, entityId, body));
+                    }
                 }
                 if (pageData.hasNext()) {
                     pageLink = pageLink.nextPageLink();
@@ -450,11 +455,12 @@ public abstract class BaseEdgeProcessor {
         EdgeEventType type = EdgeEventType.valueOf(edgeNotificationMsg.getType());
         EdgeEventActionType actionType = EdgeEventActionType.valueOf(edgeNotificationMsg.getAction());
         EntityId entityId = EntityIdFactory.getByEdgeEventTypeAndUuid(type, new UUID(edgeNotificationMsg.getEntityIdMSB(), edgeNotificationMsg.getEntityIdLSB()));
+        EdgeId sourceEdgeId = safeGetEdgeId(edgeNotificationMsg.getSourceEdgeIdMSB(), edgeNotificationMsg.getSourceEdgeIdLSB());
         if (type.isAllEdgesRelated()) {
-            return processEntityNotificationForAllEdges(tenantId, type, actionType, entityId);
+            return processEntityNotificationForAllEdges(tenantId, type, actionType, entityId, sourceEdgeId);
         } else {
             JsonNode body = JacksonUtil.toJsonNode(edgeNotificationMsg.getBody());
-            EdgeId edgeId = safeGetEdgeId(edgeNotificationMsg);
+            EdgeId edgeId = safeGetEdgeId(edgeNotificationMsg.getEdgeIdMSB(), edgeNotificationMsg.getEdgeIdLSB());
             switch (actionType) {
                 case UPDATED:
                 case CREDENTIALS_UPDATED:
@@ -463,41 +469,46 @@ public abstract class BaseEdgeProcessor {
                     if (edgeId != null) {
                         return saveEdgeEvent(tenantId, edgeId, type, actionType, entityId, body);
                     } else {
-                        return processNotificationToRelatedEdges(tenantId, entityId, type, actionType);
+                        return processNotificationToRelatedEdges(tenantId, entityId, type, actionType, sourceEdgeId);
                     }
                 case DELETED:
                     EdgeEventActionType deleted = EdgeEventActionType.DELETED;
                     if (edgeId != null) {
                         return saveEdgeEvent(tenantId, edgeId, type, deleted, entityId, body);
                     } else {
-                        return Futures.transform(Futures.allAsList(processActionForAllEdgesByTenantId(tenantId, type, deleted, entityId, body)),
+                        return Futures.transform(Futures.allAsList(processActionForAllEdgesByTenantId(tenantId, type, deleted, entityId, body, sourceEdgeId)),
                                 voids -> null, dbCallbackExecutorService);
                     }
                 case ASSIGNED_TO_EDGE:
                 case UNASSIGNED_FROM_EDGE:
-                    ListenableFuture<Void> future = saveEdgeEvent(tenantId, edgeId, type, actionType, entityId, body);
-                    return Futures.transformAsync(future, unused -> {
-                        if (type.equals(EdgeEventType.RULE_CHAIN)) {
-                            return updateDependentRuleChains(tenantId, new RuleChainId(entityId.getId()), edgeId);
-                        } else {
-                            return Futures.immediateFuture(null);
-                        }
-                    }, dbCallbackExecutorService);
+                    if (sourceEdgeId == null) {
+                        ListenableFuture<Void> future = saveEdgeEvent(tenantId, edgeId, type, actionType, entityId, body);
+                        return Futures.transformAsync(future, unused -> {
+                            if (type.equals(EdgeEventType.RULE_CHAIN)) {
+                                return updateDependentRuleChains(tenantId, new RuleChainId(entityId.getId()), edgeId);
+                            } else {
+                                return Futures.immediateFuture(null);
+                            }
+                        }, dbCallbackExecutorService);
+                    } else {
+                        return Futures.immediateFuture(null);
+                    }
                 default:
                     return Futures.immediateFuture(null);
             }
         }
     }
 
-    private EdgeId safeGetEdgeId(TransportProtos.EdgeNotificationMsgProto edgeNotificationMsg) {
-        if (edgeNotificationMsg.getEdgeIdMSB() != 0 && edgeNotificationMsg.getEdgeIdLSB() != 0) {
-            return new EdgeId(new UUID(edgeNotificationMsg.getEdgeIdMSB(), edgeNotificationMsg.getEdgeIdLSB()));
+    protected EdgeId safeGetEdgeId(long edgeIdMSB, long edgeIdLSB) {
+        if (edgeIdMSB != 0 && edgeIdLSB != 0) {
+            return new EdgeId(new UUID(edgeIdMSB, edgeIdLSB));
         } else {
             return null;
         }
     }
 
-    private ListenableFuture<Void> processNotificationToRelatedEdges(TenantId tenantId, EntityId entityId, EdgeEventType type, EdgeEventActionType actionType) {
+    private ListenableFuture<Void> processNotificationToRelatedEdges(TenantId tenantId, EntityId entityId, EdgeEventType type,
+                                                                     EdgeEventActionType actionType, EdgeId sourceEdgeId) {
         PageLink pageLink = new PageLink(DEFAULT_PAGE_SIZE);
         PageData<EdgeId> pageData;
         List<ListenableFuture<Void>> futures = new ArrayList<>();
@@ -505,7 +516,9 @@ public abstract class BaseEdgeProcessor {
             pageData = edgeService.findRelatedEdgeIdsByEntityId(tenantId, entityId, pageLink);
             if (pageData != null && pageData.getData() != null && !pageData.getData().isEmpty()) {
                 for (EdgeId relatedEdgeId : pageData.getData()) {
-                    futures.add(saveEdgeEvent(tenantId, relatedEdgeId, type, actionType, entityId, null));
+                    if (!relatedEdgeId.equals(sourceEdgeId)) {
+                        futures.add(saveEdgeEvent(tenantId, relatedEdgeId, type, actionType, entityId, null));
+                    }
                 }
                 if (pageData.hasNext()) {
                     pageLink = pageLink.nextPageLink();
@@ -548,13 +561,13 @@ public abstract class BaseEdgeProcessor {
         return Futures.transform(Futures.allAsList(futures), voids -> null, dbCallbackExecutorService);
     }
 
-    private ListenableFuture<Void> processEntityNotificationForAllEdges(TenantId tenantId, EdgeEventType type, EdgeEventActionType actionType, EntityId entityId) {
+    private ListenableFuture<Void> processEntityNotificationForAllEdges(TenantId tenantId, EdgeEventType type, EdgeEventActionType actionType, EntityId entityId, EdgeId sourceEdgeId) {
         switch (actionType) {
             case ADDED:
             case UPDATED:
             case DELETED:
             case CREDENTIALS_UPDATED: // used by USER entity
-                return processActionForAllEdges(tenantId, type, actionType, entityId);
+                return processActionForAllEdges(tenantId, type, actionType, entityId, sourceEdgeId);
             default:
                 return Futures.immediateFuture(null);
         }

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

@@ -15,7 +15,6 @@
  */
 package org.thingsboard.server.service.edge.rpc.processor.alarm;
 
-import com.fasterxml.jackson.core.JsonProcessingException;
 import com.fasterxml.jackson.databind.JsonNode;
 import com.google.common.util.concurrent.Futures;
 import com.google.common.util.concurrent.ListenableFuture;
@@ -47,6 +46,16 @@ import java.util.UUID;
 @TbCoreComponent
 public class AlarmEdgeProcessor extends BaseAlarmProcessor {
 
+    public ListenableFuture<Void> processAlarmMsgFromEdge(TenantId tenantId, EdgeId edgeId, AlarmUpdateMsg alarmUpdateMsg) {
+        log.trace("[{}] processAlarmMsgFromEdge [{}]", tenantId, alarmUpdateMsg);
+        try {
+            edgeSynchronizationManager.getEdgeId().set(edgeId);
+            return processAlarmMsg(tenantId, alarmUpdateMsg);
+        } finally {
+            edgeSynchronizationManager.getEdgeId().remove();
+        }
+    }
+
     public DownlinkMsg convertAlarmEventToDownlink(EdgeEvent edgeEvent) {
         AlarmUpdateMsg alarmUpdateMsg =
                 convertAlarmEventToAlarmMsg(edgeEvent.getTenantId(), edgeEvent.getEntityId(), edgeEvent.getAction(), edgeEvent.getBody());
@@ -59,13 +68,18 @@ public class AlarmEdgeProcessor extends BaseAlarmProcessor {
         return null;
     }
 
-    public ListenableFuture<Void> processAlarmNotification(TenantId tenantId, TransportProtos.EdgeNotificationMsgProto edgeNotificationMsg) throws JsonProcessingException {
+    public ListenableFuture<Void> processAlarmNotification(TenantId tenantId, TransportProtos.EdgeNotificationMsgProto edgeNotificationMsg) {
         EdgeEventActionType actionType = EdgeEventActionType.valueOf(edgeNotificationMsg.getAction());
         AlarmId alarmId = new AlarmId(new UUID(edgeNotificationMsg.getEntityIdMSB(), edgeNotificationMsg.getEntityIdLSB()));
+        EdgeId sourceEdgeId = safeGetEdgeId(edgeNotificationMsg.getSourceEdgeIdMSB(), edgeNotificationMsg.getSourceEdgeIdLSB());
         switch (actionType) {
             case DELETED:
-                Alarm deletedAlarm = JacksonUtil.OBJECT_MAPPER.readValue(edgeNotificationMsg.getBody(), Alarm.class);
-                List<ListenableFuture<Void>> delFutures = pushEventToAllRelatedEdges(tenantId, deletedAlarm.getOriginator(), alarmId, actionType, JacksonUtil.OBJECT_MAPPER.valueToTree(deletedAlarm));
+                Alarm deletedAlarm = JacksonUtil.fromString(edgeNotificationMsg.getBody(), Alarm.class);
+                if (deletedAlarm == null) {
+                    return Futures.immediateFuture(null);
+                }
+                List<ListenableFuture<Void>> delFutures = pushEventToAllRelatedEdges(tenantId, deletedAlarm.getOriginator(),
+                        alarmId, actionType, JacksonUtil.valueToTree(deletedAlarm), sourceEdgeId);
                 return Futures.transform(Futures.allAsList(delFutures), voids -> null, dbCallbackExecutorService);
             default:
                 ListenableFuture<Alarm> alarmFuture = alarmService.findAlarmByIdAsync(tenantId, alarmId);
@@ -77,13 +91,14 @@ public class AlarmEdgeProcessor extends BaseAlarmProcessor {
                     if (type == null) {
                         return Futures.immediateFuture(null);
                     }
-                    List<ListenableFuture<Void>>  futures = pushEventToAllRelatedEdges(tenantId, alarm.getOriginator(), alarmId, actionType, null);
+                    List<ListenableFuture<Void>> futures = pushEventToAllRelatedEdges(tenantId, alarm.getOriginator(),
+                            alarmId, actionType, null, sourceEdgeId);
                     return Futures.transform(Futures.allAsList(futures), voids -> null, dbCallbackExecutorService);
                 }, dbCallbackExecutorService);
         }
     }
 
-    private List<ListenableFuture<Void>> pushEventToAllRelatedEdges(TenantId tenantId, EntityId originatorId, AlarmId alarmId, EdgeEventActionType actionType, JsonNode body) {
+    private List<ListenableFuture<Void>> pushEventToAllRelatedEdges(TenantId tenantId, EntityId originatorId, AlarmId alarmId, EdgeEventActionType actionType, JsonNode body, EdgeId sourceEdgeId) {
         PageLink pageLink = new PageLink(DEFAULT_PAGE_SIZE);
         PageData<EdgeId> pageData;
         List<ListenableFuture<Void>> futures = new ArrayList<>();
@@ -91,12 +106,14 @@ public class AlarmEdgeProcessor extends BaseAlarmProcessor {
             pageData = edgeService.findRelatedEdgeIdsByEntityId(tenantId, originatorId, pageLink);
             if (pageData != null && pageData.getData() != null && !pageData.getData().isEmpty()) {
                 for (EdgeId relatedEdgeId : pageData.getData()) {
-                    futures.add(saveEdgeEvent(tenantId,
-                            relatedEdgeId,
-                            EdgeEventType.ALARM,
-                            actionType,
-                            alarmId,
-                            body));
+                    if (!relatedEdgeId.equals(sourceEdgeId)) {
+                        futures.add(saveEdgeEvent(tenantId,
+                                relatedEdgeId,
+                                EdgeEventType.ALARM,
+                                actionType,
+                                alarmId,
+                                body));
+                    }
                 }
                 if (pageData.hasNext()) {
                     pageLink = pageLink.nextPageLink();

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

@@ -45,8 +45,7 @@ import java.util.UUID;
 @Slf4j
 public abstract class BaseAlarmProcessor extends BaseEdgeProcessor {
 
-    public ListenableFuture<Void> processAlarmMsg(TenantId tenantId, AlarmUpdateMsg alarmUpdateMsg) {
-        log.trace("[{}] processAlarmMsg [{}]", tenantId, alarmUpdateMsg);
+    protected ListenableFuture<Void> processAlarmMsg(TenantId tenantId, AlarmUpdateMsg alarmUpdateMsg) {
         EntityId originatorId = getAlarmOriginator(tenantId, alarmUpdateMsg.getOriginatorName(),
                 EntityType.valueOf(alarmUpdateMsg.getOriginatorType()));
         AlarmId alarmId = new AlarmId(new UUID(alarmUpdateMsg.getIdMSB(), alarmUpdateMsg.getIdLSB()));
@@ -55,7 +54,7 @@ public abstract class BaseAlarmProcessor extends BaseEdgeProcessor {
             return Futures.immediateFuture(null);
         }
         try {
-            edgeSynchronizationManager.getSync().set(true);
+
             switch (alarmUpdateMsg.getMsgType()) {
                 case ENTITY_CREATED_RPC_MESSAGE:
                 case ENTITY_UPDATED_RPC_MESSAGE:
@@ -73,7 +72,7 @@ public abstract class BaseAlarmProcessor extends BaseEdgeProcessor {
                     alarm.setAcknowledged(alarmStatus.isAck());
                     alarm.setAckTs(alarmUpdateMsg.getAckTs());
                     alarm.setEndTs(alarmUpdateMsg.getEndTs());
-                    alarm.setDetails(JacksonUtil.OBJECT_MAPPER.readTree(alarmUpdateMsg.getDetails()));
+                    alarm.setDetails(JacksonUtil.toJsonNode(alarmUpdateMsg.getDetails()));
                     if (UpdateMsgType.ENTITY_CREATED_RPC_MESSAGE.equals(alarmUpdateMsg.getMsgType())) {
                         alarmService.createAlarm(AlarmCreateOrUpdateActiveRequest.fromAlarm(alarm, null, alarmId));
                     } else {
@@ -90,7 +89,7 @@ public abstract class BaseAlarmProcessor extends BaseEdgeProcessor {
                     Alarm alarmToClear = alarmService.findAlarmById(tenantId, alarmId);
                     if (alarmToClear != null) {
                         alarmService.clearAlarm(tenantId, alarmId, alarmUpdateMsg.getClearTs(),
-                                JacksonUtil.OBJECT_MAPPER.readTree(alarmUpdateMsg.getDetails()));
+                                JacksonUtil.toJsonNode(alarmUpdateMsg.getDetails()));
                     }
                     break;
                 case ENTITY_DELETED_RPC_MESSAGE:
@@ -106,26 +105,11 @@ public abstract class BaseAlarmProcessor extends BaseEdgeProcessor {
         } catch (Exception e) {
             log.error("[{}] Failed to process alarm update msg [{}]", tenantId, alarmUpdateMsg, e);
             return Futures.immediateFailedFuture(e);
-        } finally {
-            edgeSynchronizationManager.getSync().remove();
         }
         return Futures.immediateFuture(null);
     }
 
-    private EntityId getAlarmOriginator(TenantId tenantId, String entityName, EntityType entityType) {
-        switch (entityType) {
-            case DEVICE:
-                return deviceService.findDeviceByTenantIdAndName(tenantId, entityName).getId();
-            case ASSET:
-                return assetService.findAssetByTenantIdAndName(tenantId, entityName).getId();
-            case ENTITY_VIEW:
-                return entityViewService.findEntityViewByTenantIdAndName(tenantId, entityName).getId();
-            default:
-                return null;
-        }
-    }
-
-    public AlarmUpdateMsg convertAlarmEventToAlarmMsg(TenantId tenantId, UUID entityId, EdgeEventActionType actionType, JsonNode body) {
+    protected AlarmUpdateMsg convertAlarmEventToAlarmMsg(TenantId tenantId, UUID entityId, EdgeEventActionType actionType, JsonNode body) {
         AlarmId alarmId = new AlarmId(entityId);
         UpdateMsgType msgType = getUpdateMsgType(actionType);
         switch (actionType) {
@@ -139,12 +123,25 @@ public abstract class BaseAlarmProcessor extends BaseEdgeProcessor {
                 }
                 break;
             case DELETED:
-                Alarm deletedAlarm = JacksonUtil.OBJECT_MAPPER.convertValue(body, Alarm.class);
+                Alarm deletedAlarm = JacksonUtil.convertValue(body, Alarm.class);
                 return alarmMsgConstructor.constructAlarmUpdatedMsg(msgType, deletedAlarm, findOriginatorEntityName(tenantId, deletedAlarm));
         }
         return null;
     }
 
+    private EntityId getAlarmOriginator(TenantId tenantId, String entityName, EntityType entityType) {
+        switch (entityType) {
+            case DEVICE:
+                return deviceService.findDeviceByTenantIdAndName(tenantId, entityName).getId();
+            case ASSET:
+                return assetService.findAssetByTenantIdAndName(tenantId, entityName).getId();
+            case ENTITY_VIEW:
+                return entityViewService.findEntityViewByTenantIdAndName(tenantId, entityName).getId();
+            default:
+                return null;
+        }
+    }
+
     private String findOriginatorEntityName(TenantId tenantId, Alarm alarm) {
         String entityName = null;
         switch (alarm.getOriginator().getEntityType()) {

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

@@ -50,10 +50,10 @@ import java.util.UUID;
 public class AssetEdgeProcessor extends BaseAssetProcessor {
 
     public ListenableFuture<Void> processAssetMsgFromEdge(TenantId tenantId, Edge edge, AssetUpdateMsg assetUpdateMsg) {
-        log.trace("[{}] executing processAssetMsgFromEdge [{}] from edge [{}]", tenantId, assetUpdateMsg, edge.getName());
+        log.trace("[{}] executing processAssetMsgFromEdge [{}] from edge [{}]", tenantId, assetUpdateMsg, edge.getId());
         AssetId assetId = new AssetId(new UUID(assetUpdateMsg.getIdMSB(), assetUpdateMsg.getIdLSB()));
         try {
-            edgeSynchronizationManager.getSync().set(true);
+            edgeSynchronizationManager.getEdgeId().set(edge.getId());
 
             switch (assetUpdateMsg.getMsgType()) {
                 case ENTITY_CREATED_RPC_MESSAGE:
@@ -78,7 +78,7 @@ public class AssetEdgeProcessor extends BaseAssetProcessor {
                 return Futures.immediateFailedFuture(e);
             }
         } finally {
-            edgeSynchronizationManager.getSync().remove();
+            edgeSynchronizationManager.getEdgeId().remove();
         }
     }
 

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

@@ -49,10 +49,10 @@ import java.util.UUID;
 public class AssetProfileEdgeProcessor extends BaseAssetProfileProcessor {
 
     public ListenableFuture<Void> processAssetProfileMsgFromEdge(TenantId tenantId, Edge edge, AssetProfileUpdateMsg assetProfileUpdateMsg) {
-        log.trace("[{}] executing processAssetProfileMsgFromEdge [{}] from edge [{}]", tenantId, assetProfileUpdateMsg, edge.getName());
+        log.trace("[{}] executing processAssetProfileMsgFromEdge [{}] from edge [{}]", tenantId, assetProfileUpdateMsg, edge.getId());
         AssetProfileId assetProfileId = new AssetProfileId(new UUID(assetProfileUpdateMsg.getIdMSB(), assetProfileUpdateMsg.getIdLSB()));
         try {
-            edgeSynchronizationManager.getSync().set(true);
+            edgeSynchronizationManager.getEdgeId().set(edge.getId());
 
             switch (assetProfileUpdateMsg.getMsgType()) {
                 case ENTITY_CREATED_RPC_MESSAGE:
@@ -68,7 +68,7 @@ public class AssetProfileEdgeProcessor extends BaseAssetProfileProcessor {
             log.warn("[{}] Failed to process AssetProfileUpdateMsg from Edge [{}]", tenantId, assetProfileUpdateMsg, e);
             return Futures.immediateFailedFuture(e);
         } finally {
-            edgeSynchronizationManager.getSync().remove();
+            edgeSynchronizationManager.getEdgeId().remove();
         }
     }
 

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

@@ -45,10 +45,10 @@ import java.util.UUID;
 public class DashboardEdgeProcessor extends BaseDashboardProcessor {
 
     public ListenableFuture<Void> processDashboardMsgFromEdge(TenantId tenantId, Edge edge, DashboardUpdateMsg dashboardUpdateMsg) {
-        log.trace("[{}] executing processDashboardMsgFromEdge [{}] from edge [{}]", tenantId, dashboardUpdateMsg, edge.getName());
+        log.trace("[{}] executing processDashboardMsgFromEdge [{}] from edge [{}]", tenantId, dashboardUpdateMsg, edge.getId());
         DashboardId dashboardId = new DashboardId(new UUID(dashboardUpdateMsg.getIdMSB(), dashboardUpdateMsg.getIdLSB()));
         try {
-            edgeSynchronizationManager.getSync().set(true);
+            edgeSynchronizationManager.getEdgeId().set(edge.getId());
 
             switch (dashboardUpdateMsg.getMsgType()) {
                 case ENTITY_CREATED_RPC_MESSAGE:
@@ -73,7 +73,7 @@ public class DashboardEdgeProcessor extends BaseDashboardProcessor {
                 return Futures.immediateFailedFuture(e);
             }
         } finally {
-            edgeSynchronizationManager.getSync().remove();
+            edgeSynchronizationManager.getEdgeId().remove();
         }
     }
 

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

@@ -16,7 +16,6 @@
 package org.thingsboard.server.service.edge.rpc.processor.device;
 
 import com.datastax.oss.driver.api.core.uuid.Uuids;
-import com.google.common.util.concurrent.ListenableFuture;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.data.util.Pair;
@@ -107,34 +106,27 @@ public abstract class BaseDeviceProcessor extends BaseEdgeProcessor {
         return Pair.of(created, deviceNameUpdated);
     }
 
-    public ListenableFuture<Void> processDeviceCredentialsMsg(TenantId tenantId, DeviceCredentialsUpdateMsg deviceCredentialsUpdateMsg) {
-        log.debug("[{}] Executing processDeviceCredentialsMsg, deviceCredentialsUpdateMsg [{}]", tenantId, deviceCredentialsUpdateMsg);
+    protected void updateDeviceCredentials(TenantId tenantId, DeviceCredentialsUpdateMsg deviceCredentialsUpdateMsg) {
         DeviceId deviceId = new DeviceId(new UUID(deviceCredentialsUpdateMsg.getDeviceIdMSB(), deviceCredentialsUpdateMsg.getDeviceIdLSB()));
-        return dbCallbackExecutorService.submit(() -> {
-            Device device = deviceService.findDeviceById(tenantId, deviceId);
-            if (device != null) {
-                log.debug("[{}] Updating device credentials for device [{}]. New device credentials Id [{}], value [{}]",
-                        tenantId, device.getName(), deviceCredentialsUpdateMsg.getCredentialsId(), deviceCredentialsUpdateMsg.getCredentialsValue());
-                try {
-                    edgeSynchronizationManager.getSync().set(true);
+        Device device = deviceService.findDeviceById(tenantId, deviceId);
+        if (device != null) {
+            log.debug("[{}] Updating device credentials for device [{}]. New device credentials Id [{}], value [{}]",
+                    tenantId, device.getName(), deviceCredentialsUpdateMsg.getCredentialsId(), deviceCredentialsUpdateMsg.getCredentialsValue());
+            try {
+                DeviceCredentials deviceCredentials = deviceCredentialsService.findDeviceCredentialsByDeviceId(tenantId, device.getId());
+                deviceCredentials.setCredentialsType(DeviceCredentialsType.valueOf(deviceCredentialsUpdateMsg.getCredentialsType()));
+                deviceCredentials.setCredentialsId(deviceCredentialsUpdateMsg.getCredentialsId());
+                deviceCredentials.setCredentialsValue(deviceCredentialsUpdateMsg.hasCredentialsValue()
+                        ? deviceCredentialsUpdateMsg.getCredentialsValue() : null);
+                deviceCredentialsService.updateDeviceCredentials(tenantId, deviceCredentials);
 
-                    DeviceCredentials deviceCredentials = deviceCredentialsService.findDeviceCredentialsByDeviceId(tenantId, device.getId());
-                    deviceCredentials.setCredentialsType(DeviceCredentialsType.valueOf(deviceCredentialsUpdateMsg.getCredentialsType()));
-                    deviceCredentials.setCredentialsId(deviceCredentialsUpdateMsg.getCredentialsId());
-                    deviceCredentials.setCredentialsValue(deviceCredentialsUpdateMsg.hasCredentialsValue()
-                            ? deviceCredentialsUpdateMsg.getCredentialsValue() : null);
-                    deviceCredentialsService.updateDeviceCredentials(tenantId, deviceCredentials);
-                } catch (Exception e) {
-                    log.error("[{}] Can't update device credentials for device [{}], deviceCredentialsUpdateMsg [{}]",
-                            tenantId, device.getName(), deviceCredentialsUpdateMsg, e);
-                    throw new RuntimeException(e);
-                } finally {
-                    edgeSynchronizationManager.getSync().remove();
-                }
-            } else {
-                log.warn("[{}] Can't find device by id [{}], deviceCredentialsUpdateMsg [{}]", tenantId, deviceId, deviceCredentialsUpdateMsg);
+            } catch (Exception e) {
+                log.error("[{}] Can't update device credentials for device [{}], deviceCredentialsUpdateMsg [{}]",
+                        tenantId, device.getName(), deviceCredentialsUpdateMsg, e);
+                throw new RuntimeException(e);
             }
-            return null;
-        });
+        } else {
+            log.warn("[{}] Can't find device by id [{}], deviceCredentialsUpdateMsg [{}]", tenantId, deviceId, deviceCredentialsUpdateMsg);
+        }
     }
 }

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

@@ -15,7 +15,6 @@
  */
 package org.thingsboard.server.service.edge.rpc.processor.device;
 
-import com.fasterxml.jackson.core.JsonProcessingException;
 import com.fasterxml.jackson.databind.node.ObjectNode;
 import com.google.common.util.concurrent.Futures;
 import com.google.common.util.concurrent.ListenableFuture;
@@ -65,10 +64,10 @@ import java.util.UUID;
 public class DeviceEdgeProcessor extends BaseDeviceProcessor {
 
     public ListenableFuture<Void> processDeviceMsgFromEdge(TenantId tenantId, Edge edge, DeviceUpdateMsg deviceUpdateMsg) {
-        log.trace("[{}] executing processDeviceMsgFromEdge [{}] from edge [{}]", tenantId, deviceUpdateMsg, edge.getName());
+        log.trace("[{}] executing processDeviceMsgFromEdge [{}] from edge [{}]", tenantId, deviceUpdateMsg, edge.getId());
         DeviceId deviceId = new DeviceId(new UUID(deviceUpdateMsg.getIdMSB(), deviceUpdateMsg.getIdLSB()));
         try {
-            edgeSynchronizationManager.getSync().set(true);
+            edgeSynchronizationManager.getEdgeId().set(edge.getId());
 
             switch (deviceUpdateMsg.getMsgType()) {
                 case ENTITY_CREATED_RPC_MESSAGE:
@@ -93,10 +92,22 @@ public class DeviceEdgeProcessor extends BaseDeviceProcessor {
                 return Futures.immediateFailedFuture(e);
             }
         } finally {
-            edgeSynchronizationManager.getSync().remove();
+            edgeSynchronizationManager.getEdgeId().remove();
         }
     }
 
+    public ListenableFuture<Void> processDeviceCredentialsMsgFromEdge(TenantId tenantId, EdgeId edgeId, DeviceCredentialsUpdateMsg deviceCredentialsUpdateMsg) {
+        log.debug("[{}] Executing processDeviceCredentialsMsgFromEdge, deviceCredentialsUpdateMsg [{}]", tenantId, deviceCredentialsUpdateMsg);
+        try {
+            edgeSynchronizationManager.getEdgeId().set(edgeId);
+
+            updateDeviceCredentials(tenantId, deviceCredentialsUpdateMsg);
+        } finally {
+            edgeSynchronizationManager.getEdgeId().remove();
+        }
+        return Futures.immediateFuture(null);
+    }
+
     private void saveOrUpdateDevice(TenantId tenantId, DeviceId deviceId, DeviceUpdateMsg deviceUpdateMsg, Edge edge) {
         CustomerId customerId = safeGetCustomerId(deviceUpdateMsg.getCustomerIdMSB(), deviceUpdateMsg.getCustomerIdLSB());
         Pair<Boolean, Boolean> resultPair = super.saveOrUpdateDevice(tenantId, deviceId, deviceUpdateMsg, customerId);
@@ -183,7 +194,7 @@ public class DeviceEdgeProcessor extends BaseDeviceProcessor {
             data.put("method", deviceRpcCallMsg.getRequestMsg().getMethod());
             data.put("params", deviceRpcCallMsg.getRequestMsg().getParams());
             TbMsg tbMsg = TbMsg.newMsg(TbMsgType.TO_SERVER_RPC_REQUEST, deviceId, null, metaData,
-                    TbMsgDataType.JSON, JacksonUtil.OBJECT_MAPPER.writeValueAsString(data));
+                    TbMsgDataType.JSON, JacksonUtil.toString(data));
             tbClusterService.pushMsgToRuleEngine(tenantId, deviceId, tbMsg, new TbQueueCallback() {
                 @Override
                 public void onSuccess(TbQueueMsgMetadata metadata) {
@@ -197,7 +208,7 @@ public class DeviceEdgeProcessor extends BaseDeviceProcessor {
                             tenantId, device, deviceRpcCallMsg, t);
                 }
             });
-        } catch (JsonProcessingException | IllegalArgumentException e) {
+        } catch (IllegalArgumentException e) {
             log.warn("[{}][{}] Failed to push TO_SERVER_RPC_REQUEST to rule engine. deviceRpcCallMsg {}", tenantId, deviceId, deviceRpcCallMsg, e);
         }
 

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

@@ -49,10 +49,10 @@ import java.util.UUID;
 public class DeviceProfileEdgeProcessor extends BaseDeviceProfileProcessor {
 
     public ListenableFuture<Void> processDeviceProfileMsgFromEdge(TenantId tenantId, Edge edge, DeviceProfileUpdateMsg deviceProfileUpdateMsg) {
-        log.trace("[{}] executing processDeviceProfileMsgFromEdge [{}] from edge [{}]", tenantId, deviceProfileUpdateMsg, edge.getName());
+        log.trace("[{}] executing processDeviceProfileMsgFromEdge [{}] from edge [{}]", tenantId, deviceProfileUpdateMsg, edge.getId());
         DeviceProfileId deviceProfileId = new DeviceProfileId(new UUID(deviceProfileUpdateMsg.getIdMSB(), deviceProfileUpdateMsg.getIdLSB()));
         try {
-            edgeSynchronizationManager.getSync().set(true);
+            edgeSynchronizationManager.getEdgeId().set(edge.getId());
 
             switch (deviceProfileUpdateMsg.getMsgType()) {
                 case ENTITY_CREATED_RPC_MESSAGE:
@@ -68,7 +68,7 @@ public class DeviceProfileEdgeProcessor extends BaseDeviceProfileProcessor {
             log.warn("[{}] Failed to process DeviceProfileUpdateMsg from Edge [{}]", tenantId, deviceProfileUpdateMsg, e);
             return Futures.immediateFailedFuture(e);
         } finally {
-            edgeSynchronizationManager.getSync().remove();
+            edgeSynchronizationManager.getEdgeId().remove();
         }
     }
 

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

@@ -72,9 +72,9 @@ public class EdgeProcessor extends BaseEdgeProcessor {
             EdgeId edgeId = new EdgeId(new UUID(edgeNotificationMsg.getEntityIdMSB(), edgeNotificationMsg.getEntityIdLSB()));
             switch (actionType) {
                 case ASSIGNED_TO_CUSTOMER:
-                    CustomerId customerId = JacksonUtil.OBJECT_MAPPER.readValue(edgeNotificationMsg.getBody(), CustomerId.class);
+                    CustomerId customerId = JacksonUtil.fromString(edgeNotificationMsg.getBody(), CustomerId.class);
                     Edge edge = edgeService.findEdgeById(tenantId, edgeId);
-                    if (edge == null || customerId.isNullUid()) {
+                    if (customerId != null && (edge == null || customerId.isNullUid())) {
                         return Futures.immediateFuture(null);
                     }
                     List<ListenableFuture<Void>> futures = new ArrayList<>();
@@ -96,9 +96,9 @@ public class EdgeProcessor extends BaseEdgeProcessor {
                     } while (pageData != null && pageData.hasNext());
                     return Futures.transform(Futures.allAsList(futures), voids -> null, dbCallbackExecutorService);
                 case UNASSIGNED_FROM_CUSTOMER:
-                    CustomerId customerIdToDelete = JacksonUtil.OBJECT_MAPPER.readValue(edgeNotificationMsg.getBody(), CustomerId.class);
+                    CustomerId customerIdToDelete = JacksonUtil.fromString(edgeNotificationMsg.getBody(), CustomerId.class);
                     edge = edgeService.findEdgeById(tenantId, edgeId);
-                    if (edge == null || customerIdToDelete.isNullUid()) {
+                    if (customerIdToDelete != null && (edge == null || customerIdToDelete.isNullUid())) {
                         return Futures.immediateFuture(null);
                     }
                     return Futures.transformAsync(saveEdgeEvent(edge.getTenantId(), edge.getId(), EdgeEventType.EDGE, EdgeEventActionType.UNASSIGNED_FROM_CUSTOMER, edgeId, null),

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

@@ -46,10 +46,10 @@ import java.util.UUID;
 public class EntityViewEdgeProcessor extends BaseEntityViewProcessor {
 
     public ListenableFuture<Void> processEntityViewMsgFromEdge(TenantId tenantId, Edge edge, EntityViewUpdateMsg entityViewUpdateMsg) {
-        log.trace("[{}] executing processEntityViewMsgFromEdge [{}] from edge [{}]", tenantId, entityViewUpdateMsg, edge.getName());
+        log.trace("[{}] executing processEntityViewMsgFromEdge [{}] from edge [{}]", tenantId, entityViewUpdateMsg, edge.getId());
         EntityViewId entityViewId = new EntityViewId(new UUID(entityViewUpdateMsg.getIdMSB(), entityViewUpdateMsg.getIdLSB()));
         try {
-            edgeSynchronizationManager.getSync().set(true);
+            edgeSynchronizationManager.getEdgeId().set(edge.getId());
 
             switch (entityViewUpdateMsg.getMsgType()) {
                 case ENTITY_CREATED_RPC_MESSAGE:
@@ -74,7 +74,7 @@ public class EntityViewEdgeProcessor extends BaseEntityViewProcessor {
                 return Futures.immediateFailedFuture(e);
             }
         } finally {
-            edgeSynchronizationManager.getSync().remove();
+            edgeSynchronizationManager.getEdgeId().remove();
         }
     }
 

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

@@ -33,10 +33,8 @@ import java.util.UUID;
 @Slf4j
 public abstract class BaseRelationProcessor extends BaseEdgeProcessor {
 
-    public ListenableFuture<Void> processRelationMsg(TenantId tenantId, RelationUpdateMsg relationUpdateMsg) {
-        log.trace("[{}] processRelationMsg [{}]", tenantId, relationUpdateMsg);
+    protected ListenableFuture<Void> processRelationMsg(TenantId tenantId, RelationUpdateMsg relationUpdateMsg) {
         try {
-            edgeSynchronizationManager.getSync().set(true);
             EntityRelation entityRelation = new EntityRelation();
 
             UUID fromUUID = new UUID(relationUpdateMsg.getFromIdMSB(), relationUpdateMsg.getFromIdLSB());
@@ -72,8 +70,6 @@ public abstract class BaseRelationProcessor extends BaseEdgeProcessor {
         } catch (Exception e) {
             log.error("[{}] Failed to process relation update msg [{}]", tenantId, relationUpdateMsg, e);
             return Futures.immediateFailedFuture(e);
-        } finally {
-            edgeSynchronizationManager.getSync().remove();
         }
         return Futures.immediateFuture(null);
     }

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

@@ -15,7 +15,6 @@
  */
 package org.thingsboard.server.service.edge.rpc.processor.relation;
 
-import com.fasterxml.jackson.core.JsonProcessingException;
 import com.google.common.util.concurrent.Futures;
 import com.google.common.util.concurrent.ListenableFuture;
 import lombok.extern.slf4j.Slf4j;
@@ -23,6 +22,7 @@ import org.springframework.stereotype.Component;
 import org.thingsboard.common.util.JacksonUtil;
 import org.thingsboard.server.common.data.EdgeUtils;
 import org.thingsboard.server.common.data.EntityType;
+import org.thingsboard.server.common.data.edge.Edge;
 import org.thingsboard.server.common.data.edge.EdgeEvent;
 import org.thingsboard.server.common.data.edge.EdgeEventActionType;
 import org.thingsboard.server.common.data.edge.EdgeEventType;
@@ -45,8 +45,19 @@ import java.util.Set;
 @TbCoreComponent
 public class RelationEdgeProcessor extends BaseRelationProcessor {
 
+    public ListenableFuture<Void> processRelationMsgFromEdge(TenantId tenantId, Edge edge, RelationUpdateMsg relationUpdateMsg) {
+        log.trace("[{}] executing processRelationMsgFromEdge [{}] from edge [{}]", tenantId, relationUpdateMsg, edge.getId());
+        try {
+            edgeSynchronizationManager.getEdgeId().set(edge.getId());
+
+            return processRelationMsg(tenantId, relationUpdateMsg);
+        } finally {
+            edgeSynchronizationManager.getEdgeId().remove();
+        }
+    }
+
     public DownlinkMsg convertRelationEventToDownlink(EdgeEvent edgeEvent) {
-        EntityRelation entityRelation = JacksonUtil.OBJECT_MAPPER.convertValue(edgeEvent.getBody(), EntityRelation.class);
+        EntityRelation entityRelation = JacksonUtil.convertValue(edgeEvent.getBody(), EntityRelation.class);
         UpdateMsgType msgType = getUpdateMsgType(edgeEvent.getAction());
         RelationUpdateMsg relationUpdateMsg = relationMsgConstructor.constructRelationUpdatedMsg(msgType, entityRelation);
         return DownlinkMsg.newBuilder()
@@ -55,10 +66,9 @@ public class RelationEdgeProcessor extends BaseRelationProcessor {
                 .build();
     }
 
-    public ListenableFuture<Void> processRelationNotification(TenantId tenantId, TransportProtos.EdgeNotificationMsgProto edgeNotificationMsg) throws JsonProcessingException {
-        EntityRelation relation = JacksonUtil.OBJECT_MAPPER.readValue(edgeNotificationMsg.getBody(), EntityRelation.class);
-        if (relation.getFrom().getEntityType().equals(EntityType.EDGE) ||
-                relation.getTo().getEntityType().equals(EntityType.EDGE)) {
+    public ListenableFuture<Void> processRelationNotification(TenantId tenantId, TransportProtos.EdgeNotificationMsgProto edgeNotificationMsg) {
+        EntityRelation relation = JacksonUtil.fromString(edgeNotificationMsg.getBody(), EntityRelation.class);
+        if (relation == null || (relation.getFrom().getEntityType().equals(EntityType.EDGE) || relation.getTo().getEntityType().equals(EntityType.EDGE))) {
             return Futures.immediateFuture(null);
         }
 
@@ -75,7 +85,7 @@ public class RelationEdgeProcessor extends BaseRelationProcessor {
                     EdgeEventType.RELATION,
                     EdgeEventActionType.valueOf(edgeNotificationMsg.getAction()),
                     null,
-                    JacksonUtil.OBJECT_MAPPER.valueToTree(relation)));
+                    JacksonUtil.valueToTree(relation)));
         }
         return Futures.transform(Futures.allAsList(futures), voids -> null, dbCallbackExecutorService);
     }

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

@@ -32,7 +32,10 @@ import org.thingsboard.server.service.edge.rpc.processor.BaseEdgeProcessor;
 public class AdminSettingsEdgeProcessor extends BaseEdgeProcessor {
 
     public DownlinkMsg convertAdminSettingsEventToDownlink(EdgeEvent edgeEvent) {
-        AdminSettings adminSettings = JacksonUtil.OBJECT_MAPPER.convertValue(edgeEvent.getBody(), AdminSettings.class);
+        AdminSettings adminSettings = JacksonUtil.convertValue(edgeEvent.getBody(), AdminSettings.class);
+        if (adminSettings == null) {
+            return null;
+        }
         AdminSettingsUpdateMsg adminSettingsUpdateMsg = adminSettingsMsgConstructor.constructAdminSettingsUpdateMsg(adminSettings);
         return DownlinkMsg.newBuilder()
                 .setDownlinkMsgId(EdgeUtils.nextPositiveInt())

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

@@ -22,7 +22,6 @@ import com.google.common.util.concurrent.Futures;
 import com.google.common.util.concurrent.ListenableFuture;
 import com.google.common.util.concurrent.SettableFuture;
 import com.google.gson.Gson;
-import com.google.gson.JsonElement;
 import com.google.gson.JsonObject;
 import com.google.gson.JsonParser;
 import lombok.extern.slf4j.Slf4j;
@@ -346,8 +345,9 @@ public abstract class BaseTelemetryProcessor extends BaseEdgeProcessor {
                 log.warn("[{}] Unsupported edge event type [{}]", tenantId, entityType);
                 return null;
         }
-        JsonElement entityData = JsonParser.parseString(JacksonUtil.OBJECT_MAPPER.writeValueAsString(body));
-        return entityDataMsgConstructor.constructEntityDataMsg(tenantId, entityId, actionType, entityData);
+        String bodyJackson = JacksonUtil.toString(body);
+        return bodyJackson == null ? null :
+                entityDataMsgConstructor.constructEntityDataMsg(tenantId, entityId, actionType, JsonParser.parseString(bodyJackson));
     }
 
 }

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

@@ -176,7 +176,7 @@ public class DefaultEdgeRequestsService implements EdgeRequestsService {
                 if (attributes.size() > 0) {
                     entityData.put("kv", attributes);
                     entityData.put("scope", scope);
-                    JsonNode body = JacksonUtil.OBJECT_MAPPER.valueToTree(entityData);
+                    JsonNode body = JacksonUtil.valueToTree(entityData);
                     log.debug("[{}] Sending attributes data msg, entityId [{}], attributes [{}]", tenantId, entityId, body);
                     future = saveEdgeEvent(tenantId, edge.getId(), entityType, EdgeEventActionType.ATTRIBUTES_UPDATED, entityId, body);
                 } else {
@@ -249,7 +249,7 @@ public class DefaultEdgeRequestsService implements EdgeRequestsService {
                                                 EdgeEventType.RELATION,
                                                 EdgeEventActionType.ADDED,
                                                 null,
-                                                JacksonUtil.OBJECT_MAPPER.valueToTree(relation)));
+                                                JacksonUtil.valueToTree(relation)));
                                     }
                                 } catch (Exception e) {
                                     String errMsg = String.format("[%s][%s] Exception during loading relation [%s] to edge on sync!", tenantId, edge.getId(), relation);

application/src/main/java/org/thingsboard/server/service/queue/DefaultTbClusterService.java

@@ -488,7 +488,7 @@ public class DefaultTbClusterService implements TbClusterService {
     }
 
     @Override
-    public void sendNotificationMsgToEdge(TenantId tenantId, EdgeId edgeId, EntityId entityId, String body, EdgeEventType type, EdgeEventActionType action) {
+    public void sendNotificationMsgToEdge(TenantId tenantId, EdgeId edgeId, EntityId entityId, String body, EdgeEventType type, EdgeEventActionType action, EdgeId sourceEdgeId) {
         if (!edgesEnabled) {
             return;
         }
@@ -521,6 +521,10 @@ public class DefaultTbClusterService implements TbClusterService {
         if (body != null) {
             builder.setBody(body);
         }
+        if (sourceEdgeId != null) {
+            builder.setSourceEdgeIdMSB(sourceEdgeId.getId().getMostSignificantBits());
+            builder.setSourceEdgeIdLSB(sourceEdgeId.getId().getLeastSignificantBits());
+        }
         TransportProtos.EdgeNotificationMsgProto msg = builder.build();
         log.trace("[{}] sending notification to edge service {}", tenantId.getId(), msg);
         pushMsgToCore(tenantId, entityId != null ? entityId : tenantId, TransportProtos.ToCoreMsg.newBuilder().setEdgeNotificationMsg(msg).build(), null);

application/src/test/java/org/thingsboard/server/controller/AbstractNotifyEntityTest.java

@@ -91,7 +91,7 @@ public abstract class AbstractNotifyEntityTest extends AbstractWebTest {
         int cntTime = 1;
         Mockito.verify(tbClusterService, times(cntTime)).sendNotificationMsgToEdge(Mockito.eq(tenantId),
                 Mockito.isNull(), Mockito.isNull(), Mockito.any(), Mockito.eq(EdgeEventType.RELATION),
-                Mockito.eq(edgeTypeByActionType(actionType)));
+                Mockito.eq(edgeTypeByActionType(actionType)), Mockito.any());
         ArgumentMatcher<EntityId> matcherOriginatorId = argument -> argument.equals(relation.getTo());
         ArgumentMatcher<HasName> matcherEntityClassEquals = Objects::isNull;
         ArgumentMatcher<CustomerId> matcherCustomerId = customerId == null ?
@@ -111,7 +111,7 @@ public abstract class AbstractNotifyEntityTest extends AbstractWebTest {
                                                    ActionType actionType, int cntTime) {
         Mockito.verify(tbClusterService, times(cntTime)).sendNotificationMsgToEdge(Mockito.eq(tenantId),
                 Mockito.isNull(), Mockito.isNull(), Mockito.any(), Mockito.eq(EdgeEventType.RELATION),
-                Mockito.eq(edgeTypeByActionType(actionType)));
+                Mockito.eq(edgeTypeByActionType(actionType)), Mockito.any());
         ArgumentMatcher<EntityId> matcherOriginatorId = argument -> argument.getClass().equals(relation.getFrom().getClass());
         ArgumentMatcher<HasName> matcherEntityClassEquals = Objects::isNull;
         ArgumentMatcher<CustomerId> matcherCustomerId = customerId == null ?
@@ -318,13 +318,13 @@ public abstract class AbstractNotifyEntityTest extends AbstractWebTest {
     private void testNotificationMsgToEdgeServiceNeverWithActionType(EntityId entityId, ActionType actionType) {
         EdgeEventActionType edgeEventActionType = ActionType.CREDENTIALS_UPDATED.equals(actionType) ?
                 EdgeEventActionType.CREDENTIALS_UPDATED : edgeTypeByActionType(actionType);
-        Mockito.verify(tbClusterService, never()).sendNotificationMsgToEdge(Mockito.any(),
-                Mockito.any(), Mockito.any(entityId.getClass()), Mockito.any(), Mockito.any(), Mockito.eq(edgeEventActionType));
+        Mockito.verify(tbClusterService, never()).sendNotificationMsgToEdge(Mockito.any(), Mockito.any(),
+                Mockito.any(entityId.getClass()), Mockito.any(), Mockito.any(), Mockito.eq(edgeEventActionType), Mockito.any());
     }
 
     private void testNotificationMsgToEdgeServiceNever(EntityId entityId) {
-        Mockito.verify(tbClusterService, never()).sendNotificationMsgToEdge(Mockito.any(),
-                Mockito.any(), Mockito.any(entityId.getClass()), Mockito.any(), Mockito.any(), Mockito.any());
+        Mockito.verify(tbClusterService, never()).sendNotificationMsgToEdge(Mockito.any(), Mockito.any(),
+                Mockito.any(entityId.getClass()), Mockito.any(), Mockito.any(), Mockito.any(), Mockito.any());
     }
 
     private void testLogEntityActionNever(EntityId entityId, HasName entity) {
@@ -358,13 +358,13 @@ public abstract class AbstractNotifyEntityTest extends AbstractWebTest {
                 argument -> argument.getClass().equals(entityId.getClass());
         Mockito.verify(tbClusterService, times(cntTime)).sendNotificationMsgToEdge(Mockito.eq(tenantId),
                 Mockito.any(), Mockito.argThat(matcherEntityId), Mockito.any(), Mockito.isNull(),
-                Mockito.eq(edgeEventActionType));
+                Mockito.eq(edgeEventActionType), Mockito.any());
     }
 
     private void testSendNotificationMsgToEdgeServiceTimeEntityEqAny(TenantId tenantId, ActionType actionType, int cntTime) {
         Mockito.verify(tbClusterService, times(cntTime)).sendNotificationMsgToEdge(Mockito.eq(tenantId),
                 Mockito.any(), Mockito.any(EntityId.class), Mockito.any(), Mockito.isNull(),
-                Mockito.eq(edgeTypeByActionType(actionType)));
+                Mockito.eq(edgeTypeByActionType(actionType)), Mockito.any());
     }
 
     protected void testBroadcastEntityStateChangeEventTime(EntityId entityId, TenantId tenantId, int cntTime) {

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

@@ -553,9 +553,9 @@ public class DeviceEdgeTest extends AbstractEdgeTest {
         Assert.assertTrue(edgeImitator.waitForResponses());
         Assert.assertTrue(edgeImitator.waitForMessages());
 
-        AbstractMessage latestMessage = edgeImitator.getLatestMessage();
-        Assert.assertTrue(latestMessage instanceof DeviceCredentialsRequestMsg);
-        DeviceCredentialsRequestMsg latestDeviceCredentialsRequestMsg = (DeviceCredentialsRequestMsg) latestMessage;
+        Optional<DeviceCredentialsRequestMsg> deviceCredentialsRequestMsgOpt = edgeImitator.findMessageByType(DeviceCredentialsRequestMsg.class);
+        Assert.assertTrue(deviceCredentialsRequestMsgOpt.isPresent());
+        DeviceCredentialsRequestMsg latestDeviceCredentialsRequestMsg = deviceCredentialsRequestMsgOpt.get();
         Assert.assertEquals(uuid.getMostSignificantBits(), latestDeviceCredentialsRequestMsg.getDeviceIdMSB());
         Assert.assertEquals(uuid.getLeastSignificantBits(), latestDeviceCredentialsRequestMsg.getDeviceIdLSB());
 

application/src/test/java/org/thingsboard/server/service/sync/ie/ExportImportServiceSqlTest.java

@@ -592,7 +592,7 @@ public class ExportImportServiceSqlTest extends BaseExportImportServiceTest {
         Customer updatedCustomer = importEntity(tenantAdmin2, updatedCustomerEntity).getSavedEntity();
         verify(entityActionService).logEntityAction(any(), eq(importedCustomer.getId()), eq(updatedCustomer),
                 any(), eq(ActionType.UPDATED), isNull());
-        verify(tbClusterService).sendNotificationMsgToEdge(any(), any(), eq(importedCustomer.getId()), any(), any(), eq(EdgeEventActionType.UPDATED));
+        verify(tbClusterService).sendNotificationMsgToEdge(any(), any(), eq(importedCustomer.getId()), any(), any(), eq(EdgeEventActionType.UPDATED), any());
 
         Mockito.reset(entityActionService);
 
@@ -609,7 +609,7 @@ public class ExportImportServiceSqlTest extends BaseExportImportServiceTest {
         verify(entityActionService).logEntityAction(any(), eq(importedAssetProfile.getId()), eq(importedAssetProfile),
                 any(), eq(ActionType.ADDED), isNull());
         verify(tbClusterService).broadcastEntityStateChangeEvent(any(), eq(importedAssetProfile.getId()), eq(ComponentLifecycleEvent.CREATED));
-        verify(tbClusterService).sendNotificationMsgToEdge(any(), any(), eq(importedAssetProfile.getId()), any(), any(), eq(EdgeEventActionType.ADDED));
+        verify(tbClusterService).sendNotificationMsgToEdge(any(), any(), eq(importedAssetProfile.getId()), any(), any(), eq(EdgeEventActionType.ADDED), any());
 
         Asset importedAsset = (Asset) importEntity(tenantAdmin2, getAndClone(entitiesExportData, EntityType.ASSET)).getSavedEntity();
         verify(entityActionService).logEntityAction(any(), eq(importedAsset.getId()), eq(importedAsset),
@@ -625,14 +625,14 @@ public class ExportImportServiceSqlTest extends BaseExportImportServiceTest {
 
         verify(entityActionService).logEntityAction(any(), eq(importedAsset.getId()), eq(updatedAsset),
                 any(), eq(ActionType.UPDATED), isNull());
-        verify(tbClusterService).sendNotificationMsgToEdge(any(), any(), eq(importedAsset.getId()), any(), any(), eq(EdgeEventActionType.UPDATED));
+        verify(tbClusterService).sendNotificationMsgToEdge(any(), any(), eq(importedAsset.getId()), any(), any(), eq(EdgeEventActionType.UPDATED), any());
 
         DeviceProfile importedDeviceProfile = (DeviceProfile) importEntity(tenantAdmin2, getAndClone(entitiesExportData, EntityType.DEVICE_PROFILE)).getSavedEntity();
         verify(entityActionService).logEntityAction(any(), eq(importedDeviceProfile.getId()), eq(importedDeviceProfile),
                 any(), eq(ActionType.ADDED), isNull());
         verify(tbClusterService).onDeviceProfileChange(eq(importedDeviceProfile), any());
         verify(tbClusterService).broadcastEntityStateChangeEvent(any(), eq(importedDeviceProfile.getId()), eq(ComponentLifecycleEvent.CREATED));
-        verify(tbClusterService).sendNotificationMsgToEdge(any(), any(), eq(importedDeviceProfile.getId()), any(), any(), eq(EdgeEventActionType.ADDED));
+        verify(tbClusterService).sendNotificationMsgToEdge(any(), any(), eq(importedDeviceProfile.getId()), any(), any(), eq(EdgeEventActionType.ADDED), any());
         verify(otaPackageStateService).update(eq(importedDeviceProfile), eq(false), eq(false));
 
         Device importedDevice = (Device) importEntity(tenantAdmin2, getAndClone(entitiesExportData, EntityType.DEVICE)).getSavedEntity();

common/cluster-api/src/main/java/org/thingsboard/server/cluster/TbClusterService.java

@@ -95,6 +95,6 @@ public interface TbClusterService extends TbQueueClusterService {
 
     void pushEdgeSyncResponseToCore(FromEdgeSyncResponse fromEdgeSyncResponse);
 
-    void sendNotificationMsgToEdge(TenantId tenantId, EdgeId edgeId, EntityId entityId, String body, EdgeEventType type, EdgeEventActionType action);
+    void sendNotificationMsgToEdge(TenantId tenantId, EdgeId edgeId, EntityId entityId, String body, EdgeEventType type, EdgeEventActionType action, EdgeId sourceEdgeId);
 
 }

common/cluster-api/src/main/proto/queue.proto

@@ -896,6 +896,8 @@ message EdgeNotificationMsgProto {
   string body = 10;
   PostTelemetryMsg postTelemetryMsg = 11;
   PostAttributeMsg postAttributesMsg = 12;
+  int64 sourceEdgeIdMSB = 13;
+  int64 sourceEdgeIdLSB = 14;
 }
 
 /**

common/dao-api/src/main/java/org/thingsboard/server/dao/edge/EdgeSynchronizationManager.java

@@ -15,9 +15,9 @@
  */
 package org.thingsboard.server.dao.edge;
 
-public interface EdgeSynchronizationManager {
+import org.thingsboard.server.common.data.id.EdgeId;
 
-    ThreadLocal<Boolean> getSync();
+public interface EdgeSynchronizationManager {
 
-    boolean isSync();
+    ThreadLocal<EdgeId> getEdgeId();
 }

dao/src/main/java/org/thingsboard/server/dao/edge/DefaultEdgeSynchronizationManager.java

@@ -18,17 +18,12 @@ package org.thingsboard.server.dao.edge;
 import lombok.Getter;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.stereotype.Component;
+import org.thingsboard.server.common.data.id.EdgeId;
 
 @Component
 @Slf4j
 public class DefaultEdgeSynchronizationManager implements EdgeSynchronizationManager {
 
     @Getter
-    private final ThreadLocal<Boolean> sync = new ThreadLocal<>();
-
-    @Override
-    public boolean isSync() {
-        Boolean sync = this.sync.get();
-        return sync != null && sync;
-    }
+    private final ThreadLocal<EdgeId> edgeId = new ThreadLocal<>();
 }

dao/src/main/java/org/thingsboard/server/dao/eventsourcing/DeleteEntityEvent.java

@@ -26,7 +26,6 @@ import org.thingsboard.server.common.data.id.TenantId;
 public class DeleteEntityEvent<T> {
     private final TenantId tenantId;
     private final EntityId entityId;
-    private final EdgeId edgeId;
     private final T entity;
 
     @Builder.Default

rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/rpc/TbSendRPCReplyNode.java

@@ -104,7 +104,7 @@ public class TbSendRPCReplyNode implements TbNode {
         body.put("requestId", requestIdStr);
         body.put("response", msg.getData());
         EdgeEvent edgeEvent = EdgeUtils.constructEdgeEvent(ctx.getTenantId(), edgeId, EdgeEventType.DEVICE,
-                        EdgeEventActionType.RPC_CALL, deviceId, JacksonUtil.OBJECT_MAPPER.valueToTree(body));
+                        EdgeEventActionType.RPC_CALL, deviceId, JacksonUtil.valueToTree(body));
         ListenableFuture<Void> future = ctx.getEdgeEventService().saveAsync(edgeEvent);
         Futures.addCallback(future, new FutureCallback<>() {
             @Override

ui-ngx/src/app/modules/home/components/relation/relation-filters.component.html

@@ -18,7 +18,7 @@
 <div class="tb-form-panel no-border no-padding">
   <div class="tb-form-table" [formGroup]="relationFiltersFormGroup">
     <div class="tb-form-table-header">
-      <div class="tb-form-table-header-cell flex-50">{{ 'relation.type' | translate }}</div>
+      <div class="tb-form-table-header-cell flex-50">{{ 'relation.relation-type' | translate }}</div>
       <div class="tb-form-table-header-cell flex-50">{{ 'entity.entity-types' | translate }}</div>
       <div class="tb-form-table-header-cell actions-header"></div>
     </div>

ui-ngx/src/app/shared/components/entity/entity-subtype-list.component.ts

@@ -166,7 +166,7 @@ export class EntitySubTypeListComponent implements ControlValueAccessor, OnInit,
           : this.translate.instant('device.any-device');
         this.secondaryPlaceholder = '+' + this.translate.instant('device.device-type');
         this.noSubtypesMathingText = 'device.no-device-types-matching';
-        this.subtypeListEmptyText = 'device.device-type-list-empty';
+        this.subtypeListEmptyText = 'device.device-profile-type-list-empty';
         this.broadcastSubscription = this.broadcast.on('deviceSaved', () => {
           this.entitySubtypes = null;
         });

ui-ngx/src/assets/locale/locale.constant-de_DE.json

@@ -212,7 +212,7 @@
     "filter-type-single-entity": "Einzelne Entität",
     "filter-type-entity-list": "Entitätsliste",
     "filter-type-entity-name": "Entitätsname",
-	"filter-type-entity-type": "Entitätstyp",
+    "filter-type-entity-type": "Entitätstyp",
     "filter-type-state-entity": "Entität aus dem Dashboard Status",
     "filter-type-state-entity-description": "Entität aus den Dashboard Status Parametern",
     "filter-type-asset-type": "Objekttyp",
@@ -768,17 +768,17 @@
     "edge": "Edge",
     "edge-instances": "Kanteninstanzen",
     "edge-file": "Edge-Datei",
-    "management": "Rand verwalten",
-    "no-edges-matching": "Keine passenden Rand '{{entity}}' gefunden.",
-    "add": "Rand hinzufügen",
-    "no-edges-text": "Kein Rand gefunden.",
-    "edge-details": "Details der Rand",
-    "add-edge-text": "Neue Rand hinzufügen",
-    "delete": "Rand löschen",
-    "delete-edge-title": "Möchten Sie des Rands wirklich löschen '{{edgeName}}'?",
-    "delete-edge-text": "Seien Sie vorsichtig, nach der Bestätigung werden der Rand und alle zugehörigen Daten nicht wiederhergestellt.",
-    "delete-edges-title": "Sind Sie sicher, dass Sie die Rand löschen möchten { count, plural, =1 {1 Rand} other {# Rand} }?",
-    "delete-edges-text": "Vorsicht, nach Bestätigung werden alle ausgewählten Rand entfernt und alle zugehörigen Daten werden nicht wiederhergestellt.",
+    "management": "Edge verwalten",
+    "no-edges-matching": "Keine passenden Edge '{{entity}}' gefunden.",
+    "add": "Edge hinzufügen",
+    "no-edges-text": "Keine Edge gefunden.",
+    "edge-details": "Details der Edge",
+    "add-edge-text": "Neue Edge hinzufügen",
+    "delete": "Edge löschen",
+    "delete-edge-title": "Möchten Sie des Edges wirklich löschen '{{edgeName}}'?",
+    "delete-edge-text": "Seien Sie vorsichtig, nach der Bestätigung werden der Edge und alle zugehörigen Daten nicht wiederhergestellt.",
+    "delete-edges-title": "Sind Sie sicher, dass Sie die Edge löschen möchten { count, plural, =1 {1 Edge} other {# Edges} }?",
+    "delete-edges-text": "Vorsicht, nach Bestätigung werden alle ausgewählten Edge entfernt und alle zugehörigen Daten werden nicht wiederhergestellt.",
     "name": "Name",
     "name-starts-with": "Der Kantenname beginnt mit",
     "name-required": "Name ist erforderlich.",
@@ -787,62 +787,62 @@
     "events": "Ereignisse",
     "copy-id": "Regelketten-ID kopieren",
     "id-copied-message": "Regelketten-ID wurde in die Zwischenablage kopiert",
-    "sync": "Sync Edge",
-    "edge-required": "Rand ist erforderlich.",
-    "edge-type": "Randtyp",
-    "edge-type-required": "Randtyp ist erforderlich.",
+    "sync": "Edge synchonisieren",
+    "edge-required": "Edge ist erforderlich.",
+    "edge-type": "Edgetyp",
+    "edge-type-required": "Edgetyp ist erforderlich.",
     "event-action": "Ereignisaktion",
     "entity-id": "Entität ID",
-    "select-edge-type": "Randtyp auswählen",
+    "select-edge-type": "Edgetyp auswählen",
     "assign-to-customer": "Einem Kunden zuordnen",
-    "assign-to-customer-text": "Bitte wählen Sie den Kunden aus, dem die Rand zugeordnet werden sollen",
-    "assign-edge-to-customer": "Rand dem Kunden zuordnen",
-    "assign-edge-to-customer-text": "Bitte wählen Sie die Rand aus, die dem Kunden zugeordnet werden sollen",
+    "assign-to-customer-text": "Bitte wählen Sie den Kunden aus, dem die Edge zugeordnet werden sollen",
+    "assign-edge-to-customer": "Edge dem Kunden zuordnen",
+    "assign-edge-to-customer-text": "Bitte wählen Sie die Edge aus, die dem Kunden zugeordnet werden sollen",
     "assignedToCustomer": "Dem Kunden zugewiesen",
     "edge-public": "Edge ist öffentlich",
     "assigned-to-customer": "Kunden Zuordnung",
     "unassign-from-customer": "Kunden Zuordnung aufgehoben",
-    "unassign-edge-title": "Sind Sie sicher, dass Sie die Zuordnung zum Rand '{{edgeName}}' wirklich aufheben möchten?",
-    "unassign-edge-text": "Nach der Bestätigung ist der Rand nicht zugeordnet und für den Kunden nicht zugänglich.",
-    "unassign-edges-title": "Sind Sie sicher, dass Sie die Zuordnung aufheben möchten { count, plural, =1 {1 Rand} other {# Rand} }?",
+    "unassign-edge-title": "Sind Sie sicher, dass Sie die Zuordnung zum Edge '{{edgeName}}' wirklich aufheben möchten?",
+    "unassign-edge-text": "Nach der Bestätigung ist der Edge nicht zugeordnet und für den Kunden nicht zugänglich.",
+    "unassign-edges-title": "Sind Sie sicher, dass Sie die Zuordnung aufheben möchten { count, plural, =1 {1 Edge} other {# Edges} }?",
     "unassign-edges-text": "Nach der Bestätigung werden alle ausgewählten Kanten nicht zugewiesen und sind für den Kunden nicht zugänglich.",
-    "make-public": "Rand öffentlich machen",
-    "make-public-edge-title": "Sind Sie sicher, dass Sie der Rand '{{edgeName}}' öffentlich machen möchten?",
+    "make-public": "Edge öffentlich machen",
+    "make-public-edge-title": "Sind Sie sicher, dass Sie der Edge '{{edgeName}}' öffentlich machen möchten?",
     "make-public-edge-text": "Nach Bestätigung wird der Rabd und alle zugehörigen Daten anderen zugänglich gemacht.",
-    "make-private": "Rand privat machen",
+    "make-private": "Edge privat machen",
     "public": "Öffentlich",
-    "make-private-edge-title": "Sind Sie sicher, dass Sie der Rand '{{edgeName}}' privat machen möchten?",
-    "make-private-edge-text": "Nach der Bestätigung werden der Rand und dessen Daten privat und sind für andere nicht mehr zugänglich.",
-    "import": "Rand importieren",
+    "make-private-edge-title": "Sind Sie sicher, dass Sie der Edge '{{edgeName}}' privat machen möchten?",
+    "make-private-edge-text": "Nach der Bestätigung werden der Edge und dessen Daten privat und sind für andere nicht mehr zugänglich.",
+    "import": "Edge importieren",
     "label": "Bezeichnung",
     "load-entity-error": "Entität nicht gefunden. Fehler beim Laden der Informationen",
-    "assign-new-edge": "Neue Rand zuordnen",
-    "unassign-from-edge": "Rand zuweisen",
-    "edge-key": "Rand Schlüssel",
-    "copy-edge-key": "Rand Schlüssel kopieren",
-    "edge-key-copied-message": "Rand Schlüssel wurde in die Zwischenablage kopiert",
-    "edge-secret": "Rand Geheimnis",
-    "copy-edge-secret": "Rand Geheimnis kopieren",
-    "edge-secret-copied-message": "Rand Geheimnis wurde in die Zwischenablage kopiert",
-    "edge-assets": "Rand-Objekte verwalten",
-    "edge-devices": "Rand-Geräte verwalten",
-    "edge-entity-views": "Rand-Entitätsansichten verwalten",
-    "edge-dashboards": "Rand-Dashboards verwalten",
-    "edge-rulechains": "Kantenregelketten",
-    "assets": "Rand Objekte",
+    "assign-new-edge": "Neue Edge zuordnen",
+    "unassign-from-edge": "Edge zuweisen",
+    "edge-key": "Edge-Schlüssel",
+    "copy-edge-key": "Edge-Schlüssel kopieren",
+    "edge-key-copied-message": "Edge-Schlüssel wurde in die Zwischenablage kopiert",
+    "edge-secret": "Edge-Geheimnis",
+    "copy-edge-secret": "Edge-Geheimnis kopieren",
+    "edge-secret-copied-message": "Edge-Geheimnis wurde in die Zwischenablage kopiert",
+    "edge-assets": "Edge-Objekte verwalten",
+    "edge-devices": "Edge-Geräte verwalten",
+    "edge-entity-views": "Edge-Entitätsansichten verwalten",
+    "edge-dashboards": "Edge-Dashboards verwalten",
+    "edge-rulechains": "Edge-Regelketten",
+    "assets": "Edge Objekte",
     "devices": "Objekte Geräte",
     "entity-views": "Objekte Entitätsansichten",
-    "dashboard": "Kanten-Dashboard",
-    "dashboards": "Rand Dashboards",
+    "dashboard": "Edge-Dashboard",
+    "dashboards": "Edge-Dashboards",
     "rulechain-templates": "Regelkettenvorlagen",
-    "rulechains": "Rand Regelketten",
-    "search": "Kanten durchsuchen",
-    "selected-edges": "{count, plural, =1 {1 Rand} other {# Rand} } ausgewählt",
+    "rulechains": "Edge-Regelketten",
+    "search": "Edges durchsuchen",
+    "selected-edges": "{count, plural, =1 {1 Edge} other {# Edges} } ausgewählt",
     "any-edge": "Beliebige Kante",
-    "no-edge-types-matching": "Es wurden keine Kantentypen gefunden, die mit '{{entitySubtype}}' übereinstimmen.",
-    "edge-type-list-empty": "Keine Kantentypen ausgewählt.",
-    "edge-types": "Kantentypen",
-    "enter-edge-type": "Geben Sie den Kantentyp ein",
+    "no-edge-types-matching": "Es wurden keine Edgetypen gefunden, die mit '{{entitySubtype}}' übereinstimmen.",
+    "edge-type-list-empty": "Keine Edgetypen ausgewählt.",
+    "edge-types": "Edgetypen",
+    "enter-edge-type": "Geben Sie den Edgetyp ein",
     "deployed": "Bereitgestellt",
     "pending": "Steht aus",
     "downlinks": "Downlinks",
@@ -855,38 +855,38 @@
   "edge-event": {
     "type-dashboard": "Dashboard",
     "type-asset": "Asset",
-    "type-device": "Device",
-    "type-device-profile": "Device Profile",
-    "type-entity-view": "Entity View",
+    "type-device": "Gerät",
+    "type-device-profile": "Geräteprofile",
+    "type-entity-view": "Entitätsansicht",
     "type-alarm": "Alarm",
-    "type-rule-chain": "Rule Chain",
-    "type-rule-chain-metadata": "Rule Chain Metadata",
+    "type-rule-chain": "Regelkette",
+    "type-rule-chain-metadata": "Regelkettenmetadaten",
     "type-edge": "Edge",
-    "type-user": "User",
-    "type-customer": "Customer",
-    "type-relation": "Relation",
-    "type-widgets-bundle": "Widgets Bundle",
-    "type-widgets-type": "Widgets Type",
-    "type-admin-settings": "Admin Settings",
-    "action-type-added": "Added",
-    "action-type-deleted": "Deleted",
-    "action-type-updated": "Updated",
-    "action-type-post-attributes": "Post Attributes",
-    "action-type-attributes-updated": "Attributes Updated",
-    "action-type-attributes-deleted": "Attributes Deleted",
-    "action-type-timeseries-updated": "Timeseries Updated",
-    "action-type-credentials-updated": "Credentials Updated",
-    "action-type-assigned-to-customer": "Assigned to Customer",
-    "action-type-unassigned-from-customer": "Unassigned from Customer",
-    "action-type-relation-add-or-update": "Relation Add or Update",
-    "action-type-relation-deleted": "Relation Deleted",
-    "action-type-rpc-call": "RPC Call",
-    "action-type-alarm-ack": "Alarm Ack",
-    "action-type-alarm-clear": "Alarm Clear",
-    "action-type-assigned-to-edge": "Assigned to Edge",
-    "action-type-unassigned-from-edge": "Unassigned from Edge",
-    "action-type-credentials-request": "Credentials Request",
-    "action-type-entity-merge-request": "Entity Merge Request"
+    "type-user": "Benutzer",
+    "type-customer": "Kunde",
+    "type-relation": "Beziehung",
+    "type-widgets-bundle": "Widgetpaket",
+    "type-widgets-type": "Widgettyp",
+    "type-admin-settings": "Admineinstellungen",
+    "action-type-added": "Hinzugefügt",
+    "action-type-deleted": "Gelöscht",
+    "action-type-updated": "Aktualisiert",
+    "action-type-post-attributes": "Attribute posten",
+    "action-type-attributes-updated": "Attribute Aktualisiert",
+    "action-type-attributes-deleted": "Attribute Gelöscht",
+    "action-type-timeseries-updated": "Timeseries Aktualisiert",
+    "action-type-credentials-updated": "Zugangsdaten Aktualisiert",
+    "action-type-assigned-to-customer": "Dem Kunden zugewiesen",
+    "action-type-unassigned-from-customer": "Kundenzuweisung aufgehoben",
+    "action-type-relation-add-or-update": "Beziehung hinzufügen oder Aktualisieren",
+    "action-type-relation-deleted": "Beziehung gelöscht",
+    "action-type-rpc-call": "RPC-Aufruf",
+    "action-type-alarm-ack": "Alarm zurkenntnisnehmen",
+    "action-type-alarm-clear": "Alarm löschen",
+    "action-type-assigned-to-edge": "Der Edge zuweisen",
+    "action-type-unassigned-from-edge": "Edge-Zuweisung aufheben",
+    "action-type-credentials-request": "Zugangsdatenabfrage",
+    "action-type-entity-merge-request": "Entitäts-Merge-Request"
   },
   "error": {
     "unable-to-connect": "Es konnte keine Verbindung zum Server hergestellt werden! Bitte überprüfen Sie Ihre Internetverbindung.",
@@ -980,10 +980,10 @@
     "type-rulenodes": "Regelknoten",
     "list-of-rulenodes": "{ count, plural, =1 {Ein Regelknoten} other {Liste von # Regelknoten} }",
     "rulenode-name-starts-with": "Regelknoten beginnend mit '{{prefix}}'",
-    "type-edge": "Randtyp",
-    "type-edges": "Randtyp",
-    "list-of-edges": "{ count, plural, =1 {1 Rand} other {# Rand} }",
-    "edge-name-starts-with": "Rand beginnend mit '{{prefix}}'",
+    "type-edge": "Edgetyp",
+    "type-edges": "Eandtyp",
+    "list-of-edges": "{ count, plural, =1 {1 Edge} other {# Edges} }",
+    "edge-name-starts-with": "Edge beginnend mit '{{prefix}}'",
     "type-current-customer": "Aktueller Kunde",
     "search": "Entitäten suchen",
     "selected-entities": "{ count, plural, =1 {Entität} other {# Entitäten} } ausgewählt",
@@ -1125,7 +1125,7 @@
   },
   "extension": {
     "extensions": "Erweiterungen",
-    "selected-extensions": "{ count, plural, =1 {Erweiterung} other {# extensions} } ausgewählt",
+    "selected-extensions": "{ count, plural, =1 {Erweiterung} other {# Erweiterungen} } ausgewählt",
     "type": "Typ",
     "key": "Schlüssel",
     "value": "Wert",
@@ -1139,7 +1139,7 @@
     "edit": "Erweiterung bearbeiten",
     "delete-extension-title": "Möchten Sie die Erweiterung '{{extensionId}}' wirklich löschen?",
     "delete-extension-text": "Vorsicht, nach Bestätigung werden die Erweiterung und alle zugehörigen Daten nicht wiederhergestellt.",
-    "delete-extensions-title": "Möchten Sie wirklich löschen? { count, plural, =1 {1 extension} other {# extensions} }?",
+    "delete-extensions-title": "Möchten Sie wirklich { count, plural, =1 {1 Erweiterung} other {# Erweiterungen} } löschen?",
     "delete-extensions-text": "Vorsicht, nach der Bestätigung werden alle ausgewählten Erweiterungen entfernt.",
     "converters": "Konverter",
     "converter-id": "Konverter-ID",
@@ -1166,7 +1166,7 @@
     "host": "Host",
     "port": "Port",
     "port-range": "Der Port sollte im Bereich von 1 bis 65535 liegen.",
-    "ssl": "Ssl",
+    "ssl": "SSL",
     "credentials": "Zugangsdaten",
     "username": "Benutzername",
     "password": "Passwort",
@@ -1229,7 +1229,7 @@
     "opc-keystore": "Schlüsselspeicher",
     "opc-type": "Typ",
     "opc-keystore-type": "Typ",
-    "opc-keystore-location": "Standort *",
+    "opc-keystore-location": "Speicherort *",
     "opc-keystore-password": "Passwort",
     "opc-keystore-alias": "Alias",
     "opc-keystore-key-password": "Schlüsselpasswort",

ui-ngx/src/assets/locale/locale.constant-en_US.json

@@ -1384,6 +1384,7 @@
         "any-device": "Any device",
         "no-device-types-matching": "No device profiles matching '{{entitySubtype}}' were found.",
         "device-type-list-empty": "No device profiles selected!",
+        "device-profile-type-list-empty": "At least one device profile should be selected.",
         "device-types": "Device types",
         "name": "Name",
         "name-required": "Name is required.",