Merge branch 'master' into develop/3.3

@@ -35,6 +35,7 @@ import org.thingsboard.server.common.data.rule.RuleChain;
 import org.thingsboard.server.common.data.rule.RuleNode;
 import org.thingsboard.server.common.msg.TbMsg;
 import org.thingsboard.server.common.msg.plugin.ComponentLifecycleMsg;
+import org.thingsboard.server.common.msg.plugin.RuleNodeUpdatedMsg;
 import org.thingsboard.server.common.msg.queue.PartitionChangeMsg;
 import org.thingsboard.server.common.msg.queue.QueueToRuleEngineMsg;
 import org.thingsboard.server.common.msg.queue.RuleEngineException;
@@ -131,7 +132,7 @@ public class RuleChainActorMessageProcessor extends ComponentMsgProcessor<RuleCh
                 } else {
                     log.trace("[{}][{}] Updating rule node [{}]: {}", entityId, ruleNode.getId(), ruleNode.getName(), ruleNode);
                     existing.setSelf(ruleNode);
-                    existing.getSelfActor().tellWithHighPriority(new ComponentLifecycleMsg(tenantId, existing.getSelf().getId(), ComponentLifecycleEvent.UPDATED));
+                    existing.getSelfActor().tellWithHighPriority(new RuleNodeUpdatedMsg(tenantId, existing.getSelf().getId()));
                 }
             }
 

@@ -26,7 +26,6 @@ import org.thingsboard.server.actors.service.ContextBasedCreator;
 import org.thingsboard.server.common.data.id.RuleChainId;
 import org.thingsboard.server.common.data.id.RuleNodeId;
 import org.thingsboard.server.common.data.id.TenantId;
-import org.thingsboard.server.common.data.rule.RuleChain;
 import org.thingsboard.server.common.msg.TbActorMsg;
 import org.thingsboard.server.common.msg.plugin.ComponentLifecycleMsg;
 import org.thingsboard.server.common.msg.queue.PartitionChangeMsg;
@@ -54,6 +53,7 @@ public class RuleNodeActor extends ComponentActor<RuleNodeId, RuleNodeActorMessa
     protected boolean doProcess(TbActorMsg msg) {
         switch (msg.getMsgType()) {
             case COMPONENT_LIFE_CYCLE_MSG:
+            case RULE_NODE_UPDATED_MSG:
                 onComponentLifecycleMsg((ComponentLifecycleMsg) msg);
                 break;
             case RULE_CHAIN_TO_RULE_MSG:

@@ -20,14 +20,13 @@ import org.thingsboard.rule.engine.api.TbNodeConfiguration;
 import org.thingsboard.server.actors.ActorSystemContext;
 import org.thingsboard.server.actors.TbActorCtx;
 import org.thingsboard.server.actors.TbActorRef;
+import org.thingsboard.server.actors.TbRuleNodeUpdateException;
 import org.thingsboard.server.actors.shared.ComponentMsgProcessor;
 import org.thingsboard.server.common.data.ApiUsageRecordKey;
-import org.thingsboard.server.common.data.TenantProfile;
 import org.thingsboard.server.common.data.id.RuleNodeId;
 import org.thingsboard.server.common.data.id.TenantId;
 import org.thingsboard.server.common.data.plugin.ComponentLifecycleState;
 import org.thingsboard.server.common.data.rule.RuleNode;
-import org.thingsboard.server.common.data.tenant.profile.TenantProfileConfiguration;
 import org.thingsboard.server.common.msg.TbMsg;
 import org.thingsboard.server.common.msg.queue.PartitionChangeMsg;
 import org.thingsboard.server.common.msg.queue.RuleNodeException;
@@ -78,7 +77,11 @@ public class RuleNodeActorMessageProcessor extends ComponentMsgProcessor<RuleNod
             if (tbNode != null) {
                 tbNode.destroy();
             }
-            start(context);
+            try {
+                start(context);
+            } catch (Exception e) {
+                throw new TbRuleNodeUpdateException("Failed to update rule node", e);
+            }
         }
     }
 

@@ -20,6 +20,7 @@ import org.thingsboard.server.actors.ActorSystemContext;
 import org.thingsboard.server.actors.TbActor;
 import org.thingsboard.server.actors.TbActorCtx;
 import org.thingsboard.server.actors.TbActorException;
+import org.thingsboard.server.actors.TbRuleNodeUpdateException;
 import org.thingsboard.server.actors.shared.ComponentMsgProcessor;
 import org.thingsboard.server.actors.stats.StatsPersistMsg;
 import org.thingsboard.server.common.data.id.EntityId;
@@ -123,6 +124,9 @@ public abstract class ComponentActor<T extends EntityId, P extends ComponentMsgP
         } catch (Exception e) {
             logAndPersist("onLifecycleMsg", e, true);
             logLifecycleEvent(msg.getEvent(), e);
+            if (e instanceof TbRuleNodeUpdateException) {
+                throw (TbRuleNodeUpdateException) e;
+            }
         }
     }
 

@@ -34,6 +34,7 @@ import org.thingsboard.server.actors.app.AppInitMsg;
 import org.thingsboard.server.actors.stats.StatsActor;
 import org.thingsboard.server.common.msg.queue.PartitionChangeMsg;
 import org.thingsboard.server.queue.discovery.PartitionChangeEvent;
+import org.thingsboard.server.queue.discovery.TbApplicationEventListener;
 
 import javax.annotation.PostConstruct;
 import javax.annotation.PreDestroy;
@@ -43,7 +44,7 @@ import java.util.concurrent.ScheduledExecutorService;
 
 @Service
 @Slf4j
-public class DefaultActorService implements ActorService {
+public class DefaultActorService extends TbApplicationEventListener<PartitionChangeEvent> implements ActorService {
 
     public static final String APP_DISPATCHER_NAME = "app-dispatcher";
     public static final String TENANT_DISPATCHER_NAME = "tenant-dispatcher";
@@ -120,10 +121,10 @@ public class DefaultActorService implements ActorService {
         appActor.tellWithHighPriority(new AppInitMsg());
     }
 
-    @EventListener(PartitionChangeEvent.class)
-    public void onApplicationEvent(PartitionChangeEvent partitionChangeEvent) {
+    @Override
+    protected void onTbApplicationEvent(PartitionChangeEvent event) {
         log.info("Received partition change event.");
-        this.appActor.tellWithHighPriority(new PartitionChangeMsg(partitionChangeEvent.getServiceQueueKey(), partitionChangeEvent.getPartitions()));
+        this.appActor.tellWithHighPriority(new PartitionChangeMsg(event.getServiceQueueKey(), event.getPartitions()));
     }
 
     @PreDestroy

@@ -27,7 +27,22 @@ import org.springframework.beans.factory.annotation.Value;
 import org.springframework.security.core.Authentication;
 import org.springframework.security.core.context.SecurityContextHolder;
 import org.springframework.web.bind.annotation.ExceptionHandler;
-import org.thingsboard.server.common.data.*;
+import org.thingsboard.server.common.data.Customer;
+import org.thingsboard.server.common.data.Dashboard;
+import org.thingsboard.server.common.data.DashboardInfo;
+import org.thingsboard.server.common.data.DataConstants;
+import org.thingsboard.server.common.data.Device;
+import org.thingsboard.server.common.data.DeviceInfo;
+import org.thingsboard.server.common.data.DeviceProfile;
+import org.thingsboard.server.common.data.EntityType;
+import org.thingsboard.server.common.data.EntityView;
+import org.thingsboard.server.common.data.EntityViewInfo;
+import org.thingsboard.server.common.data.HasName;
+import org.thingsboard.server.common.data.HasTenantId;
+import org.thingsboard.server.common.data.Tenant;
+import org.thingsboard.server.common.data.TenantInfo;
+import org.thingsboard.server.common.data.TenantProfile;
+import org.thingsboard.server.common.data.User;
 import org.thingsboard.server.common.data.alarm.Alarm;
 import org.thingsboard.server.common.data.alarm.AlarmInfo;
 import org.thingsboard.server.common.data.asset.Asset;
@@ -84,6 +99,7 @@ import org.thingsboard.server.dao.oauth2.OAuth2ConfigTemplateService;
 import org.thingsboard.server.dao.oauth2.OAuth2Service;
 import org.thingsboard.server.dao.relation.RelationService;
 import org.thingsboard.server.dao.rule.RuleChainService;
+import org.thingsboard.server.dao.tenant.TbTenantProfileCache;
 import org.thingsboard.server.dao.tenant.TenantProfileService;
 import org.thingsboard.server.dao.tenant.TenantService;
 import org.thingsboard.server.dao.user.UserService;
@@ -96,7 +112,6 @@ import org.thingsboard.server.queue.util.TbCoreComponent;
 import org.thingsboard.server.service.component.ComponentDiscoveryService;
 import org.thingsboard.server.service.lwm2m.LwM2MModelsRepository;
 import org.thingsboard.server.service.profile.TbDeviceProfileCache;
-import org.thingsboard.server.dao.tenant.TbTenantProfileCache;
 import org.thingsboard.server.service.queue.TbClusterService;
 import org.thingsboard.server.service.security.model.SecurityUser;
 import org.thingsboard.server.service.security.permission.AccessControlService;
@@ -124,6 +139,9 @@ public abstract class BaseController {
     public static final String INCORRECT_TENANT_ID = "Incorrect tenantId ";
     public static final String YOU_DON_T_HAVE_PERMISSION_TO_PERFORM_THIS_OPERATION = "You don't have permission to perform this operation!";
 
+    protected static final String DEFAULT_DASHBOARD = "defaultDashboardId";
+    protected static final String HOME_DASHBOARD = "homeDashboardId";
+
     private static final ObjectMapper json = new ObjectMapper();
 
     @Autowired
@@ -766,7 +784,7 @@ public abstract class BaseController {
                         String scope = extractParameter(String.class, 0, additionalInfo);
                         @SuppressWarnings("unchecked")
                         List<AttributeKvEntry> attributes = extractParameter(List.class, 1, additionalInfo);
-                        metaData.putValue("scope", scope);
+                        metaData.putValue(DataConstants.SCOPE, scope);
                         if (attributes != null) {
                             for (AttributeKvEntry attr : attributes) {
                                 addKvEntry(entityNode, attr);
@@ -776,7 +794,7 @@ public abstract class BaseController {
                         String scope = extractParameter(String.class, 0, additionalInfo);
                         @SuppressWarnings("unchecked")
                         List<String> keys = extractParameter(List.class, 1, additionalInfo);
-                        metaData.putValue("scope", scope);
+                        metaData.putValue(DataConstants.SCOPE, scope);
                         ArrayNode attrsArrayNode = entityNode.putArray("attributes");
                         if (keys != null) {
                             keys.forEach(attrsArrayNode::add);
@@ -862,4 +880,14 @@ public abstract class BaseController {
             }
         }
     }
+
+    protected void processDashboardIdFromAdditionalInfo(ObjectNode additionalInfo, String requiredFields) throws ThingsboardException {
+        String dashboardId = additionalInfo.has(requiredFields) ? additionalInfo.get(requiredFields).asText() : null;
+        if(dashboardId != null && !dashboardId.equals("null")) {
+            if(dashboardService.findDashboardById(getTenantId(), new DashboardId(UUID.fromString(dashboardId))) == null) {
+                additionalInfo.remove(requiredFields);
+            }
+        }
+    }
+
 }

@@ -55,7 +55,11 @@ public class CustomerController extends BaseController {
         checkParameter(CUSTOMER_ID, strCustomerId);
         try {
             CustomerId customerId = new CustomerId(toUUID(strCustomerId));
-            return checkCustomerId(customerId, Operation.READ);
+            Customer customer = checkCustomerId(customerId, Operation.READ);
+            if(!customer.getAdditionalInfo().isNull()) {
+                processDashboardIdFromAdditionalInfo((ObjectNode) customer.getAdditionalInfo(), HOME_DASHBOARD);
+            }
+            return customer;
         } catch (Exception e) {
             throw handleException(e);
         }

@@ -15,6 +15,7 @@
  */
 package org.thingsboard.server.controller;
 
+import com.fasterxml.jackson.databind.node.ObjectNode;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.http.HttpStatus;
@@ -59,7 +60,11 @@ public class TenantController extends BaseController {
         checkParameter("tenantId", strTenantId);
         try {
             TenantId tenantId = new TenantId(toUUID(strTenantId));
-            return checkTenantId(tenantId, Operation.READ);
+            Tenant tenant = checkTenantId(tenantId, Operation.READ);
+            if(!tenant.getAdditionalInfo().isNull()) {
+                processDashboardIdFromAdditionalInfo((ObjectNode) tenant.getAdditionalInfo(), HOME_DASHBOARD);
+            }
+            return tenant;
         } catch (Exception e) {
             throw handleException(e);
         }

@@ -53,7 +53,6 @@ import org.thingsboard.server.service.security.model.token.JwtTokenFactory;
 import org.thingsboard.server.service.security.permission.Operation;
 import org.thingsboard.server.service.security.permission.Resource;
 import org.thingsboard.server.service.security.system.SystemSecurityService;
-import org.thingsboard.server.utils.MiscUtils;
 
 import javax.servlet.http.HttpServletRequest;
 
@@ -90,7 +89,12 @@ public class UserController extends BaseController {
         checkParameter(USER_ID, strUserId);
         try {
             UserId userId = new UserId(toUUID(strUserId));
-            return checkUserId(userId, Operation.READ);
+            User user = checkUserId(userId, Operation.READ);
+            if(!user.getAdditionalInfo().isNull()) {
+                processDashboardIdFromAdditionalInfo((ObjectNode) user.getAdditionalInfo(), DEFAULT_DASHBOARD);
+                processDashboardIdFromAdditionalInfo((ObjectNode) user.getAdditionalInfo(), HOME_DASHBOARD);
+            }
+            return user;
         } catch (Exception e) {
             throw handleException(e);
         }
@@ -329,4 +333,5 @@ public class UserController extends BaseController {
             throw handleException(e);
         }
     }
+
 }

@@ -54,6 +54,7 @@ import org.thingsboard.server.gen.transport.TransportProtos.UsageStatsKVProto;
 import org.thingsboard.server.queue.common.TbProtoQueueMsg;
 import org.thingsboard.server.queue.discovery.PartitionChangeEvent;
 import org.thingsboard.server.queue.discovery.PartitionService;
+import org.thingsboard.server.queue.discovery.TbApplicationEventListener;
 import org.thingsboard.server.queue.scheduler.SchedulerComponent;
 import org.thingsboard.server.service.queue.TbClusterService;
 import org.thingsboard.server.service.telemetry.InternalTelemetryService;
@@ -78,7 +79,7 @@ import java.util.stream.Collectors;
 
 @Slf4j
 @Service
-public class DefaultTbApiUsageStateService implements TbApiUsageStateService {
+public class DefaultTbApiUsageStateService extends TbApplicationEventListener<PartitionChangeEvent> implements TbApiUsageStateService {
 
     public static final String HOURLY = "Hourly";
     public static final FutureCallback<Integer> VOID_CALLBACK = new FutureCallback<Integer>() {
@@ -188,7 +189,7 @@ public class DefaultTbApiUsageStateService implements TbApiUsageStateService {
     }
 
     @Override
-    public void onApplicationEvent(PartitionChangeEvent partitionChangeEvent) {
+    protected void onTbApplicationEvent(PartitionChangeEvent partitionChangeEvent) {
         if (partitionChangeEvent.getServiceType().equals(ServiceType.TB_CORE)) {
             myTenantStates.entrySet().removeIf(entry -> !partitionService.resolve(ServiceType.TB_CORE, entry.getKey(), entry.getKey()).isMyPartition());
             otherTenantStates.entrySet().removeIf(entry -> partitionService.resolve(ServiceType.TB_CORE, entry.getKey(), entry.getKey()).isMyPartition());

@@ -151,12 +151,12 @@ public class DefaultTbCoreConsumerService extends AbstractConsumerService<ToCore
     }
 
     @Override
-    public void onApplicationEvent(PartitionChangeEvent partitionChangeEvent) {
-        if (partitionChangeEvent.getServiceType().equals(getServiceType())) {
-            log.info("Subscribing to partitions: {}", partitionChangeEvent.getPartitions());
-            this.mainConsumer.subscribe(partitionChangeEvent.getPartitions());
+    protected void onTbApplicationEvent(PartitionChangeEvent event) {
+        if (event.getServiceType().equals(getServiceType())) {
+            log.info("Subscribing to partitions: {}", event.getPartitions());
+            this.mainConsumer.subscribe(event.getPartitions());
             this.usageStatsConsumer.subscribe(
-                    partitionChangeEvent
+                    event
                             .getPartitions()
                             .stream()
                             .map(tpi -> tpi.newByTopic(usageStatsConsumer.getTopic()))

@@ -140,11 +140,11 @@ public class DefaultTbRuleEngineConsumerService extends AbstractConsumerService<
     }
 
     @Override
-    public void onApplicationEvent(PartitionChangeEvent partitionChangeEvent) {
-        if (partitionChangeEvent.getServiceType().equals(getServiceType())) {
-            ServiceQueue serviceQueue = partitionChangeEvent.getServiceQueueKey().getServiceQueue();
-            log.info("[{}] Subscribing to partitions: {}", serviceQueue.getQueue(), partitionChangeEvent.getPartitions());
-            consumers.get(serviceQueue.getQueue()).subscribe(partitionChangeEvent.getPartitions());
+    protected void onTbApplicationEvent(PartitionChangeEvent event) {
+        if (event.getServiceType().equals(getServiceType())) {
+            ServiceQueue serviceQueue = event.getServiceQueueKey().getServiceQueue();
+            log.info("[{}] Subscribing to partitions: {}", serviceQueue.getQueue(), event.getPartitions());
+            consumers.get(serviceQueue.getQueue()).subscribe(event.getPartitions());
         }
     }
 

@@ -36,6 +36,7 @@ import org.thingsboard.server.queue.TbQueueConsumer;
 import org.thingsboard.server.queue.common.TbProtoQueueMsg;
 import org.thingsboard.server.queue.discovery.PartitionChangeEvent;
 import org.thingsboard.server.common.transport.util.DataDecodingEncodingService;
+import org.thingsboard.server.queue.discovery.TbApplicationEventListener;
 import org.thingsboard.server.service.apiusage.TbApiUsageStateService;
 import org.thingsboard.server.service.profile.TbDeviceProfileCache;
 import org.thingsboard.server.dao.tenant.TbTenantProfileCache;
@@ -56,7 +57,7 @@ import java.util.function.Function;
 import java.util.stream.Collectors;
 
 @Slf4j
-public abstract class AbstractConsumerService<N extends com.google.protobuf.GeneratedMessageV3> implements ApplicationListener<PartitionChangeEvent> {
+public abstract class AbstractConsumerService<N extends com.google.protobuf.GeneratedMessageV3> extends TbApplicationEventListener<PartitionChangeEvent> {
 
     protected volatile ExecutorService consumersExecutor;
     protected volatile ExecutorService notificationsConsumerExecutor;

@@ -56,6 +56,7 @@ import org.thingsboard.server.dao.util.mapping.JacksonUtil;
 import org.thingsboard.server.gen.transport.TransportProtos;
 import org.thingsboard.server.queue.discovery.PartitionChangeEvent;
 import org.thingsboard.server.queue.discovery.PartitionService;
+import org.thingsboard.server.queue.discovery.TbApplicationEventListener;
 import org.thingsboard.server.queue.util.TbCoreComponent;
 import org.thingsboard.server.service.queue.TbClusterService;
 import org.thingsboard.server.service.telemetry.TelemetrySubscriptionService;
@@ -90,7 +91,7 @@ import static org.thingsboard.server.common.data.DataConstants.SERVER_SCOPE;
 @Service
 @TbCoreComponent
 @Slf4j
-public class DefaultDeviceStateService implements DeviceStateService {
+public class DefaultDeviceStateService extends TbApplicationEventListener<PartitionChangeEvent> implements DeviceStateService {
 
     public static final String ACTIVITY_STATE = "active";
     public static final String LAST_CONNECT_TIME = "lastConnectTime";
@@ -206,7 +207,6 @@ public class DefaultDeviceStateService implements DeviceStateService {
                 if (!state.isActive()) {
                     state.setActive(true);
                     save(deviceId, ACTIVITY_STATE, state.isActive());
-                    stateData.getMetaData().putValue("scope", SERVER_SCOPE);
                     pushRuleEngineMessage(stateData, ACTIVITY_EVENT);
                 }
             }
@@ -295,7 +295,7 @@ public class DefaultDeviceStateService implements DeviceStateService {
     }
 
     @Override
-    public void onApplicationEvent(PartitionChangeEvent partitionChangeEvent) {
+    protected void onTbApplicationEvent(PartitionChangeEvent partitionChangeEvent) {
         if (ServiceType.TB_CORE.equals(partitionChangeEvent.getServiceType())) {
             deduplicationExecutor.submit(partitionChangeEvent.getPartitions());
         }
@@ -447,7 +447,7 @@ public class DefaultDeviceStateService implements DeviceStateService {
     }
 
     private <T extends KvEntry> Function<List<T>, DeviceStateData> extractDeviceStateData(Device device) {
-        return new Function<List<T>, DeviceStateData>() {
+        return new Function<>() {
             @Nullable
             @Override
             public DeviceStateData apply(@Nullable List<T> data) {
@@ -503,7 +503,11 @@ public class DefaultDeviceStateService implements DeviceStateService {
             } else {
                 data = JacksonUtil.toString(state);
             }
-            TbMsg tbMsg = TbMsg.newMsg(msgType, stateData.getDeviceId(), stateData.getMetaData().copy(), TbMsgDataType.JSON, data);
+            TbMsgMetaData md = stateData.getMetaData().copy();
+            if(!persistToTelemetry){
+                md.putValue(DataConstants.SCOPE, SERVER_SCOPE);
+            }
+            TbMsg tbMsg = TbMsg.newMsg(msgType, stateData.getDeviceId(), md, TbMsgDataType.JSON, data);
             clusterService.pushMsgToRuleEngine(stateData.getTenantId(), stateData.getDeviceId(), tbMsg, null);
         } catch (Exception e) {
             log.warn("[{}] Failed to push inactivity alarm: {}", stateData.getDeviceId(), state, e);

@@ -48,6 +48,7 @@ import org.thingsboard.server.queue.TbQueueProducer;
 import org.thingsboard.server.queue.common.TbProtoQueueMsg;
 import org.thingsboard.server.queue.discovery.PartitionChangeEvent;
 import org.thingsboard.server.queue.discovery.PartitionService;
+import org.thingsboard.server.queue.discovery.TbApplicationEventListener;
 import org.thingsboard.server.queue.discovery.TbServiceInfoProvider;
 import org.thingsboard.server.queue.provider.TbQueueProducerProvider;
 import org.thingsboard.server.queue.util.TbCoreComponent;
@@ -76,7 +77,7 @@ import java.util.function.Predicate;
 @Slf4j
 @TbCoreComponent
 @Service
-public class DefaultSubscriptionManagerService implements SubscriptionManagerService {
+public class DefaultSubscriptionManagerService extends TbApplicationEventListener<PartitionChangeEvent> implements SubscriptionManagerService {
 
     @Autowired
     private AttributesService attrService;
@@ -178,7 +179,7 @@ public class DefaultSubscriptionManagerService implements SubscriptionManagerSer
     }
 
     @Override
-    public void onApplicationEvent(PartitionChangeEvent partitionChangeEvent) {
+    protected void onTbApplicationEvent(PartitionChangeEvent partitionChangeEvent) {
         if (ServiceType.TB_CORE.equals(partitionChangeEvent.getServiceType())) {
             Set<TopicPartitionInfo> removedPartitions = new HashSet<>(currentPartitions);
             removedPartitions.removeAll(partitionChangeEvent.getPartitions());

@@ -476,7 +476,7 @@ public class DefaultTbEntityDataSubscriptionService implements TbEntityDataSubsc
     public void cancelAllSessionSubscriptions(String sessionId) {
         Map<Integer, TbAbstractDataSubCtx> sessionSubs = subscriptionsBySessionId.remove(sessionId);
         if (sessionSubs != null) {
-            sessionSubs.values().stream().filter(sub -> sub instanceof TbEntityDataSubCtx).map(sub -> (TbEntityDataSubCtx) sub).forEach(this::cleanupAndCancel);
+            sessionSubs.values().forEach(this::cleanupAndCancel);
         }
     }
 

@@ -28,6 +28,7 @@ import org.thingsboard.server.queue.discovery.PartitionService;
 import org.thingsboard.server.common.msg.queue.ServiceType;
 import org.thingsboard.server.common.msg.queue.TopicPartitionInfo;
 import org.thingsboard.server.common.msg.queue.TbCallback;
+import org.thingsboard.server.queue.discovery.TbApplicationEventListener;
 import org.thingsboard.server.queue.util.TbCoreComponent;
 import org.thingsboard.server.service.queue.TbClusterService;
 import org.thingsboard.server.service.telemetry.sub.AlarmSubscriptionUpdate;
@@ -62,6 +63,34 @@ public class DefaultTbLocalSubscriptionService implements TbLocalSubscriptionSer
     private SubscriptionManagerService subscriptionManagerService;
 
     private ExecutorService subscriptionUpdateExecutor;
+    
+    private TbApplicationEventListener<PartitionChangeEvent> partitionChangeListener = new TbApplicationEventListener<>() {
+        @Override
+        protected void onTbApplicationEvent(PartitionChangeEvent event) {
+            if (ServiceType.TB_CORE.equals(event.getServiceType())) {
+                currentPartitions.clear();
+                currentPartitions.addAll(event.getPartitions());
+            }
+        }
+    };
+
+    private TbApplicationEventListener<ClusterTopologyChangeEvent> clusterTopologyChangeListener = new TbApplicationEventListener<>() {
+        @Override
+        protected void onTbApplicationEvent(ClusterTopologyChangeEvent event) {
+            if (event.getServiceQueueKeys().stream().anyMatch(key -> ServiceType.TB_CORE.equals(key.getServiceType()))) {
+                /*
+                 * If the cluster topology has changed, we need to push all current subscriptions to SubscriptionManagerService again.
+                 * Otherwise, the SubscriptionManagerService may "forget" those subscriptions in case of restart.
+                 * Although this is resource consuming operation, it is cheaper than sending ping/pong commands periodically
+                 * It is also cheaper then caching the subscriptions by entity id and then lookup of those caches every time we have new telemetry in SubscriptionManagerService.
+                 * Even if we cache locally the list of active subscriptions by entity id, it is still time consuming operation to get them from cache
+                 * Since number of subscriptions is usually much less then number of devices that are pushing data.
+                 */
+                subscriptionsBySessionId.values().forEach(map -> map.values()
+                        .forEach(sub -> pushSubscriptionToManagerService(sub, true)));
+            }
+        }
+    };
 
     @PostConstruct
     public void initExecutor() {
@@ -77,28 +106,14 @@ public class DefaultTbLocalSubscriptionService implements TbLocalSubscriptionSer
 
     @Override
     @EventListener(PartitionChangeEvent.class)
-    public void onApplicationEvent(PartitionChangeEvent partitionChangeEvent) {
-        if (ServiceType.TB_CORE.equals(partitionChangeEvent.getServiceType())) {
-            currentPartitions.clear();
-            currentPartitions.addAll(partitionChangeEvent.getPartitions());
-        }
+    public void onApplicationEvent(PartitionChangeEvent event) {
+        partitionChangeListener.onApplicationEvent(event);
     }
 
     @Override
     @EventListener(ClusterTopologyChangeEvent.class)
     public void onApplicationEvent(ClusterTopologyChangeEvent event) {
-        if (event.getServiceQueueKeys().stream().anyMatch(key -> ServiceType.TB_CORE.equals(key.getServiceType()))) {
-            /*
-             * If the cluster topology has changed, we need to push all current subscriptions to SubscriptionManagerService again.
-             * Otherwise, the SubscriptionManagerService may "forget" those subscriptions in case of restart.
-             * Although this is resource consuming operation, it is cheaper than sending ping/pong commands periodically
-             * It is also cheaper then caching the subscriptions by entity id and then lookup of those caches every time we have new telemetry in SubscriptionManagerService.
-             * Even if we cache locally the list of active subscriptions by entity id, it is still time consuming operation to get them from cache
-             * Since number of subscriptions is usually much less then number of devices that are pushing data.
-             */
-            subscriptionsBySessionId.values().forEach(map -> map.values()
-                    .forEach(sub -> pushSubscriptionToManagerService(sub, true)));
-        }
+        clusterTopologyChangeListener.onApplicationEvent(event);
     }
 
     //TODO 3.1: replace null callbacks with callbacks from websocket service.

@@ -41,6 +41,7 @@ import org.thingsboard.server.dao.timeseries.TimeseriesService;
 import org.thingsboard.server.gen.transport.TransportProtos;
 import org.thingsboard.server.queue.discovery.PartitionChangeEvent;
 import org.thingsboard.server.queue.discovery.PartitionService;
+import org.thingsboard.server.queue.discovery.TbApplicationEventListener;
 import org.thingsboard.server.service.queue.TbClusterService;
 import org.thingsboard.server.service.subscription.SubscriptionManagerService;
 import org.thingsboard.server.service.subscription.TbSubscriptionUtils;
@@ -61,7 +62,7 @@ import java.util.function.Consumer;
  * Created by ashvayka on 27.03.18.
  */
 @Slf4j
-public abstract class AbstractSubscriptionService implements ApplicationListener<PartitionChangeEvent> {
+public abstract class AbstractSubscriptionService extends TbApplicationEventListener<PartitionChangeEvent>{
 
     protected final Set<TopicPartitionInfo> currentPartitions = ConcurrentHashMap.newKeySet();
 
@@ -97,8 +98,7 @@ public abstract class AbstractSubscriptionService implements ApplicationListener
     }
 
     @Override
-    @EventListener(PartitionChangeEvent.class)
-    public void onApplicationEvent(PartitionChangeEvent partitionChangeEvent) {
+    protected void onTbApplicationEvent(PartitionChangeEvent partitionChangeEvent) {
         if (ServiceType.TB_CORE.equals(partitionChangeEvent.getServiceType())) {
             currentPartitions.clear();
             currentPartitions.addAll(partitionChangeEvent.getPartitions());

@@ -118,6 +118,15 @@ security:
     githubMapper:
       emailUrl: "${SECURITY_OAUTH2_GITHUB_MAPPER_EMAIL_URL_KEY:https://api.github.com/user/emails}"
 
+# Usage statistics parameters
+usage:
+  stats:
+    report:
+      enabled: "${USAGE_STATS_REPORT_ENABLED:true}"
+      interval: "${USAGE_STATS_REPORT_INTERVAL:10}"
+    check:
+      cycle: "${USAGE_STATS_CHECK_CYCLE:60000}"
+
 # Dashboard parameters
 dashboard:
   # Maximum allowed datapoints fetched by widgets
@@ -675,6 +684,10 @@ queue:
       transport-api: "${TB_QUEUE_KAFKA_TA_TOPIC_PROPERTIES:retention.ms:604800000;segment.bytes:26214400;retention.bytes:1048576000;partitions:1}"
       notifications: "${TB_QUEUE_KAFKA_NOTIFICATIONS_TOPIC_PROPERTIES:retention.ms:604800000;segment.bytes:26214400;retention.bytes:1048576000;partitions:1}"
       js-executor: "${TB_QUEUE_KAFKA_JE_TOPIC_PROPERTIES:retention.ms:604800000;segment.bytes:26214400;retention.bytes:104857600;partitions:100}"
+    consumer-stats:
+      enabled: "${TB_QUEUE_KAFKA_CONSUMER_STATS_ENABLED:true}"
+      print-interval-ms: "${TB_QUEUE_KAFKA_CONSUMER_STATS_MIN_PRINT_INTERVAL_MS:60000}"
+      kafka-response-timeout-ms: "${TB_QUEUE_KAFKA_CONSUMER_STATS_RESPONSE_TIMEOUT_MS:1000}"
   aws_sqs:
     use_default_credential_provider_chain: "${TB_QUEUE_AWS_SQS_USE_DEFAULT_CREDENTIAL_PROVIDER_CHAIN:false}"
     access_key_id: "${TB_QUEUE_AWS_SQS_ACCESS_KEY_ID:YOUR_KEY}"

@@ -17,6 +17,7 @@ package org.thingsboard.server.actors;
 
 import lombok.Data;
 import lombok.extern.slf4j.Slf4j;
+import org.thingsboard.server.common.msg.MsgType;
 import org.thingsboard.server.common.msg.TbActorMsg;
 import org.thingsboard.server.common.msg.TbActorStopReason;
 
@@ -73,7 +74,7 @@ public final class TbActorMailbox implements TbActorCtx {
             if (strategy.isStop() || (settings.getMaxActorInitAttempts() > 0 && attemptIdx > settings.getMaxActorInitAttempts())) {
                 log.info("[{}] Failed to init actor, attempt {}, going to stop attempts.", selfId, attempt, t);
                 stopReason = TbActorStopReason.INIT_FAILED;
-                system.stop(selfId);
+                destroy();
             } else if (strategy.getRetryDelay() > 0) {
                 log.info("[{}] Failed to init actor, attempt {}, going to retry in attempts in {}ms", selfId, attempt, strategy.getRetryDelay());
                 log.debug("[{}] Error", selfId, t);
@@ -95,7 +96,19 @@ public final class TbActorMailbox implements TbActorCtx {
             }
             tryProcessQueue(true);
         } else {
-            msg.onTbActorStopped(stopReason);
+            if (highPriority && msg.getMsgType().equals(MsgType.RULE_NODE_UPDATED_MSG)) {
+                synchronized (this) {
+                    if (stopReason == TbActorStopReason.INIT_FAILED) {
+                        destroyInProgress.set(false);
+                        stopReason = null;
+                        initActor();
+                    } else {
+                        msg.onTbActorStopped(stopReason);
+                    }
+                }
+            } else {
+                msg.onTbActorStopped(stopReason);
+            }
         }
     }
 
@@ -126,6 +139,9 @@ public final class TbActorMailbox implements TbActorCtx {
                 try {
                     log.debug("[{}] Going to process message: {}", selfId, msg);
                     actor.process(msg);
+                } catch (TbRuleNodeUpdateException updateException){
+                    stopReason = TbActorStopReason.INIT_FAILED;
+                    destroy();
                 } catch (Throwable t) {
                     log.debug("[{}] Failed to process message: {}", selfId, msg, t);
                     ProcessFailureStrategy strategy = actor.onProcessFailure(t);

+/**
+ * Copyright © 2016-2021 The Thingsboard Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.thingsboard.server.actors;
+
+public class TbRuleNodeUpdateException extends RuntimeException {
+
+    private static final long serialVersionUID = 8209771144711980882L;
+
+    public TbRuleNodeUpdateException(String message, Throwable cause) {
+        super(message, cause);
+    }
+}
+

@@ -24,6 +24,7 @@ public class DataConstants {
     public static final String CUSTOMER = "CUSTOMER";
     public static final String DEVICE = "DEVICE";
 
+    public static final String SCOPE = "scope";
     public static final String CLIENT_SCOPE = "CLIENT_SCOPE";
     public static final String SERVER_SCOPE = "SERVER_SCOPE";
     public static final String SHARED_SCOPE = "SHARED_SCOPE";

@@ -17,17 +17,23 @@ package org.thingsboard.server.common.data.query;
 
 import com.fasterxml.jackson.annotation.JsonIgnore;
 import lombok.Data;
-import lombok.Getter;
+import lombok.RequiredArgsConstructor;
 
 @Data
+@RequiredArgsConstructor
 public class DynamicValue<T> {
 
     @JsonIgnore
     private T resolvedValue;
 
-    @Getter
     private final DynamicValueSourceType sourceType;
-    @Getter
     private final String sourceAttribute;
+    private final boolean inherit;
+
+    public DynamicValue(DynamicValueSourceType sourceType, String sourceAttribute) {
+        this.sourceAttribute = sourceAttribute;
+        this.sourceType = sourceType;
+        this.inherit = false;
+    }
 
 }

@@ -40,6 +40,11 @@ public enum MsgType {
     COMPONENT_LIFE_CYCLE_MSG,
 
     /**
+     * Special message to indicate rule node update request
+     */
+    RULE_NODE_UPDATED_MSG,
+
+    /**
      * Misc messages consumed from the Queue and forwarded to Rule Engine Actor.
      *
      * See {@link QueueToRuleEngineMsg}

+/**
+ * Copyright © 2016-2021 The Thingsboard Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.thingsboard.server.common.msg.plugin;
+
+import lombok.ToString;
+import org.thingsboard.server.common.data.id.EntityId;
+import org.thingsboard.server.common.data.id.TenantId;
+import org.thingsboard.server.common.data.plugin.ComponentLifecycleEvent;
+import org.thingsboard.server.common.msg.MsgType;
+
+import java.util.Optional;
+
+/**
+ * @author Andrew Shvayka
+ */
+@ToString
+public class RuleNodeUpdatedMsg extends ComponentLifecycleMsg {
+
+    public RuleNodeUpdatedMsg(TenantId tenantId, EntityId entityId) {
+        super(tenantId, entityId, ComponentLifecycleEvent.UPDATED);
+    }
+
+    @Override
+    public MsgType getMsgType() {
+        return MsgType.RULE_NODE_UPDATED_MSG;
+    }
+}
\ No newline at end of file

@@ -22,7 +22,9 @@ import org.thingsboard.server.common.msg.queue.ServiceQueueKey;
 import java.util.Set;
 
 
-public class ClusterTopologyChangeEvent extends ApplicationEvent {
+public class ClusterTopologyChangeEvent extends TbApplicationEvent {
+
+    private static final long serialVersionUID = -2441739930040282254L;
 
     @Getter
     private final Set<ServiceQueueKey> serviceQueueKeys;

@@ -126,7 +126,7 @@ public class HashPartitionService implements PartitionService {
     }
 
     @Override
-    public void recalculatePartitions(ServiceInfo currentService, List<ServiceInfo> otherServices) {
+    public synchronized void recalculatePartitions(ServiceInfo currentService, List<ServiceInfo> otherServices) {
         logServiceInfo(currentService);
         otherServices.forEach(this::logServiceInfo);
         Map<ServiceQueueKey, List<ServiceInfo>> queueServicesMap = new HashMap<>();
@@ -134,7 +134,7 @@ public class HashPartitionService implements PartitionService {
         for (ServiceInfo other : otherServices) {
             addNode(queueServicesMap, other);
         }
-        queueServicesMap.values().forEach(list -> list.sort((a, b) -> a.getServiceId().compareTo(b.getServiceId())));
+        queueServicesMap.values().forEach(list -> list.sort(Comparator.comparing(ServiceInfo::getServiceId)));
 
         ConcurrentMap<ServiceQueueKey, List<Integer>> oldPartitions = myPartitions;
         TenantId myIsolatedOrSystemTenantId = getSystemOrIsolatedTenantId(currentService);

@@ -24,7 +24,9 @@ import org.thingsboard.server.common.msg.queue.TopicPartitionInfo;
 import java.util.Set;
 
 
-public class PartitionChangeEvent extends ApplicationEvent {
+public class PartitionChangeEvent extends TbApplicationEvent {
+
+    private static final long serialVersionUID = -8731788167026510559L;
 
     @Getter
     private final ServiceQueueKey serviceQueueKey;

+/**
+ * Copyright © 2016-2021 The Thingsboard Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.thingsboard.server.queue.discovery;
+
+import lombok.Getter;
+import org.springframework.context.ApplicationEvent;
+
+import java.util.concurrent.atomic.AtomicInteger;
+
+public class TbApplicationEvent extends ApplicationEvent {
+
+    private static final long serialVersionUID = 3884264064887765146L;
+
+    private static final AtomicInteger sequence = new AtomicInteger();
+
+    @Getter
+    private final int sequenceNumber;
+
+    public TbApplicationEvent(Object source) {
+        super(source);
+        sequenceNumber = sequence.incrementAndGet();
+    }
+
+}

+/**
+ * Copyright © 2016-2021 The Thingsboard Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.thingsboard.server.queue.discovery;
+
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.context.ApplicationListener;
+
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReentrantLock;
+
+@Slf4j
+public abstract class TbApplicationEventListener<T extends TbApplicationEvent> implements ApplicationListener<T> {
+
+    private int lastProcessedSequenceNumber = Integer.MIN_VALUE;
+    private final Lock seqNumberLock = new ReentrantLock();
+
+    @Override
+    public void onApplicationEvent(T event) {
+        boolean validUpdate = false;
+        seqNumberLock.lock();
+        try {
+            if (event.getSequenceNumber() > lastProcessedSequenceNumber) {
+                validUpdate = true;
+                lastProcessedSequenceNumber = event.getSequenceNumber();
+            }
+        } finally {
+            seqNumberLock.unlock();
+        }
+        if (validUpdate) {
+            onTbApplicationEvent(event);
+        } else {
+            log.info("Application event ignored due to invalid sequence number ({} > {}). Event: {}", lastProcessedSequenceNumber, event.getSequenceNumber(), event);
+        }
+    }
+
+    protected abstract void onTbApplicationEvent(T event);
+
+
+}

+/**
+ * Copyright © 2016-2021 The Thingsboard Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.thingsboard.server.queue.kafka;
+
+import lombok.AllArgsConstructor;
+import lombok.Getter;
+import lombok.NoArgsConstructor;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
+import org.springframework.stereotype.Component;
+
+@Component
+@ConditionalOnProperty(prefix = "queue", value = "type", havingValue = "kafka")
+@Getter
+@AllArgsConstructor
+@NoArgsConstructor
+public class TbKafkaConsumerStatisticConfig {
+    @Value("${queue.kafka.consumer-stats.enabled:true}")
+    private Boolean enabled;
+    @Value("${queue.kafka.consumer-stats.print-interval-ms:60000}")
+    private Long printIntervalMs;
+    @Value("${queue.kafka.consumer-stats.kafka-response-timeout-ms:1000}")
+    private Long kafkaResponseTimeoutMs;
+}

+/**
+ * Copyright © 2016-2021 The Thingsboard Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.thingsboard.server.queue.kafka;
+
+import lombok.Builder;
+import lombok.Data;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.kafka.clients.admin.AdminClient;
+import org.apache.kafka.clients.consumer.Consumer;
+import org.apache.kafka.clients.consumer.ConsumerConfig;
+import org.apache.kafka.clients.consumer.KafkaConsumer;
+import org.apache.kafka.clients.consumer.OffsetAndMetadata;
+import org.apache.kafka.common.TopicPartition;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
+import org.springframework.stereotype.Component;
+import org.springframework.util.StringUtils;
+import org.thingsboard.common.util.ThingsBoardThreadFactory;
+import org.thingsboard.server.common.data.id.TenantId;
+import org.thingsboard.server.common.msg.queue.ServiceType;
+import org.thingsboard.server.queue.discovery.PartitionService;
+
+import javax.annotation.PostConstruct;
+import javax.annotation.PreDestroy;
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+
+@Slf4j
+@Component
+@RequiredArgsConstructor
+@ConditionalOnProperty(prefix = "queue", value = "type", havingValue = "kafka")
+public class TbKafkaConsumerStatsService {
+    private final Set<String> monitoredGroups = ConcurrentHashMap.newKeySet();
+
+    private final TbKafkaSettings kafkaSettings;
+    private final TbKafkaConsumerStatisticConfig statsConfig;
+    private final PartitionService partitionService;
+
+    private AdminClient adminClient;
+    private Consumer<String, byte[]> consumer;
+    private ScheduledExecutorService statsPrintScheduler;
+
+    @PostConstruct
+    public void init() {
+        if (!statsConfig.getEnabled()) {
+            return;
+        }
+        this.adminClient = AdminClient.create(kafkaSettings.toAdminProps());
+        this.statsPrintScheduler = Executors.newSingleThreadScheduledExecutor(ThingsBoardThreadFactory.forName("kafka-consumer-stats"));
+
+        Properties consumerProps = kafkaSettings.toConsumerProps();
+        consumerProps.put(ConsumerConfig.CLIENT_ID_CONFIG, "consumer-stats-loader-client");
+        consumerProps.put(ConsumerConfig.GROUP_ID_CONFIG, "consumer-stats-loader-client-group");
+        this.consumer = new KafkaConsumer<>(consumerProps);
+
+        startLogScheduling();
+    }
+
+    private void startLogScheduling() {
+        Duration timeoutDuration = Duration.ofMillis(statsConfig.getKafkaResponseTimeoutMs());
+        statsPrintScheduler.scheduleWithFixedDelay(() -> {
+            if (!isStatsPrintRequired()) {
+                return;
+            }
+            for (String groupId : monitoredGroups) {
+                try {
+                    Map<TopicPartition, OffsetAndMetadata> groupOffsets = adminClient.listConsumerGroupOffsets(groupId).partitionsToOffsetAndMetadata()
+                            .get(statsConfig.getKafkaResponseTimeoutMs(), TimeUnit.MILLISECONDS);
+                    Map<TopicPartition, Long> endOffsets = consumer.endOffsets(groupOffsets.keySet(), timeoutDuration);
+
+                    List<GroupTopicStats> lagTopicsStats = getTopicsStatsWithLag(groupOffsets, endOffsets);
+                    if (!lagTopicsStats.isEmpty()) {
+                        StringBuilder builder = new StringBuilder();
+                        for (int i = 0; i < lagTopicsStats.size(); i++) {
+                            builder.append(lagTopicsStats.get(i).toString());
+                            if (i != lagTopicsStats.size() - 1) {
+                                builder.append(", ");
+                            }
+                        }
+                        log.info("[{}] Topic partitions with lag: [{}].", groupId, builder.toString());
+                    }
+                } catch (Exception e) {
+                    log.warn("[{}] Failed to get consumer group stats. Reason - {}.", groupId, e.getMessage());
+                    log.trace("Detailed error: ", e);
+                }
+            }
+
+        }, statsConfig.getPrintIntervalMs(), statsConfig.getPrintIntervalMs(), TimeUnit.MILLISECONDS);
+    }
+
+    private boolean isStatsPrintRequired() {
+        boolean isMyRuleEnginePartition = partitionService.resolve(ServiceType.TB_RULE_ENGINE, TenantId.SYS_TENANT_ID, TenantId.SYS_TENANT_ID).isMyPartition();
+        boolean isMyCorePartition = partitionService.resolve(ServiceType.TB_CORE, TenantId.SYS_TENANT_ID, TenantId.SYS_TENANT_ID).isMyPartition();
+        return log.isInfoEnabled() && (isMyRuleEnginePartition || isMyCorePartition);
+    }
+
+    private List<GroupTopicStats> getTopicsStatsWithLag(Map<TopicPartition, OffsetAndMetadata> groupOffsets, Map<TopicPartition, Long> endOffsets) {
+        List<GroupTopicStats> consumerGroupStats = new ArrayList<>();
+        for (TopicPartition topicPartition : groupOffsets.keySet()) {
+            long endOffset = endOffsets.get(topicPartition);
+            long committedOffset = groupOffsets.get(topicPartition).offset();
+            long lag = endOffset - committedOffset;
+            if (lag != 0) {
+                GroupTopicStats groupTopicStats = GroupTopicStats.builder()
+                        .topic(topicPartition.topic())
+                        .partition(topicPartition.partition())
+                        .committedOffset(committedOffset)
+                        .endOffset(endOffset)
+                        .lag(lag)
+                        .build();
+                consumerGroupStats.add(groupTopicStats);
+            }
+        }
+        return consumerGroupStats;
+    }
+
+    public void registerClientGroup(String groupId) {
+        if (statsConfig.getEnabled() && !StringUtils.isEmpty(groupId)) {
+            monitoredGroups.add(groupId);
+        }
+    }
+
+    public void unregisterClientGroup(String groupId) {
+        if (statsConfig.getEnabled() && !StringUtils.isEmpty(groupId)) {
+            monitoredGroups.remove(groupId);
+        }
+    }
+
+    @PreDestroy
+    public void destroy() {
+        if (statsPrintScheduler != null) {
+            statsPrintScheduler.shutdownNow();
+        }
+        if (adminClient != null) {
+            adminClient.close();
+        }
+        if (consumer != null) {
+            consumer.close();
+        }
+    }
+
+
+    @Builder
+    @Data
+    private static class GroupTopicStats {
+        private String topic;
+        private int partition;
+        private long committedOffset;
+        private long endOffset;
+        private long lag;
+
+        @Override
+        public String toString() {
+            return "[" +
+                    "topic=[" + topic + ']' +
+                    ", partition=[" + partition + "]" +
+                    ", committedOffset=[" + committedOffset + "]" +
+                    ", endOffset=[" + endOffset + "]" +
+                    ", lag=[" + lag + "]" +
+                    "]";
+        }
+    }
+}

@@ -42,10 +42,13 @@ public class TbKafkaConsumerTemplate<T extends TbQueueMsg> extends AbstractTbQue
     private final KafkaConsumer<String, byte[]> consumer;
     private final TbKafkaDecoder<T> decoder;
 
+    private final TbKafkaConsumerStatsService statsService;
+    private final String groupId;
+
     @Builder
     private TbKafkaConsumerTemplate(TbKafkaSettings settings, TbKafkaDecoder<T> decoder,
                                     String clientId, String groupId, String topic,
-                                    TbQueueAdmin admin) {
+                                    TbQueueAdmin admin, TbKafkaConsumerStatsService statsService) {
         super(topic);
         Properties props = settings.toConsumerProps();
         props.put(ConsumerConfig.CLIENT_ID_CONFIG, clientId);
@@ -53,6 +56,13 @@ public class TbKafkaConsumerTemplate<T extends TbQueueMsg> extends AbstractTbQue
             props.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);
         }
 
+        this.statsService = statsService;
+        this.groupId = groupId;
+
+        if (statsService != null) {
+            statsService.registerClientGroup(groupId);
+        }
+
         this.admin = admin;
         this.consumer = new KafkaConsumer<>(props);
         this.decoder = decoder;
@@ -96,6 +106,8 @@ public class TbKafkaConsumerTemplate<T extends TbQueueMsg> extends AbstractTbQue
             consumer.unsubscribe();
             consumer.close();
         }
+        if (statsService != null) {
+            statsService.unregisterClientGroup(groupId);
+        }
     }
-
 }

@@ -39,6 +39,7 @@ import org.thingsboard.server.queue.common.TbProtoQueueMsg;
 import org.thingsboard.server.queue.discovery.PartitionService;
 import org.thingsboard.server.queue.discovery.TbServiceInfoProvider;
 import org.thingsboard.server.queue.kafka.TbKafkaAdmin;
+import org.thingsboard.server.queue.kafka.TbKafkaConsumerStatsService;
 import org.thingsboard.server.queue.kafka.TbKafkaConsumerTemplate;
 import org.thingsboard.server.queue.kafka.TbKafkaProducerTemplate;
 import org.thingsboard.server.queue.kafka.TbKafkaSettings;
@@ -65,6 +66,7 @@ public class KafkaMonolithQueueFactory implements TbCoreQueueFactory, TbRuleEngi
     private final TbQueueTransportApiSettings transportApiSettings;
     private final TbQueueTransportNotificationSettings transportNotificationSettings;
     private final TbQueueRemoteJsInvokeSettings jsInvokeSettings;
+    private final TbKafkaConsumerStatsService consumerStatsService;
 
     private final TbQueueAdmin coreAdmin;
     private final TbQueueAdmin ruleEngineAdmin;
@@ -79,6 +81,7 @@ public class KafkaMonolithQueueFactory implements TbCoreQueueFactory, TbRuleEngi
                                      TbQueueTransportApiSettings transportApiSettings,
                                      TbQueueTransportNotificationSettings transportNotificationSettings,
                                      TbQueueRemoteJsInvokeSettings jsInvokeSettings,
+                                     TbKafkaConsumerStatsService consumerStatsService,
                                      TbKafkaTopicConfigs kafkaTopicConfigs) {
         this.partitionService = partitionService;
         this.kafkaSettings = kafkaSettings;
@@ -88,6 +91,7 @@ public class KafkaMonolithQueueFactory implements TbCoreQueueFactory, TbRuleEngi
         this.transportApiSettings = transportApiSettings;
         this.transportNotificationSettings = transportNotificationSettings;
         this.jsInvokeSettings = jsInvokeSettings;
+        this.consumerStatsService = consumerStatsService;
 
         this.coreAdmin = new TbKafkaAdmin(kafkaSettings, kafkaTopicConfigs.getCoreConfigs());
         this.ruleEngineAdmin = new TbKafkaAdmin(kafkaSettings, kafkaTopicConfigs.getRuleEngineConfigs());
@@ -156,6 +160,7 @@ public class KafkaMonolithQueueFactory implements TbCoreQueueFactory, TbRuleEngi
         consumerBuilder.groupId("re-" + queueName + "-consumer");
         consumerBuilder.decoder(msg -> new TbProtoQueueMsg<>(msg.getKey(), ToRuleEngineMsg.parseFrom(msg.getData()), msg.getHeaders()));
         consumerBuilder.admin(ruleEngineAdmin);
+        consumerBuilder.statsService(consumerStatsService);
         return consumerBuilder.build();
     }
 
@@ -168,6 +173,7 @@ public class KafkaMonolithQueueFactory implements TbCoreQueueFactory, TbRuleEngi
         consumerBuilder.groupId("monolith-rule-engine-notifications-consumer-" + serviceInfoProvider.getServiceId());
         consumerBuilder.decoder(msg -> new TbProtoQueueMsg<>(msg.getKey(), ToRuleEngineNotificationMsg.parseFrom(msg.getData()), msg.getHeaders()));
         consumerBuilder.admin(notificationAdmin);
+        consumerBuilder.statsService(consumerStatsService);
         return consumerBuilder.build();
     }
 
@@ -180,6 +186,7 @@ public class KafkaMonolithQueueFactory implements TbCoreQueueFactory, TbRuleEngi
         consumerBuilder.groupId("monolith-core-consumer");
         consumerBuilder.decoder(msg -> new TbProtoQueueMsg<>(msg.getKey(), ToCoreMsg.parseFrom(msg.getData()), msg.getHeaders()));
         consumerBuilder.admin(coreAdmin);
+        consumerBuilder.statsService(consumerStatsService);
         return consumerBuilder.build();
     }
 
@@ -192,6 +199,7 @@ public class KafkaMonolithQueueFactory implements TbCoreQueueFactory, TbRuleEngi
         consumerBuilder.groupId("monolith-core-notifications-consumer-" + serviceInfoProvider.getServiceId());
         consumerBuilder.decoder(msg -> new TbProtoQueueMsg<>(msg.getKey(), ToCoreNotificationMsg.parseFrom(msg.getData()), msg.getHeaders()));
         consumerBuilder.admin(notificationAdmin);
+        consumerBuilder.statsService(consumerStatsService);
         return consumerBuilder.build();
     }
 
@@ -204,6 +212,7 @@ public class KafkaMonolithQueueFactory implements TbCoreQueueFactory, TbRuleEngi
         consumerBuilder.groupId("monolith-transport-api-consumer");
         consumerBuilder.decoder(msg -> new TbProtoQueueMsg<>(msg.getKey(), TransportApiRequestMsg.parseFrom(msg.getData()), msg.getHeaders()));
         consumerBuilder.admin(transportApiAdmin);
+        consumerBuilder.statsService(consumerStatsService);
         return consumerBuilder.build();
     }
 
@@ -237,6 +246,7 @@ public class KafkaMonolithQueueFactory implements TbCoreQueueFactory, TbRuleEngi
                     return new TbProtoQueueMsg<>(msg.getKey(), builder.build(), msg.getHeaders());
                 }
         );
+        responseBuilder.statsService(consumerStatsService);
         responseBuilder.admin(jsExecutorAdmin);
 
         DefaultTbQueueRequestTemplate.DefaultTbQueueRequestTemplateBuilder
@@ -259,6 +269,7 @@ public class KafkaMonolithQueueFactory implements TbCoreQueueFactory, TbRuleEngi
         consumerBuilder.groupId("monolith-us-consumer");
         consumerBuilder.decoder(msg -> new TbProtoQueueMsg<>(msg.getKey(), ToUsageStatsServiceMsg.parseFrom(msg.getData()), msg.getHeaders()));
         consumerBuilder.admin(coreAdmin);
+        consumerBuilder.statsService(consumerStatsService);
         return consumerBuilder.build();
     }
 

@@ -39,6 +39,7 @@ import org.thingsboard.server.queue.common.TbProtoQueueMsg;
 import org.thingsboard.server.queue.discovery.PartitionService;
 import org.thingsboard.server.queue.discovery.TbServiceInfoProvider;
 import org.thingsboard.server.queue.kafka.TbKafkaAdmin;
+import org.thingsboard.server.queue.kafka.TbKafkaConsumerStatsService;
 import org.thingsboard.server.queue.kafka.TbKafkaConsumerTemplate;
 import org.thingsboard.server.queue.kafka.TbKafkaProducerTemplate;
 import org.thingsboard.server.queue.kafka.TbKafkaSettings;
@@ -62,6 +63,7 @@ public class KafkaTbCoreQueueFactory implements TbCoreQueueFactory {
     private final TbQueueRuleEngineSettings ruleEngineSettings;
     private final TbQueueTransportApiSettings transportApiSettings;
     private final TbQueueRemoteJsInvokeSettings jsInvokeSettings;
+    private final TbKafkaConsumerStatsService consumerStatsService;
 
     private final TbQueueAdmin coreAdmin;
     private final TbQueueAdmin ruleEngineAdmin;
@@ -75,6 +77,7 @@ public class KafkaTbCoreQueueFactory implements TbCoreQueueFactory {
                                    TbQueueRuleEngineSettings ruleEngineSettings,
                                    TbQueueTransportApiSettings transportApiSettings,
                                    TbQueueRemoteJsInvokeSettings jsInvokeSettings,
+                                   TbKafkaConsumerStatsService consumerStatsService,
                                    TbKafkaTopicConfigs kafkaTopicConfigs) {
         this.partitionService = partitionService;
         this.kafkaSettings = kafkaSettings;
@@ -83,6 +86,7 @@ public class KafkaTbCoreQueueFactory implements TbCoreQueueFactory {
         this.ruleEngineSettings = ruleEngineSettings;
         this.transportApiSettings = transportApiSettings;
         this.jsInvokeSettings = jsInvokeSettings;
+        this.consumerStatsService = consumerStatsService;
 
         this.coreAdmin = new TbKafkaAdmin(kafkaSettings, kafkaTopicConfigs.getCoreConfigs());
         this.ruleEngineAdmin = new TbKafkaAdmin(kafkaSettings, kafkaTopicConfigs.getRuleEngineConfigs());
@@ -150,6 +154,7 @@ public class KafkaTbCoreQueueFactory implements TbCoreQueueFactory {
         consumerBuilder.groupId("tb-core-node");
         consumerBuilder.decoder(msg -> new TbProtoQueueMsg<>(msg.getKey(), ToCoreMsg.parseFrom(msg.getData()), msg.getHeaders()));
         consumerBuilder.admin(coreAdmin);
+        consumerBuilder.statsService(consumerStatsService);
         return consumerBuilder.build();
     }
 
@@ -162,6 +167,7 @@ public class KafkaTbCoreQueueFactory implements TbCoreQueueFactory {
         consumerBuilder.groupId("tb-core-notifications-node-" + serviceInfoProvider.getServiceId());
         consumerBuilder.decoder(msg -> new TbProtoQueueMsg<>(msg.getKey(), ToCoreNotificationMsg.parseFrom(msg.getData()), msg.getHeaders()));
         consumerBuilder.admin(notificationAdmin);
+        consumerBuilder.statsService(consumerStatsService);
         return consumerBuilder.build();
     }
 
@@ -174,6 +180,7 @@ public class KafkaTbCoreQueueFactory implements TbCoreQueueFactory {
         consumerBuilder.groupId("tb-core-transport-api-consumer");
         consumerBuilder.decoder(msg -> new TbProtoQueueMsg<>(msg.getKey(), TransportApiRequestMsg.parseFrom(msg.getData()), msg.getHeaders()));
         consumerBuilder.admin(transportApiAdmin);
+        consumerBuilder.statsService(consumerStatsService);
         return consumerBuilder.build();
     }
 
@@ -208,6 +215,7 @@ public class KafkaTbCoreQueueFactory implements TbCoreQueueFactory {
                 }
         );
         responseBuilder.admin(jsExecutorAdmin);
+        responseBuilder.statsService(consumerStatsService);
 
         DefaultTbQueueRequestTemplate.DefaultTbQueueRequestTemplateBuilder
                 <TbProtoJsQueueMsg<JsInvokeProtos.RemoteJsRequest>, TbProtoQueueMsg<JsInvokeProtos.RemoteJsResponse>> builder = DefaultTbQueueRequestTemplate.builder();
@@ -229,6 +237,7 @@ public class KafkaTbCoreQueueFactory implements TbCoreQueueFactory {
         consumerBuilder.groupId("tb-core-us-consumer");
         consumerBuilder.decoder(msg -> new TbProtoQueueMsg<>(msg.getKey(), ToUsageStatsServiceMsg.parseFrom(msg.getData()), msg.getHeaders()));
         consumerBuilder.admin(coreAdmin);
+        consumerBuilder.statsService(consumerStatsService);
         return consumerBuilder.build();
     }
 

@@ -37,6 +37,7 @@ import org.thingsboard.server.queue.common.TbProtoQueueMsg;
 import org.thingsboard.server.queue.discovery.PartitionService;
 import org.thingsboard.server.queue.discovery.TbServiceInfoProvider;
 import org.thingsboard.server.queue.kafka.TbKafkaAdmin;
+import org.thingsboard.server.queue.kafka.TbKafkaConsumerStatsService;
 import org.thingsboard.server.queue.kafka.TbKafkaConsumerTemplate;
 import org.thingsboard.server.queue.kafka.TbKafkaProducerTemplate;
 import org.thingsboard.server.queue.kafka.TbKafkaSettings;
@@ -59,6 +60,7 @@ public class KafkaTbRuleEngineQueueFactory implements TbRuleEngineQueueFactory {
     private final TbQueueCoreSettings coreSettings;
     private final TbQueueRuleEngineSettings ruleEngineSettings;
     private final TbQueueRemoteJsInvokeSettings jsInvokeSettings;
+    private final TbKafkaConsumerStatsService consumerStatsService;
 
     private final TbQueueAdmin coreAdmin;
     private final TbQueueAdmin ruleEngineAdmin;
@@ -70,6 +72,7 @@ public class KafkaTbRuleEngineQueueFactory implements TbRuleEngineQueueFactory {
                                          TbQueueCoreSettings coreSettings,
                                          TbQueueRuleEngineSettings ruleEngineSettings,
                                          TbQueueRemoteJsInvokeSettings jsInvokeSettings,
+                                         TbKafkaConsumerStatsService consumerStatsService,
                                          TbKafkaTopicConfigs kafkaTopicConfigs) {
         this.partitionService = partitionService;
         this.kafkaSettings = kafkaSettings;
@@ -77,6 +80,7 @@ public class KafkaTbRuleEngineQueueFactory implements TbRuleEngineQueueFactory {
         this.coreSettings = coreSettings;
         this.ruleEngineSettings = ruleEngineSettings;
         this.jsInvokeSettings = jsInvokeSettings;
+        this.consumerStatsService = consumerStatsService;
 
         this.coreAdmin = new TbKafkaAdmin(kafkaSettings, kafkaTopicConfigs.getCoreConfigs());
         this.ruleEngineAdmin = new TbKafkaAdmin(kafkaSettings, kafkaTopicConfigs.getRuleEngineConfigs());
@@ -145,6 +149,7 @@ public class KafkaTbRuleEngineQueueFactory implements TbRuleEngineQueueFactory {
         consumerBuilder.groupId("re-" + queueName + "-consumer");
         consumerBuilder.decoder(msg -> new TbProtoQueueMsg<>(msg.getKey(), ToRuleEngineMsg.parseFrom(msg.getData()), msg.getHeaders()));
         consumerBuilder.admin(ruleEngineAdmin);
+        consumerBuilder.statsService(consumerStatsService);
         return consumerBuilder.build();
     }
 
@@ -157,6 +162,7 @@ public class KafkaTbRuleEngineQueueFactory implements TbRuleEngineQueueFactory {
         consumerBuilder.groupId("tb-rule-engine-notifications-node-" + serviceInfoProvider.getServiceId());
         consumerBuilder.decoder(msg -> new TbProtoQueueMsg<>(msg.getKey(), ToRuleEngineNotificationMsg.parseFrom(msg.getData()), msg.getHeaders()));
         consumerBuilder.admin(notificationAdmin);
+        consumerBuilder.statsService(consumerStatsService);
         return consumerBuilder.build();
     }
 
@@ -181,6 +187,7 @@ public class KafkaTbRuleEngineQueueFactory implements TbRuleEngineQueueFactory {
                 }
         );
         responseBuilder.admin(jsExecutorAdmin);
+        responseBuilder.statsService(consumerStatsService);
 
         DefaultTbQueueRequestTemplate.DefaultTbQueueRequestTemplateBuilder
                 <TbProtoJsQueueMsg<JsInvokeProtos.RemoteJsRequest>, TbProtoQueueMsg<JsInvokeProtos.RemoteJsResponse>> builder = DefaultTbQueueRequestTemplate.builder();

@@ -32,6 +32,7 @@ import org.thingsboard.server.queue.common.DefaultTbQueueRequestTemplate;
 import org.thingsboard.server.queue.common.TbProtoQueueMsg;
 import org.thingsboard.server.queue.discovery.TbServiceInfoProvider;
 import org.thingsboard.server.queue.kafka.TbKafkaAdmin;
+import org.thingsboard.server.queue.kafka.TbKafkaConsumerStatsService;
 import org.thingsboard.server.queue.kafka.TbKafkaConsumerTemplate;
 import org.thingsboard.server.queue.kafka.TbKafkaProducerTemplate;
 import org.thingsboard.server.queue.kafka.TbKafkaSettings;
@@ -54,6 +55,7 @@ public class KafkaTbTransportQueueFactory implements TbTransportQueueFactory {
     private final TbQueueRuleEngineSettings ruleEngineSettings;
     private final TbQueueTransportApiSettings transportApiSettings;
     private final TbQueueTransportNotificationSettings transportNotificationSettings;
+    private final TbKafkaConsumerStatsService consumerStatsService;
 
     private final TbQueueAdmin coreAdmin;
     private final TbQueueAdmin ruleEngineAdmin;
@@ -66,6 +68,7 @@ public class KafkaTbTransportQueueFactory implements TbTransportQueueFactory {
                                         TbQueueRuleEngineSettings ruleEngineSettings,
                                         TbQueueTransportApiSettings transportApiSettings,
                                         TbQueueTransportNotificationSettings transportNotificationSettings,
+                                        TbKafkaConsumerStatsService consumerStatsService,
                                         TbKafkaTopicConfigs kafkaTopicConfigs) {
         this.kafkaSettings = kafkaSettings;
         this.serviceInfoProvider = serviceInfoProvider;
@@ -73,6 +76,7 @@ public class KafkaTbTransportQueueFactory implements TbTransportQueueFactory {
         this.ruleEngineSettings = ruleEngineSettings;
         this.transportApiSettings = transportApiSettings;
         this.transportNotificationSettings = transportNotificationSettings;
+        this.consumerStatsService = consumerStatsService;
 
         this.coreAdmin = new TbKafkaAdmin(kafkaSettings, kafkaTopicConfigs.getCoreConfigs());
         this.ruleEngineAdmin = new TbKafkaAdmin(kafkaSettings, kafkaTopicConfigs.getRuleEngineConfigs());
@@ -95,6 +99,7 @@ public class KafkaTbTransportQueueFactory implements TbTransportQueueFactory {
         responseBuilder.groupId("transport-node-" + serviceInfoProvider.getServiceId());
         responseBuilder.decoder(msg -> new TbProtoQueueMsg<>(msg.getKey(), TransportApiResponseMsg.parseFrom(msg.getData()), msg.getHeaders()));
         responseBuilder.admin(transportApiAdmin);
+        responseBuilder.statsService(consumerStatsService);
 
         DefaultTbQueueRequestTemplate.DefaultTbQueueRequestTemplateBuilder
                 <TbProtoQueueMsg<TransportApiRequestMsg>, TbProtoQueueMsg<TransportApiResponseMsg>> templateBuilder = DefaultTbQueueRequestTemplate.builder();
@@ -136,6 +141,7 @@ public class KafkaTbTransportQueueFactory implements TbTransportQueueFactory {
         responseBuilder.groupId("transport-node-" + serviceInfoProvider.getServiceId());
         responseBuilder.decoder(msg -> new TbProtoQueueMsg<>(msg.getKey(), ToTransportMsg.parseFrom(msg.getData()), msg.getHeaders()));
         responseBuilder.admin(notificationAdmin);
+        responseBuilder.statsService(consumerStatsService);
         return responseBuilder.build();
     }
 

@@ -51,6 +51,7 @@ import org.thingsboard.server.gen.transport.TransportProtos.ValidateBasicMqttCre
 import org.thingsboard.server.gen.transport.TransportProtos.ValidateDeviceTokenRequestMsg;
 import org.thingsboard.server.gen.transport.TransportProtos.ValidateDeviceX509CertRequestMsg;
 
+import java.math.BigDecimal;
 import java.util.ArrayList;
 import java.util.HashMap;
 import java.util.HashSet;
@@ -244,12 +245,25 @@ public class JsonConverter {
     }
 
     private static void parseNumericValue(List<KvEntry> result, Entry<String, JsonElement> valueEntry, JsonPrimitive value) {
-        if (value.getAsString().contains(".")) {
-            result.add(new DoubleDataEntry(valueEntry.getKey(), value.getAsDouble()));
+        String valueAsString = value.getAsString();
+        String key = valueEntry.getKey();
+        if (valueAsString.contains("e") || valueAsString.contains("E")) {
+            var bd = new BigDecimal(valueAsString);
+            if (bd.stripTrailingZeros().scale() <= 0) {
+                try {
+                    result.add(new LongDataEntry(key, bd.longValueExact()));
+                } catch (ArithmeticException e) {
+                    result.add(new DoubleDataEntry(key, bd.doubleValue()));
+                }
+            } else {
+                result.add(new DoubleDataEntry(key, bd.doubleValue()));
+            }
+        } else if (valueAsString.contains(".")) {
+            result.add(new DoubleDataEntry(key, value.getAsDouble()));
         } else {
             try {
                 long longValue = Long.parseLong(value.getAsString());
-                result.add(new LongDataEntry(valueEntry.getKey(), longValue));
+                result.add(new LongDataEntry(key, longValue));
             } catch (NumberFormatException e) {
                 throw new JsonSyntaxException("Big integer values are not supported!");
             }
@@ -284,7 +298,8 @@ public class JsonConverter {
         return result;
     }
 
-    public static JsonObject getJsonObjectForGateway(String deviceName, TransportProtos.GetAttributeResponseMsg responseMsg) {
+    public static JsonObject getJsonObjectForGateway(String deviceName, TransportProtos.GetAttributeResponseMsg
+            responseMsg) {
         JsonObject result = new JsonObject();
         result.addProperty("id", responseMsg.getRequestId());
         result.addProperty(DEVICE_PROPERTY, deviceName);
@@ -297,7 +312,8 @@ public class JsonConverter {
         return result;
     }
 
-    public static JsonObject getJsonObjectForGateway(String deviceName, AttributeUpdateNotificationMsg notificationMsg) {
+    public static JsonObject getJsonObjectForGateway(String deviceName, AttributeUpdateNotificationMsg
+            notificationMsg) {
         JsonObject result = new JsonObject();
         result.addProperty(DEVICE_PROPERTY, deviceName);
         result.add("data", toJson(notificationMsg));
@@ -447,7 +463,8 @@ public class JsonConverter {
         return result;
     }
 
-    public static JsonElement toGatewayJson(String deviceName, TransportProtos.ProvisionDeviceResponseMsg responseRequest) {
+    public static JsonElement toGatewayJson(String deviceName, TransportProtos.ProvisionDeviceResponseMsg
+            responseRequest) {
         JsonObject result = new JsonObject();
         result.addProperty(DEVICE_PROPERTY, deviceName);
         result.add("data", JsonConverter.toJson(responseRequest));
@@ -497,15 +514,18 @@ public class JsonConverter {
         return result;
     }
 
-    public static Map<Long, List<KvEntry>> convertToTelemetry(JsonElement jsonElement, long systemTs) throws JsonSyntaxException {
+    public static Map<Long, List<KvEntry>> convertToTelemetry(JsonElement jsonElement, long systemTs) throws
+            JsonSyntaxException {
         return convertToTelemetry(jsonElement, systemTs, false);
     }
 
-    public static Map<Long, List<KvEntry>> convertToSortedTelemetry(JsonElement jsonElement, long systemTs) throws JsonSyntaxException {
+    public static Map<Long, List<KvEntry>> convertToSortedTelemetry(JsonElement jsonElement, long systemTs) throws
+            JsonSyntaxException {
         return convertToTelemetry(jsonElement, systemTs, true);
     }
 
-    public static Map<Long, List<KvEntry>> convertToTelemetry(JsonElement jsonElement, long systemTs, boolean sorted) throws JsonSyntaxException {
+    public static Map<Long, List<KvEntry>> convertToTelemetry(JsonElement jsonElement, long systemTs, boolean sorted) throws
+            JsonSyntaxException {
         Map<Long, List<KvEntry>> result = sorted ? new TreeMap<>() : new HashMap<>();
         convertToTelemetry(jsonElement, systemTs, result, null);
         return result;
@@ -575,7 +595,8 @@ public class JsonConverter {
                 .build();
     }
 
-    private static TransportProtos.ProvisionDeviceCredentialsMsg buildProvisionDeviceCredentialsMsg(String provisionKey, String provisionSecret) {
+    private static TransportProtos.ProvisionDeviceCredentialsMsg buildProvisionDeviceCredentialsMsg(String
+                                                                                                            provisionKey, String provisionSecret) {
         return TransportProtos.ProvisionDeviceCredentialsMsg.newBuilder()
                 .setProvisionDeviceKey(provisionKey)
                 .setProvisionDeviceSecret(provisionSecret)

@@ -784,8 +784,8 @@ public class DefaultTransportService implements TransportService {
                 wrappedCallback);
     }
 
-    protected void sendToRuleEngine(TenantId tenantId, TbMsg tbMsg, TbQueueCallback callback) {
-        TopicPartitionInfo tpi = partitionService.resolve(ServiceType.TB_RULE_ENGINE, tenantId, tbMsg.getOriginator());
+    private void sendToRuleEngine(TenantId tenantId, TbMsg tbMsg, TbQueueCallback callback) {
+        TopicPartitionInfo tpi = partitionService.resolve(ServiceType.TB_RULE_ENGINE, tbMsg.getQueueName(), tenantId, tbMsg.getOriginator());
         if (log.isTraceEnabled()) {
             log.trace("[{}][{}] Pushing to topic {} message {}", tenantId, tbMsg.getOriginator(), tpi.getFullTopicName(), tbMsg);
         }
@@ -797,7 +797,7 @@ public class DefaultTransportService implements TransportService {
         ruleEngineMsgProducer.send(tpi, new TbProtoQueueMsg<>(tbMsg.getId(), msg), wrappedCallback);
     }
 
-    protected void sendToRuleEngine(TenantId tenantId, DeviceId deviceId, TransportProtos.SessionInfoProto sessionInfo, JsonObject json,
+    private void sendToRuleEngine(TenantId tenantId, DeviceId deviceId, TransportProtos.SessionInfoProto sessionInfo, JsonObject json,
                                     TbMsgMetaData metaData, SessionMsgType sessionMsgType, TbQueueCallback callback) {
         DeviceProfileId deviceProfileId = new DeviceProfileId(new UUID(sessionInfo.getDeviceProfileIdMSB(), sessionInfo.getDeviceProfileIdLSB()));
         DeviceProfile deviceProfile = deviceProfileCache.get(deviceProfileId);

+/**
+ * Copyright © 2016-2021 The Thingsboard Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+import com.google.gson.JsonParser;
+import org.junit.Assert;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.mockito.junit.MockitoJUnitRunner;
+import org.thingsboard.server.common.transport.adaptor.JsonConverter;
+
+@RunWith(MockitoJUnitRunner.class)
+public class JsonConverterTest {
+
+    private static final JsonParser JSON_PARSER = new JsonParser();
+
+    @Test
+    public void testParseBigDecimalAsLong() {
+        var result = JsonConverter.convertToTelemetry(JSON_PARSER.parse("{\"meterReadingDelta\": 1E+1}"), 0L);
+        Assert.assertEquals(10L, result.get(0L).get(0).getLongValue().get().longValue());
+    }
+
+    @Test
+    public void testParseBigDecimalAsDouble() {
+        var result = JsonConverter.convertToTelemetry(JSON_PARSER.parse("{\"meterReadingDelta\": 101E-1}"), 0L);
+        Assert.assertEquals(10.1, result.get(0L).get(0).getDoubleValue().get(), 0.0);
+    }
+
+    @Test
+    public void testParseAsDouble() {
+        var result = JsonConverter.convertToTelemetry(JSON_PARSER.parse("{\"meterReadingDelta\": 1.1}"), 0L);
+        Assert.assertEquals(1.1, result.get(0L).get(0).getDoubleValue().get(), 0.0);
+    }
+
+    @Test
+    public void testParseAsLong() {
+        var result = JsonConverter.convertToTelemetry(JSON_PARSER.parse("{\"meterReadingDelta\": 11}"), 0L);
+        Assert.assertEquals(11L, result.get(0L).get(0).getLongValue().get().longValue());
+    }
+
+}

@@ -76,7 +76,7 @@ public class TbCopyAttributesToEntityViewNode implements TbNode {
             if (!msg.getMetaData().getData().isEmpty()) {
                 long now = System.currentTimeMillis();
                 String scope = msg.getType().equals(SessionMsgType.POST_ATTRIBUTES_REQUEST.name()) ?
-                        DataConstants.CLIENT_SCOPE : msg.getMetaData().getValue("scope");
+                        DataConstants.CLIENT_SCOPE : msg.getMetaData().getValue(DataConstants.SCOPE);
 
                 ListenableFuture<List<EntityView>> entityViewsFuture =
                         ctx.getEntityViewService().findEntityViewsByTenantIdAndEntityIdAsync(ctx.getTenantId(), msg.getOriginator());

@@ -104,9 +104,10 @@ public class TbMsgGeneratorNode implements TbNode {
                         }
                     },
                     t -> {
-                        if (initialized) {
+                        if (initialized && (config.getMsgCount() == TbMsgGeneratorNodeConfiguration.UNLIMITED_MSG_COUNT || currentMsgCount < config.getMsgCount())) {
                             ctx.tellFailure(msg, t);
                             scheduleTickMsg(ctx);
+                            currentMsgCount++;
                         }
                     });
         }

@@ -33,6 +33,7 @@ import org.thingsboard.server.common.data.id.EntityId;
 import org.thingsboard.server.common.data.kv.TsKvEntry;
 import org.thingsboard.server.common.data.plugin.ComponentType;
 import org.thingsboard.server.common.msg.TbMsg;
+import org.thingsboard.server.common.msg.session.SessionMsgType;
 import org.thingsboard.server.dao.timeseries.TimeseriesService;
 import org.thingsboard.server.dao.util.mapping.JacksonUtil;
 
@@ -44,11 +45,13 @@ import java.util.concurrent.ConcurrentHashMap;
 
 @Slf4j
 @RuleNode(type = ComponentType.ENRICHMENT,
-        name = "calculate delta",
+        name = "calculate delta", relationTypes = {"Success", "Failure", "Other"},
         configClazz = CalculateDeltaNodeConfiguration.class,
         nodeDescription = "Calculates and adds 'delta' value into message based on the incoming and previous value",
         nodeDetails = "Calculates delta and period based on the previous time-series reading and current data. " +
-                "Delta calculation is done in scope of the message originator, e.g. device, asset or customer.",
+                "Delta calculation is done in scope of the message originator, e.g. device, asset or customer. " +
+                "If there is input key, the output relation will be 'Success' unless delta is negative and corresponding configuration parameter is set. " +
+                "If there is no input value key in the incoming message, the output relation will be 'Other'.",
         uiResources = {"static/rulenode/rulenode-core-config.js"},
         configDirective = "tbEnrichmentNodeCalculateDeltaConfig")
 public class CalculateDeltaNode implements TbNode {
@@ -72,43 +75,50 @@ public class CalculateDeltaNode implements TbNode {
 
     @Override
     public void onMsg(TbContext ctx, TbMsg msg) {
-        JsonNode json = JacksonUtil.toJsonNode(msg.getData());
-        String inputKey = config.getInputValueKey();
-        if (json.has(inputKey)) {
-            DonAsynchron.withCallback(getLastValue(msg.getOriginator()),
-                    previousData -> {
-                        double currentValue = json.get(inputKey).asDouble();
-                        long currentTs = TbMsgTimeseriesNode.getTs(msg);
-
-                        if (useCache) {
-                            cache.put(msg.getOriginator(), new ValueWithTs(currentTs, currentValue));
-                        }
-
-                        BigDecimal delta = BigDecimal.valueOf(previousData != null ? currentValue - previousData.value : 0.0);
-
-                        if (config.isTellFailureIfDeltaIsNegative() && delta.doubleValue() < 0) {
-                            ctx.tellNext(msg, TbRelationTypes.FAILURE);
-                            return;
-                        }
-
-                        if (config.getRound() != null) {
-                            delta = delta.setScale(config.getRound(), RoundingMode.HALF_UP);
-                        }
-
-                        ObjectNode result = (ObjectNode) json;
-                        result.put(config.getOutputValueKey(), delta);
-
-                        if (config.isAddPeriodBetweenMsgs()) {
-                            long period = previousData != null ? currentTs - previousData.ts : 0;
-                            result.put(config.getPeriodValueKey(), period);
-                        }
-                        ctx.tellSuccess(TbMsg.transformMsg(msg, msg.getType(), msg.getOriginator(), msg.getMetaData(), JacksonUtil.toString(result)));
-                    },
-                    t -> ctx.tellFailure(msg, t), ctx.getDbCallbackExecutor());
-        } else if (config.isTellFailureIfInputValueKeyIsAbsent()) {
-            ctx.tellNext(msg, TbRelationTypes.FAILURE);
+        if (msg.getType().equals(SessionMsgType.POST_TELEMETRY_REQUEST.name())) {
+            JsonNode json = JacksonUtil.toJsonNode(msg.getData());
+            String inputKey = config.getInputValueKey();
+            if (json.has(inputKey)) {
+                DonAsynchron.withCallback(getLastValue(msg.getOriginator()),
+                        previousData -> {
+                            double currentValue = json.get(inputKey).asDouble();
+                            long currentTs = TbMsgTimeseriesNode.getTs(msg);
+
+                            if (useCache) {
+                                cache.put(msg.getOriginator(), new ValueWithTs(currentTs, currentValue));
+                            }
+
+                            BigDecimal delta = BigDecimal.valueOf(previousData != null ? currentValue - previousData.value : 0.0);
+
+                            if (config.isTellFailureIfDeltaIsNegative() && delta.doubleValue() < 0) {
+                                ctx.tellNext(msg, TbRelationTypes.FAILURE);
+                                return;
+                            }
+
+
+                            if (config.getRound() != null) {
+                                delta = delta.setScale(config.getRound(), RoundingMode.HALF_UP);
+                            }
+
+                            ObjectNode result = (ObjectNode) json;
+                            if (delta.stripTrailingZeros().scale() > 0) {
+                                result.put(config.getOutputValueKey(), delta.doubleValue());
+                            } else {
+                                result.put(config.getOutputValueKey(), delta.longValueExact());
+                            }
+
+                            if (config.isAddPeriodBetweenMsgs()) {
+                                long period = previousData != null ? currentTs - previousData.ts : 0;
+                                result.put(config.getPeriodValueKey(), period);
+                            }
+                            ctx.tellSuccess(TbMsg.transformMsg(msg, msg.getType(), msg.getOriginator(), msg.getMetaData(), JacksonUtil.toString(result)));
+                        },
+                        t -> ctx.tellFailure(msg, t), ctx.getDbCallbackExecutor());
+            } else {
+                ctx.tellNext(msg, "Other");
+            }
         } else {
-            ctx.tellSuccess(msg);
+            ctx.tellNext(msg, "Other");
         }
     }
 

@@ -15,10 +15,12 @@
  */
 package org.thingsboard.rule.engine.metadata;
 
+import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
 import lombok.Data;
 import org.thingsboard.rule.engine.api.NodeConfiguration;
 
 @Data
+@JsonIgnoreProperties(ignoreUnknown = true)
 public class CalculateDeltaNodeConfiguration implements NodeConfiguration<CalculateDeltaNodeConfiguration> {
     private String inputValueKey;
     private String outputValueKey;
@@ -26,7 +28,6 @@ public class CalculateDeltaNodeConfiguration implements NodeConfiguration<Calcul
     private boolean addPeriodBetweenMsgs;
     private String periodValueKey;
     private Integer round;
-    private boolean tellFailureIfInputValueKeyIsAbsent;
     private boolean tellFailureIfDeltaIsNegative;
 
     @Override
@@ -37,7 +38,6 @@ public class CalculateDeltaNodeConfiguration implements NodeConfiguration<Calcul
         configuration.setUseCache(true);
         configuration.setAddPeriodBetweenMsgs(false);
         configuration.setPeriodValueKey("periodInMs");
-        configuration.setTellFailureIfInputValueKeyIsAbsent(true);
         configuration.setTellFailureIfDeltaIsNegative(true);
         return configuration;
     }

@@ -388,12 +388,6 @@ class AlarmRuleState {
         EntityKeyValue ekv = null;
         if (value.getDynamicValue() != null) {
             switch (value.getDynamicValue().getSourceType()) {
-                case CURRENT_TENANT:
-                    ekv = dynamicPredicateValueCtx.getTenantValue(value.getDynamicValue().getSourceAttribute());
-                    break;
-                case CURRENT_CUSTOMER:
-                    ekv = dynamicPredicateValueCtx.getCustomerValue(value.getDynamicValue().getSourceAttribute());
-                    break;
                 case CURRENT_DEVICE:
                     ekv = data.getValue(new EntityKey(EntityKeyType.ATTRIBUTE, value.getDynamicValue().getSourceAttribute()));
                     if (ekv == null) {
@@ -405,6 +399,16 @@ class AlarmRuleState {
                             }
                         }
                     }
+                    if(ekv != null || !value.getDynamicValue().isInherit()) {
+                        break;
+                    }
+                case CURRENT_CUSTOMER:
+                    ekv = dynamicPredicateValueCtx.getCustomerValue(value.getDynamicValue().getSourceAttribute());
+                    if(ekv != null || !value.getDynamicValue().isInherit()) {
+                       break;
+                    }
+                case CURRENT_TENANT:
+                    ekv = dynamicPredicateValueCtx.getTenantValue(value.getDynamicValue().getSourceAttribute());
             }
         }
         return ekv;

@@ -138,6 +138,8 @@ class DeviceState {
             stateChanged = processTelemetry(ctx, msg);
         } else if (msg.getType().equals(SessionMsgType.POST_ATTRIBUTES_REQUEST.name())) {
             stateChanged = processAttributesUpdateRequest(ctx, msg);
+        } else if (msg.getType().equals(DataConstants.ACTIVITY_EVENT) || msg.getType().equals(DataConstants.INACTIVITY_EVENT)) {
+            stateChanged = processDeviceActivityEvent(ctx, msg);
         } else if (msg.getType().equals(DataConstants.ATTRIBUTES_UPDATED)) {
             stateChanged = processAttributesUpdateNotification(ctx, msg);
         } else if (msg.getType().equals(DataConstants.ATTRIBUTES_DELETED)) {
@@ -158,6 +160,15 @@ class DeviceState {
         }
     }
 
+    private boolean processDeviceActivityEvent(TbContext ctx, TbMsg msg) throws ExecutionException, InterruptedException {
+        String scope = msg.getMetaData().getValue(DataConstants.SCOPE);
+        if (StringUtils.isEmpty(scope)) {
+            return processTelemetry(ctx, msg);
+        } else {
+            return processAttributes(ctx, msg, scope);
+        }
+    }
+
     private boolean processAlarmClearNotification(TbContext ctx, TbMsg msg) {
         boolean stateChanged = false;
         Alarm alarmNf = JacksonUtil.fromString(msg.getData(), Alarm.class);
@@ -181,19 +192,18 @@ class DeviceState {
     }
 
     private boolean processAttributesUpdateNotification(TbContext ctx, TbMsg msg) throws ExecutionException, InterruptedException {
-        Set<AttributeKvEntry> attributes = JsonConverter.convertToAttributes(new JsonParser().parse(msg.getData()));
-        String scope = msg.getMetaData().getValue("scope");
+        String scope = msg.getMetaData().getValue(DataConstants.SCOPE);
         if (StringUtils.isEmpty(scope)) {
             scope = DataConstants.CLIENT_SCOPE;
         }
-        return processAttributesUpdate(ctx, msg, attributes, scope);
+        return processAttributes(ctx, msg, scope);
     }
 
     private boolean processAttributesDeleteNotification(TbContext ctx, TbMsg msg) throws ExecutionException, InterruptedException {
         boolean stateChanged = false;
         List<String> keys = new ArrayList<>();
         new JsonParser().parse(msg.getData()).getAsJsonObject().get("attributes").getAsJsonArray().forEach(e -> keys.add(e.getAsString()));
-        String scope = msg.getMetaData().getValue("scope");
+        String scope = msg.getMetaData().getValue(DataConstants.SCOPE);
         if (StringUtils.isEmpty(scope)) {
             scope = DataConstants.CLIENT_SCOPE;
         }
@@ -211,12 +221,12 @@ class DeviceState {
     }
 
     protected boolean processAttributesUpdateRequest(TbContext ctx, TbMsg msg) throws ExecutionException, InterruptedException {
-        Set<AttributeKvEntry> attributes = JsonConverter.convertToAttributes(new JsonParser().parse(msg.getData()));
-        return processAttributesUpdate(ctx, msg, attributes, DataConstants.CLIENT_SCOPE);
+        return processAttributes(ctx, msg, DataConstants.CLIENT_SCOPE);
     }
 
-    private boolean processAttributesUpdate(TbContext ctx, TbMsg msg, Set<AttributeKvEntry> attributes, String scope) throws ExecutionException, InterruptedException {
+    private boolean processAttributes(TbContext ctx, TbMsg msg, String scope) throws ExecutionException, InterruptedException {
         boolean stateChanged = false;
+        Set<AttributeKvEntry> attributes = JsonConverter.convertToAttributes(new JsonParser().parse(msg.getData()));
         if (!attributes.isEmpty()) {
             SnapshotUpdate update = merge(latestValues, attributes, scope);
             for (DeviceProfileAlarm alarm : deviceProfile.getAlarmSettings()) {

@@ -134,7 +134,8 @@ public class TbDeviceProfileNode implements TbNode {
                     if (deviceState != null) {
                         deviceState.process(ctx, msg);
                     } else {
-                        ctx.tellFailure(msg, new IllegalStateException("Device profile for device [" + deviceId + "] not found!"));
+                        log.info("Device was not found! Most probably device [" + deviceId + "] has been removed from the database. Acknowledging msg.");
+                        ctx.ack(msg);
                     }
                 }
             } else {

@@ -434,11 +434,152 @@ public class TbDeviceProfileNodeTest {
         verify(ctx, Mockito.never()).tellFailure(Mockito.any(), Mockito.any());
     }
 
-    private void init() throws TbNodeException {
+    @Test
+    public void testTenantInheritModeForDynamicValues() throws Exception {
+        init();
+
+        DeviceProfile deviceProfile = new DeviceProfile();
+        DeviceProfileData deviceProfileData = new DeviceProfileData();
+
+        AttributeKvCompositeKey compositeKey = new AttributeKvCompositeKey(
+                EntityType.TENANT, deviceId.getId(), "SERVER_SCOPE", "tenantAttribute"
+        );
+
+        AttributeKvEntity attributeKvEntity = new AttributeKvEntity();
+        attributeKvEntity.setId(compositeKey);
+        attributeKvEntity.setLongValue(100L);
+        attributeKvEntity.setLastUpdateTs(0L);
+
+        AttributeKvEntry entry = attributeKvEntity.toData();
+        ListenableFuture<List<AttributeKvEntry>> listListenableFutureWithLess =
+                Futures.immediateFuture(Collections.singletonList(entry));
+
+        KeyFilter lowTempFilter = new KeyFilter();
+        lowTempFilter.setKey(new EntityKey(EntityKeyType.TIME_SERIES, "temperature"));
+        lowTempFilter.setValueType(EntityKeyValueType.NUMERIC);
+        NumericFilterPredicate lowTempPredicate = new NumericFilterPredicate();
+        lowTempPredicate.setOperation(NumericFilterPredicate.NumericOperation.GREATER);
+        lowTempPredicate.setValue(
+                new FilterPredicateValue<>(
+                        0.0,
+                        null,
+                        new DynamicValue<>(DynamicValueSourceType.CURRENT_DEVICE, "tenantAttribute", true))
+        );
+        lowTempFilter.setPredicate(lowTempPredicate);
+        AlarmCondition alarmCondition = new AlarmCondition();
+        alarmCondition.setCondition(Collections.singletonList(lowTempFilter));
+        AlarmRule alarmRule = new AlarmRule();
+        alarmRule.setCondition(alarmCondition);
+        DeviceProfileAlarm dpa = new DeviceProfileAlarm();
+        dpa.setId("lesstempID");
+        dpa.setAlarmType("lessTemperatureAlarm");
+        dpa.setCreateRules(new TreeMap<>(Collections.singletonMap(AlarmSeverity.CRITICAL, alarmRule)));
+
+        deviceProfileData.setAlarms(Collections.singletonList(dpa));
+        deviceProfile.setProfileData(deviceProfileData);
 
-        UUID uuid = new UUID(6041557255264276971L, -9019477126543226049L);
-        System.out.println(uuid);
+        Mockito.when(cache.get(tenantId, deviceId)).thenReturn(deviceProfile);
+        Mockito.when(timeseriesService.findLatest(tenantId, deviceId, Collections.singleton("temperature")))
+                .thenReturn(Futures.immediateFuture(Collections.emptyList()));
+        Mockito.when(alarmService.findLatestByOriginatorAndType(tenantId, deviceId, "lessTemperatureAlarm"))
+                .thenReturn(Futures.immediateFuture(null));
+        Mockito.when(alarmService.createOrUpdateAlarm(Mockito.any()))
+                .thenAnswer(AdditionalAnswers.returnsFirstArg());
+        Mockito.when(ctx.getAttributesService()).thenReturn(attributesService);
+        Mockito.when(attributesService.find(eq(tenantId), eq(deviceId), Mockito.anyString(), Mockito.anySet()))
+                .thenReturn(listListenableFutureWithLess);
+
+        TbMsg theMsg = TbMsg.newMsg("ALARM", deviceId, new TbMsgMetaData(), "");
+        Mockito.when(ctx.newMsg(Mockito.anyString(), Mockito.anyString(), Mockito.any(), Mockito.any(), Mockito.anyString()))
+                .thenReturn(theMsg);
+
+        ObjectNode data = mapper.createObjectNode();
+        data.put("temperature", 150L);
+        TbMsg msg = TbMsg.newMsg(SessionMsgType.POST_TELEMETRY_REQUEST.name(), deviceId, new TbMsgMetaData(),
+                TbMsgDataType.JSON, mapper.writeValueAsString(data), null, null);
+
+        node.onMsg(ctx, msg);
+        verify(ctx).tellSuccess(msg);
+        verify(ctx).tellNext(theMsg, "Alarm Created");
+        verify(ctx, Mockito.never()).tellFailure(Mockito.any(), Mockito.any());
+
+    }
+
+    @Test
+    public void testCustomerInheritModeForDynamicValues() throws Exception {
+        init();
 
+        DeviceProfile deviceProfile = new DeviceProfile();
+        DeviceProfileData deviceProfileData = new DeviceProfileData();
+
+        AttributeKvCompositeKey compositeKey = new AttributeKvCompositeKey(
+                EntityType.TENANT, deviceId.getId(), "SERVER_SCOPE", "customerAttribute"
+        );
+
+        AttributeKvEntity attributeKvEntity = new AttributeKvEntity();
+        attributeKvEntity.setId(compositeKey);
+        attributeKvEntity.setLongValue(100L);
+        attributeKvEntity.setLastUpdateTs(0L);
+
+        AttributeKvEntry entry = attributeKvEntity.toData();
+        ListenableFuture<List<AttributeKvEntry>> listListenableFutureWithLess =
+                Futures.immediateFuture(Collections.singletonList(entry));
+        ListenableFuture<Optional<AttributeKvEntry>> optionalListenableFutureWithLess =
+                Futures.immediateFuture(Optional.of(entry));
+
+        KeyFilter lowTempFilter = new KeyFilter();
+        lowTempFilter.setKey(new EntityKey(EntityKeyType.TIME_SERIES, "temperature"));
+        lowTempFilter.setValueType(EntityKeyValueType.NUMERIC);
+        NumericFilterPredicate lowTempPredicate = new NumericFilterPredicate();
+        lowTempPredicate.setOperation(NumericFilterPredicate.NumericOperation.GREATER);
+        lowTempPredicate.setValue(
+                new FilterPredicateValue<>(
+                        0.0,
+                        null,
+                        new DynamicValue<>(DynamicValueSourceType.CURRENT_CUSTOMER, "customerAttribute", true))
+        );
+        lowTempFilter.setPredicate(lowTempPredicate);
+        AlarmCondition alarmCondition = new AlarmCondition();
+        alarmCondition.setCondition(Collections.singletonList(lowTempFilter));
+        AlarmRule alarmRule = new AlarmRule();
+        alarmRule.setCondition(alarmCondition);
+        DeviceProfileAlarm dpa = new DeviceProfileAlarm();
+        dpa.setId("lesstempID");
+        dpa.setAlarmType("lessTemperatureAlarm");
+        dpa.setCreateRules(new TreeMap<>(Collections.singletonMap(AlarmSeverity.CRITICAL, alarmRule)));
+
+        deviceProfileData.setAlarms(Collections.singletonList(dpa));
+        deviceProfile.setProfileData(deviceProfileData);
+
+        Mockito.when(cache.get(tenantId, deviceId)).thenReturn(deviceProfile);
+        Mockito.when(timeseriesService.findLatest(tenantId, deviceId, Collections.singleton("temperature")))
+                .thenReturn(Futures.immediateFuture(Collections.emptyList()));
+        Mockito.when(alarmService.findLatestByOriginatorAndType(tenantId, deviceId, "lessTemperatureAlarm"))
+                .thenReturn(Futures.immediateFuture(null));
+        Mockito.when(alarmService.createOrUpdateAlarm(Mockito.any()))
+                .thenAnswer(AdditionalAnswers.returnsFirstArg());
+        Mockito.when(ctx.getAttributesService()).thenReturn(attributesService);
+        Mockito.when(attributesService.find(eq(tenantId), eq(deviceId), Mockito.anyString(), Mockito.anySet()))
+                .thenReturn(listListenableFutureWithLess);
+        Mockito.when(attributesService.find(eq(tenantId), eq(tenantId), eq(DataConstants.SERVER_SCOPE),  Mockito.anyString()))
+                .thenReturn(optionalListenableFutureWithLess);
+
+        TbMsg theMsg = TbMsg.newMsg("ALARM", deviceId, new TbMsgMetaData(), "");
+        Mockito.when(ctx.newMsg(Mockito.anyString(), Mockito.anyString(), Mockito.any(), Mockito.any(), Mockito.anyString()))
+                .thenReturn(theMsg);
+
+        ObjectNode data = mapper.createObjectNode();
+        data.put("temperature", 150L);
+        TbMsg msg = TbMsg.newMsg(SessionMsgType.POST_TELEMETRY_REQUEST.name(), deviceId, new TbMsgMetaData(),
+                TbMsgDataType.JSON, mapper.writeValueAsString(data), null, null);
+
+        node.onMsg(ctx, msg);
+        verify(ctx).tellSuccess(msg);
+        verify(ctx).tellNext(theMsg, "Alarm Created");
+        verify(ctx, Mockito.never()).tellFailure(Mockito.any(), Mockito.any());
+    }
+
+    private void init() throws TbNodeException {
         Mockito.when(ctx.getTenantId()).thenReturn(tenantId);
         Mockito.when(ctx.getDeviceProfileCache()).thenReturn(cache);
         Mockito.when(ctx.getTimeseriesService()).thenReturn(timeseriesService);

@@ -43,6 +43,7 @@ import {
   AlarmDetailsDialogComponent,
   AlarmDetailsDialogData
 } from '@home/components/alarm/alarm-details-dialog.component';
+import { DAY, historyInterval } from '@shared/models/time/time.models';
 
 export class AlarmTableConfig extends EntityTableConfig<AlarmInfo, TimePageLink> {
 
@@ -59,6 +60,7 @@ export class AlarmTableConfig extends EntityTableConfig<AlarmInfo, TimePageLink>
     this.loadDataOnInit = false;
     this.tableTitle = '';
     this.useTimePageLink = true;
+    this.defaultTimewindowInterval = historyInterval(DAY * 30);
     this.detailsPanelEnabled = false;
     this.selectionEnabled = false;
     this.searchEnabled = true;

@@ -55,11 +55,11 @@ import { EntityTypeTranslation } from '@shared/models/entity-type.models';
 import { DialogService } from '@core/services/dialog.service';
 import { AddEntityDialogComponent } from './add-entity-dialog.component';
 import { AddEntityDialogData, EntityAction } from '@home/models/entity/entity-component.models';
-import { DAY, historyInterval, HistoryWindowType, Timewindow } from '@shared/models/time/time.models';
+import { HistoryWindowType, Timewindow } from '@shared/models/time/time.models';
 import { DomSanitizer, SafeHtml } from '@angular/platform-browser';
 import { TbAnchorComponent } from '@shared/components/tb-anchor.component';
 import { isDefined, isUndefined } from '@core/utils';
-import { HasUUID } from '../../../../shared/models/id/has-uuid';
+import { HasUUID } from '@shared/models/id/has-uuid';
 
 @Component({
   selector: 'tb-entities-table',
@@ -202,7 +202,7 @@ export class EntitiesTableComponent extends PageComponent implements AfterViewIn
     this.pageSizeOptions = [this.defaultPageSize, this.defaultPageSize * 2, this.defaultPageSize * 3];
 
     if (this.entitiesTableConfig.useTimePageLink) {
-      this.timewindow = historyInterval(DAY);
+      this.timewindow = this.entitiesTableConfig.defaultTimewindowInterval;
       const currentTime = Date.now();
       this.pageLink = new TimePageLink(10, 0, null, sortOrder,
         currentTime - this.timewindow.history.timewindowMs, currentTime);
@@ -446,7 +446,7 @@ export class EntitiesTableComponent extends PageComponent implements AfterViewIn
   resetSortAndFilter(update: boolean = true, preserveTimewindow: boolean = false) {
     this.pageLink.textSearch = null;
     if (this.entitiesTableConfig.useTimePageLink && !preserveTimewindow) {
-      this.timewindow = historyInterval(DAY);
+      this.timewindow = this.entitiesTableConfig.defaultTimewindowInterval;
     }
     if (this.displayPagination) {
       this.paginator.pageIndex = 0;

@@ -16,7 +16,7 @@
 
 -->
 <div fxFlex fxLayout="row" fxLayoutAlign="start start" fxLayoutGap="8px" [formGroup]="booleanFilterPredicateFormGroup">
-  <mat-form-field floatLabel="always" hideRequiredMarker fxFlex="40" class="mat-block">
+  <mat-form-field floatLabel="always" hideRequiredMarker fxFlex="30" class="mat-block">
     <mat-label></mat-label>
     <mat-select required formControlName="operation" placeholder="{{'filter.operation.operation' | translate}}">
       <mat-option *ngFor="let operation of booleanOperations" [value]="operation">
@@ -25,7 +25,7 @@
     </mat-select>
   </mat-form-field>
   <tb-filter-predicate-value [allowUserDynamicSource]="allowUserDynamicSource"
-                             fxFlex="60"
+                             fxFlex="70"
                              [valueType]="valueTypeEnum.BOOLEAN"
                              formControlName="value">
   </tb-filter-predicate-value>

@@ -26,12 +26,12 @@
       <span fxFlex="8"></span>
       <div fxLayout="row" fxLayoutAlign="start center" fxLayoutGap="8px" fxFlex="92">
         <div fxFlex fxLayout="row" fxLayoutGap="8px">
-          <div fxFlex="40" fxLayout="row" fxLayoutAlign="start center" fxLayoutGap="8px">
+          <div fxFlex="30" fxLayout="row" fxLayoutAlign="start center" fxLayoutGap="8px">
             <label fxFlex translate class="tb-title no-padding">filter.operation.operation</label>
             <label *ngIf="valueType === valueTypeEnum.STRING"
                    translate class="tb-title no-padding" style="min-width: 70px;">filter.ignore-case</label>
           </div>
-          <label fxFlex="60" translate class="tb-title no-padding">filter.value</label>
+          <label fxFlex="70" translate class="tb-title no-padding">filter.value</label>
         </div>
         <label *ngIf="displayUserParameters"
                translate class="tb-title no-padding" style="width: 60px;">filter.user-parameters</label>

@@ -47,7 +47,7 @@
   </div>
   <div fxFlex fxLayout="column" [fxShow]="dynamicMode">
     <div formGroupName="dynamicValue" fxLayout="row" fxLayoutAlign="start center" fxLayoutGap="8px">
-      <div fxFlex fxLayout="column">
+      <div fxFlex="35" fxLayout="column">
         <mat-form-field floatLabel="always" hideRequiredMarker class="mat-block">
           <mat-label></mat-label>
           <mat-select formControlName="sourceType" placeholder="{{'filter.dynamic-source-type' | translate}}">
@@ -68,6 +68,14 @@
         </mat-form-field>
         <div class="tb-hint" translate>filter.source-attribute</div>
       </div>
+      <div *ngIf="!allow && inheritMode"
+           fxLayout="column"
+           style="padding-top: 6px">
+        <mat-checkbox formControlName="inherit">
+          {{ 'filter.inherit-owner' | translate}}
+        </mat-checkbox>
+        <div class="tb-hint" translate>filter.source-attribute-not-set</div>
+      </div>
     </div>
   </div>
   <button mat-icon-button

@@ -44,6 +44,10 @@ import {
 })
 export class FilterPredicateValueComponent implements ControlValueAccessor, OnInit {
 
+  private readonly inheritModeForSources: DynamicValueSourceType[] = [
+    DynamicValueSourceType.CURRENT_CUSTOMER,
+    DynamicValueSourceType.CURRENT_DEVICE];
+
   @Input() disabled: boolean;
 
   @Input()
@@ -72,6 +76,8 @@ export class FilterPredicateValueComponent implements ControlValueAccessor, OnIn
 
   dynamicMode = false;
 
+  inheritMode = false;
+
   allow = true;
 
   private propagateChange = null;
@@ -105,7 +111,8 @@ export class FilterPredicateValueComponent implements ControlValueAccessor, OnIn
       dynamicValue: this.fb.group(
         {
           sourceType: [null],
-          sourceAttribute: [null]
+          sourceAttribute: [null],
+          inherit: [false]
         }
       )
     });
@@ -114,6 +121,7 @@ export class FilterPredicateValueComponent implements ControlValueAccessor, OnIn
         if (!sourceType) {
           this.filterPredicateValueFormGroup.get('dynamicValue').get('sourceAttribute').patchValue(null, {emitEvent: false});
         }
+        this.updateShowInheritMode(sourceType);
       }
     );
     this.filterPredicateValueFormGroup.valueChanges.subscribe(() => {
@@ -139,10 +147,13 @@ export class FilterPredicateValueComponent implements ControlValueAccessor, OnIn
 
   writeValue(predicateValue: FilterPredicateValue<string | number | boolean>): void {
     this.filterPredicateValueFormGroup.get('defaultValue').patchValue(predicateValue.defaultValue, {emitEvent: false});
-    this.filterPredicateValueFormGroup.get('dynamicValue').get('sourceType').patchValue(predicateValue.dynamicValue ?
+    this.filterPredicateValueFormGroup.get('dynamicValue.sourceType').patchValue(predicateValue.dynamicValue ?
       predicateValue.dynamicValue.sourceType : null, {emitEvent: false});
-    this.filterPredicateValueFormGroup.get('dynamicValue').get('sourceAttribute').patchValue(predicateValue.dynamicValue ?
+    this.filterPredicateValueFormGroup.get('dynamicValue.sourceAttribute').patchValue(predicateValue.dynamicValue ?
       predicateValue.dynamicValue.sourceAttribute : null, {emitEvent: false});
+    this.filterPredicateValueFormGroup.get('dynamicValue.inherit').patchValue(predicateValue.dynamicValue ?
+      predicateValue.dynamicValue.inherit : false, {emitEvent: false});
+    this.updateShowInheritMode(predicateValue?.dynamicValue?.sourceType);
   }
 
   private updateModel() {
@@ -158,4 +169,12 @@ export class FilterPredicateValueComponent implements ControlValueAccessor, OnIn
     this.propagateChange(predicateValue);
   }
 
+  private updateShowInheritMode(sourceType: DynamicValueSourceType) {
+    if (this.inheritModeForSources.includes(sourceType)) {
+      this.inheritMode = true;
+    } else {
+      this.filterPredicateValueFormGroup.get('dynamicValue.inherit').patchValue(false, {emitEvent: false});
+      this.inheritMode = false;
+    }
+  }
 }

@@ -16,7 +16,7 @@
 
 -->
 <div fxFlex fxLayout="row" fxLayoutAlign="start start" fxLayoutGap="8px" [formGroup]="numericFilterPredicateFormGroup">
-  <mat-form-field floatLabel="always" hideRequiredMarker fxFlex="40" class="mat-block">
+  <mat-form-field floatLabel="always" hideRequiredMarker fxFlex="30" class="mat-block">
     <mat-label></mat-label>
     <mat-select required formControlName="operation" placeholder="{{'filter.operation.operation' | translate}}">
       <mat-option *ngFor="let operation of numericOperations" [value]="operation">
@@ -25,7 +25,7 @@
     </mat-select>
   </mat-form-field>
   <tb-filter-predicate-value [allowUserDynamicSource]="allowUserDynamicSource"
-                             fxFlex="60"
+                             fxFlex="70"
                              [valueType]="valueType"
                              formControlName="value">
   </tb-filter-predicate-value>

@@ -16,7 +16,7 @@
 
 -->
 <div fxFlex fxLayout="row" fxLayoutAlign="start start" fxLayoutGap="8px" [formGroup]="stringFilterPredicateFormGroup">
-  <div fxFlex="40" fxLayout="row" fxLayoutAlign="start center" fxLayoutGap="8px">
+  <div fxFlex="30" fxLayout="row" fxLayoutAlign="start center" fxLayoutGap="8px">
     <mat-form-field floatLabel="always" hideRequiredMarker fxFlex class="mat-block">
       <mat-label></mat-label>
       <mat-select required formControlName="operation" placeholder="{{'filter.operation.operation' | translate}}">
@@ -29,7 +29,7 @@
     </mat-checkbox>
   </div>
   <tb-filter-predicate-value [allowUserDynamicSource]="allowUserDynamicSource"
-                             fxFlex="60"
+                             fxFlex="70"
                              [valueType]="valueTypeEnum.STRING"
                              formControlName="value">
   </tb-filter-predicate-value>

@@ -30,6 +30,7 @@ import { EntitiesTableComponent } from '@home/components/entity/entities-table.c
 import { EntityTableHeaderComponent } from '@home/components/entity/entity-table-header.component';
 import { ActivatedRoute } from '@angular/router';
 import { EntityTabsComponent } from '../../components/entity/entity-tabs.component';
+import { DAY, historyInterval } from '@shared/models/time/time.models';
 
 export type EntityBooleanFunction<T extends BaseData<HasId>> = (entity: T) => boolean;
 export type EntityStringFunction<T extends BaseData<HasId>> = (entity: T) => string;
@@ -135,6 +136,7 @@ export class EntityTableConfig<T extends BaseData<HasId>, P extends PageLink = P
   onLoadAction: (route: ActivatedRoute) => void = null;
   table: EntitiesTableComponent = null;
   useTimePageLink = false;
+  defaultTimewindowInterval = historyInterval(DAY);
   entityType: EntityType = null;
   tableTitle = '';
   selectionEnabled = true;
@@ -162,7 +164,7 @@ export class EntityTableConfig<T extends BaseData<HasId>, P extends PageLink = P
   dataSource: (dataLoadedFunction: (col?: number, row?: number) => void)
     => EntitiesDataSource<L> = (dataLoadedFunction: (col?: number, row?: number) => void) => {
     return new EntitiesDataSource(this.entitiesFetchFunction, this.entitySelectionEnabled, dataLoadedFunction);
-  };
+  }
   detailsReadonly: EntityBooleanFunction<T> = () => false;
   entitySelectionEnabled: EntityBooleanFunction<L> = () => true;
   deleteEnabled: EntityBooleanFunction<T | L> = () => true;

@@ -285,6 +285,7 @@ export const dynamicValueSourceTypeTranslationMap = new Map<DynamicValueSourceTy
 export interface DynamicValue<T> {
   sourceType: DynamicValueSourceType;
   sourceAttribute: string;
+  inherit?: boolean;
 }
 
 export interface FilterPredicateValue<T> {

@@ -1667,7 +1667,9 @@
          "no-dynamic-value": "No dynamic value",
          "source-attribute": "Source attribute",
          "switch-to-dynamic-value": "Switch to dynamic value",
-         "switch-to-default-value": "Switch to default value"
+         "switch-to-default-value": "Switch to default value",
+         "inherit-owner": "Inherit from owner",
+         "source-attribute-not-set": "If source attribute isn't set"
     },
     "fullscreen": {
         "expand": "Expand to fullscreen",