Merge remote-tracking branch 'upstream/feature/entity-alarm-rules' into feature/propagation-cf

application/src/main/java/org/thingsboard/server/actors/ActorSystemContext.java

@@ -664,10 +664,6 @@ public class ActorSystemContext {
     @Getter
     private long cfCalculationResultTimeout;
 
-    @Value("${actors.alarms.reevaluation_interval:60}")
-    @Getter
-    private long alarmsReevaluationInterval;
-
     @Autowired
     @Getter
     private MqttClientSettings mqttClientSettings;
@@ -895,12 +891,13 @@ public class ActorSystemContext {
         return getScheduler().scheduleWithFixedDelay(() -> ctx.tell(msg), delayInMs, periodInMs, TimeUnit.MILLISECONDS);
     }
 
-    public void scheduleMsgWithDelay(TbActorRef ctx, TbActorMsg msg, long delayInMs) {
+    public ScheduledFuture<?> scheduleMsgWithDelay(TbActorRef ctx, TbActorMsg msg, long delayInMs) {
         log.debug("Scheduling msg {} with delay {} ms", msg, delayInMs);
         if (delayInMs > 0) {
-            getScheduler().schedule(() -> ctx.tell(msg), delayInMs, TimeUnit.MILLISECONDS);
+            return getScheduler().schedule(() -> ctx.tell(msg), delayInMs, TimeUnit.MILLISECONDS);
         } else {
             ctx.tell(msg);
+            return null;
         }
     }
 

application/src/main/java/org/thingsboard/server/actors/app/AppActor.java

@@ -88,6 +88,7 @@ public class AppActor extends ContextAwareActor {
                 break;
             case PARTITION_CHANGE_MSG:
             case CF_PARTITIONS_CHANGE_MSG:
+            case CF_STATE_PARTITION_RESTORE_MSG:
                 ctx.broadcastToChildren(msg, true);
                 break;
             case COMPONENT_LIFE_CYCLE_MSG:

application/src/main/java/org/thingsboard/server/actors/calculatedField/CalculatedFieldEntityActor.java

@@ -21,6 +21,7 @@ import org.thingsboard.server.actors.TbActorCtx;
 import org.thingsboard.server.actors.TbActorException;
 import org.thingsboard.server.common.data.id.EntityId;
 import org.thingsboard.server.common.data.id.TenantId;
+import org.thingsboard.server.common.msg.CalculatedFieldStatePartitionRestoreMsg;
 import org.thingsboard.server.common.msg.TbActorStopReason;
 import org.thingsboard.server.common.msg.ToCalculatedFieldSystemMsg;
 import org.thingsboard.server.common.msg.cf.CalculatedFieldPartitionChangeMsg;
@@ -63,6 +64,9 @@ public class CalculatedFieldEntityActor extends AbstractCalculatedFieldActor {
             case CF_STATE_RESTORE_MSG:
                 processor.process((CalculatedFieldStateRestoreMsg) msg);
                 break;
+            case CF_STATE_PARTITION_RESTORE_MSG:
+                processor.process((CalculatedFieldStatePartitionRestoreMsg) msg);
+                break;
             case CF_ENTITY_INIT_CF_MSG:
                 processor.process((EntityInitCalculatedFieldMsg) msg);
                 break;

application/src/main/java/org/thingsboard/server/actors/calculatedField/CalculatedFieldEntityMessageProcessor.java

@@ -36,6 +36,7 @@ import org.thingsboard.server.common.data.id.EntityId;
 import org.thingsboard.server.common.data.id.TenantId;
 import org.thingsboard.server.common.data.kv.StringDataEntry;
 import org.thingsboard.server.common.data.msg.TbMsgType;
+import org.thingsboard.server.common.msg.CalculatedFieldStatePartitionRestoreMsg;
 import org.thingsboard.server.common.msg.cf.CalculatedFieldPartitionChangeMsg;
 import org.thingsboard.server.common.msg.queue.ServiceType;
 import org.thingsboard.server.common.msg.queue.TbCallback;
@@ -83,7 +84,7 @@ public class CalculatedFieldEntityMessageProcessor extends AbstractContextAwareM
     final CalculatedFieldProcessingService cfService;
     final CalculatedFieldStateService cfStateService;
 
-    TbActorCtx ctx;
+    TbActorCtx actorCtx;
     Map<CalculatedFieldId, CalculatedFieldState> states = new HashMap<>();
 
     CalculatedFieldEntityMessageProcessor(ActorSystemContext systemContext, TenantId tenantId, EntityId entityId) {
@@ -95,7 +96,7 @@ public class CalculatedFieldEntityMessageProcessor extends AbstractContextAwareM
     }
 
     void init(TbActorCtx ctx) {
-        this.ctx = ctx;
+        this.actorCtx = ctx;
     }
 
     public void stop(boolean partitionChanged) {
@@ -104,7 +105,7 @@ public class CalculatedFieldEntityMessageProcessor extends AbstractContextAwareM
                         "[{}][{}] Stopping entity actor.",
                 tenantId, entityId);
         states.clear();
-        ctx.stop(ctx.getSelf());
+        actorCtx.stop(actorCtx.getSelf());
     }
 
     public void process(CalculatedFieldPartitionChangeMsg msg) {
@@ -116,13 +117,25 @@ public class CalculatedFieldEntityMessageProcessor extends AbstractContextAwareM
     public void process(CalculatedFieldStateRestoreMsg msg) {
         CalculatedFieldId cfId = msg.getId().cfId();
         log.debug("[{}] [{}] Processing CF state restore msg.", msg.getId().entityId(), cfId);
-        if (msg.getState() != null) {
-            states.put(cfId, msg.getState());
+        CalculatedFieldState state = msg.getState();
+        if (state != null) {
+            state.setCtx(msg.getCtx(), actorCtx);
+            state.setPartition(msg.getPartition());
+            states.put(cfId, state);
         } else {
             states.remove(cfId);
         }
     }
 
+    public void process(CalculatedFieldStatePartitionRestoreMsg msg) {
+        log.debug("Processing CF state partition restore msg: {}", msg);
+        for (CalculatedFieldState state : states.values()) {
+            if (msg.getPartition().equals(state.getPartition())) {
+                state.init();
+            }
+        }
+    }
+
     public void process(EntityInitCalculatedFieldMsg msg) throws CalculatedFieldException {
         log.debug("[{}] Processing entity init CF msg: {}", msg.getCtx().getCfId(), msg);
         var ctx = msg.getCtx();
@@ -138,10 +151,11 @@ public class CalculatedFieldEntityMessageProcessor extends AbstractContextAwareM
                 state = createState(ctx);
             } else if (msg.getStateAction() == StateAction.REINIT) {
                 log.debug("Force reinitialization of CF: [{}].", ctx.getCfId());
-                state.reset(ctx);
+                state.reset();
                 initState(state, ctx);
             } else {
-                state.init(ctx);
+                state.setCtx(ctx, actorCtx);
+                state.init();
             }
             if (state.isSizeOk()) {
                 processStateIfReady(state, Collections.emptyMap(), ctx, Collections.singletonList(ctx.getCfId()), null, null, msg.getCallback());
@@ -183,7 +197,7 @@ public class CalculatedFieldEntityMessageProcessor extends AbstractContextAwareM
             } else {
                 MultipleTbCallback multipleTbCallback = new MultipleTbCallback(states.size(), msg.getCallback());
                 states.forEach((cfId, state) -> cfStateService.removeState(new CalculatedFieldEntityCtxId(tenantId, cfId, entityId), multipleTbCallback));
-                ctx.stop(ctx.getSelf());
+                actorCtx.stop(actorCtx.getSelf());
             }
         } else {
             var cfId = new CalculatedFieldId(msg.getEntityId().getId());
@@ -266,30 +280,30 @@ public class CalculatedFieldEntityMessageProcessor extends AbstractContextAwareM
     }
 
     public void process(CalculatedFieldReevaluateMsg msg) throws CalculatedFieldException {
-        CalculatedFieldId cfId = msg.getCfCtx().getCfId();
+        CalculatedFieldId cfId = msg.getCtx().getCfId();
         CalculatedFieldState state = states.get(cfId);
         if (state == null) {
             log.debug("[{}][{}] Failed to find CF state for entity to handle {}", entityId, cfId, msg);
         } else {
             if (state.isSizeOk()) {
                 log.debug("[{}][{}] Reevaluating CF state", entityId, cfId);
-                processStateIfReady(state, null, msg.getCfCtx(), Collections.singletonList(cfId), null, null, msg.getCallback());
+                processStateIfReady(state, null, msg.getCtx(), Collections.singletonList(cfId), null, null, msg.getCallback());
             } else {
-                throw new RuntimeException(msg.getCfCtx().getSizeExceedsLimitMessage());
+                throw new RuntimeException(msg.getCtx().getSizeExceedsLimitMessage());
             }
         }
     }
 
     public void process(CalculatedFieldAlarmActionMsg msg) {
         log.debug("[{}] Processing alarm action event msg: {}", entityId, msg);
-        states.values().forEach(state -> {
+        for (CalculatedFieldState state : states.values()) {
             if (state instanceof AlarmCalculatedFieldState alarmCfState) {
                 Alarm stateAlarm = alarmCfState.getCurrentAlarm();
                 if (stateAlarm != null && stateAlarm.getId().equals(msg.getAlarm().getId())) {
                     alarmCfState.processAlarmAction(msg.getAlarm(), msg.getAction());
                 }
             }
-        });
+        }
         msg.getCallback().onSuccess();
     }
 
@@ -354,7 +368,8 @@ public class CalculatedFieldEntityMessageProcessor extends AbstractContextAwareM
     }
 
     private void initState(CalculatedFieldState state, CalculatedFieldCtx ctx) {
-        state.init(ctx);
+        state.setCtx(ctx, actorCtx);
+        state.init();
 
         if (ctx.getCfType() == CalculatedFieldType.GEOFENCING && ctx.isRelationQueryDynamicArguments()) {
             GeofencingCalculatedFieldState geofencingState = (GeofencingCalculatedFieldState) state;

application/src/main/java/org/thingsboard/server/actors/calculatedField/CalculatedFieldManagerActor.java

@@ -20,6 +20,7 @@ import org.thingsboard.server.actors.ActorSystemContext;
 import org.thingsboard.server.actors.TbActorCtx;
 import org.thingsboard.server.actors.TbActorException;
 import org.thingsboard.server.common.data.id.TenantId;
+import org.thingsboard.server.common.msg.CalculatedFieldStatePartitionRestoreMsg;
 import org.thingsboard.server.common.msg.TbActorStopReason;
 import org.thingsboard.server.common.msg.ToCalculatedFieldSystemMsg;
 import org.thingsboard.server.common.msg.cf.CalculatedFieldCacheInitMsg;
@@ -70,6 +71,9 @@ public class CalculatedFieldManagerActor extends AbstractCalculatedFieldActor {
             case CF_STATE_RESTORE_MSG:
                 processor.onStateRestoreMsg((CalculatedFieldStateRestoreMsg) msg);
                 break;
+            case CF_STATE_PARTITION_RESTORE_MSG:
+                processor.onStatePartitionRestoreMsg((CalculatedFieldStatePartitionRestoreMsg) msg);
+                break;
             case CF_ENTITY_LIFECYCLE_MSG:
                 processor.onEntityLifecycleMsg((CalculatedFieldEntityLifecycleMsg) msg);
                 break;

application/src/main/java/org/thingsboard/server/actors/calculatedField/CalculatedFieldManagerMessageProcessor.java

@@ -24,6 +24,7 @@ import org.thingsboard.server.actors.TbCalculatedFieldEntityActorId;
 import org.thingsboard.server.actors.calculatedField.EntityInitCalculatedFieldMsg.StateAction;
 import org.thingsboard.server.actors.service.DefaultActorService;
 import org.thingsboard.server.actors.shared.AbstractContextAwareMsgProcessor;
+import org.thingsboard.server.common.data.Customer;
 import org.thingsboard.server.common.data.DataConstants;
 import org.thingsboard.server.common.data.EntityType;
 import org.thingsboard.server.common.data.ProfileEntityIdInfo;
@@ -36,6 +37,7 @@ import org.thingsboard.server.common.data.id.DeviceId;
 import org.thingsboard.server.common.data.id.EntityId;
 import org.thingsboard.server.common.data.id.TenantId;
 import org.thingsboard.server.common.data.page.PageDataIterable;
+import org.thingsboard.server.common.msg.CalculatedFieldStatePartitionRestoreMsg;
 import org.thingsboard.server.common.msg.cf.CalculatedFieldCacheInitMsg;
 import org.thingsboard.server.common.msg.cf.CalculatedFieldEntityLifecycleMsg;
 import org.thingsboard.server.common.msg.cf.CalculatedFieldPartitionChangeMsg;
@@ -44,6 +46,7 @@ import org.thingsboard.server.common.msg.queue.ServiceType;
 import org.thingsboard.server.common.msg.queue.TbCallback;
 import org.thingsboard.server.dao.asset.AssetService;
 import org.thingsboard.server.dao.cf.CalculatedFieldService;
+import org.thingsboard.server.dao.customer.CustomerService;
 import org.thingsboard.server.dao.device.DeviceService;
 import org.thingsboard.server.queue.settings.TbQueueCalculatedFieldSettings;
 import org.thingsboard.server.service.cf.CalculatedFieldProcessingService;
@@ -63,8 +66,6 @@ import java.util.List;
 import java.util.Map;
 import java.util.Set;
 import java.util.concurrent.CopyOnWriteArrayList;
-import java.util.concurrent.ScheduledFuture;
-import java.util.concurrent.TimeUnit;
 import java.util.function.BiConsumer;
 
 import static org.thingsboard.server.utils.CalculatedFieldUtils.fromProto;
@@ -79,13 +80,13 @@ public class CalculatedFieldManagerMessageProcessor extends AbstractContextAware
     private final Map<EntityId, List<CalculatedFieldCtx>> entityIdCalculatedFields = new HashMap<>();
     private final Map<EntityId, List<CalculatedFieldLink>> entityIdCalculatedFieldLinks = new HashMap<>();
     private final Map<EntityId, Set<EntityId>> ownerEntities = new HashMap<>();
-    private ScheduledFuture<?> cfsReevaluationTask;
 
     private final CalculatedFieldProcessingService cfExecService;
     private final CalculatedFieldStateService cfStateService;
     private final CalculatedFieldService cfDaoService;
     private final DeviceService deviceService;
     private final AssetService assetService;
+    private final CustomerService customerService;
     private final TbAssetProfileCache assetProfileCache;
     private final TbDeviceProfileCache deviceProfileCache;
     private final TenantEntityProfileCache entityProfileCache;
@@ -102,6 +103,7 @@ public class CalculatedFieldManagerMessageProcessor extends AbstractContextAware
         this.cfDaoService = systemContext.getCalculatedFieldService();
         this.deviceService = systemContext.getDeviceService();
         this.assetService = systemContext.getAssetService();
+        this.customerService = systemContext.getCustomerService();
         this.assetProfileCache = systemContext.getAssetProfileCache();
         this.deviceProfileCache = systemContext.getDeviceProfileCache();
         this.entityProfileCache = new TenantEntityProfileCache();
@@ -120,10 +122,6 @@ public class CalculatedFieldManagerMessageProcessor extends AbstractContextAware
         calculatedFields.clear();
         entityIdCalculatedFields.clear();
         entityIdCalculatedFieldLinks.clear();
-        if (cfsReevaluationTask != null) {
-            cfsReevaluationTask.cancel(true);
-            cfsReevaluationTask = null;
-        }
         ctx.stop(ctx.getSelf());
     }
 
@@ -131,7 +129,6 @@ public class CalculatedFieldManagerMessageProcessor extends AbstractContextAware
         log.debug("[{}] Processing CF actor init message.", msg.getTenantId().getId());
         initEntitiesCache();
         initCalculatedFields();
-        scheduleCfsReevaluation();
         msg.getCallback().onSuccess();
     }
 
@@ -140,9 +137,7 @@ public class CalculatedFieldManagerMessageProcessor extends AbstractContextAware
         var ctx = calculatedFields.get(cfId);
 
         if (ctx != null) {
-            if (msg.getState() != null) {
-                msg.getState().init(ctx);
-            }
+            msg.setCtx(ctx);
             log.debug("Pushing CF state restore msg to specific actor [{}]", msg.getId().entityId());
             getOrCreateActor(msg.getId().entityId()).tell(msg);
         } else {
@@ -150,21 +145,8 @@ public class CalculatedFieldManagerMessageProcessor extends AbstractContextAware
         }
     }
 
-    private void scheduleCfsReevaluation() {
-        cfsReevaluationTask = systemContext.getScheduler().scheduleWithFixedDelay(() -> {
-            try {
-                calculatedFields.values().forEach(cf -> {
-                    if (cf.isRequiresScheduledReevaluation()) {
-                        applyToTargetCfEntityActors(cf, TbCallback.EMPTY, (entityId, callback) -> {
-                            log.debug("[{}][{}] Pushing scheduled CF reevaluate msg", entityId, cf.getCfId());
-                            getOrCreateActor(entityId).tell(new CalculatedFieldReevaluateMsg(tenantId, cf));
-                        });
-                    }
-                });
-            } catch (Exception e) {
-                log.warn("[{}] Failed to trigger CFs reevaluation", tenantId, e);
-            }
-        }, systemContext.getAlarmsReevaluationInterval(), systemContext.getAlarmsReevaluationInterval(), TimeUnit.SECONDS);
+    public void onStatePartitionRestoreMsg(CalculatedFieldStatePartitionRestoreMsg msg) {
+        ctx.broadcastToChildren(msg, true);
     }
 
     public void onEntityLifecycleMsg(CalculatedFieldEntityLifecycleMsg msg) throws CalculatedFieldException {
@@ -172,56 +154,29 @@ public class CalculatedFieldManagerMessageProcessor extends AbstractContextAware
         var entityType = msg.getData().getEntityId().getEntityType();
         var event = msg.getData().getEvent();
         switch (entityType) {
-            case CALCULATED_FIELD: {
+            case CALCULATED_FIELD -> {
                 switch (event) {
-                    case CREATED:
-                        onCfCreated(msg.getData(), msg.getCallback());
-                        break;
-                    case UPDATED:
-                        onCfUpdated(msg.getData(), msg.getCallback());
-                        break;
-                    case DELETED:
-                        onCfDeleted(msg.getData(), msg.getCallback());
-                        break;
-                    default:
-                        msg.getCallback().onSuccess();
-                        break;
+                    case CREATED -> onCfCreated(msg.getData(), msg.getCallback());
+                    case UPDATED -> onCfUpdated(msg.getData(), msg.getCallback());
+                    case DELETED -> onCfDeleted(msg.getData(), msg.getCallback());
+                    default -> msg.getCallback().onSuccess();
                 }
-                break;
             }
-            case DEVICE:
-            case ASSET: {
+            case DEVICE, ASSET, CUSTOMER -> {
                 switch (event) {
-                    case CREATED:
-                        onEntityCreated(msg.getData(), msg.getCallback());
-                        break;
-                    case UPDATED:
-                        onEntityUpdated(msg.getData(), msg.getCallback());
-                        break;
-                    case DELETED:
-                        onEntityDeleted(msg.getData(), msg.getCallback());
-                        break;
-                    default:
-                        msg.getCallback().onSuccess();
-                        break;
+                    case CREATED -> onEntityCreated(msg.getData(), msg.getCallback());
+                    case UPDATED -> onEntityUpdated(msg.getData(), msg.getCallback());
+                    case DELETED -> onEntityDeleted(msg.getData(), msg.getCallback());
+                    default -> msg.getCallback().onSuccess();
                 }
-                break;
             }
-            case DEVICE_PROFILE:
-            case ASSET_PROFILE: {
+            case DEVICE_PROFILE, ASSET_PROFILE -> {
                 switch (event) {
-                    case DELETED:
-                        onProfileDeleted(msg.getData(), msg.getCallback());
-                        break;
-                    default:
-                        msg.getCallback().onSuccess();
-                        break;
+                    case DELETED -> onProfileDeleted(msg.getData(), msg.getCallback());
+                    default -> msg.getCallback().onSuccess();
                 }
-                break;
-            }
-            default: {
-                msg.getCallback().onSuccess();
             }
+            default -> msg.getCallback().onSuccess();
         }
     }
 
@@ -293,7 +248,10 @@ public class CalculatedFieldManagerMessageProcessor extends AbstractContextAware
     }
 
     private void onEntityDeleted(ComponentLifecycleMsg msg, TbCallback callback) {
-        entityProfileCache.removeEntityId(msg.getEntityId());
+        switch (msg.getEntityId().getEntityType()) {
+            case DEVICE, ASSET -> entityProfileCache.removeEntityId(msg.getEntityId());
+            case CUSTOMER -> ownerEntities.remove(msg.getEntityId());
+        }
         ownerEntities.values().forEach(entities -> entities.remove(msg.getEntityId()));
         if (isMyPartition(msg.getEntityId(), callback)) {
             log.debug("Pushing entity lifecycle msg to specific actor [{}]", msg.getEntityId());
@@ -426,9 +384,9 @@ public class CalculatedFieldManagerMessageProcessor extends AbstractContextAware
         }
         // process all cfs related to owner entity
         if (entityId.getEntityType().isOneOf(EntityType.TENANT, EntityType.CUSTOMER)) {
-            List<CalculatedFieldEntityCtxId> ownerCFs = filterOwnerEntitiesCFs(msg);
-            if (!ownerCFs.isEmpty()) {
-                cfExecService.pushMsgToLinks(msg, ownerCFs, callback);
+            List<CalculatedFieldEntityCtxId> ownedEntitiesCFs = filterOwnedEntitiesCFs(msg);
+            if (!ownedEntitiesCFs.isEmpty()) {
+                cfExecService.pushMsgToLinks(msg, ownedEntitiesCFs, callback);
             } else {
                 callback.onSuccess();
             }
@@ -495,8 +453,8 @@ public class CalculatedFieldManagerMessageProcessor extends AbstractContextAware
         return result;
     }
 
-    private List<CalculatedFieldEntityCtxId> filterOwnerEntitiesCFs(CalculatedFieldTelemetryMsg msg) {
-        Set<EntityId> entities = getOwnerEntities(msg.getEntityId());
+    private List<CalculatedFieldEntityCtxId> filterOwnedEntitiesCFs(CalculatedFieldTelemetryMsg msg) {
+        Set<EntityId> entities = getOwnedEntities(msg.getEntityId());
         var proto = msg.getProto();
         List<CalculatedFieldEntityCtxId> result = new ArrayList<>();
         for (var entityId : entities) {
@@ -538,7 +496,7 @@ public class CalculatedFieldManagerMessageProcessor extends AbstractContextAware
         return result;
     }
 
-    private Set<EntityId> getOwnerEntities(EntityId entityId) {
+    private Set<EntityId> getOwnedEntities(EntityId entityId) {
         if (entityId == null) {
             return Collections.emptySet();
         }
@@ -649,21 +607,32 @@ public class CalculatedFieldManagerMessageProcessor extends AbstractContextAware
             log.trace("Processing device record: {}", idInfo);
             try {
                 entityProfileCache.add(idInfo.getProfileId(), idInfo.getEntityId());
-                ownerEntities.computeIfAbsent(idInfo.getOwnerId(), ownerId -> new HashSet<>()).add(idInfo.getEntityId());
+                ownerEntities.computeIfAbsent(idInfo.getOwnerId(), __ -> new HashSet<>()).add(idInfo.getEntityId());
             } catch (Exception e) {
                 log.error("Failed to process device record: {}", idInfo, e);
             }
         }
+
         PageDataIterable<ProfileEntityIdInfo> assetIdInfos = new PageDataIterable<>(pageLink -> assetService.findProfileEntityIdInfosByTenantId(tenantId, pageLink), cfSettings.getInitTenantFetchPackSize());
         for (ProfileEntityIdInfo idInfo : assetIdInfos) {
             log.trace("Processing asset record: {}", idInfo);
             try {
                 entityProfileCache.add(idInfo.getProfileId(), idInfo.getEntityId());
-                ownerEntities.computeIfAbsent(idInfo.getOwnerId(), ownerId -> new HashSet<>()).add(idInfo.getEntityId());
+                ownerEntities.computeIfAbsent(idInfo.getOwnerId(), __ -> new HashSet<>()).add(idInfo.getEntityId());
             } catch (Exception e) {
                 log.error("Failed to process asset record: {}", idInfo, e);
             }
         }
+
+        PageDataIterable<Customer> customers = new PageDataIterable<>(pageLink -> customerService.findCustomersByTenantId(tenantId, pageLink), cfSettings.getInitTenantFetchPackSize());
+        for (Customer customer : customers) {
+            log.trace("Processing customer record: {}", customer);
+            try {
+                ownerEntities.computeIfAbsent(customer.getTenantId(), __ -> new HashSet<>()).add(customer.getId());
+            } catch (Exception e) {
+                log.error("Failed to process customer record: {}", customer, e);
+            }
+        }
     }
 
     private void updateEntityOwner(EntityId entityId) {

application/src/main/java/org/thingsboard/server/actors/calculatedField/CalculatedFieldReevaluateMsg.java

@@ -25,7 +25,7 @@ import org.thingsboard.server.service.cf.ctx.state.CalculatedFieldCtx;
 public class CalculatedFieldReevaluateMsg implements ToCalculatedFieldSystemMsg {
 
     private final TenantId tenantId;
-    private final CalculatedFieldCtx cfCtx;
+    private final CalculatedFieldCtx ctx;
 
     @Override
     public MsgType getMsgType() {

application/src/main/java/org/thingsboard/server/actors/calculatedField/CalculatedFieldStateRestoreMsg.java

@@ -19,7 +19,9 @@ import lombok.Data;
 import org.thingsboard.server.common.data.id.TenantId;
 import org.thingsboard.server.common.msg.MsgType;
 import org.thingsboard.server.common.msg.ToCalculatedFieldSystemMsg;
+import org.thingsboard.server.common.msg.queue.TopicPartitionInfo;
 import org.thingsboard.server.service.cf.ctx.CalculatedFieldEntityCtxId;
+import org.thingsboard.server.service.cf.ctx.state.CalculatedFieldCtx;
 import org.thingsboard.server.service.cf.ctx.state.CalculatedFieldState;
 
 @Data
@@ -27,6 +29,8 @@ public class CalculatedFieldStateRestoreMsg implements ToCalculatedFieldSystemMs
 
     private final CalculatedFieldEntityCtxId id;
     private final CalculatedFieldState state;
+    private final TopicPartitionInfo partition;
+    private CalculatedFieldCtx ctx;
 
     @Override
     public MsgType getMsgType() {

application/src/main/java/org/thingsboard/server/actors/tenant/TenantActor.java

@@ -186,6 +186,7 @@ public class TenantActor extends RuleChainManagerActor {
             case CF_CACHE_INIT_MSG:
             case CF_STATE_RESTORE_MSG:
             case CF_PARTITIONS_CHANGE_MSG:
+            case CF_STATE_PARTITION_RESTORE_MSG:
                 forwardToCfActor((ToCalculatedFieldSystemMsg) msg, true);
                 break;
             case CF_TELEMETRY_MSG:
@@ -349,7 +350,7 @@ public class TenantActor extends RuleChainManagerActor {
                 }
             }
             if (cfActor != null) {
-                if (msg.getEntityId().getEntityType().isOneOf(EntityType.CALCULATED_FIELD, EntityType.DEVICE, EntityType.ASSET)) {
+                if (msg.getEntityId().getEntityType().isOneOf(EntityType.CALCULATED_FIELD, EntityType.DEVICE, EntityType.ASSET, EntityType.CUSTOMER)) {
                     cfActor.tellWithHighPriority(new CalculatedFieldEntityLifecycleMsg(tenantId, msg));
                 }
             }
@@ -394,6 +395,7 @@ public class TenantActor extends RuleChainManagerActor {
         public TbActor createActor() {
             return new TenantActor(context, tenantId);
         }
+
     }
 
 }

application/src/main/java/org/thingsboard/server/controller/CalculatedFieldController.java

@@ -44,6 +44,7 @@ import org.thingsboard.script.api.tbel.TbelInvokeService;
 import org.thingsboard.server.common.data.EntityType;
 import org.thingsboard.server.common.data.EventInfo;
 import org.thingsboard.server.common.data.cf.CalculatedField;
+import org.thingsboard.server.common.data.cf.CalculatedFieldType;
 import org.thingsboard.server.common.data.cf.configuration.CalculatedFieldConfiguration;
 import org.thingsboard.server.common.data.event.EventType;
 import org.thingsboard.server.common.data.exception.ThingsboardException;
@@ -159,19 +160,27 @@ public class CalculatedFieldController extends BaseController {
     )
     @PreAuthorize("hasAnyAuthority('TENANT_ADMIN')")
     @GetMapping(value = "/{entityType}/{entityId}/calculatedFields", params = {"pageSize", "page"})
-    public PageData<CalculatedField> getCalculatedFieldsByEntityId(
-            @Parameter(description = ENTITY_TYPE_PARAM_DESCRIPTION, required = true, schema = @Schema(defaultValue = "DEVICE")) @PathVariable("entityType") String entityType,
-            @Parameter(description = ENTITY_ID_PARAM_DESCRIPTION, required = true) @PathVariable("entityId") String entityIdStr,
-            @Parameter(description = PAGE_SIZE_DESCRIPTION, required = true) @RequestParam int pageSize,
-            @Parameter(description = PAGE_NUMBER_DESCRIPTION, required = true) @RequestParam int page,
-            @Parameter(description = CF_TEXT_SEARCH_DESCRIPTION) @RequestParam(required = false) String textSearch,
-            @Parameter(description = SORT_PROPERTY_DESCRIPTION, schema = @Schema(allowableValues = {"createdTime", "name"})) @RequestParam(required = false) String sortProperty,
-            @Parameter(description = SORT_ORDER_DESCRIPTION, schema = @Schema(allowableValues = {"ASC", "DESC"})) @RequestParam(required = false) String sortOrder) throws ThingsboardException {
+    public PageData<CalculatedField> getCalculatedFieldsByEntityId(@Parameter(description = ENTITY_TYPE_PARAM_DESCRIPTION, required = true, schema = @Schema(defaultValue = "DEVICE"))
+                                                                   @PathVariable("entityType") String entityType,
+                                                                   @Parameter(description = ENTITY_ID_PARAM_DESCRIPTION, required = true)
+                                                                   @PathVariable("entityId") String entityIdStr,
+                                                                   @Parameter(description = PAGE_SIZE_DESCRIPTION, required = true)
+                                                                   @RequestParam int pageSize,
+                                                                   @Parameter(description = PAGE_NUMBER_DESCRIPTION, required = true)
+                                                                   @RequestParam int page,
+                                                                   @Parameter(description = "Calculated field type. If not specified, all types will be returned.")
+                                                                   @RequestParam(required = false) CalculatedFieldType type,
+                                                                   @Parameter(description = CF_TEXT_SEARCH_DESCRIPTION)
+                                                                   @RequestParam(required = false) String textSearch,
+                                                                   @Parameter(description = SORT_PROPERTY_DESCRIPTION, schema = @Schema(allowableValues = {"createdTime", "name"}))
+                                                                   @RequestParam(required = false) String sortProperty,
+                                                                   @Parameter(description = SORT_ORDER_DESCRIPTION, schema = @Schema(allowableValues = {"ASC", "DESC"}))
+                                                                   @RequestParam(required = false) String sortOrder) throws ThingsboardException {
         PageLink pageLink = createPageLink(pageSize, page, textSearch, sortProperty, sortOrder);
         checkParameter("entityId", entityIdStr);
         EntityId entityId = EntityIdFactory.getByTypeAndUuid(entityType, entityIdStr);
         checkEntityId(entityId, Operation.READ_CALCULATED_FIELD);
-        return checkNotNull(tbCalculatedFieldService.findAllByTenantIdAndEntityId(entityId, getCurrentUser(), pageLink));
+        return checkNotNull(tbCalculatedFieldService.findByTenantIdAndEntityId(getTenantId(), entityId, type, pageLink));
     }
 
     @ApiOperation(value = "Delete Calculated Field (deleteCalculatedField)",

application/src/main/java/org/thingsboard/server/service/cf/AbstractCalculatedFieldStateService.java

@@ -15,10 +15,15 @@
  */
 package org.thingsboard.server.service.cf;
 
+import lombok.extern.slf4j.Slf4j;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.context.annotation.Lazy;
 import org.thingsboard.server.actors.ActorSystemContext;
 import org.thingsboard.server.actors.calculatedField.CalculatedFieldStateRestoreMsg;
+import org.thingsboard.server.common.data.DataConstants;
+import org.thingsboard.server.common.data.exception.TenantNotFoundException;
+import org.thingsboard.server.common.msg.CalculatedFieldStatePartitionRestoreMsg;
+import org.thingsboard.server.common.msg.queue.ServiceType;
 import org.thingsboard.server.common.msg.queue.TbCallback;
 import org.thingsboard.server.common.msg.queue.TopicPartitionInfo;
 import org.thingsboard.server.exception.CalculatedFieldStateException;
@@ -37,6 +42,7 @@ import java.util.stream.Collectors;
 import static org.thingsboard.server.utils.CalculatedFieldUtils.fromProto;
 import static org.thingsboard.server.utils.CalculatedFieldUtils.toProto;
 
+@Slf4j
 public abstract class AbstractCalculatedFieldStateService implements CalculatedFieldStateService {
 
     @Autowired
@@ -62,19 +68,38 @@ public abstract class AbstractCalculatedFieldStateService implements CalculatedF
 
     protected abstract void doRemove(CalculatedFieldEntityCtxId stateId, TbCallback callback);
 
-    protected void processRestoredState(CalculatedFieldStateProto stateMsg) {
+    protected void processRestoredState(CalculatedFieldStateProto stateMsg, TopicPartitionInfo partition) {
         var id = fromProto(stateMsg.getId());
+        if (partition == null) {
+            try {
+                partition = actorSystemContext.resolve(ServiceType.TB_RULE_ENGINE, DataConstants.CF_QUEUE_NAME, id.tenantId(), id.entityId());
+            } catch (TenantNotFoundException e) {
+                log.debug("Skipping CF state msg for non-existing tenant {}", id.tenantId());
+                return;
+            }
+        }
         var state = fromProto(id, stateMsg);
-        processRestoredState(id, state);
+        processRestoredState(id, state, partition);
     }
 
-    protected void processRestoredState(CalculatedFieldEntityCtxId id, CalculatedFieldState state) {
-        actorSystemContext.tell(new CalculatedFieldStateRestoreMsg(id, state));
+    protected void processRestoredState(CalculatedFieldEntityCtxId id, CalculatedFieldState state, TopicPartitionInfo partition) {
+        partition = partition.withTopic(DataConstants.CF_STATES_QUEUE_NAME);
+        actorSystemContext.tell(new CalculatedFieldStateRestoreMsg(id, state, partition));
     }
 
     @Override
     public void restore(QueueKey queueKey, Set<TopicPartitionInfo> partitions) {
-        stateService.update(queueKey, partitions, null);
+        stateService.update(queueKey, partitions, new QueueStateService.RestoreCallback() {
+            @Override
+            public void onAllPartitionsRestored() {
+            }
+
+            @Override
+            public void onPartitionRestored(TopicPartitionInfo partition) {
+                partition = partition.withTopic(DataConstants.CF_STATES_QUEUE_NAME);
+                actorSystemContext.tellWithHighPriority(new CalculatedFieldStatePartitionRestoreMsg(partition));
+            }
+        });
     }
 
     @Override

application/src/main/java/org/thingsboard/server/service/cf/DefaultCalculatedFieldQueueService.java

@@ -25,6 +25,7 @@ import org.thingsboard.rule.engine.api.TimeseriesDeleteRequest;
 import org.thingsboard.rule.engine.api.TimeseriesSaveRequest;
 import org.thingsboard.server.cluster.TbClusterService;
 import org.thingsboard.server.common.data.EntityType;
+import org.thingsboard.server.common.data.cf.CalculatedField;
 import org.thingsboard.server.common.data.cf.CalculatedFieldLink;
 import org.thingsboard.server.common.data.id.CalculatedFieldId;
 import org.thingsboard.server.common.data.id.EntityId;
@@ -45,9 +46,7 @@ import org.thingsboard.server.queue.TbQueueMsgMetadata;
 import org.thingsboard.server.service.cf.ctx.state.CalculatedFieldCtx;
 
 import java.util.Collections;
-import java.util.EnumSet;
 import java.util.List;
-import java.util.Set;
 import java.util.UUID;
 import java.util.function.Predicate;
 import java.util.function.Supplier;
@@ -73,10 +72,6 @@ public class DefaultCalculatedFieldQueueService implements CalculatedFieldQueueS
     private final CalculatedFieldCache calculatedFieldCache;
     private final TbClusterService clusterService;
 
-    private static final Set<EntityType> supportedReferencedEntities = EnumSet.of(
-            EntityType.DEVICE, EntityType.ASSET, EntityType.CUSTOMER, EntityType.TENANT
-    );
-
     @Override
     public void pushRequestToQueue(TimeseriesSaveRequest request, TimeseriesSaveResult result, FutureCallback<Void> callback) {
         var tenantId = request.getTenantId();
@@ -155,7 +150,7 @@ public class DefaultCalculatedFieldQueueService implements CalculatedFieldQueueS
     }
 
     private boolean checkEntityForCalculatedFields(TenantId tenantId, EntityId entityId, Predicate<CalculatedFieldCtx> filter, Predicate<CalculatedFieldCtx> linkedEntityFilter, Predicate<CalculatedFieldCtx> dynamicSourceFilter) {
-        if (!supportedReferencedEntities.contains(entityId.getEntityType())) {
+        if (!CalculatedField.SUPPORTED_REFERENCED_ENTITIES.contains(entityId.getEntityType())) {
             return false;
         }
 

application/src/main/java/org/thingsboard/server/service/cf/OwnerService.java

@@ -17,6 +17,7 @@ package org.thingsboard.server.service.cf;
 
 import lombok.RequiredArgsConstructor;
 import org.springframework.stereotype.Service;
+import org.thingsboard.server.common.data.Customer;
 import org.thingsboard.server.common.data.DeviceInfo;
 import org.thingsboard.server.common.data.DeviceInfoFilter;
 import org.thingsboard.server.common.data.EntityType;
@@ -28,6 +29,7 @@ import org.thingsboard.server.common.data.id.EntityId;
 import org.thingsboard.server.common.data.id.TenantId;
 import org.thingsboard.server.common.data.page.PageDataIterable;
 import org.thingsboard.server.dao.asset.AssetService;
+import org.thingsboard.server.dao.customer.CustomerService;
 import org.thingsboard.server.dao.device.DeviceService;
 
 import java.util.HashSet;
@@ -39,29 +41,36 @@ public class OwnerService {
 
     private final DeviceService deviceService;
     private final AssetService assetService;
+    private final CustomerService customerService;
 
     public EntityId getOwner(TenantId tenantId, EntityId entityId) {
         return switch (entityId.getEntityType()) {
             case DEVICE -> deviceService.findDeviceById(tenantId, (DeviceId) entityId).getOwnerId();
             case ASSET -> assetService.findAssetById(tenantId, (AssetId) entityId).getOwnerId();
+            case CUSTOMER -> tenantId;
             default -> throw new UnsupportedOperationException();
         };
     }
 
     public Set<EntityId> getOwnedEntities(TenantId tenantId, EntityId ownerId) {
-        Set<EntityId> ownerEntities = new HashSet<>();
+        Set<EntityId> ownedEntities = new HashSet<>();
         if (EntityType.CUSTOMER.equals(ownerId.getEntityType())) {
             PageDataIterable<DeviceInfo> deviceIdInfos = new PageDataIterable<>(pageLink -> deviceService.findDeviceInfosByFilter(DeviceInfoFilter.builder().tenantId(tenantId).customerId((CustomerId) ownerId).build(), pageLink), 1000);
-            deviceIdInfos.forEach(deviceInfo -> ownerEntities.add(deviceInfo.getId()));
+            deviceIdInfos.forEach(deviceInfo -> ownedEntities.add(deviceInfo.getId()));
+
             PageDataIterable<Asset> assets = new PageDataIterable<>(pageLink -> assetService.findAssetsByTenantIdAndCustomerId(tenantId, (CustomerId) ownerId, pageLink), 1000);
-            assets.forEach(asset -> ownerEntities.add(asset.getId()));
+            assets.forEach(asset -> ownedEntities.add(asset.getId()));
         } else if (EntityType.TENANT.equals(ownerId.getEntityType())) {
             PageDataIterable<DeviceInfo> deviceIdInfos = new PageDataIterable<>(pageLink -> deviceService.findDeviceInfosByFilter(DeviceInfoFilter.builder().tenantId((TenantId) ownerId).customerId(new CustomerId(CustomerId.NULL_UUID)).build(), pageLink), 1000);
-            deviceIdInfos.forEach(deviceInfo -> ownerEntities.add(deviceInfo.getId()));
+            deviceIdInfos.forEach(deviceInfo -> ownedEntities.add(deviceInfo.getId()));
+
             PageDataIterable<Asset> assets = new PageDataIterable<>(pageLink -> assetService.findAssetsByTenantIdAndCustomerId((TenantId) ownerId, new CustomerId(CustomerId.NULL_UUID), pageLink), 1000);
-            assets.forEach(asset -> ownerEntities.add(asset.getId()));
+            assets.forEach(asset -> ownedEntities.add(asset.getId()));
+
+            PageDataIterable<Customer> customers = new PageDataIterable<>(pageLink -> customerService.findCustomersByTenantId((TenantId) ownerId, pageLink), 1000);
+            customers.forEach(customer -> ownedEntities.add(customer.getId()));
         }
-        return ownerEntities;
+        return ownedEntities;
     }
 
 }

application/src/main/java/org/thingsboard/server/service/cf/ctx/state/BaseCalculatedFieldState.java

@@ -17,8 +17,11 @@ package org.thingsboard.server.service.cf.ctx.state;
 
 import com.fasterxml.jackson.databind.node.ObjectNode;
 import lombok.Getter;
+import lombok.Setter;
 import org.thingsboard.common.util.JacksonUtil;
+import org.thingsboard.server.actors.TbActorRef;
 import org.thingsboard.server.common.data.id.EntityId;
+import org.thingsboard.server.common.msg.queue.TopicPartitionInfo;
 import org.thingsboard.server.service.cf.ctx.CalculatedFieldEntityCtxId;
 import org.thingsboard.server.utils.CalculatedFieldUtils;
 
@@ -31,21 +34,32 @@ import java.util.Map;
 public abstract class BaseCalculatedFieldState implements CalculatedFieldState {
 
     protected final EntityId entityId;
+    protected CalculatedFieldCtx ctx;
+    protected TbActorRef actorCtx;
     protected List<String> requiredArguments;
 
     protected Map<String, ArgumentEntry> arguments = new HashMap<>();
     protected boolean sizeExceedsLimit;
     protected long latestTimestamp = -1;
 
+    @Setter
+    private TopicPartitionInfo partition;
+
     public BaseCalculatedFieldState(EntityId entityId) {
         this.entityId = entityId;
     }
 
     @Override
-    public void init(CalculatedFieldCtx ctx) {
+    public void setCtx(CalculatedFieldCtx ctx, TbActorRef actorCtx) {
+        this.ctx = ctx;
+        this.actorCtx = actorCtx;
         this.requiredArguments = ctx.getArgNames();
     }
 
+    @Override
+    public void init() {
+    }
+
     @Override
     public Map<String, ArgumentEntry> update(Map<String, ArgumentEntry> argumentValues, CalculatedFieldCtx ctx) {
         Map<String, ArgumentEntry> updatedArguments = null;
@@ -84,7 +98,7 @@ public abstract class BaseCalculatedFieldState implements CalculatedFieldState {
     }
 
     @Override
-    public void reset(CalculatedFieldCtx ctx) { // must reset everything dependent on arguments
+    public void reset() { // must reset everything dependent on arguments
         requiredArguments = null;
         arguments.clear();
         sizeExceedsLimit = false;

application/src/main/java/org/thingsboard/server/service/cf/ctx/state/CalculatedFieldCtx.java

@@ -17,6 +17,7 @@ package org.thingsboard.server.service.cf.ctx.state;
 
 import com.google.common.util.concurrent.ListenableFuture;
 import lombok.Data;
+import lombok.extern.slf4j.Slf4j;
 import net.objecthunter.exp4j.Expression;
 import org.mvel2.MVEL;
 import org.thingsboard.common.util.ExpressionUtils;
@@ -25,6 +26,8 @@ import org.thingsboard.script.api.tbel.TbelCfCtx;
 import org.thingsboard.script.api.tbel.TbelCfSingleValueArg;
 import org.thingsboard.script.api.tbel.TbelInvokeService;
 import org.thingsboard.server.actors.ActorSystemContext;
+import org.thingsboard.server.actors.TbActorRef;
+import org.thingsboard.server.actors.calculatedField.CalculatedFieldReevaluateMsg;
 import org.thingsboard.server.common.data.AttributeScope;
 import org.thingsboard.server.common.data.alarm.rule.AlarmRule;
 import org.thingsboard.server.common.data.alarm.rule.condition.expression.TbelAlarmConditionExpression;
@@ -61,10 +64,12 @@ import java.util.LinkedHashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
+import java.util.concurrent.ScheduledFuture;
 import java.util.concurrent.TimeUnit;
 import java.util.stream.Stream;
 
 @Data
+@Slf4j
 public class CalculatedFieldCtx {
 
     private CalculatedField calculatedField;
@@ -81,8 +86,8 @@ public class CalculatedFieldCtx {
     private Output output;
     private String expression;
     private boolean useLatestTs;
-    private boolean requiresScheduledReevaluation;
 
+    private ActorSystemContext systemContext;
     private TbelInvokeService tbelInvokeService;
     private RelationService relationService;
     private AlarmSubscriptionService alarmService;
@@ -167,7 +172,7 @@ public class CalculatedFieldCtx {
         if (calculatedField.getConfiguration() instanceof ScheduledUpdateSupportedCalculatedFieldConfiguration scheduledConfig) {
             this.scheduledUpdateIntervalMillis = scheduledConfig.isScheduledUpdateEnabled() ? TimeUnit.SECONDS.toMillis(scheduledConfig.getScheduledUpdateInterval()) : -1L;
         }
-        this.requiresScheduledReevaluation = calculatedField.getConfiguration().requiresScheduledReevaluation();
+        this.systemContext = systemContext;
         this.tbelInvokeService = systemContext.getTbelInvokeService();
         this.relationService = systemContext.getRelationService();
         this.alarmService = systemContext.getAlarmService();
@@ -254,6 +259,12 @@ public class CalculatedFieldCtx {
         return tbelExpressions.get(expression).executeScriptAsync(args.toArray());
     }
 
+    public ScheduledFuture<?> scheduleReevaluation(long delayMs, TbActorRef actorCtx) {
+        log.debug("[{}] Scheduling CF reevaluation in {} ms", cfId, delayMs);
+        // TODO: use single lazy-loaded instance of CalculatedFieldReevaluateMsg
+        return systemContext.scheduleMsgWithDelay(actorCtx, new CalculatedFieldReevaluateMsg(tenantId, this), delayMs);
+    }
+
     private TbelCfArg toTbelArgument(String key, CalculatedFieldState state) {
         return state.getArguments().get(key).toTbelCfArg();
     }
@@ -487,8 +498,12 @@ public class CalculatedFieldCtx {
     }
 
     public boolean hasContextOnlyChanges(CalculatedFieldCtx other) { // has changes that do not require state reinit and will be picked up by the state on the fly
-        if (calculatedField.getConfiguration() instanceof ExpressionBasedCalculatedFieldConfiguration && !expression.equals(other.expression)) {
-            return true;
+        if (calculatedField.getConfiguration() instanceof ExpressionBasedCalculatedFieldConfiguration expressionConfig) {
+            boolean shouldCompareExpression = !(expressionConfig instanceof PropagationCalculatedFieldConfiguration propagationConfig)
+                                              || propagationConfig.isApplyExpressionToResolvedArguments();
+            if (shouldCompareExpression && !expression.equals(other.expression)) {
+                return true;
+            }
         }
         if (!output.equals(other.output)) {
             return true;
@@ -496,10 +511,7 @@ public class CalculatedFieldCtx {
         if (cfType == CalculatedFieldType.ALARM && !calculatedField.getName().equals(other.getCalculatedField().getName())) {
             return true;
         }
-        if (scheduledUpdateIntervalMillis != other.scheduledUpdateIntervalMillis) {
-            return true;
-        }
-        return false;
+        return scheduledUpdateIntervalMillis != other.scheduledUpdateIntervalMillis;
     }
 
     public boolean hasStateChanges(CalculatedFieldCtx other) { // has changes that require state reinit (will trigger state.reset() and re-fetch arguments)

application/src/main/java/org/thingsboard/server/service/cf/ctx/state/CalculatedFieldState.java

@@ -20,7 +20,9 @@ import com.fasterxml.jackson.annotation.JsonSubTypes;
 import com.fasterxml.jackson.annotation.JsonSubTypes.Type;
 import com.fasterxml.jackson.annotation.JsonTypeInfo;
 import com.google.common.util.concurrent.ListenableFuture;
+import org.thingsboard.server.actors.TbActorRef;
 import org.thingsboard.server.common.data.cf.CalculatedFieldType;
+import org.thingsboard.server.common.msg.queue.TopicPartitionInfo;
 import org.thingsboard.server.service.cf.CalculatedFieldResult;
 import org.thingsboard.server.service.cf.ctx.CalculatedFieldEntityCtxId;
 import org.thingsboard.server.service.cf.ctx.state.alarm.AlarmCalculatedFieldState;
@@ -49,11 +51,13 @@ public interface CalculatedFieldState {
 
     long getLatestTimestamp();
 
-    void init(CalculatedFieldCtx ctx);
+    void setCtx(CalculatedFieldCtx ctx, TbActorRef actorCtx);
+
+    void init();
 
     Map<String, ArgumentEntry> update(Map<String, ArgumentEntry> arguments, CalculatedFieldCtx ctx);
 
-    void reset(CalculatedFieldCtx ctx);
+    void reset();
 
     ListenableFuture<CalculatedFieldResult> performCalculation(Map<String, ArgumentEntry> updatedArgs, CalculatedFieldCtx ctx);
 
@@ -67,6 +71,10 @@ public interface CalculatedFieldState {
         return !isSizeExceedsLimit();
     }
 
+    TopicPartitionInfo getPartition();
+
+    void setPartition(TopicPartitionInfo partition);
+
     void checkStateSize(CalculatedFieldEntityCtxId ctxId, long maxStateSize);
 
     default void checkArgumentSize(String name, ArgumentEntry entry, CalculatedFieldCtx ctx) {

application/src/main/java/org/thingsboard/server/service/cf/ctx/state/KafkaCalculatedFieldStateService.java

@@ -43,6 +43,7 @@ import org.thingsboard.server.queue.provider.TbRuleEngineQueueFactory;
 import org.thingsboard.server.service.cf.AbstractCalculatedFieldStateService;
 import org.thingsboard.server.service.cf.ctx.CalculatedFieldEntityCtxId;
 
+import java.util.Set;
 import java.util.concurrent.atomic.AtomicInteger;
 
 import static org.thingsboard.server.queue.common.AbstractTbQueueTemplate.bytesToString;
@@ -77,9 +78,9 @@ public class KafkaCalculatedFieldStateService extends AbstractCalculatedFieldSta
                     for (TbProtoQueueMsg<CalculatedFieldStateProto> msg : msgs) {
                         try {
                             if (msg.getValue() != null) {
-                                processRestoredState(msg.getValue());
+                                processRestoredState(msg.getValue(), consumerKey.partition());
                             } else {
-                                processRestoredState(getStateId(msg.getHeaders()), null);
+                                processRestoredState(getStateId(msg.getHeaders()), null, consumerKey.partition());
                             }
                         } catch (Throwable t) {
                             log.error("Failed to process state message: {}", msg, t);
@@ -104,6 +105,11 @@ public class KafkaCalculatedFieldStateService extends AbstractCalculatedFieldSta
         this.stateProducer = (TbKafkaProducerTemplate<TbProtoQueueMsg<CalculatedFieldStateProto>>) queueFactory.createCalculatedFieldStateProducer();
     }
 
+    @Override
+    public void restore(QueueKey queueKey, Set<TopicPartitionInfo> partitions) {
+        stateService.update(queueKey, partitions, null);
+    }
+
     @Override
     protected void doPersist(CalculatedFieldEntityCtxId stateId, CalculatedFieldStateProto stateMsgProto, TbCallback callback) {
         TopicPartitionInfo tpi = partitionService.resolve(ServiceType.TB_RULE_ENGINE, DataConstants.CF_STATES_QUEUE_NAME, stateId.tenantId(), stateId.entityId());

application/src/main/java/org/thingsboard/server/service/cf/ctx/state/RocksDBCalculatedFieldStateService.java

@@ -15,7 +15,6 @@
  */
 package org.thingsboard.server.service.cf.ctx.state;
 
-import com.google.protobuf.InvalidProtocolBufferException;
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression;
@@ -64,8 +63,8 @@ public class RocksDBCalculatedFieldStateService extends AbstractCalculatedFieldS
         if (stateService.getPartitions().isEmpty()) {
             cfRocksDb.forEach((key, value) -> {
                 try {
-                    processRestoredState(CalculatedFieldStateProto.parseFrom(value));
-                } catch (InvalidProtocolBufferException e) {
+                    processRestoredState(CalculatedFieldStateProto.parseFrom(value), null);
+                } catch (Exception e) {
                     log.error("[{}] Failed to process restored state", key, e);
                 }
             });

application/src/main/java/org/thingsboard/server/service/cf/ctx/state/alarm/AlarmCalculatedFieldState.java

@@ -21,11 +21,13 @@ import com.google.common.util.concurrent.Futures;
 import com.google.common.util.concurrent.ListenableFuture;
 import lombok.EqualsAndHashCode;
 import lombok.Getter;
+import lombok.Setter;
 import lombok.SneakyThrows;
 import lombok.extern.slf4j.Slf4j;
 import org.thingsboard.common.util.JacksonUtil;
 import org.thingsboard.common.util.KvUtil;
 import org.thingsboard.rule.engine.action.TbAlarmResult;
+import org.thingsboard.server.actors.TbActorRef;
 import org.thingsboard.server.common.data.StringUtils;
 import org.thingsboard.server.common.data.alarm.Alarm;
 import org.thingsboard.server.common.data.alarm.AlarmApiCallResult;
@@ -61,10 +63,15 @@ import org.thingsboard.server.service.cf.ctx.state.SingleValueArgumentEntry;
 import java.util.Comparator;
 import java.util.Map;
 import java.util.TreeMap;
+import java.util.concurrent.ScheduledFuture;
+import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.function.Function;
 
 import static org.thingsboard.server.common.data.StringUtils.equalsAny;
 import static org.thingsboard.server.common.data.StringUtils.splitByCommaWithoutQuotes;
+import static org.thingsboard.server.service.cf.ctx.state.alarm.AlarmEvalResult.Status.FALSE;
+import static org.thingsboard.server.service.cf.ctx.state.alarm.AlarmEvalResult.Status.NOT_YET_TRUE;
+import static org.thingsboard.server.service.cf.ctx.state.alarm.AlarmEvalResult.Status.TRUE;
 
 @EqualsAndHashCode(callSuper = true)
 @Slf4j
@@ -76,6 +83,7 @@ public class AlarmCalculatedFieldState extends BaseCalculatedFieldState {
     @Getter
     private final Map<AlarmSeverity, AlarmRuleState> createRuleStates = new TreeMap<>(Comparator.comparing(Enum::ordinal));
     @Getter
+    @Setter
     private AlarmRuleState clearRuleState;
 
     @Getter
@@ -87,36 +95,71 @@ public class AlarmCalculatedFieldState extends BaseCalculatedFieldState {
     }
 
     @Override
-    public void init(CalculatedFieldCtx ctx) {
-        super.init(ctx);
-
+    public void setCtx(CalculatedFieldCtx ctx, TbActorRef actorCtx) {
+        super.setCtx(ctx, actorCtx);
         this.alarmType = ctx.getCalculatedField().getName();
         this.configuration = getConfiguration(ctx);
+    }
 
+    @Override
+    public void init() { // todo: properly close state!
+        super.init();
+        AtomicBoolean reevalNeeded = new AtomicBoolean(false);
         Map<AlarmSeverity, AlarmRule> createRules = configuration.getCreateRules();
-        createRules.forEach((severity, rule) -> {
-            AlarmRuleState ruleState = createRuleStates.get(severity);
-            if (ruleState == null) {
-                ruleState = new AlarmRuleState(severity, rule, this);
-                createRuleStates.put(severity, ruleState);
-            } else { // can be null if was restored
-                ruleState.setAlarmRule(rule);
-                // todo: is it enough to just set new alarm rule to alarm rule state? is it ok to leave the state as were??
+        for (AlarmSeverity severity : AlarmSeverity.values()) {
+            AlarmRule rule = createRules.get(severity);
+            if (rule != null) {
+                createRuleStates.compute(severity, (__, ruleState) -> {
+                    return initRuleState(severity, rule, ruleState, reevalNeeded);
+                });
+            } else {
+                AlarmRuleState state = createRuleStates.remove(severity);
+                if (state != null) {
+                    clearState(state);
+                }
             }
-        });
-        createRuleStates.keySet().removeIf(severity -> !createRules.containsKey(severity));
+        }
 
         AlarmRule clearRule = configuration.getClearRule();
         if (clearRule != null) {
-            if (clearRuleState == null) {
-                clearRuleState = new AlarmRuleState(null, clearRule, this);
-            } else {
-                clearRuleState.setAlarmRule(clearRule);
+            clearRuleState = initRuleState(null, clearRule, clearRuleState, reevalNeeded);
+        } else {
+            if (clearRuleState != null) {
+                clearState(clearRuleState);
+                clearRuleState = null;
             }
+        }
+        log.debug("Initialized create rule states {} and clear rule state {} for {}", createRuleStates, clearRuleState, configuration);
+
+        if (reevalNeeded.get()) {
+            initCurrentAlarm(ctx);
+            createOrClearAlarms(state -> {
+                if (state.getCondition().getType() == AlarmConditionType.DURATION) {
+                    AlarmEvalResult evalResult = state.reeval(System.currentTimeMillis());
+                    if (evalResult.getStatus() == TRUE || evalResult.getStatus() == NOT_YET_TRUE) {
+                        ScheduledFuture<?> future = ctx.scheduleReevaluation(evalResult.getLeftDuration(), actorCtx);
+                        // TODO: use single task for multiple durations if durations are close enough. but be careful when cancelling the task in one of the states
+                        if (future != null) {
+                            state.setDurationCheckFuture(future);
+                        }
+                    }
+                }
+                return AlarmEvalResult.NOT_YET_TRUE;
+            }, ctx);
+        }
+    }
+
+    private AlarmRuleState initRuleState(AlarmSeverity severity, AlarmRule rule, AlarmRuleState ruleState, AtomicBoolean reevalNeeded) {
+        if (ruleState == null) {
+            ruleState = new AlarmRuleState(severity, rule, this);
         } else {
-            clearRuleState = null;
+            // when restored
+            ruleState.setAlarmRule(rule);
+            if (rule.getCondition().getType() == AlarmConditionType.DURATION && !ruleState.isEmpty()) {
+                reevalNeeded.set(true);
+            }
         }
-        log.debug("Initialized create rule states {} and clear rule state {} for {}", createRuleStates, clearRuleState, ctx.getCalculatedField());
+        return ruleState;
     }
 
     @Override
@@ -125,8 +168,12 @@ public class AlarmCalculatedFieldState extends BaseCalculatedFieldState {
     }
 
     @Override
-    public void reset(CalculatedFieldCtx ctx) {
-        super.reset(ctx);
+    public void reset() {
+        super.reset();
+        createRuleStates.values().forEach(AlarmRuleState::clear);
+        if (clearRuleState != null) {
+            clearRuleState.clear();
+        }
     }
 
     @Override
@@ -135,9 +182,19 @@ public class AlarmCalculatedFieldState extends BaseCalculatedFieldState {
         TbAlarmResult result = createOrClearAlarms(state -> {
             if (updatedArgs != null) {
                 boolean newEvent = !updatedArgs.isEmpty();
-                return state.eval(newEvent, ctx);
+                AlarmEvalResult evalResult = state.eval(newEvent, ctx);
+                if (evalResult.getStatus() == NOT_YET_TRUE && evalResult.getLeftDuration() > 0) {
+                    // rounding up to the closest second
+//                    long leftDuration = (long) Math.ceil(evalResult.getLeftDuration() / 1000.0) * 1000;
+                    long leftDuration = evalResult.getLeftDuration();
+                    ScheduledFuture<?> future = ctx.scheduleReevaluation(leftDuration, actorCtx); // TODO: use single task for multiple durations if durations are close enough. but be careful when cancelling the task in one of the states
+                    if (future != null) {
+                        state.setDurationCheckFuture(future);
+                    }
+                }
+                return evalResult;
             } else {
-                return state.eval(System.currentTimeMillis());
+                return state.reeval(System.currentTimeMillis());
             }
         }, ctx);
         return Futures.immediateFuture(AlarmCalculatedFieldResult.builder()
@@ -177,11 +234,11 @@ public class AlarmCalculatedFieldState extends BaseCalculatedFieldState {
         for (AlarmRuleState state : createRuleStates.values()) {
             AlarmEvalResult evalResult = evalFunction.apply(state);
             log.debug("Evaluated create rule {} with args {}. Result: {}", state, arguments, evalResult);
-            if (evalResult == AlarmEvalResult.TRUE) {
+            if (evalResult.getStatus() == TRUE) {
                 resultState = state;
                 break;
-            } else if (evalResult == AlarmEvalResult.FALSE) {
-                clearAlarmState(state);
+            } else if (evalResult.getStatus() == FALSE) {
+                clearState(state);
             }
         }
 
@@ -189,15 +246,15 @@ public class AlarmCalculatedFieldState extends BaseCalculatedFieldState {
             result = calculateAlarmResult(resultState, ctx);
             resultStateInfo = resultState.getStateInfo();
             log.debug("Alarm result for state {}: {}", resultState, result);
-            clearAlarmState(clearRuleState);
+            clearState(clearRuleState);
         } else if (currentAlarm != null && clearRuleState != null) {
             AlarmEvalResult evalResult = evalFunction.apply(clearRuleState);
             log.debug("Evaluated clear rule {} with args {}. Result: {}", clearRuleState, arguments, evalResult);
-            if (evalResult == AlarmEvalResult.TRUE) {
+            if (evalResult.getStatus() == TRUE) {
                 resultStateInfo = clearRuleState.getStateInfo();
-                clearAlarmState(clearRuleState);
+                clearState(clearRuleState);
                 for (AlarmRuleState state : createRuleStates.values()) {
-                    clearAlarmState(state);
+                    clearState(state);
                 }
                 AlarmApiCallResult clearResult = ctx.getAlarmService().clearAlarm(
                         ctx.getTenantId(), currentAlarm.getId(), System.currentTimeMillis(), createDetails(clearRuleState), true
@@ -207,12 +264,11 @@ public class AlarmCalculatedFieldState extends BaseCalculatedFieldState {
                             .isCleared(true)
                             .alarm(clearResult.getAlarm())
                             .build();
-                    addStateInfo(result, clearRuleState);
                     resultState = clearRuleState;
                 }
                 currentAlarm = null;
-            } else if (evalResult == AlarmEvalResult.FALSE) {
-                clearAlarmState(clearRuleState);
+            } else if (evalResult.getStatus() == FALSE) {
+                clearState(clearRuleState);
             }
         }
         if (result != null && resultState != null) {
@@ -222,8 +278,9 @@ public class AlarmCalculatedFieldState extends BaseCalculatedFieldState {
         return result;
     }
 
-    private void clearAlarmState(AlarmRuleState state) {
+    private void clearState(AlarmRuleState state) {
         if (state != null) {
+            log.debug("Clearing rule state {}", state);
             state.clear();
         }
     }
@@ -283,14 +340,6 @@ public class AlarmCalculatedFieldState extends BaseCalculatedFieldState {
         }
     }
 
-    private void addStateInfo(TbAlarmResult alarmResult, AlarmRuleState ruleState) {
-        if (ruleState.getCondition().getType() == AlarmConditionType.REPEATING) {
-            alarmResult.setConditionRepeats(ruleState.getEventCount());
-        } else if (ruleState.getCondition().getType() == AlarmConditionType.DURATION) {
-            alarmResult.setConditionDuration(ruleState.getDuration());
-        }
-    }
-
     private JsonNode createDetails(AlarmRuleState ruleState) {
         JsonNode alarmDetails;
         String alarmDetailsStr = ruleState.getAlarmRule().getAlarmDetails();

application/src/main/java/org/thingsboard/server/service/cf/ctx/state/alarm/AlarmEvalResult.java

@@ -15,8 +15,31 @@
  */
 package org.thingsboard.server.service.cf.ctx.state.alarm;
 
-public enum AlarmEvalResult {
+import lombok.Data;
+import lombok.RequiredArgsConstructor;
 
-    FALSE, NOT_YET_TRUE, TRUE;
+@Data
+@RequiredArgsConstructor
+public class AlarmEvalResult {
+
+    public static final AlarmEvalResult TRUE = new AlarmEvalResult(Status.TRUE);
+    public static final AlarmEvalResult FALSE = new AlarmEvalResult(Status.FALSE);
+    public static final AlarmEvalResult NOT_YET_TRUE = new AlarmEvalResult(Status.NOT_YET_TRUE);
+
+    private final Status status;
+    private final long leftDuration;
+    private final long leftEvents;
+
+    public AlarmEvalResult(Status status) {
+        this(status, 0, 0);
+    }
+
+    public static AlarmEvalResult notYetTrue(long leftEvents, long leftDuration) {
+        return new AlarmEvalResult(Status.NOT_YET_TRUE, leftDuration, leftEvents);
+    }
+
+    public enum Status {
+        FALSE, NOT_YET_TRUE, TRUE;
+    }
 
 }

application/src/main/java/org/thingsboard/server/service/cf/ctx/state/alarm/AlarmRuleState.java

@@ -38,6 +38,7 @@ import java.time.Instant;
 import java.time.ZoneId;
 import java.time.ZonedDateTime;
 import java.util.Optional;
+import java.util.concurrent.ScheduledFuture;
 
 @Data
 @Slf4j
@@ -49,9 +50,11 @@ public class AlarmRuleState {
 
     private AlarmCondition condition;
 
-    private long lastEventTs;
-    private long duration;
     private long eventCount;
+    private long firstEventTs; // when duration condition started
+    private long lastEventTs;
+    private transient long duration;
+    private ScheduledFuture<?> durationCheckFuture;
 
     public AlarmRuleState(AlarmSeverity severity, AlarmRule alarmRule, AlarmCalculatedFieldState state) {
         this.severity = severity;
@@ -70,17 +73,22 @@ public class AlarmRuleState {
         };
     }
 
-    public AlarmEvalResult eval(long ts) { // on schedule
+    public AlarmEvalResult reeval(long ts) {
         switch (condition.getType()) {
             case SIMPLE, REPEATING -> {
                 return AlarmEvalResult.NOT_YET_TRUE;
             }
             case DURATION -> {
-                long requiredDurationInMs = getRequiredDurationInMs();
-                if (requiredDurationInMs > 0 && lastEventTs > 0 && ts > lastEventTs) {
-                    long duration = this.duration + (ts - lastEventTs);
+                long requiredDuration = getRequiredDurationInMs();
+                if (requiredDuration > 0 && lastEventTs > 0 && ts > lastEventTs) {
+                    duration = ts - firstEventTs;
                     if (isActive(ts)) {
-                        return duration > requiredDurationInMs ? AlarmEvalResult.TRUE : AlarmEvalResult.NOT_YET_TRUE;
+                        long leftDuration = requiredDuration - duration;
+                        if (leftDuration <= 0) {
+                            return AlarmEvalResult.TRUE;
+                        } else {
+                            return AlarmEvalResult.notYetTrue(0, leftDuration);
+                        }
                     } else {
                         return AlarmEvalResult.FALSE;
                     }
@@ -101,7 +109,8 @@ public class AlarmRuleState {
                 eventCount++;
             }
             long requiredRepeats = getIntValue(((RepeatingAlarmCondition) condition).getCount());
-            return eventCount >= requiredRepeats ? AlarmEvalResult.TRUE : AlarmEvalResult.NOT_YET_TRUE;
+            long leftRepeats = requiredRepeats - eventCount;
+            return leftRepeats <= 0 ? AlarmEvalResult.TRUE : AlarmEvalResult.notYetTrue(leftRepeats, 0);
         } else {
             return AlarmEvalResult.FALSE;
         }
@@ -109,17 +118,26 @@ public class AlarmRuleState {
 
     private AlarmEvalResult evalDuration(boolean active, CalculatedFieldCtx ctx) {
         if (active && eval(condition.getExpression(), ctx)) {
+            long eventTs = state.getLatestTimestamp();
             if (lastEventTs > 0) {
-                if (state.getLatestTimestamp() > lastEventTs) {
-                    duration = duration + (state.getLatestTimestamp() - lastEventTs);
-                    lastEventTs = state.getLatestTimestamp();
+                if (eventTs > lastEventTs) {
+                    if (firstEventTs == 0) {
+                        firstEventTs = lastEventTs;
+                    }
+                    lastEventTs = eventTs;
                 }
             } else {
-                lastEventTs = state.getLatestTimestamp();
-                duration = 0L;
+                firstEventTs = eventTs;
+                lastEventTs = eventTs;
+            }
+            duration = lastEventTs - firstEventTs;
+            long requiredDuration = getRequiredDurationInMs();
+            long leftDuration = requiredDuration - duration;
+            if (leftDuration <= 0) {
+                return AlarmEvalResult.TRUE;
+            } else {
+                return AlarmEvalResult.notYetTrue(0, leftDuration);
             }
-            long requiredDurationInMs = getRequiredDurationInMs();
-            return duration > requiredDurationInMs ? AlarmEvalResult.TRUE : AlarmEvalResult.NOT_YET_TRUE;
         } else {
             return AlarmEvalResult.FALSE;
         }
@@ -190,8 +208,17 @@ public class AlarmRuleState {
 
     public void clear() {
         eventCount = 0L;
+        firstEventTs = 0L;
         lastEventTs = 0L;
         duration = 0L;
+        if (durationCheckFuture != null) {
+            durationCheckFuture.cancel(true);
+            durationCheckFuture = null;
+        }
+    }
+
+    public boolean isEmpty() {
+        return eventCount == 0L && firstEventTs == 0L && lastEventTs == 0L && durationCheckFuture == null;
     }
 
     private Integer getIntValue(AlarmConditionValue<Integer> value) {
@@ -216,7 +243,7 @@ public class AlarmRuleState {
         if (condition.getType() == AlarmConditionType.REPEATING) {
             return new StateInfo(eventCount, null);
         } else if (condition.getType() == AlarmConditionType.DURATION) {
-            return new StateInfo(null, duration + (System.currentTimeMillis() - lastEventTs));
+            return new StateInfo(null, duration);
         } else {
             return StateInfo.EMPTY;
         }
@@ -227,9 +254,11 @@ public class AlarmRuleState {
         return "AlarmRuleState{" +
                "severity=" + severity +
                ", condition=" + condition +
+               ", eventCount=" + eventCount +
+               ", firstEventTs=" + firstEventTs +
                ", lastEventTs=" + lastEventTs +
                ", duration=" + duration +
-               ", eventCount=" + eventCount +
+               ", durationCheckFuture=" + durationCheckFuture +
                '}';
     }
 

application/src/main/java/org/thingsboard/server/service/cf/ctx/state/geofencing/GeofencingCalculatedFieldState.java

@@ -141,8 +141,8 @@ public class GeofencingCalculatedFieldState extends BaseCalculatedFieldState {
     }
 
     @Override
-    public void reset(CalculatedFieldCtx ctx) {
-        super.reset(ctx);
+    public void reset() {
+        super.reset();
         lastDynamicArgumentsRefreshTs = -1;
     }
 

application/src/main/java/org/thingsboard/server/service/entitiy/EntityStateSourcingListener.java

@@ -25,6 +25,7 @@ import org.thingsboard.common.util.JacksonUtil;
 import org.thingsboard.rule.engine.api.JobManager;
 import org.thingsboard.server.cluster.TbClusterService;
 import org.thingsboard.server.common.data.ApiUsageState;
+import org.thingsboard.server.common.data.Customer;
 import org.thingsboard.server.common.data.Device;
 import org.thingsboard.server.common.data.DeviceProfile;
 import org.thingsboard.server.common.data.EntityType;
@@ -148,6 +149,9 @@ public class EntityStateSourcingListener {
             case JOB -> {
                 onJobUpdate((Job) event.getEntity());
             }
+            case CUSTOMER -> {
+                tbClusterService.onCustomerUpdated((Customer) event.getEntity(), (Customer) event.getOldEntity());
+            }
             default -> {
             }
         }

application/src/main/java/org/thingsboard/server/service/entitiy/cf/DefaultTbCalculatedFieldService.java

@@ -22,6 +22,7 @@ import org.springframework.transaction.annotation.Transactional;
 import org.thingsboard.server.common.data.EntityType;
 import org.thingsboard.server.common.data.audit.ActionType;
 import org.thingsboard.server.common.data.cf.CalculatedField;
+import org.thingsboard.server.common.data.cf.CalculatedFieldType;
 import org.thingsboard.server.common.data.exception.ThingsboardException;
 import org.thingsboard.server.common.data.id.CalculatedFieldId;
 import org.thingsboard.server.common.data.id.EntityId;
@@ -33,7 +34,7 @@ import org.thingsboard.server.queue.util.TbCoreComponent;
 import org.thingsboard.server.service.entitiy.AbstractTbEntityService;
 import org.thingsboard.server.service.security.model.SecurityUser;
 
-import java.util.Optional;
+import java.util.Set;
 
 @TbCoreComponent
 @Service
@@ -52,7 +53,7 @@ public class DefaultTbCalculatedFieldService extends AbstractTbEntityService imp
                 CalculatedField existingCf = calculatedFieldService.findById(tenantId, calculatedField.getId());
                 checkForEntityChange(existingCf, calculatedField);
             }
-            checkEntityExistence(tenantId, calculatedField.getEntityId());
+            checkEntity(tenantId, calculatedField.getEntityId(), calculatedField.getType());
             CalculatedField savedCalculatedField = checkNotNull(calculatedFieldService.save(calculatedField));
             logEntityActionService.logEntityAction(tenantId, savedCalculatedField.getId(), savedCalculatedField, actionType, user);
             return savedCalculatedField;
@@ -68,10 +69,9 @@ public class DefaultTbCalculatedFieldService extends AbstractTbEntityService imp
     }
 
     @Override
-    public PageData<CalculatedField> findAllByTenantIdAndEntityId(EntityId entityId, SecurityUser user, PageLink pageLink) {
-        TenantId tenantId = user.getTenantId();
-        checkEntityExistence(tenantId, entityId);
-        return calculatedFieldService.findAllCalculatedFieldsByEntityId(tenantId, entityId, pageLink);
+    public PageData<CalculatedField> findByTenantIdAndEntityId(TenantId tenantId, EntityId entityId, CalculatedFieldType type, PageLink pageLink) {
+        checkEntity(tenantId, entityId, type);
+        return calculatedFieldService.findCalculatedFieldsByEntityId(tenantId, entityId, type, pageLink);
     }
 
     @Override
@@ -95,11 +95,15 @@ public class DefaultTbCalculatedFieldService extends AbstractTbEntityService imp
         }
     }
 
-    private void checkEntityExistence(TenantId tenantId, EntityId entityId) {
-        switch (entityId.getEntityType()) {
-            case ASSET, DEVICE, ASSET_PROFILE, DEVICE_PROFILE -> Optional.ofNullable(entityService.fetchEntity(tenantId, entityId))
-                    .orElseThrow(() -> new IllegalArgumentException(entityId.getEntityType().getNormalName() + " with id [" + entityId.getId() + "] does not exist."));
-            default -> throw new IllegalArgumentException("Entity type '" + entityId.getEntityType() + "' does not support calculated fields.");
+    private void checkEntity(TenantId tenantId, EntityId entityId, CalculatedFieldType type) {
+        EntityType entityType = entityId.getEntityType();
+        Set<CalculatedFieldType> supportedTypes = CalculatedField.SUPPORTED_ENTITIES.get(entityType);
+        if (supportedTypes == null || supportedTypes.isEmpty()) {
+            throw new IllegalArgumentException("Entity type '" + entityType + "' does not support calculated fields");
+        } else if (type != null && !supportedTypes.contains(type)) {
+            throw new IllegalArgumentException("Entity type '" + entityType + "' does not support '" + type + "' calculated fields");
+        } else if (entityService.fetchEntity(tenantId, entityId).isEmpty()) {
+            throw new IllegalArgumentException(entityType.getNormalName() + " with id [" + entityId.getId() + "] does not exist.");
         }
     }
 

application/src/main/java/org/thingsboard/server/service/entitiy/cf/TbCalculatedFieldService.java

@@ -16,9 +16,11 @@
 package org.thingsboard.server.service.entitiy.cf;
 
 import org.thingsboard.server.common.data.cf.CalculatedField;
+import org.thingsboard.server.common.data.cf.CalculatedFieldType;
 import org.thingsboard.server.common.data.exception.ThingsboardException;
 import org.thingsboard.server.common.data.id.CalculatedFieldId;
 import org.thingsboard.server.common.data.id.EntityId;
+import org.thingsboard.server.common.data.id.TenantId;
 import org.thingsboard.server.common.data.page.PageData;
 import org.thingsboard.server.common.data.page.PageLink;
 import org.thingsboard.server.service.security.model.SecurityUser;
@@ -29,7 +31,7 @@ public interface TbCalculatedFieldService {
 
     CalculatedField findById(CalculatedFieldId calculatedFieldId, SecurityUser user);
 
-    PageData<CalculatedField> findAllByTenantIdAndEntityId(EntityId entityId, SecurityUser user, PageLink pageLink);
+    PageData<CalculatedField> findByTenantIdAndEntityId(TenantId tenantId, EntityId entityId, CalculatedFieldType type, PageLink pageLink);
 
     void delete(CalculatedField calculatedField, SecurityUser user);
 

application/src/main/java/org/thingsboard/server/service/install/ProjectInfo.java

@@ -19,14 +19,17 @@ import lombok.RequiredArgsConstructor;
 import org.springframework.boot.info.BuildProperties;
 import org.springframework.stereotype.Component;
 
+import java.util.Optional;
+
 @Component
 @RequiredArgsConstructor
 public class ProjectInfo {
 
-    private final BuildProperties buildProperties;
+    private final Optional<BuildProperties> buildProperties;
 
     public String getProjectVersion() {
-        return buildProperties.getVersion().replaceAll("[^\\d.]", "");
+        return buildProperties.orElseThrow(() -> new IllegalStateException("Build properties are missing. Please rebuild the project with maven"))
+                .getVersion().replaceAll("[^\\d.]", "");
     }
 
     public String getProductType() {

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

@@ -26,6 +26,7 @@ import org.thingsboard.common.util.JacksonUtil;
 import org.thingsboard.server.cache.TbTransactionalCache;
 import org.thingsboard.server.cluster.TbClusterService;
 import org.thingsboard.server.common.data.ApiUsageState;
+import org.thingsboard.server.common.data.Customer;
 import org.thingsboard.server.common.data.DataConstants;
 import org.thingsboard.server.common.data.Device;
 import org.thingsboard.server.common.data.DeviceProfile;
@@ -468,6 +469,17 @@ public class DefaultTbClusterService implements TbClusterService {
         broadcastEntityStateChangeEvent(resource.getTenantId(), resource.getId(), ComponentLifecycleEvent.DELETED);
     }
 
+    @Override
+    public void onCustomerUpdated(Customer customer, Customer oldCustomer) {
+        ComponentLifecycleMsg msg = ComponentLifecycleMsg.builder()
+                .tenantId(customer.getTenantId())
+                .entityId(customer.getId())
+                .event(oldCustomer == null ? ComponentLifecycleEvent.CREATED : ComponentLifecycleEvent.UPDATED)
+                .ownerChanged(false) // for compatibility with PE
+                .build();
+        broadcast(msg);
+    }
+
     private <T> void broadcastEntityChangeToTransport(TenantId tenantId, EntityId entityid, T entity, TbQueueCallback callback) {
         String entityName = (entity instanceof HasName) ? ((HasName) entity).getName() : entity.getClass().getName();
         log.trace("[{}][{}][{}] Processing [{}] change event", tenantId, entityid.getEntityType(), entityid.getId(), entityName);
@@ -597,7 +609,8 @@ public class DefaultTbClusterService implements TbClusterService {
                 EntityType.DEVICE_PROFILE,
                 EntityType.ASSET_PROFILE,
                 EntityType.JOB,
-                EntityType.TB_RESOURCE)
+                EntityType.TB_RESOURCE,
+                EntityType.CUSTOMER)
                 || (entityType == EntityType.ASSET && msg.getEvent() == ComponentLifecycleEvent.UPDATED)
                 || (entityType == EntityType.DEVICE && msg.getEvent() == ComponentLifecycleEvent.UPDATED)
         ) {

application/src/main/java/org/thingsboard/server/service/queue/processing/AbstractConsumerService.java

@@ -209,9 +209,14 @@ public abstract class AbstractConsumerService<N extends com.google.protobuf.Gene
         } else if (EntityType.API_USAGE_STATE.equals(componentLifecycleMsg.getEntityId().getEntityType())) {
             apiUsageStateService.onApiUsageStateUpdate(tenantId);
         } else if (EntityType.CUSTOMER.equals(componentLifecycleMsg.getEntityId().getEntityType())) {
-            if (componentLifecycleMsg.getEvent() == ComponentLifecycleEvent.DELETED) {
+            if (componentLifecycleMsg.getEvent().equals(ComponentLifecycleEvent.CREATED)) {
+                calculatedFieldCache.addOwnerEntity(tenantId, componentLifecycleMsg.getEntityId());
+            } else if (componentLifecycleMsg.getEvent().equals(ComponentLifecycleEvent.UPDATED) && componentLifecycleMsg.isOwnerChanged()) {
+                calculatedFieldCache.updateOwnerEntity(tenantId, componentLifecycleMsg.getEntityId());
+            } else if (componentLifecycleMsg.getEvent() == ComponentLifecycleEvent.DELETED) {
                 apiUsageStateService.onCustomerDelete((CustomerId) componentLifecycleMsg.getEntityId());
                 calculatedFieldCache.evictOwner(componentLifecycleMsg.getEntityId());
+                calculatedFieldCache.evictEntity(componentLifecycleMsg.getEntityId());
             }
         } else if (EntityType.CALCULATED_FIELD.equals(componentLifecycleMsg.getEntityId().getEntityType())) {
             if (componentLifecycleMsg.getEvent() == ComponentLifecycleEvent.CREATED) {
@@ -221,7 +226,7 @@ public abstract class AbstractConsumerService<N extends com.google.protobuf.Gene
             } else {
                 calculatedFieldCache.evict((CalculatedFieldId) componentLifecycleMsg.getEntityId());
             }
-        }  else if (EntityType.TB_RESOURCE.equals(componentLifecycleMsg.getEntityId().getEntityType())) {
+        } else if (EntityType.TB_RESOURCE.equals(componentLifecycleMsg.getEntityId().getEntityType())) {
             tbResourceDataCache.evictResourceData(tenantId, new TbResourceId(componentLifecycleMsg.getEntityId().getId()));
         }
 

application/src/main/java/org/thingsboard/server/service/queue/ruleengine/TbRuleEngineQueueConsumerManager.java

@@ -38,6 +38,7 @@ import org.thingsboard.server.queue.common.consumer.MainQueueConsumerManager;
 import org.thingsboard.server.queue.common.consumer.TbQueueConsumerManagerTask;
 import org.thingsboard.server.queue.common.consumer.TbQueueConsumerManagerTask.DeleteQueueTask;
 import org.thingsboard.server.queue.common.consumer.TbQueueConsumerTask;
+import org.thingsboard.server.queue.common.consumer.TbQueueConsumerTask.ConsumerKey;
 import org.thingsboard.server.queue.discovery.QueueKey;
 import org.thingsboard.server.service.queue.TbMsgPackCallback;
 import org.thingsboard.server.service.queue.TbMsgPackProcessingContext;
@@ -127,7 +128,7 @@ public class TbRuleEngineQueueConsumerManager extends MainQueueConsumerManager<T
     @Override
     protected void processMsgs(List<TbProtoQueueMsg<ToRuleEngineMsg>> msgs,
                                TbQueueConsumer<TbProtoQueueMsg<ToRuleEngineMsg>> consumer,
-                               Object consumerKey,
+                               ConsumerKey consumerKey,
                                Queue queue) throws Exception {
         TbRuleEngineSubmitStrategy submitStrategy = getSubmitStrategy(queue);
         TbRuleEngineProcessingStrategy ackStrategy = getProcessingStrategy(queue);

application/src/main/java/org/thingsboard/server/service/telemetry/DefaultTelemetrySubscriptionService.java

@@ -169,7 +169,6 @@ public class DefaultTelemetrySubscriptionService extends AbstractSubscriptionSer
 
         addMainCallback(resultFuture, result -> {
             if (strategy.processCalculatedFields()) {
-                // TODO: divide CFs and alarm rules processing
                 calculatedFieldQueueService.pushRequestToQueue(request, result, request.getCallback());
             } else {
                 request.getCallback().onSuccess(null);

application/src/main/java/org/thingsboard/server/utils/CalculatedFieldUtils.java

@@ -114,12 +114,21 @@ public class CalculatedFieldUtils {
     private static AlarmRuleStateProto toAlarmRuleStateProto(AlarmRuleState ruleState) {
         return AlarmRuleStateProto.newBuilder()
                 .setSeverity(Optional.ofNullable(ruleState.getSeverity()).map(Enum::name).orElse(""))
-                .setLastEventTs(ruleState.getLastEventTs())
-                .setDuration(ruleState.getDuration())
                 .setEventCount(ruleState.getEventCount())
+                .setFirstEventTs(ruleState.getFirstEventTs())
+                .setLastEventTs(ruleState.getLastEventTs())
                 .build();
     }
 
+    private static AlarmRuleState fromAlarmRuleStateProto(AlarmRuleStateProto proto, AlarmCalculatedFieldState state) {
+        AlarmSeverity severity = StringUtils.isNotEmpty(proto.getSeverity()) ? AlarmSeverity.valueOf(proto.getSeverity()) : null;
+        AlarmRuleState ruleState = new AlarmRuleState(severity, null, state);
+        ruleState.setEventCount(proto.getEventCount());
+        ruleState.setFirstEventTs(proto.getFirstEventTs());
+        ruleState.setLastEventTs(proto.getLastEventTs());
+        return ruleState;
+    }
+
     public static SingleValueArgumentProto toSingleValueArgumentProto(String argName, SingleValueArgumentEntry entry) {
         SingleValueArgumentProto.Builder builder = SingleValueArgumentProto.newBuilder()
                 .setArgName(argName);
@@ -196,12 +205,11 @@ public class CalculatedFieldUtils {
                 AlarmCalculatedFieldState alarmState = (AlarmCalculatedFieldState) state;
                 AlarmStateProto alarmStateProto = proto.getAlarmState();
                 for (AlarmRuleStateProto ruleStateProto : alarmStateProto.getCreateRuleStatesList()) {
-                    AlarmSeverity severity = StringUtils.isNotEmpty(ruleStateProto.getSeverity()) ? AlarmSeverity.valueOf(ruleStateProto.getSeverity()) : null;
-                    AlarmRuleState ruleState = new AlarmRuleState(severity, null, alarmState);
-                    ruleState.setLastEventTs(ruleStateProto.getLastEventTs());
-                    ruleState.setDuration(ruleStateProto.getDuration());
-                    ruleState.setEventCount(ruleStateProto.getEventCount());
-                    alarmState.getCreateRuleStates().put(severity, ruleState);
+                    AlarmRuleState ruleState = fromAlarmRuleStateProto(ruleStateProto, alarmState);
+                    alarmState.getCreateRuleStates().put(ruleState.getSeverity(), ruleState);
+                }
+                if (alarmStateProto.hasClearRuleState()) {
+                    alarmState.setClearRuleState(fromAlarmRuleStateProto(alarmStateProto.getClearRuleState(), alarmState));
                 }
             }
         }

application/src/main/resources/thingsboard.yml

@@ -526,9 +526,6 @@ actors:
       configuration: "${ACTORS_CALCULATED_FIELD_DEBUG_MODE_RATE_LIMITS_PER_TENANT_CONFIGURATION:50000:3600}"
     # Time in seconds to receive calculation result.
     calculation_timeout: "${ACTORS_CALCULATION_TIMEOUT_SEC:5}"
-  alarms:
-    # Interval in seconds to re-evaluate Alarm rules with duration condition
-    reevaluation_interval: "${ACTORS_ALARMS_REEVALUATION_INTERVAL_SEC:60}"
 
 debug:
   settings:

application/src/test/java/org/thingsboard/server/cf/AlarmRulesTest.java

@@ -17,11 +17,9 @@ package org.thingsboard.server.cf;
 
 import lombok.Getter;
 import lombok.extern.slf4j.Slf4j;
-import org.assertj.core.api.Assertions;
 import org.junit.Before;
 import org.junit.Test;
 import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.test.context.TestPropertySource;
 import org.springframework.test.context.bean.override.mockito.MockitoSpyBean;
 import org.thingsboard.common.util.JacksonUtil;
 import org.thingsboard.rule.engine.action.TbAlarmResult;
@@ -43,6 +41,8 @@ import org.thingsboard.server.common.data.alarm.rule.condition.expression.Simple
 import org.thingsboard.server.common.data.alarm.rule.condition.expression.TbelAlarmConditionExpression;
 import org.thingsboard.server.common.data.alarm.rule.condition.expression.predicate.NumericFilterPredicate;
 import org.thingsboard.server.common.data.alarm.rule.condition.expression.predicate.NumericFilterPredicate.NumericOperation;
+import org.thingsboard.server.common.data.alarm.rule.condition.expression.predicate.StringFilterPredicate;
+import org.thingsboard.server.common.data.alarm.rule.condition.expression.predicate.StringFilterPredicate.StringOperation;
 import org.thingsboard.server.common.data.cf.CalculatedField;
 import org.thingsboard.server.common.data.cf.CalculatedFieldType;
 import org.thingsboard.server.common.data.cf.configuration.AlarmCalculatedFieldConfiguration;
@@ -57,6 +57,7 @@ import org.thingsboard.server.common.data.id.CalculatedFieldId;
 import org.thingsboard.server.common.data.id.DeviceId;
 import org.thingsboard.server.common.data.id.EntityId;
 import org.thingsboard.server.common.data.id.EventId;
+import org.thingsboard.server.common.data.page.PageLink;
 import org.thingsboard.server.controller.AbstractControllerTest;
 import org.thingsboard.server.dao.event.EventDao;
 import org.thingsboard.server.dao.service.DaoSqlTest;
@@ -72,9 +73,6 @@ import static org.testcontainers.shaded.org.awaitility.Awaitility.await;
 
 @Slf4j
 @DaoSqlTest
-@TestPropertySource(properties = {
-        "actors.alarms.reevaluation_interval=1"
-})
 public class AlarmRulesTest extends AbstractControllerTest {
 
     @MockitoSpyBean
@@ -85,6 +83,7 @@ public class AlarmRulesTest extends AbstractControllerTest {
 
     private Device device;
     private DeviceId deviceId;
+    private EntityId originatorId;
     private EventId latestEventId;
 
     @Before
@@ -92,6 +91,7 @@ public class AlarmRulesTest extends AbstractControllerTest {
         loginTenantAdmin();
         device = createDevice("Device A", "aaa");
         deviceId = device.getId();
+        originatorId = deviceId;
     }
 
     @Test
@@ -111,6 +111,8 @@ public class AlarmRulesTest extends AbstractControllerTest {
         Condition clearRule = new Condition("return temperature <= 25;", null, null);
         CalculatedField calculatedField = createAlarmCf(deviceId, "High Temperature Alarm",
                 arguments, createRules, clearRule);
+        assertThat(getCalculatedFields(deviceId, CalculatedFieldType.ALARM, new PageLink(1)).getData())
+                .singleElement().isEqualTo(calculatedField);
 
         postTelemetry(deviceId, "{\"temperature\":50}");
         checkAlarmResult(calculatedField, alarmResult -> {
@@ -177,7 +179,7 @@ public class AlarmRulesTest extends AbstractControllerTest {
     }
 
     /*
-     * todo: state restore (event count)
+     * todo: test state restore (event count)
      * */
     @Test
     public void testCreateAlarmForRepeatingCondition() throws Exception {
@@ -235,10 +237,9 @@ public class AlarmRulesTest extends AbstractControllerTest {
         Map<AlarmSeverity, Condition> createRules = Map.of(
                 AlarmSeverity.CRITICAL, new Condition("return powerConsumption >= 3000;", null, createDurationMs)
         );
-        long clearDurationMs = 2000L;
         Condition clearRule = new Condition("return powerConsumption < 3000;", null, createDurationMs);
 
-        CalculatedField calculatedField = createAlarmCf(deviceId, "High power consumption during 3 seconds",
+        CalculatedField calculatedField = createAlarmCf(deviceId, "High power consumption during 5 seconds",
                 arguments, createRules, clearRule);
         postTelemetry(deviceId, "{\"powerConsumption\":3500}");
         Thread.sleep(createDurationMs - 2000);
@@ -286,6 +287,53 @@ public class AlarmRulesTest extends AbstractControllerTest {
         });
     }
 
+    @Test
+    public void testCreateAndClearAlarm_customerAlarmRule_simpleExpression() throws Exception {
+        Argument locationArgument = new Argument();
+        locationArgument.setRefEntityKey(new ReferencedEntityKey("location", ArgumentType.ATTRIBUTE, AttributeScope.SERVER_SCOPE));
+        locationArgument.setDefaultValue("unknown");
+        originatorId = customerId;
+
+        Argument locationFilterArgument = new Argument();
+        locationFilterArgument.setRefEntityKey(new ReferencedEntityKey("locationFilter", ArgumentType.ATTRIBUTE, AttributeScope.SERVER_SCOPE));
+        locationFilterArgument.setRefDynamicSourceConfiguration(new CurrentOwnerDynamicSourceConfiguration());
+        locationFilterArgument.setDefaultValue("None");
+
+        Map<String, Argument> arguments = Map.of(
+                "location", locationArgument,
+                "locationFilter", locationFilterArgument
+        );
+
+        Map<AlarmSeverity, Condition> createRules = Map.of(
+                AlarmSeverity.INDETERMINATE, new Condition(createSimpleExpression(
+                        "location", StringOperation.CONTAINS, new AlarmConditionValue<>(null, "locationFilter")
+                ), null, null)
+        );
+        Condition clearRule = new Condition(createSimpleExpression(
+                "location", StringOperation.NOT_CONTAINS, new AlarmConditionValue<>(null, "locationFilter")
+        ), null, null);
+
+        CalculatedField calculatedField = createAlarmCf(customerId, "New resident",
+                arguments, createRules, clearRule);
+
+        loginSysAdmin();
+        postAttributes(tenantId, AttributeScope.SERVER_SCOPE, "{\"locationFilter\":\"Kyiv\"}");
+        loginTenantAdmin();
+        postAttributes(customerId, AttributeScope.SERVER_SCOPE, "{\"location\":\"Ukraine, Kyiv\"}");
+        checkAlarmResult(calculatedField, alarmResult -> {
+            assertThat(alarmResult.isCreated()).isTrue();
+            assertThat(alarmResult.getAlarm().getSeverity()).isEqualTo(AlarmSeverity.INDETERMINATE);
+            assertThat(alarmResult.getAlarm().getStatus()).isEqualTo(AlarmStatus.ACTIVE_UNACK);
+        });
+
+        postAttributes(customerId, AttributeScope.SERVER_SCOPE, "{\"location\":\"Ukraine, Lviv\"}");
+        checkAlarmResult(calculatedField, alarmResult -> {
+            assertThat(alarmResult.isCleared()).isTrue();
+            assertThat(alarmResult.getAlarm().getSeverity()).isEqualTo(AlarmSeverity.INDETERMINATE);
+            assertThat(alarmResult.getAlarm().getStatus()).isEqualTo(AlarmStatus.CLEARED_UNACK);
+        });
+    }
+
     private void checkAlarmResult(CalculatedField calculatedField, Consumer<TbAlarmResult> assertion) {
         await().atMost(TIMEOUT, TimeUnit.SECONDS).untilAsserted(() -> {
             TbAlarmResult alarmResult = getLatestAlarmResult(calculatedField.getId());
@@ -293,7 +341,7 @@ public class AlarmRulesTest extends AbstractControllerTest {
             assertion.accept(alarmResult);
 
             Alarm alarm = alarmResult.getAlarm();
-            assertThat(alarm.getOriginator()).isEqualTo(deviceId);
+            assertThat(alarm.getOriginator()).isEqualTo(originatorId);
             assertThat(alarm.getType()).isEqualTo(calculatedField.getName());
         });
     }
@@ -305,8 +353,7 @@ public class AlarmRulesTest extends AbstractControllerTest {
         }
         CalculatedFieldDebugEvent debugEvent = debugEvents.get(0);
         if (debugEvent.getError() != null) {
-            System.err.println("CF error: " + debugEvent.getError());
-            Assertions.fail();
+            throw new RuntimeException(debugEvent.getError());
         }
         if (debugEvent.getId().equals(latestEventId)) {
             return null;
@@ -383,6 +430,18 @@ public class AlarmRulesTest extends AbstractControllerTest {
         return rule;
     }
 
+    private SimpleAlarmConditionExpression createSimpleExpression(String argument, StringOperation stringOperation, AlarmConditionValue<String> conditionValue) {
+        SimpleAlarmConditionExpression simpleExpression = new SimpleAlarmConditionExpression();
+        AlarmConditionFilter filter = new AlarmConditionFilter();
+        filter.setArgument(argument);
+        StringFilterPredicate predicate = new StringFilterPredicate();
+        predicate.setOperation(stringOperation);
+        predicate.setValue(conditionValue);
+        filter.setPredicate(predicate);
+        simpleExpression.setFilters(List.of(filter));
+        return simpleExpression;
+    }
+
     private List<CalculatedFieldDebugEvent> getDebugEvents(CalculatedFieldId calculatedFieldId, int limit) {
         return eventDao.findLatestEvents(tenantId.getId(), calculatedFieldId.getId(), EventType.DEBUG_CALCULATED_FIELD, limit).stream()
                 .map(e -> (CalculatedFieldDebugEvent) e).toList();

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

@@ -93,6 +93,7 @@ import org.thingsboard.server.common.data.TenantProfile;
 import org.thingsboard.server.common.data.User;
 import org.thingsboard.server.common.data.asset.AssetProfile;
 import org.thingsboard.server.common.data.cf.CalculatedField;
+import org.thingsboard.server.common.data.cf.CalculatedFieldType;
 import org.thingsboard.server.common.data.device.data.DefaultDeviceConfiguration;
 import org.thingsboard.server.common.data.device.data.DefaultDeviceTransportConfiguration;
 import org.thingsboard.server.common.data.device.data.DeviceData;
@@ -1333,6 +1334,11 @@ public abstract class AbstractWebTest extends AbstractInMemoryStorageTest {
         return doPost("/api/calculatedField", calculatedField, CalculatedField.class);
     }
 
+    protected PageData<CalculatedField> getCalculatedFields(EntityId entityId, CalculatedFieldType type, PageLink pageLink) throws Exception {
+        return doGetTypedWithPageLink("/api/" + entityId.getEntityType() + "/" + entityId.getId() + "/calculatedFields" +
+                                      (type != null ? "?type=" + type.name() + "&" : "?"), new TypeReference<>() {}, pageLink);
+    }
+
     protected PageData<EventInfo> getDebugEvents(TenantId tenantId, EntityId entityId, int limit) throws Exception {
         return getEvents(tenantId, entityId, EventType.DEBUG_RULE_NODE, limit);
     }

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

@@ -35,7 +35,8 @@ import org.thingsboard.server.common.data.cf.configuration.SimpleCalculatedField
 import org.thingsboard.server.common.data.cf.configuration.geofencing.EntityCoordinates;
 import org.thingsboard.server.common.data.cf.configuration.geofencing.GeofencingCalculatedFieldConfiguration;
 import org.thingsboard.server.common.data.cf.configuration.geofencing.ZoneGroupConfiguration;
-import org.thingsboard.server.common.data.id.DeviceId;
+import org.thingsboard.server.common.data.id.EntityId;
+import org.thingsboard.server.common.data.page.PageLink;
 import org.thingsboard.server.common.data.relation.EntityRelation;
 import org.thingsboard.server.common.data.relation.EntitySearchDirection;
 import org.thingsboard.server.common.data.relation.RelationPathLevel;
@@ -190,6 +191,18 @@ public class CalculatedFieldControllerTest extends AbstractControllerTest {
                 .andExpect(status().isOk());
     }
 
+    @Test
+    public void testGetCalculatedFields() throws Exception {
+        Device testDevice = createDevice("Test device", "1234567890");
+        CalculatedField calculatedField = getSimpleCalculatedField(testDevice.getId());
+        calculatedField = doPost("/api/calculatedField", calculatedField, CalculatedField.class);
+
+        assertThat(getCalculatedFields(testDevice.getId(), null, new PageLink(10)).getData())
+                .singleElement().isEqualTo(calculatedField);
+        assertThat(getCalculatedFields(testDevice.getId(), CalculatedFieldType.SIMPLE, new PageLink(10)).getData())
+                .singleElement().isEqualTo(calculatedField);
+    }
+
     @Test
     public void testDeleteCalculatedField() throws Exception {
         Device testDevice = createDevice("Test device", "1234567890");
@@ -204,17 +217,17 @@ public class CalculatedFieldControllerTest extends AbstractControllerTest {
         doGet("/api/calculatedField/" + savedCalculatedField.getId().getId()).andExpect(status().isNotFound());
     }
 
-    private CalculatedField getSimpleCalculatedField(DeviceId deviceId) {
-        return getCalculatedField(deviceId, CalculatedFieldType.SIMPLE);
+    private CalculatedField getSimpleCalculatedField(EntityId entityId) {
+        return getCalculatedField(entityId, CalculatedFieldType.SIMPLE);
     }
 
-    private CalculatedField getCalculatedField(DeviceId deviceId, CalculatedFieldType cfType) {
-        return getCalculatedField(deviceId, cfType, null);
+    private CalculatedField getCalculatedField(EntityId entityId, CalculatedFieldType cfType) {
+        return getCalculatedField(entityId, cfType, null);
     }
 
-    private CalculatedField getCalculatedField(DeviceId deviceId, CalculatedFieldType cfType, CalculatedFieldConfiguration customConfiguration) {
+    private CalculatedField getCalculatedField(EntityId entityId, CalculatedFieldType cfType, CalculatedFieldConfiguration customConfiguration) {
         CalculatedField calculatedField = new CalculatedField();
-        calculatedField.setEntityId(deviceId);
+        calculatedField.setEntityId(entityId);
         calculatedField.setType(cfType);
         calculatedField.setName("Test Calculated Field");
         calculatedField.setConfigurationVersion(1);

application/src/test/java/org/thingsboard/server/service/cf/ctx/state/GeofencingCalculatedFieldStateTest.java

@@ -104,7 +104,8 @@ public class GeofencingCalculatedFieldStateTest {
         ctx = new CalculatedFieldCtx(getCalculatedField(), systemContext);
         ctx.init();
         state = new GeofencingCalculatedFieldState(ctx.getEntityId());
-        state.init(ctx);
+        state.setCtx(ctx, null);
+        state.init();
     }
 
     @Test

application/src/test/java/org/thingsboard/server/service/cf/ctx/state/PropagationCalculatedFieldStateTest.java

@@ -103,7 +103,8 @@ public class PropagationCalculatedFieldStateTest {
         ctx.init();
 
         state = new PropagationCalculatedFieldState(ctx.getEntityId());
-        state.init(ctx);
+        state.setCtx(ctx, null);
+        state.init();
     }
 
     @Test

application/src/test/java/org/thingsboard/server/service/cf/ctx/state/ScriptCalculatedFieldStateTest.java

@@ -87,7 +87,8 @@ public class ScriptCalculatedFieldStateTest {
         ctx = new CalculatedFieldCtx(getCalculatedField(), systemContext);
         ctx.init();
         state = new ScriptCalculatedFieldState(ctx.getEntityId());
-        state.init(ctx);
+        state.setCtx(ctx, null);
+        state.init();
     }
 
     @Test

application/src/test/java/org/thingsboard/server/service/cf/ctx/state/SimpleCalculatedFieldStateTest.java

@@ -79,7 +79,8 @@ public class SimpleCalculatedFieldStateTest {
         ctx = new CalculatedFieldCtx(getCalculatedField(), systemContext);
         ctx.init();
         state = new SimpleCalculatedFieldState(ctx.getEntityId());
-        state.init(ctx);
+        state.setCtx(ctx, null);
+        state.init();
     }
 
     @Test

application/src/test/java/org/thingsboard/server/service/queue/ruleengine/TbRuleEngineQueueConsumerManagerTest.java

@@ -595,7 +595,7 @@ public class TbRuleEngineQueueConsumerManagerTest {
             await().atMost(5, TimeUnit.SECONDS).until(() -> {
                 for (TopicPartitionInfo partition : expectedPartitions) {
                     if (consumers.stream().noneMatch(consumer -> consumer.subscribed &&
-                            consumer.pollingStarted && Set.of(partition).equals(consumer.getPartitions()))) {
+                                                                 consumer.pollingStarted && Set.of(partition).equals(consumer.getPartitions()))) {
                         return false;
                     }
                 }
@@ -605,7 +605,7 @@ public class TbRuleEngineQueueConsumerManagerTest {
             await().atMost(5, TimeUnit.SECONDS).until(() -> {
                 return consumers.size() == 1 && consumers.stream()
                         .anyMatch(consumer -> consumer.subscribed && consumer.pollingStarted &&
-                                expectedPartitions.equals(consumer.getPartitions()));
+                                              expectedPartitions.equals(consumer.getPartitions()));
             });
         }
         Mockito.reset(ruleEngineConsumerContext.getSubmitStrategyFactory());
@@ -667,8 +667,8 @@ public class TbRuleEngineQueueConsumerManagerTest {
         return await().atMost(5, TimeUnit.SECONDS)
                 .until(() -> consumers.stream()
                         .filter(consumer -> consumer.getPartitions() != null &&
-                                consumer.getPartitions().size() == 1 &&
-                                consumer.getPartitions().contains(tpi))
+                                            consumer.getPartitions().size() == 1 &&
+                                            consumer.getPartitions().contains(tpi))
                         .findFirst().orElse(null), Objects::nonNull);
     }
 
@@ -676,9 +676,9 @@ public class TbRuleEngineQueueConsumerManagerTest {
         return await().atMost(5, TimeUnit.SECONDS)
                 .until(() -> consumers.stream()
                         .filter(consumer -> consumer.getPartitions() != null &&
-                                consumer.getPartitions().size() == 1 &&
-                                consumer.getPartitions().stream()
-                                        .anyMatch(tpi -> tpi.getPartition().get().equals(partition)))
+                                            consumer.getPartitions().size() == 1 &&
+                                            consumer.getPartitions().stream()
+                                                    .anyMatch(tpi -> tpi.getPartition().get().equals(partition)))
                         .findFirst().orElse(null), Objects::nonNull);
     }
 
@@ -778,10 +778,6 @@ public class TbRuleEngineQueueConsumerManagerTest {
             return false;
         }
 
-        public Set<TopicPartitionInfo> getPartitions() {
-            return partitions;
-        }
-
         public void setUpTestMsg() {
             testMsg = TbMsg.newMsg()
                     .type(TbMsgType.POST_TELEMETRY_REQUEST)
@@ -790,6 +786,7 @@ public class TbRuleEngineQueueConsumerManagerTest {
                     .data("{}")
                     .build();
         }
+
     }
 
 }

application/src/test/java/org/thingsboard/server/service/queue/ruleengine/TbRuleEngineStrategyTest.java

@@ -43,6 +43,7 @@ import org.thingsboard.server.common.msg.queue.ServiceType;
 import org.thingsboard.server.gen.transport.TransportProtos.ToRuleEngineMsg;
 import org.thingsboard.server.queue.TbQueueConsumer;
 import org.thingsboard.server.queue.common.TbProtoQueueMsg;
+import org.thingsboard.server.queue.common.consumer.TbQueueConsumerTask.ConsumerKey;
 import org.thingsboard.server.queue.discovery.QueueKey;
 import org.thingsboard.server.service.queue.processing.TbRuleEngineProcessingStrategyFactory;
 import org.thingsboard.server.service.queue.processing.TbRuleEngineSubmitStrategyFactory;
@@ -191,6 +192,7 @@ public class TbRuleEngineStrategyTest {
         queue.setProcessingStrategy(processingStrategy);
 
         QueueKey queueKey = new QueueKey(ServiceType.TB_RULE_ENGINE, queue);
+        ConsumerKey consumerKey = new ConsumerKey(queueKey, null);
         var consumerManager = TbRuleEngineQueueConsumerManager.create()
                 .ctx(ruleEngineConsumerContext)
                 .queueKey(queueKey)
@@ -238,7 +240,7 @@ public class TbRuleEngineStrategyTest {
                 .map(this::toProto)
                 .toList();
 
-        consumerManager.processMsgs(protoMsgs, consumer, queueKey, queue);
+        consumerManager.processMsgs(protoMsgs, consumer, consumerKey, queue);
 
         processingData.forEach(data -> {
             verify(actorContext, times(data.attempts)).tell(argThat(msg ->

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

@@ -16,6 +16,7 @@
 package org.thingsboard.server.cluster;
 
 import org.thingsboard.server.common.data.ApiUsageState;
+import org.thingsboard.server.common.data.Customer;
 import org.thingsboard.server.common.data.Device;
 import org.thingsboard.server.common.data.DeviceProfile;
 import org.thingsboard.server.common.data.TbResourceInfo;
@@ -130,6 +131,8 @@ public interface TbClusterService extends TbQueueClusterService {
 
     void sendNotificationMsgToEdge(TenantId tenantId, EdgeId edgeId, EntityId entityId, String body, EdgeEventType type, EdgeEventActionType action, EdgeId sourceEdgeId);
 
+    void onCustomerUpdated(Customer customer, Customer oldCustomer);
+
     void onCalculatedFieldUpdated(CalculatedField calculatedField, CalculatedField oldCalculatedField, TbQueueCallback callback);
 
     void onCalculatedFieldDeleted(CalculatedField calculatedField, TbQueueCallback callback);

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

@@ -38,6 +38,8 @@ public interface TbQueueConsumer<T extends TbQueueMsg> {
 
     boolean isStopped();
 
+    Set<TopicPartitionInfo> getPartitions();
+
     List<String> getFullTopicNames();
 
 }

common/dao-api/src/main/java/org/thingsboard/server/dao/cf/CalculatedFieldService.java

@@ -17,6 +17,7 @@ package org.thingsboard.server.dao.cf;
 
 import org.thingsboard.server.common.data.cf.CalculatedField;
 import org.thingsboard.server.common.data.cf.CalculatedFieldLink;
+import org.thingsboard.server.common.data.cf.CalculatedFieldType;
 import org.thingsboard.server.common.data.id.CalculatedFieldId;
 import org.thingsboard.server.common.data.id.CalculatedFieldLinkId;
 import org.thingsboard.server.common.data.id.EntityId;
@@ -45,7 +46,7 @@ public interface CalculatedFieldService extends EntityDaoService {
 
     PageData<CalculatedField> findCalculatedFieldsByTenantId(TenantId tenantId, PageLink pageLink);
 
-    PageData<CalculatedField> findAllCalculatedFieldsByEntityId(TenantId tenantId, EntityId entityId, PageLink pageLink);
+    PageData<CalculatedField> findCalculatedFieldsByEntityId(TenantId tenantId, EntityId entityId, CalculatedFieldType type, PageLink pageLink);
 
     void deleteCalculatedField(TenantId tenantId, CalculatedFieldId calculatedFieldId);
 

common/data/src/main/java/org/thingsboard/server/common/data/alarm/rule/condition/AlarmConditionValue.java

@@ -15,9 +15,13 @@
  */
 package org.thingsboard.server.common.data.alarm.rule.condition;
 
+import lombok.AllArgsConstructor;
 import lombok.Data;
+import lombok.NoArgsConstructor;
 
 @Data
+@AllArgsConstructor
+@NoArgsConstructor
 public class AlarmConditionValue<T> {
 
     private T staticValue;

common/data/src/main/java/org/thingsboard/server/common/data/cf/CalculatedField.java

@@ -25,6 +25,7 @@ import lombok.EqualsAndHashCode;
 import lombok.Getter;
 import lombok.Setter;
 import org.thingsboard.server.common.data.BaseData;
+import org.thingsboard.server.common.data.EntityType;
 import org.thingsboard.server.common.data.HasDebugSettings;
 import org.thingsboard.server.common.data.HasName;
 import org.thingsboard.server.common.data.HasTenantId;
@@ -39,6 +40,10 @@ import org.thingsboard.server.common.data.validation.Length;
 import org.thingsboard.server.common.data.validation.NoXss;
 
 import java.io.Serial;
+import java.util.Collections;
+import java.util.EnumSet;
+import java.util.Map;
+import java.util.Set;
 
 @Schema
 @Data
@@ -48,6 +53,18 @@ public class CalculatedField extends BaseData<CalculatedFieldId> implements HasN
     @Serial
     private static final long serialVersionUID = 4491966747773381420L;
 
+    public static final Map<EntityType, Set<CalculatedFieldType>> SUPPORTED_ENTITIES = Map.of(
+            EntityType.DEVICE, CalculatedFieldType.all,
+            EntityType.ASSET, CalculatedFieldType.all,
+            EntityType.DEVICE_PROFILE, CalculatedFieldType.all,
+            EntityType.ASSET_PROFILE, CalculatedFieldType.all,
+            EntityType.CUSTOMER, Set.of(CalculatedFieldType.ALARM)
+    );
+
+    public static final Set<EntityType> SUPPORTED_REFERENCED_ENTITIES = Collections.unmodifiableSet(EnumSet.of(
+            EntityType.DEVICE, EntityType.ASSET, EntityType.CUSTOMER, EntityType.TENANT
+    ));
+
     private TenantId tenantId;
     private EntityId entityId;
 

common/data/src/main/java/org/thingsboard/server/common/data/cf/CalculatedFieldType.java

@@ -15,10 +15,18 @@
  */
 package org.thingsboard.server.common.data.cf;
 
+import java.util.Collections;
+import java.util.EnumSet;
+import java.util.Set;
+
 public enum CalculatedFieldType {
+
     SIMPLE,
     SCRIPT,
     GEOFENCING,
     ALARM,
-    PROPAGATION
+    PROPAGATION;
+
+    public static final Set<CalculatedFieldType> all = Collections.unmodifiableSet(EnumSet.allOf(CalculatedFieldType.class));
+
 }

common/data/src/main/java/org/thingsboard/server/common/data/cf/configuration/AlarmCalculatedFieldConfiguration.java

@@ -20,7 +20,6 @@ import jakarta.validation.constraints.NotEmpty;
 import lombok.Data;
 import org.thingsboard.server.common.data.alarm.AlarmSeverity;
 import org.thingsboard.server.common.data.alarm.rule.AlarmRule;
-import org.thingsboard.server.common.data.alarm.rule.condition.AlarmConditionType;
 import org.thingsboard.server.common.data.cf.CalculatedFieldType;
 
 import java.util.List;
@@ -57,10 +56,4 @@ public class AlarmCalculatedFieldConfiguration implements ArgumentsBasedCalculat
 
     }
 
-    @Override
-    public boolean requiresScheduledReevaluation() {
-        return createRules.values().stream().anyMatch(rule -> rule.getCondition().getType() == AlarmConditionType.DURATION) ||
-               (clearRule != null && clearRule.getCondition().getType() == AlarmConditionType.DURATION);
-    }
-
 }

common/data/src/main/java/org/thingsboard/server/common/data/cf/configuration/CalculatedFieldConfiguration.java

@@ -72,8 +72,4 @@ public interface CalculatedFieldConfiguration {
                 .collect(Collectors.toList());
     }
 
-    default boolean requiresScheduledReevaluation() {
-        return false;
-    }
-
 }

common/message/src/main/java/org/thingsboard/server/common/msg/CalculatedFieldStatePartitionRestoreMsg.java

@@ -0,0 +1,37 @@
+/**
+ * Copyright © 2016-2025 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;
+
+import lombok.Data;
+import org.thingsboard.server.common.data.id.TenantId;
+import org.thingsboard.server.common.msg.queue.TopicPartitionInfo;
+
+@Data
+public class CalculatedFieldStatePartitionRestoreMsg implements ToCalculatedFieldSystemMsg {
+
+    private final TopicPartitionInfo partition;
+
+    @Override
+    public TenantId getTenantId() {
+        return TenantId.SYS_TENANT_ID;
+    }
+
+    @Override
+    public MsgType getMsgType() {
+        return MsgType.CF_STATE_PARTITION_RESTORE_MSG;
+    }
+
+}

common/message/src/main/java/org/thingsboard/server/common/msg/MsgType.java

@@ -138,6 +138,7 @@ public enum MsgType {
 
     CF_CACHE_INIT_MSG, // Sent to init caches for CF actor;
     CF_STATE_RESTORE_MSG, // Sent to restore particular calculated field entity state;
+    CF_STATE_PARTITION_RESTORE_MSG,
     CF_PARTITIONS_CHANGE_MSG, // Sent when cluster event occures;
 
     CF_ENTITY_LIFECYCLE_MSG, // Sent on CF/Device/Asset create/update/delete;

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

@@ -1914,7 +1914,7 @@ message AlarmStateProto {
 
 message AlarmRuleStateProto {
   string severity = 1;
-  int64 lastEventTs = 2;
-  int64 duration = 3;
-  int64 eventCount = 4;
+  int64 eventCount = 2;
+  int64 firstEventTs = 3;
+  int64 lastEventTs = 4;
 }

common/queue/src/main/java/org/thingsboard/server/queue/common/AbstractTbQueueConsumerTemplate.java

@@ -194,6 +194,11 @@ public abstract class AbstractTbQueueConsumerTemplate<R, T extends TbQueueMsg> i
 
     abstract protected void doUnsubscribe();
 
+    @Override
+    public Set<TopicPartitionInfo> getPartitions() {
+        return partitions;
+    }
+
     @Override
     public List<String> getFullTopicNames() {
         if (partitions == null) {

common/queue/src/main/java/org/thingsboard/server/queue/common/consumer/MainQueueConsumerManager.java

@@ -25,6 +25,7 @@ import org.thingsboard.server.queue.TbQueueConsumer;
 import org.thingsboard.server.queue.TbQueueMsg;
 import org.thingsboard.server.queue.common.consumer.TbQueueConsumerManagerTask.UpdateConfigTask;
 import org.thingsboard.server.queue.common.consumer.TbQueueConsumerManagerTask.UpdatePartitionsTask;
+import org.thingsboard.server.queue.common.consumer.TbQueueConsumerTask.ConsumerKey;
 import org.thingsboard.server.queue.kafka.TbKafkaConsumerTemplate;
 
 import java.util.Collection;
@@ -218,7 +219,7 @@ public class MainQueueConsumerManager<M extends TbQueueMsg, C extends QueueConfi
         consumerTask.setTask(consumerLoop);
     }
 
-    private void consumerLoop(Object consumerKey, TbQueueConsumer<M> consumer) {
+    private void consumerLoop(ConsumerKey consumerKey, TbQueueConsumer<M> consumer) {
         try {
             while (!stopped && !consumer.isStopped()) {
                 try {
@@ -250,7 +251,7 @@ public class MainQueueConsumerManager<M extends TbQueueMsg, C extends QueueConfi
         }
     }
 
-    protected void processMsgs(List<M> msgs, TbQueueConsumer<M> consumer, Object consumerKey, C config) throws Exception {
+    protected void processMsgs(List<M> msgs, TbQueueConsumer<M> consumer, ConsumerKey consumerKey, C config) throws Exception {
         log.trace("Processing {} messages", msgs.size());
         msgPackProcessor.process(msgs, consumer, consumerKey, config);
         log.trace("Processed {} messages", msgs.size());
@@ -273,7 +274,7 @@ public class MainQueueConsumerManager<M extends TbQueueMsg, C extends QueueConfi
     }
 
     public interface MsgPackProcessor<M extends TbQueueMsg, C extends QueueConfig> {
-        void process(List<M> msgs, TbQueueConsumer<M> consumer, Object consumerKey, C config) throws Exception;
+        void process(List<M> msgs, TbQueueConsumer<M> consumer, ConsumerKey consumerKey, C config) throws Exception;
     }
 
     public interface ConsumerWrapper<M extends TbQueueMsg> {
@@ -285,6 +286,7 @@ public class MainQueueConsumerManager<M extends TbQueueMsg, C extends QueueConfi
     }
 
     class ConsumerPerPartitionWrapper implements ConsumerWrapper<M> {
+
         private final Map<TopicPartitionInfo, TbQueueConsumerTask<M>> consumers = new HashMap<>();
 
         @Override
@@ -307,8 +309,7 @@ public class MainQueueConsumerManager<M extends TbQueueMsg, C extends QueueConfi
 
         protected void addPartitions(Set<TopicPartitionInfo> partitions, Consumer<TopicPartitionInfo> onStop, Function<String, Long> startOffsetProvider) {
             partitions.forEach(tpi -> {
-                Integer partitionId = tpi.getPartition().orElse(-1);
-                String key = queueKey + "-" + partitionId;
+                ConsumerKey key = new ConsumerKey(queueKey, tpi);
                 Runnable callback = onStop != null ? () -> onStop.accept(tpi) : null;
 
                 TbQueueConsumerTask<M> consumer = new TbQueueConsumerTask<>(key, () -> {
@@ -328,9 +329,11 @@ public class MainQueueConsumerManager<M extends TbQueueMsg, C extends QueueConfi
         public Collection<TbQueueConsumerTask<M>> getConsumers() {
             return consumers.values();
         }
+
     }
 
     class SingleConsumerWrapper implements ConsumerWrapper<M> {
+
         private TbQueueConsumerTask<M> consumer;
 
         @Override
@@ -346,7 +349,7 @@ public class MainQueueConsumerManager<M extends TbQueueMsg, C extends QueueConfi
             }
 
             if (consumer == null) {
-                consumer = new TbQueueConsumerTask<>(queueKey, () -> consumerCreator.apply(config, null), null); // no partitionId passed
+                consumer = new TbQueueConsumerTask<>(new ConsumerKey(queueKey, null), () -> consumerCreator.apply(config, null), null); // no partitionId passed
             }
             consumer.subscribe(partitions);
             if (!consumer.isRunning()) {
@@ -361,5 +364,7 @@ public class MainQueueConsumerManager<M extends TbQueueMsg, C extends QueueConfi
             }
             return List.of(consumer);
         }
+
     }
+
 }

common/queue/src/main/java/org/thingsboard/server/queue/common/consumer/TbQueueConsumerTask.java

@@ -32,7 +32,7 @@ import java.util.function.Supplier;
 public class TbQueueConsumerTask<M extends TbQueueMsg> {
 
     @Getter
-    private final Object key;
+    private final ConsumerKey key;
     private volatile TbQueueConsumer<M> consumer;
     private volatile Supplier<TbQueueConsumer<M>> consumerSupplier;
     @Getter
@@ -41,7 +41,7 @@ public class TbQueueConsumerTask<M extends TbQueueMsg> {
     @Setter
     private Future<?> task;
 
-    public TbQueueConsumerTask(Object key, Supplier<TbQueueConsumer<M>> consumerSupplier, Runnable callback) {
+    public TbQueueConsumerTask(ConsumerKey key, Supplier<TbQueueConsumer<M>> consumerSupplier, Runnable callback) {
         this.key = key;
         this.consumer = null;
         this.consumerSupplier = consumerSupplier;
@@ -97,4 +97,18 @@ public class TbQueueConsumerTask<M extends TbQueueMsg> {
         return task != null;
     }
 
+    public record ConsumerKey(Object queueKey, TopicPartitionInfo partition) {
+
+        @Override
+        public String toString() {
+            if (partition != null) {
+                Integer partitionId = partition.getPartition().orElse(-1);
+                return queueKey + "-" + partitionId;
+            } else {
+                return queueKey.toString();
+            }
+        }
+
+    }
+
 }

common/queue/src/main/java/org/thingsboard/server/queue/common/state/DefaultQueueStateService.java

@@ -15,10 +15,15 @@
  */
 package org.thingsboard.server.queue.common.state;
 
+import org.thingsboard.server.common.msg.queue.TopicPartitionInfo;
 import org.thingsboard.server.queue.TbQueueMsg;
 import org.thingsboard.server.queue.common.consumer.PartitionedQueueConsumerManager;
+import org.thingsboard.server.queue.discovery.QueueKey;
 
 import java.util.Collections;
+import java.util.Set;
+
+import static org.thingsboard.server.common.msg.queue.TopicPartitionInfo.withTopic;
 
 public class DefaultQueueStateService<E extends TbQueueMsg, S extends TbQueueMsg> extends QueueStateService<E, S> {
 
@@ -26,4 +31,18 @@ public class DefaultQueueStateService<E extends TbQueueMsg, S extends TbQueueMsg
         super(eventConsumer, Collections.emptyList());
     }
 
+    @Override
+    protected void addPartitions(QueueKey queueKey, Set<TopicPartitionInfo> partitions, RestoreCallback callback) {
+        if (callback != null) {
+            for (TopicPartitionInfo partition : partitions) {
+                callback.onPartitionRestored(partition);
+            }
+            callback.onAllPartitionsRestored();
+        }
+        eventConsumer.addPartitions(partitions);
+        for (PartitionedQueueConsumerManager<?> consumer : otherConsumers) {
+            consumer.addPartitions(withTopic(partitions, consumer.getTopic()));
+        }
+    }
+
 }

common/queue/src/main/java/org/thingsboard/server/queue/common/state/KafkaQueueStateService.java

@@ -50,7 +50,7 @@ public class KafkaQueueStateService<E extends TbQueueMsg, S extends TbQueueMsg>
     }
 
     @Override
-    protected void addPartitions(QueueKey queueKey, Set<TopicPartitionInfo> partitions, Runnable whenAllProcessed) {
+    protected void addPartitions(QueueKey queueKey, Set<TopicPartitionInfo> partitions, RestoreCallback callback) {
         Map<String, Long> eventsStartOffsets = eventsStartOffsetsProvider != null ? eventsStartOffsetsProvider.get() : null; // remembering the offsets before subscribing to states
 
         Set<TopicPartitionInfo> statePartitions = withTopic(partitions, stateConsumer.getTopic());
@@ -61,10 +61,13 @@ public class KafkaQueueStateService<E extends TbQueueMsg, S extends TbQueueMsg>
             try {
                 partitionsInProgress.remove(statePartition);
                 log.info("Finished partition {} (still in progress: {})", statePartition, partitionsInProgress);
+                if (callback != null) {
+                    callback.onPartitionRestored(statePartition);
+                }
                 if (partitionsInProgress.isEmpty()) {
                     log.info("All partitions processed");
-                    if (whenAllProcessed != null) {
-                        whenAllProcessed.run();
+                    if (callback != null) {
+                        callback.onAllPartitionsRestored();
                     }
                 }
 

common/queue/src/main/java/org/thingsboard/server/queue/common/state/QueueStateService.java

@@ -49,7 +49,7 @@ public abstract class QueueStateService<E extends TbQueueMsg, S extends TbQueueM
         this.otherConsumers = otherConsumers;
     }
 
-    public void update(QueueKey queueKey, Set<TopicPartitionInfo> newPartitions, Runnable whenAllProcessed) {
+    public void update(QueueKey queueKey, Set<TopicPartitionInfo> newPartitions, RestoreCallback callback) {
         newPartitions = withTopic(newPartitions, eventConsumer.getTopic());
         var writeLock = partitionsLock.writeLock();
         writeLock.lock();
@@ -71,23 +71,15 @@ public abstract class QueueStateService<E extends TbQueueMsg, S extends TbQueueM
         }
 
         if (!addedPartitions.isEmpty()) {
-            addPartitions(queueKey, addedPartitions, whenAllProcessed);
+            addPartitions(queueKey, addedPartitions, callback);
         } else {
-            if (whenAllProcessed != null) {
-                whenAllProcessed.run();
+            if (callback != null) {
+                callback.onAllPartitionsRestored();
             }
         }
     }
 
-    protected void addPartitions(QueueKey queueKey, Set<TopicPartitionInfo> partitions, Runnable whenAllProcessed) {
-        if (whenAllProcessed != null) {
-            whenAllProcessed.run();
-        }
-        eventConsumer.addPartitions(partitions);
-        for (PartitionedQueueConsumerManager<?> consumer : otherConsumers) {
-            consumer.addPartitions(withTopic(partitions, consumer.getTopic()));
-        }
-    }
+    protected abstract void addPartitions(QueueKey queueKey, Set<TopicPartitionInfo> partitions, RestoreCallback callback) ;
 
     protected void removePartitions(QueueKey queueKey, Set<TopicPartitionInfo> partitions) {
         eventConsumer.removePartitions(partitions);
@@ -122,4 +114,12 @@ public abstract class QueueStateService<E extends TbQueueMsg, S extends TbQueueM
         eventConsumer.awaitStop();
     }
 
+    public interface RestoreCallback {
+
+        void onAllPartitionsRestored();
+
+        default void onPartitionRestored(TopicPartitionInfo partition) {}
+
+    }
+
 }

common/queue/src/main/java/org/thingsboard/server/queue/memory/InMemoryTbQueueConsumer.java

@@ -109,6 +109,11 @@ public class InMemoryTbQueueConsumer<T extends TbQueueMsg> implements TbQueueCon
         return stopped;
     }
 
+    @Override
+    public Set<TopicPartitionInfo> getPartitions() {
+        return partitions;
+    }
+
     @Override
     public List<String> getFullTopicNames() {
         return partitions.stream().map(TopicPartitionInfo::getFullTopicName).collect(Collectors.toList());

dao/src/main/java/org/thingsboard/server/dao/cf/BaseCalculatedFieldService.java

@@ -21,6 +21,7 @@ import org.springframework.stereotype.Service;
 import org.thingsboard.server.common.data.EntityType;
 import org.thingsboard.server.common.data.cf.CalculatedField;
 import org.thingsboard.server.common.data.cf.CalculatedFieldLink;
+import org.thingsboard.server.common.data.cf.CalculatedFieldType;
 import org.thingsboard.server.common.data.cf.configuration.CalculatedFieldConfiguration;
 import org.thingsboard.server.common.data.id.CalculatedFieldId;
 import org.thingsboard.server.common.data.id.CalculatedFieldLinkId;
@@ -33,10 +34,13 @@ import org.thingsboard.server.dao.entity.AbstractEntityService;
 import org.thingsboard.server.dao.eventsourcing.DeleteEntityEvent;
 import org.thingsboard.server.dao.eventsourcing.SaveEntityEvent;
 import org.thingsboard.server.dao.exception.IncorrectParameterException;
-import org.thingsboard.server.dao.service.DataValidator;
+import org.thingsboard.server.dao.service.validator.CalculatedFieldDataValidator;
+import org.thingsboard.server.dao.service.validator.CalculatedFieldLinkDataValidator;
 
+import java.util.EnumSet;
 import java.util.List;
 import java.util.Optional;
+import java.util.Set;
 
 import static org.thingsboard.server.dao.service.Validator.validateId;
 import static org.thingsboard.server.dao.service.Validator.validatePageLink;
@@ -52,8 +56,8 @@ public class BaseCalculatedFieldService extends AbstractEntityService implements
 
     private final CalculatedFieldDao calculatedFieldDao;
     private final CalculatedFieldLinkDao calculatedFieldLinkDao;
-    private final DataValidator<CalculatedField> calculatedFieldDataValidator;
-    private final DataValidator<CalculatedFieldLink> calculatedFieldLinkDataValidator;
+    private final CalculatedFieldDataValidator calculatedFieldDataValidator;
+    private final CalculatedFieldLinkDataValidator calculatedFieldLinkDataValidator;
 
     @Override
     public CalculatedField save(CalculatedField calculatedField) {
@@ -135,11 +139,18 @@ public class BaseCalculatedFieldService extends AbstractEntityService implements
     }
 
     @Override
-    public PageData<CalculatedField> findAllCalculatedFieldsByEntityId(TenantId tenantId, EntityId entityId, PageLink pageLink) {
+    public PageData<CalculatedField> findCalculatedFieldsByEntityId(TenantId tenantId, EntityId entityId, CalculatedFieldType type, PageLink pageLink) {
         log.trace("Executing findAllByEntityId, entityId [{}], pageLink [{}]", entityId, pageLink);
         validateId(entityId.getId(), id -> INCORRECT_ENTITY_ID + id);
         validatePageLink(pageLink);
-        return calculatedFieldDao.findAllByEntityId(tenantId, entityId, pageLink);
+        Set<CalculatedFieldType> types;
+        if (type == null) {
+            types = EnumSet.allOf(CalculatedFieldType.class);
+            types.remove(CalculatedFieldType.ALARM);
+        } else {
+            types = Set.of(type);
+        }
+        return calculatedFieldDao.findByEntityIdAndTypes(tenantId, entityId, types, pageLink);
     }
 
     @Override

dao/src/main/java/org/thingsboard/server/dao/cf/CalculatedFieldDao.java

@@ -16,6 +16,7 @@
 package org.thingsboard.server.dao.cf;
 
 import org.thingsboard.server.common.data.cf.CalculatedField;
+import org.thingsboard.server.common.data.cf.CalculatedFieldType;
 import org.thingsboard.server.common.data.id.CalculatedFieldId;
 import org.thingsboard.server.common.data.id.EntityId;
 import org.thingsboard.server.common.data.id.TenantId;
@@ -24,6 +25,7 @@ import org.thingsboard.server.common.data.page.PageLink;
 import org.thingsboard.server.dao.Dao;
 
 import java.util.List;
+import java.util.Set;
 
 public interface CalculatedFieldDao extends Dao<CalculatedField> {
 
@@ -41,7 +43,7 @@ public interface CalculatedFieldDao extends Dao<CalculatedField> {
 
     PageData<CalculatedField> findAllByTenantId(TenantId tenantId, PageLink pageLink);
 
-    PageData<CalculatedField> findAllByEntityId(TenantId tenantId, EntityId entityId, PageLink pageLink);
+    PageData<CalculatedField> findByEntityIdAndTypes(TenantId tenantId, EntityId entityId, Set<CalculatedFieldType> types, PageLink pageLink);
 
     List<CalculatedField> removeAllByEntityId(TenantId tenantId, EntityId entityId);
 

dao/src/main/java/org/thingsboard/server/dao/customer/CustomerServiceImpl.java

@@ -144,9 +144,10 @@ public class CustomerServiceImpl extends AbstractCachedEntityService<CustomerCac
 
     private Customer saveCustomer(Customer customer, boolean doValidate) {
         log.trace("Executing saveCustomer [{}]", customer);
+        Customer oldCustomer = null;
         String oldCustomerTitle = null;
         if (doValidate) {
-            Customer oldCustomer = customerValidator.validate(customer, Customer::getTenantId);
+            oldCustomer = customerValidator.validate(customer, Customer::getTenantId);
             if (oldCustomer != null) {
                 oldCustomerTitle = oldCustomer.getTitle();
             }
@@ -161,8 +162,13 @@ public class CustomerServiceImpl extends AbstractCachedEntityService<CustomerCac
                 countService.publishCountEntityEvictEvent(savedCustomer.getTenantId(), EntityType.CUSTOMER);
             }
             publishEvictEvent(evictEvent);
-            eventPublisher.publishEvent(SaveEntityEvent.builder().tenantId(savedCustomer.getTenantId())
-                    .entityId(savedCustomer.getId()).entity(savedCustomer).created(customer.getId() == null).build());
+            eventPublisher.publishEvent(SaveEntityEvent.builder()
+                    .tenantId(savedCustomer.getTenantId())
+                    .entityId(savedCustomer.getId())
+                    .entity(savedCustomer)
+                    .oldEntity(oldCustomer)
+                    .created(customer.getId() == null)
+                    .build());
             return savedCustomer;
         } catch (Exception e) {
             handleEvictEvent(evictEvent);

dao/src/main/java/org/thingsboard/server/dao/sql/cf/CalculatedFieldRepository.java

@@ -38,9 +38,9 @@ public interface CalculatedFieldRepository extends JpaRepository<CalculatedField
     Page<CalculatedFieldEntity> findAllByTenantId(UUID tenantId, Pageable pageable);
 
     @Query("SELECT cf FROM CalculatedFieldEntity cf WHERE cf.tenantId = :tenantId " +
-            "AND cf.entityId = :entityId " +
-            "AND (:textSearch IS NULL OR ilike(cf.name, CONCAT('%', :textSearch, '%')) = true)")
-    Page<CalculatedFieldEntity> findAllByTenantIdAndEntityId(UUID tenantId, UUID entityId, String textSearch, Pageable pageable);
+           "AND cf.entityId = :entityId AND cf.type IN :types " +
+           "AND (:textSearch IS NULL OR ilike(cf.name, CONCAT('%', :textSearch, '%')) = true)")
+    Page<CalculatedFieldEntity> findByTenantIdAndEntityIdAndTypes(UUID tenantId, UUID entityId, List<String> types, String textSearch, Pageable pageable);
 
     List<CalculatedFieldEntity> findAllByTenantId(UUID tenantId);
 

dao/src/main/java/org/thingsboard/server/dao/sql/cf/JpaCalculatedFieldDao.java

@@ -22,6 +22,7 @@ import org.springframework.data.jpa.repository.JpaRepository;
 import org.springframework.stereotype.Component;
 import org.thingsboard.server.common.data.EntityType;
 import org.thingsboard.server.common.data.cf.CalculatedField;
+import org.thingsboard.server.common.data.cf.CalculatedFieldType;
 import org.thingsboard.server.common.data.id.CalculatedFieldId;
 import org.thingsboard.server.common.data.id.EntityId;
 import org.thingsboard.server.common.data.id.TenantId;
@@ -34,6 +35,7 @@ import org.thingsboard.server.dao.sql.JpaAbstractDao;
 import org.thingsboard.server.dao.util.SqlDao;
 
 import java.util.List;
+import java.util.Set;
 import java.util.UUID;
 
 @Slf4j
@@ -83,9 +85,10 @@ public class JpaCalculatedFieldDao extends JpaAbstractDao<CalculatedFieldEntity,
     }
 
     @Override
-    public PageData<CalculatedField> findAllByEntityId(TenantId tenantId, EntityId entityId, PageLink pageLink) {
-        log.debug("Try to find calculated fields by entityId[{}] and pageLink [{}]", entityId, pageLink);
-        return DaoUtil.toPageData(calculatedFieldRepository.findAllByTenantIdAndEntityId(tenantId.getId(), entityId.getId(), pageLink.getTextSearch(), DaoUtil.toPageable(pageLink)));
+    public PageData<CalculatedField> findByEntityIdAndTypes(TenantId tenantId, EntityId entityId, Set<CalculatedFieldType> types, PageLink pageLink) {
+        log.debug("Try to find calculated fields by entityId [{}] and type [{}] and pageLink [{}]", entityId, types, pageLink);
+        return DaoUtil.toPageData(calculatedFieldRepository.findByTenantIdAndEntityIdAndTypes(tenantId.getId(), entityId.getId(),
+                types.stream().map(Enum::name).toList(), pageLink.getTextSearch(), DaoUtil.toPageable(pageLink)));
     }
 
     @Override

pom.xml

@@ -1929,9 +1929,6 @@
             <name>Typesafe Repository</name>
             <url>https://repo.typesafe.com/typesafe/releases/</url>
         </repository>
-        <repository>
-            <id>sonatype</id>
-            <url>https://oss.sonatype.org/content/groups/public</url>
-        </repository>
     </repositories>
+
 </project>

rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/TbCalculatedFieldsNode.java

@@ -44,13 +44,13 @@ import static org.thingsboard.server.common.data.DataConstants.SCOPE;
 @Slf4j
 @RuleNode(
         type = ComponentType.ACTION,
-        name = "calculated fields",
+        name = "calculated fields and alarm rules",
         configClazz = EmptyNodeConfiguration.class,
-        nodeDescription = "Pushes incoming messages to calculated fields service",
-        nodeDetails = "Node enables the processing of calculated fields without persisting incoming messages to the database. " +
-                "By default, the processing of calculated fields is triggered by the <b>save attributes</b> and <b>save time series</b> nodes. " +
+        nodeDescription = "Pushes incoming messages to calculated fields and alarm rules services",
+        nodeDetails = "Node enables the processing of calculated fields and alarm rules without persisting incoming messages to the database. " +
+                "By default, the processing of calculated fields and alarm rules is triggered by the <b>save attributes</b> and <b>save time series</b> nodes. " +
                 "This rule node accepts the same messages as these nodes but allows you to trigger the processing of calculated " +
-                "fields independently, ensuring that derived data can be computed and utilized in real time without storing the original message in the database.",
+                "fields or alarm rules independently, ensuring that derived data can be computed and utilized in real time without storing the original message in the database.",
         configDirective = "tbNodeEmptyConfig",
         icon = "published_with_changes"
 )

ui-ngx/src/app/modules/home/components/rule-node/action/advanced-processing-setting.component.html

@@ -39,6 +39,6 @@
   ></tb-advanced-processing-setting-row>
   <tb-advanced-processing-setting-row *ngIf="calculatedFields"
     formControlName="calculatedFields"
-    title="{{ 'rule-node-config.save-time-series.calculated-fields' | translate }}"
+    title="{{ 'rule-node-config.save-time-series.calculated-fields-and-alarm-rules' | translate }}"
   ></tb-advanced-processing-setting-row>
 </section>

ui-ngx/src/app/modules/home/components/widget/config/basic/chart/doughnut-basic-config.component.ts

@@ -86,9 +86,11 @@ export class DoughnutBasicConfigComponent extends LatestChartBasicConfigComponen
     if (totalEnabled) {
       latestChartWidgetConfigForm.get('totalValueFont').enable();
       latestChartWidgetConfigForm.get('totalValueColor').enable();
+      latestChartWidgetConfigForm.get('legendShowTotal').disable();
     } else {
       latestChartWidgetConfigForm.get('totalValueFont').disable();
       latestChartWidgetConfigForm.get('totalValueColor').disable();
+      latestChartWidgetConfigForm.get('legendShowTotal').enable();
     }
   }
 }

ui-ngx/src/app/modules/home/components/widget/lib/settings/chart/doughnut-widget-settings.component.ts

@@ -69,9 +69,11 @@ export class DoughnutWidgetSettingsComponent extends LatestChartWidgetSettingsCo
     if (totalEnabled) {
       latestChartWidgetSettingsForm.get('totalValueFont').enable();
       latestChartWidgetSettingsForm.get('totalValueColor').enable();
+      latestChartWidgetSettingsForm.get('legendShowTotal').disable();
     } else {
       latestChartWidgetSettingsForm.get('totalValueFont').disable();
       latestChartWidgetSettingsForm.get('totalValueColor').disable();
+      latestChartWidgetSettingsForm.get('legendShowTotal').enable();
     }
   }
 }

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

@@ -5505,7 +5505,7 @@
             "time-series": "Time series",
             "latest": "Latest values",
             "web-sockets": "WebSockets",
-            "calculated-fields": "Calculated fields"
+            "calculated-fields-and-alarm-rules": "Calculated fields and alarm rules"
         },
         "save-attribute": {
             "processing-settings": "Processing settings",