Merge pull request #13040 from irynamatveieva/cf-fixes

CF: fixes

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

@@ -30,9 +30,9 @@ import org.springframework.context.annotation.Lazy;
 import org.springframework.data.redis.core.RedisTemplate;
 import org.springframework.stereotype.Component;
 import org.thingsboard.common.util.JacksonUtil;
+import org.thingsboard.rule.engine.api.DeviceStateManager;
 import org.thingsboard.rule.engine.api.MailService;
 import org.thingsboard.rule.engine.api.NotificationCenter;
-import org.thingsboard.rule.engine.api.DeviceStateManager;
 import org.thingsboard.rule.engine.api.SmsService;
 import org.thingsboard.rule.engine.api.notification.SlackService;
 import org.thingsboard.rule.engine.api.sms.SmsSenderFactory;
@@ -110,7 +110,6 @@ import org.thingsboard.server.service.apiusage.TbApiUsageStateService;
 import org.thingsboard.server.service.cf.CalculatedFieldProcessingService;
 import org.thingsboard.server.service.cf.CalculatedFieldQueueService;
 import org.thingsboard.server.service.cf.CalculatedFieldStateService;
-import org.thingsboard.server.service.cf.cache.CalculatedFieldEntityProfileCache;
 import org.thingsboard.server.service.cf.ctx.state.ArgumentEntry;
 import org.thingsboard.server.service.component.ComponentDiscoveryService;
 import org.thingsboard.server.service.edge.rpc.EdgeRpcService;
@@ -546,11 +545,6 @@ public class ActorSystemContext {
     @Getter
     private CalculatedFieldQueueService calculatedFieldQueueService;
 
-    @Lazy
-    @Autowired(required = false)
-    @Getter
-    private CalculatedFieldEntityProfileCache calculatedFieldEntityProfileCache;
-
     @Value("${actors.session.max_concurrent_sessions_per_device:1}")
     @Getter
     private int maxConcurrentSessionsPerDevice;

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

@@ -113,6 +113,8 @@ public class AppActor extends ContextAwareActor {
             case SESSION_TIMEOUT_MSG:
                 ctx.broadcastToChildrenByType(msg, EntityType.TENANT);
                 break;
+            case CF_CACHE_INIT_MSG:
+            case CF_INIT_PROFILE_ENTITY_MSG:
             case CF_INIT_MSG:
             case CF_LINK_INIT_MSG:
             case CF_STATE_RESTORE_MSG:

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

@@ -22,8 +22,10 @@ import org.thingsboard.server.actors.TbActorException;
 import org.thingsboard.server.common.data.id.TenantId;
 import org.thingsboard.server.common.msg.TbActorStopReason;
 import org.thingsboard.server.common.msg.ToCalculatedFieldSystemMsg;
+import org.thingsboard.server.common.msg.cf.CalculatedFieldCacheInitMsg;
 import org.thingsboard.server.common.msg.cf.CalculatedFieldEntityLifecycleMsg;
 import org.thingsboard.server.common.msg.cf.CalculatedFieldInitMsg;
+import org.thingsboard.server.common.msg.cf.CalculatedFieldInitProfileEntityMsg;
 import org.thingsboard.server.common.msg.cf.CalculatedFieldLinkInitMsg;
 import org.thingsboard.server.common.msg.cf.CalculatedFieldPartitionChangeMsg;
 
@@ -65,6 +67,12 @@ public class CalculatedFieldManagerActor extends AbstractCalculatedFieldActor {
             case CF_PARTITIONS_CHANGE_MSG:
                 processor.onPartitionChange((CalculatedFieldPartitionChangeMsg) msg);
                 break;
+            case CF_CACHE_INIT_MSG:
+                processor.onCacheInitMsg((CalculatedFieldCacheInitMsg) msg);
+                break;
+            case CF_INIT_PROFILE_ENTITY_MSG:
+                processor.onProfileEntityMsg((CalculatedFieldInitProfileEntityMsg) msg);
+                break;
             case CF_INIT_MSG:
                 processor.onFieldInitMsg((CalculatedFieldInitMsg) msg);
                 break;

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

@@ -15,7 +15,9 @@
  */
 package org.thingsboard.server.actors.calculatedField;
 
+import lombok.Getter;
 import lombok.extern.slf4j.Slf4j;
+import org.springframework.beans.factory.annotation.Value;
 import org.thingsboard.server.actors.ActorSystemContext;
 import org.thingsboard.server.actors.TbActorCtx;
 import org.thingsboard.server.actors.TbActorRef;
@@ -24,6 +26,7 @@ import org.thingsboard.server.actors.service.DefaultActorService;
 import org.thingsboard.server.actors.shared.AbstractContextAwareMsgProcessor;
 import org.thingsboard.server.common.data.DataConstants;
 import org.thingsboard.server.common.data.EntityType;
+import org.thingsboard.server.common.data.ProfileEntityIdInfo;
 import org.thingsboard.server.common.data.cf.CalculatedField;
 import org.thingsboard.server.common.data.cf.CalculatedFieldLink;
 import org.thingsboard.server.common.data.id.AssetId;
@@ -31,17 +34,22 @@ 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.TenantId;
+import org.thingsboard.server.common.data.page.PageDataIterable;
+import org.thingsboard.server.common.msg.cf.CalculatedFieldCacheInitMsg;
 import org.thingsboard.server.common.msg.cf.CalculatedFieldEntityLifecycleMsg;
 import org.thingsboard.server.common.msg.cf.CalculatedFieldInitMsg;
+import org.thingsboard.server.common.msg.cf.CalculatedFieldInitProfileEntityMsg;
 import org.thingsboard.server.common.msg.cf.CalculatedFieldLinkInitMsg;
 import org.thingsboard.server.common.msg.cf.CalculatedFieldPartitionChangeMsg;
 import org.thingsboard.server.common.msg.plugin.ComponentLifecycleMsg;
 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.device.DeviceService;
 import org.thingsboard.server.service.cf.CalculatedFieldProcessingService;
 import org.thingsboard.server.service.cf.CalculatedFieldStateService;
-import org.thingsboard.server.service.cf.cache.CalculatedFieldEntityProfileCache;
+import org.thingsboard.server.service.cf.cache.TenantEntityProfileCache;
 import org.thingsboard.server.service.cf.ctx.CalculatedFieldEntityCtxId;
 import org.thingsboard.server.service.cf.ctx.state.CalculatedFieldCtx;
 import org.thingsboard.server.service.profile.TbAssetProfileCache;
@@ -68,22 +76,30 @@ public class CalculatedFieldManagerMessageProcessor extends AbstractContextAware
 
     private final CalculatedFieldProcessingService cfExecService;
     private final CalculatedFieldStateService cfStateService;
-    private final CalculatedFieldEntityProfileCache cfEntityCache;
     private final CalculatedFieldService cfDaoService;
+    private final DeviceService deviceService;
+    private final AssetService assetService;
     private final TbAssetProfileCache assetProfileCache;
     private final TbDeviceProfileCache deviceProfileCache;
+    private final TenantEntityProfileCache entityProfileCache;
     protected final TenantId tenantId;
 
     protected TbActorCtx ctx;
 
+    @Value("${queue.calculated_fields.init_tenant_fetch_pack_size:1000}")
+    @Getter
+    private int initFetchPackSize;
+
     CalculatedFieldManagerMessageProcessor(ActorSystemContext systemContext, TenantId tenantId) {
         super(systemContext);
-        this.cfEntityCache = systemContext.getCalculatedFieldEntityProfileCache();
         this.cfExecService = systemContext.getCalculatedFieldProcessingService();
         this.cfStateService = systemContext.getCalculatedFieldStateService();
         this.cfDaoService = systemContext.getCalculatedFieldService();
+        this.deviceService = systemContext.getDeviceService();
+        this.assetService = systemContext.getAssetService();
         this.assetProfileCache = systemContext.getAssetProfileCache();
         this.deviceProfileCache = systemContext.getDeviceProfileCache();
+        this.entityProfileCache = new TenantEntityProfileCache();
         this.tenantId = tenantId;
     }
 
@@ -100,6 +116,19 @@ public class CalculatedFieldManagerMessageProcessor extends AbstractContextAware
         ctx.stop(ctx.getSelf());
     }
 
+    public void onCacheInitMsg(CalculatedFieldCacheInitMsg msg) {
+        log.debug("[{}] Processing CF actor init message.", msg.getTenantId().getId());
+        initEntityProfileCache();
+        initCalculatedFields();
+        msg.getCallback().onSuccess();
+    }
+
+    public void onProfileEntityMsg(CalculatedFieldInitProfileEntityMsg msg) {
+        log.debug("[{}] Processing profile entity message.", msg.getTenantId().getId());
+        entityProfileCache.add(msg.getProfileEntityId(), msg.getEntityId());
+        msg.getCallback().onSuccess();
+    }
+
     public void onFieldInitMsg(CalculatedFieldInitMsg msg) throws CalculatedFieldException {
         log.debug("[{}] Processing CF init message.", msg.getCf().getId());
         var cf = msg.getCf();
@@ -180,16 +209,35 @@ public class CalculatedFieldManagerMessageProcessor extends AbstractContextAware
                 }
                 break;
             }
+            case DEVICE_PROFILE:
+            case ASSET_PROFILE: {
+                switch (event) {
+                    case DELETED:
+                        onProfileDeleted(msg.getData(), msg.getCallback());
+                        break;
+                    default:
+                        msg.getCallback().onSuccess();
+                        break;
+                }
+                break;
+            }
             default: {
                 msg.getCallback().onSuccess();
             }
         }
     }
 
+    private void onProfileDeleted(ComponentLifecycleMsg msg, TbCallback callback) {
+        entityProfileCache.removeProfileId(msg.getEntityId());
+        callback.onSuccess();
+    }
+
     private void onEntityCreated(ComponentLifecycleMsg msg, TbCallback callback) {
         EntityId entityId = msg.getEntityId();
         EntityId profileId = getProfileId(tenantId, entityId);
-        cfEntityCache.add(tenantId, profileId, entityId);
+        if (profileId != null) {
+            entityProfileCache.add(profileId, entityId);
+        }
         if (!isMyPartition(entityId, callback)) {
             return;
         }
@@ -207,7 +255,7 @@ public class CalculatedFieldManagerMessageProcessor extends AbstractContextAware
 
     private void onEntityUpdated(ComponentLifecycleMsg msg, TbCallback callback) {
         if (msg.getOldProfileId() != null && !msg.getOldProfileId().equals(msg.getProfileId())) {
-            cfEntityCache.update(tenantId, msg.getOldProfileId(), msg.getProfileId(), msg.getEntityId());
+            entityProfileCache.update(msg.getOldProfileId(), msg.getProfileId(), msg.getEntityId());
             if (!isMyPartition(msg.getEntityId(), callback)) {
                 return;
             }
@@ -226,7 +274,7 @@ public class CalculatedFieldManagerMessageProcessor extends AbstractContextAware
     }
 
     private void onEntityDeleted(ComponentLifecycleMsg msg, TbCallback callback) {
-        cfEntityCache.evict(tenantId, msg.getEntityId());
+        entityProfileCache.removeEntityId(msg.getEntityId());
         if (isMyPartition(msg.getEntityId(), callback)) {
             log.debug("Pushing entity lifecycle msg to specific actor [{}]", msg.getEntityId());
             getOrCreateActor(msg.getEntityId()).tell(new CalculatedFieldEntityDeleteMsg(tenantId, msg.getEntityId(), callback));
@@ -322,11 +370,15 @@ public class CalculatedFieldManagerMessageProcessor extends AbstractContextAware
             EntityId entityId = cfCtx.getEntityId();
             EntityType entityType = cfCtx.getEntityId().getEntityType();
             if (isProfileEntity(entityType)) {
-                var entityIds = cfEntityCache.getMyEntityIdsByProfileId(tenantId, entityId);
+                var entityIds = entityProfileCache.getEntityIdsByProfileId(entityId);
                 if (!entityIds.isEmpty()) {
                     //TODO: no need to do this if we cache all created actors and know which one belong to us;
                     var multiCallback = new MultipleTbCallback(entityIds.size(), callback);
-                    entityIds.forEach(id -> deleteCfForEntity(id, cfId, multiCallback));
+                    entityIds.forEach(id -> {
+                        if (isMyPartition(id, multiCallback)) {
+                            deleteCfForEntity(id, cfId, multiCallback);
+                        }
+                    });
                 } else {
                     callback.onSuccess();
                 }
@@ -366,10 +418,11 @@ public class CalculatedFieldManagerMessageProcessor extends AbstractContextAware
         EntityId sourceEntityId = msg.getEntityId();
         log.debug("Received linked telemetry msg from entity [{}]", sourceEntityId);
         var proto = msg.getProto();
+        var callback = msg.getCallback();
         var linksList = proto.getLinksList();
         if (linksList.isEmpty()) {
             log.debug("[{}] No CF links to process new telemetry.", msg.getTenantId());
-            msg.getCallback().onSuccess();
+            callback.onSuccess();
         }
         for (var linkProto : linksList) {
             var link = fromProto(linkProto);
@@ -378,21 +431,23 @@ public class CalculatedFieldManagerMessageProcessor extends AbstractContextAware
             var cf = calculatedFields.get(link.cfId());
             if (EntityType.DEVICE_PROFILE.equals(targetEntityType) || EntityType.ASSET_PROFILE.equals(targetEntityType)) {
                 // iterate over all entities that belong to profile and push the message for corresponding CF
-                var entityIds = cfEntityCache.getMyEntityIdsByProfileId(tenantId, targetEntityId);
+                var entityIds = entityProfileCache.getEntityIdsByProfileId(targetEntityId);
                 if (!entityIds.isEmpty()) {
-                    MultipleTbCallback callback = new MultipleTbCallback(entityIds.size(), msg.getCallback());
+                    MultipleTbCallback multipleCallback = new MultipleTbCallback(entityIds.size(), callback);
-                    var newMsg = new EntityCalculatedFieldLinkedTelemetryMsg(tenantId, sourceEntityId, proto.getMsg(), cf, callback);
+                    var newMsg = new EntityCalculatedFieldLinkedTelemetryMsg(tenantId, sourceEntityId, proto.getMsg(), cf, multipleCallback);
                     entityIds.forEach(entityId -> {
-                        log.debug("Pushing linked telemetry msg to specific actor [{}]", entityId);
+                        if (isMyPartition(entityId, multipleCallback)) {
-                        getOrCreateActor(entityId).tell(newMsg);
+                            log.debug("Pushing linked telemetry msg to specific actor [{}]", entityId);
+                            getOrCreateActor(entityId).tell(newMsg);
+                        }
                     });
                 } else {
-                    msg.getCallback().onSuccess();
+                    callback.onSuccess();
                 }
             } else {
-                if (isMyPartition(targetEntityId, msg.getCallback())) {
+                if (isMyPartition(targetEntityId, callback)) {
                     log.debug("Pushing linked telemetry msg to specific actor [{}]", targetEntityId);
-                    var newMsg = new EntityCalculatedFieldLinkedTelemetryMsg(tenantId, sourceEntityId, proto.getMsg(), cf, msg.getCallback());
+                    var newMsg = new EntityCalculatedFieldLinkedTelemetryMsg(tenantId, sourceEntityId, proto.getMsg(), cf, callback);
                     getOrCreateActor(targetEntityId).tell(newMsg);
                 }
             }
@@ -438,10 +493,14 @@ public class CalculatedFieldManagerMessageProcessor extends AbstractContextAware
         EntityId entityId = cfCtx.getEntityId();
         EntityType entityType = cfCtx.getEntityId().getEntityType();
         if (isProfileEntity(entityType)) {
-            var entityIds = cfEntityCache.getMyEntityIdsByProfileId(tenantId, entityId);
+            var entityIds = entityProfileCache.getEntityIdsByProfileId(entityId);
             if (!entityIds.isEmpty()) {
                 var multiCallback = new MultipleTbCallback(entityIds.size(), callback);
-                entityIds.forEach(id -> initCfForEntity(id, cfCtx, forceStateReinit, multiCallback));
+                entityIds.forEach(id -> {
+                    if (isMyPartition(id, multiCallback)) {
+                        initCfForEntity(id, cfCtx, forceStateReinit, multiCallback);
+                    }
+                });
             } else {
                 callback.onSuccess();
             }
@@ -479,7 +538,8 @@ public class CalculatedFieldManagerMessageProcessor extends AbstractContextAware
         return switch (entityId.getEntityType()) {
             case ASSET -> assetProfileCache.get(tenantId, (AssetId) entityId).getId();
             case DEVICE -> deviceProfileCache.get(tenantId, (DeviceId) entityId).getId();
-            default -> null;
+            default ->
+                    throw new IllegalArgumentException("'" + entityId.getEntityType() + "' is not profile entity." + entityId);
         };
     }
 
@@ -504,4 +564,45 @@ public class CalculatedFieldManagerMessageProcessor extends AbstractContextAware
     public void onPartitionChange(CalculatedFieldPartitionChangeMsg msg) {
         ctx.broadcastToChildren(msg, true);
     }
+
+    public void initCalculatedFields() {
+        PageDataIterable<CalculatedField> cfs = new PageDataIterable<>(pageLink -> cfDaoService.findCalculatedFieldsByTenantId(tenantId, pageLink), initFetchPackSize);
+        cfs.forEach(cf -> {
+            log.trace("Processing calculated field record: {}", cf);
+            try {
+                onFieldInitMsg(new CalculatedFieldInitMsg(cf.getTenantId(), cf));
+            } catch (CalculatedFieldException e) {
+                log.error("Failed to process calculated field record: {}", cf, e);
+            }
+        });
+        calculatedFields.values().forEach(cf -> {
+            entityIdCalculatedFields.computeIfAbsent(cf.getEntityId(), id -> new CopyOnWriteArrayList<>()).add(cf);
+        });
+        PageDataIterable<CalculatedFieldLink> cfls = new PageDataIterable<>(pageLink -> cfDaoService.findAllCalculatedFieldLinksByTenantId(tenantId, pageLink), initFetchPackSize);
+        cfls.forEach(link -> {
+            onLinkInitMsg(new CalculatedFieldLinkInitMsg(link.getTenantId(), link));
+        });
+    }
+
+    private void initEntityProfileCache() {
+        PageDataIterable<ProfileEntityIdInfo> deviceIdInfos = new PageDataIterable<>(pageLink -> deviceService.findProfileEntityIdInfosByTenantId(tenantId, pageLink), initFetchPackSize);
+        for (ProfileEntityIdInfo idInfo : deviceIdInfos) {
+            log.trace("Processing device record: {}", idInfo);
+            try {
+                entityProfileCache.add(idInfo.getProfileId(), idInfo.getEntityId());
+            } catch (Exception e) {
+                log.error("Failed to process device record: {}", idInfo, e);
+            }
+        }
+        PageDataIterable<ProfileEntityIdInfo> assetIdInfos = new PageDataIterable<>(pageLink -> assetService.findProfileEntityIdInfosByTenantId(tenantId, pageLink), initFetchPackSize);
+        for (ProfileEntityIdInfo idInfo : assetIdInfos) {
+            log.trace("Processing asset record: {}", idInfo);
+            try {
+                entityProfileCache.add(idInfo.getProfileId(), idInfo.getEntityId());
+            } catch (Exception e) {
+                log.error("Failed to process asset record: {}", idInfo, e);
+            }
+        }
+    }
+
 }

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

@@ -50,6 +50,7 @@ import org.thingsboard.server.common.msg.TbMsg;
 import org.thingsboard.server.common.msg.ToCalculatedFieldSystemMsg;
 import org.thingsboard.server.common.msg.aware.DeviceAwareMsg;
 import org.thingsboard.server.common.msg.aware.RuleChainAwareMsg;
+import org.thingsboard.server.common.msg.cf.CalculatedFieldCacheInitMsg;
 import org.thingsboard.server.common.msg.cf.CalculatedFieldEntityLifecycleMsg;
 import org.thingsboard.server.common.msg.plugin.ComponentLifecycleMsg;
 import org.thingsboard.server.common.msg.queue.PartitionChangeMsg;
@@ -176,6 +177,8 @@ public class TenantActor extends RuleChainManagerActor {
             case RULE_CHAIN_TO_RULE_CHAIN_MSG:
                 onRuleChainMsg((RuleChainAwareMsg) msg);
                 break;
+            case CF_CACHE_INIT_MSG:
+            case CF_INIT_PROFILE_ENTITY_MSG:
             case CF_INIT_MSG:
             case CF_LINK_INIT_MSG:
             case CF_STATE_RESTORE_MSG:
@@ -199,6 +202,7 @@ public class TenantActor extends RuleChainManagerActor {
             } else {
                 log.debug("[{}] CF Actor is not initialized. ToCalculatedFieldSystemMsg: [{}]", tenantId, msg);
             }
+            msg.getCallback().onSuccess();
             return;
         }
         if (priority) {
@@ -274,10 +278,12 @@ public class TenantActor extends RuleChainManagerActor {
                                 () -> DefaultActorService.CF_MANAGER_DISPATCHER_NAME,
                                 () -> new CalculatedFieldManagerActorCreator(systemContext, tenantId),
                                 () -> true);
+                        cfActor.tellWithHighPriority(new CalculatedFieldCacheInitMsg(tenantId));
                     } catch (Exception e) {
                         log.info("[{}] Failed to init CF Actor.", tenantId, e);
                     }
                 }
+                cfActor.tellWithHighPriority(msg);
                 if (!ruleChainsInitialized) {
                     log.info("Tenant {} is now managed by this service, initializing rule chains", tenantId);
                     initRuleChains();

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

@@ -64,7 +64,7 @@ public class DefaultCalculatedFieldCache implements CalculatedFieldCache {
     private final ConcurrentMap<EntityId, List<CalculatedFieldLink>> entityIdCalculatedFieldLinks = new ConcurrentHashMap<>();
     private final ConcurrentMap<CalculatedFieldId, CalculatedFieldCtx> calculatedFieldsCtx = new ConcurrentHashMap<>();
 
-    @Value("${calculatedField.initFetchPackSize:50000}")
+    @Value("${queue.calculated_fields.init_fetch_pack_size:50000}")
     @Getter
     private int initFetchPackSize;
 

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

@@ -20,13 +20,14 @@ import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.beans.factory.annotation.Value;
 import org.springframework.stereotype.Service;
+import org.thingsboard.server.actors.ActorSystemContext;
 import org.thingsboard.server.common.data.ProfileEntityIdInfo;
 import org.thingsboard.server.common.data.page.PageDataIterable;
+import org.thingsboard.server.common.msg.cf.CalculatedFieldInitProfileEntityMsg;
 import org.thingsboard.server.dao.asset.AssetService;
 import org.thingsboard.server.dao.device.DeviceService;
 import org.thingsboard.server.queue.util.AfterStartUp;
 import org.thingsboard.server.queue.util.TbRuleEngineComponent;
-import org.thingsboard.server.service.cf.cache.CalculatedFieldEntityProfileCache;
 
 @Slf4j
 @Service
@@ -34,11 +35,11 @@ import org.thingsboard.server.service.cf.cache.CalculatedFieldEntityProfileCache
 @RequiredArgsConstructor
 public class DefaultCalculatedFieldInitService implements CalculatedFieldInitService {
 
-    private final CalculatedFieldEntityProfileCache entityProfileCache;
     private final AssetService assetService;
     private final DeviceService deviceService;
+    private final ActorSystemContext actorSystemContext;
 
-    @Value("${calculated_fields.init_fetch_pack_size:50000}")
+    @Value("${queue.calculated_fields.init_fetch_pack_size:50000}")
     @Getter
     private int initFetchPackSize;
 
@@ -48,7 +49,7 @@ public class DefaultCalculatedFieldInitService implements CalculatedFieldInitSer
         for (ProfileEntityIdInfo idInfo : deviceIdInfos) {
             log.trace("Processing device record: {}", idInfo);
             try {
-                entityProfileCache.add(idInfo.getTenantId(), idInfo.getProfileId(), idInfo.getEntityId());
+                actorSystemContext.tell(new CalculatedFieldInitProfileEntityMsg(idInfo.getTenantId(), idInfo.getProfileId(), idInfo.getEntityId()));
             } catch (Exception e) {
                 log.error("Failed to process device record: {}", idInfo, e);
             }
@@ -57,7 +58,7 @@ public class DefaultCalculatedFieldInitService implements CalculatedFieldInitSer
         for (ProfileEntityIdInfo idInfo : assetIdInfos) {
             log.trace("Processing asset record: {}", idInfo);
             try {
-                entityProfileCache.add(idInfo.getTenantId(), idInfo.getProfileId(), idInfo.getEntityId());
+                actorSystemContext.tell(new CalculatedFieldInitProfileEntityMsg(idInfo.getTenantId(), idInfo.getProfileId(), idInfo.getEntityId()));
             } catch (Exception e) {
                 log.error("Failed to process asset record: {}", idInfo, e);
             }

application/src/main/java/org/thingsboard/server/service/cf/cache/DefaultCalculatedFieldEntityProfileCache.java

@@ -1,95 +0,0 @@
-/**
- * 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.service.cf.cache;
-
-import lombok.RequiredArgsConstructor;
-import lombok.extern.slf4j.Slf4j;
-import org.springframework.stereotype.Service;
-import org.thingsboard.server.common.data.DataConstants;
-import org.thingsboard.server.common.data.id.EntityId;
-import org.thingsboard.server.common.data.id.TenantId;
-import org.thingsboard.server.common.msg.queue.ServiceType;
-import org.thingsboard.server.common.msg.queue.TopicPartitionInfo;
-import org.thingsboard.server.queue.discovery.PartitionService;
-import org.thingsboard.server.queue.discovery.QueueKey;
-import org.thingsboard.server.queue.discovery.TbApplicationEventListener;
-import org.thingsboard.server.queue.discovery.event.PartitionChangeEvent;
-import org.thingsboard.server.queue.util.TbRuleEngineComponent;
-
-import java.util.Collection;
-import java.util.Collections;
-import java.util.List;
-import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.ConcurrentMap;
-import java.util.stream.Collectors;
-
-@TbRuleEngineComponent
-@Service
-@Slf4j
-@RequiredArgsConstructor
-//TODO ashvayka: remove and use TenantEntityProfileCache in each CalculatedFieldManagerMessageProcessor;
-public class DefaultCalculatedFieldEntityProfileCache extends TbApplicationEventListener<PartitionChangeEvent> implements CalculatedFieldEntityProfileCache {
-
-    private static final Integer UNKNOWN = 0;
-    private final ConcurrentMap<TenantId, TenantEntityProfileCache> tenantCache = new ConcurrentHashMap<>();
-    private final PartitionService partitionService;
-    private volatile List<Integer> myPartitions = Collections.emptyList();
-
-    @Override
-    protected void onTbApplicationEvent(PartitionChangeEvent event) {
-        myPartitions = event.getCfPartitions().stream()
-                .filter(TopicPartitionInfo::isMyPartition)
-                .map(tpi -> tpi.getPartition().orElse(UNKNOWN)).collect(Collectors.toList());
-        //Naive approach that need to be improved.
-        tenantCache.values().forEach(cache -> cache.setMyPartitions(myPartitions));
-    }
-
-    @Override
-    public void add(TenantId tenantId, EntityId profileId, EntityId entityId) {
-        var tpi = partitionService.resolve(ServiceType.TB_RULE_ENGINE, DataConstants.CF_QUEUE_NAME, tenantId, entityId);
-        var partition = tpi.getPartition().orElse(UNKNOWN);
-        tenantCache.computeIfAbsent(tenantId, id -> new TenantEntityProfileCache())
-                .add(profileId, entityId, partition, tpi.isMyPartition());
-    }
-
-    @Override
-    public void update(TenantId tenantId, EntityId oldProfileId, EntityId newProfileId, EntityId entityId) {
-        var tpi = partitionService.resolve(ServiceType.TB_RULE_ENGINE, DataConstants.CF_QUEUE_NAME, tenantId, entityId);
-        var partition = tpi.getPartition().orElse(UNKNOWN);
-        var cache = tenantCache.computeIfAbsent(tenantId, id -> new TenantEntityProfileCache());
-        //TODO: make this method atomic;
-        cache.remove(oldProfileId, entityId);
-        cache.add(newProfileId, entityId, partition, tpi.isMyPartition());
-    }
-
-    @Override
-    public void evict(TenantId tenantId, EntityId entityId) {
-        var cache = tenantCache.computeIfAbsent(tenantId, id -> new TenantEntityProfileCache());
-        cache.removeEntityId(entityId);
-    }
-
-    @Override
-    public Collection<EntityId> getMyEntityIdsByProfileId(TenantId tenantId, EntityId profileId) {
-        return tenantCache.computeIfAbsent(tenantId, id -> new TenantEntityProfileCache()).getMyEntityIdsByProfileId(profileId);
-    }
-
-    @Override
-    public int getEntityIdPartition(TenantId tenantId, EntityId entityId) {
-        var tpi = partitionService.resolve(ServiceType.TB_RULE_ENGINE, DataConstants.CF_QUEUE_NAME, tenantId, entityId);
-        return tpi.getPartition().orElse(UNKNOWN);
-    }
-
-}

application/src/main/java/org/thingsboard/server/service/cf/cache/TenantEntityProfileCache.java

@@ -32,31 +32,13 @@ import java.util.concurrent.locks.ReentrantReadWriteLock;
 public class TenantEntityProfileCache {
 
     private final ReadWriteLock lock = new ReentrantReadWriteLock();
-    private final Map<Integer, Map<EntityId, Set<EntityId>>> allEntities = new HashMap<>();
+    private final Map<EntityId, Set<EntityId>> allEntities = new HashMap<>();
-    private final Map<EntityId, Set<EntityId>> myEntities = new HashMap<>();
-
-    public void setMyPartitions(List<Integer> myPartitions) {
-        lock.writeLock().lock();
-        try {
-            myEntities.clear();
-            myPartitions.forEach(partitionId -> {
-                var map = allEntities.get(partitionId);
-                if (map != null) {
-                    map.forEach((profileId, entityIds) -> myEntities.computeIfAbsent(profileId, k -> new HashSet<>()).addAll(entityIds));
-                }
-            });
-        } finally {
-            lock.writeLock().unlock();
-        }
-    }
 
     public void removeProfileId(EntityId profileId) {
         lock.writeLock().lock();
         try {
             // Remove from allEntities
-            allEntities.values().forEach(map -> map.remove(profileId));
+            allEntities.remove(profileId);
-            // Remove from myEntities
-            myEntities.remove(profileId);
         } finally {
             lock.writeLock().unlock();
         }
@@ -66,9 +48,7 @@ public class TenantEntityProfileCache {
         lock.writeLock().lock();
         try {
             // Remove from allEntities
-            allEntities.values().forEach(map -> map.values().forEach(set -> set.remove(entityId)));
+            allEntities.values().forEach(set -> set.remove(entityId));
-            // Remove from myEntities
-            myEntities.values().forEach(set -> set.remove(entityId));
         } finally {
             lock.writeLock().unlock();
         }
@@ -78,33 +58,33 @@ public class TenantEntityProfileCache {
         lock.writeLock().lock();
         try {
             // Remove from allEntities
-            allEntities.values().forEach(map -> removeSafely(map, profileId, entityId));
+            removeSafely(allEntities, profileId, entityId);
-            // Remove from myEntities
-            removeSafely(myEntities, profileId, entityId);
         } finally {
             lock.writeLock().unlock();
         }
     }
 
-    public void add(EntityId profileId, EntityId entityId, Integer partition, boolean mine) {
+    public void add(EntityId profileId, EntityId entityId) {
         lock.writeLock().lock();
         try {
-            if(EntityType.DEVICE.equals(profileId.getEntityType())){
+            if (EntityType.DEVICE.equals(profileId.getEntityType()) || EntityType.ASSET.equals(profileId.getEntityType())) {
-                throw new RuntimeException("WTF?");
+                throw new RuntimeException("Entity type '" + profileId.getEntityType() + "' is not a profileId.");
-            }
-            if (mine) {
-                myEntities.computeIfAbsent(profileId, k -> new HashSet<>()).add(entityId);
             }
-            allEntities.computeIfAbsent(partition, k -> new HashMap<>()).computeIfAbsent(profileId, p -> new HashSet<>()).add(entityId);
+            allEntities.computeIfAbsent(profileId, k -> new HashSet<>()).add(entityId);
         } finally {
             lock.writeLock().unlock();
         }
     }
 
-    public Collection<EntityId> getMyEntityIdsByProfileId(EntityId profileId) {
+    public void update(EntityId oldProfileId, EntityId newProfileId, EntityId entityId) {
+        remove(oldProfileId, entityId);
+        add(newProfileId, entityId);
+    }
+
+    public Collection<EntityId> getEntityIdsByProfileId(EntityId profileId) {
         lock.readLock().lock();
         try {
-            var entities = myEntities.getOrDefault(profileId, Collections.emptySet());
+            var entities = allEntities.getOrDefault(profileId, Collections.emptySet());
             List<EntityId> result = new ArrayList<>(entities.size());
             result.addAll(entities);
             return result;

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

@@ -91,7 +91,7 @@ public class KafkaCalculatedFieldStateService extends AbstractCalculatedFieldSta
                         }
                     }
                 })
-                .consumerCreator((config, partitionId) -> queueFactory.createCalculatedFieldStateConsumer())
+                .consumerCreator((queueConfig, tpi) -> queueFactory.createCalculatedFieldStateConsumer())
                 .queueAdmin(queueFactory.getCalculatedFieldQueueAdmin())
                 .consumerExecutor(eventConsumer.getConsumerExecutor())
                 .scheduler(eventConsumer.getScheduler())

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

@@ -22,11 +22,11 @@ import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression;
 import org.springframework.stereotype.Service;
 import org.thingsboard.server.common.msg.queue.TbCallback;
 import org.thingsboard.server.common.msg.queue.TopicPartitionInfo;
-import org.thingsboard.server.queue.common.state.DefaultQueueStateService;
 import org.thingsboard.server.gen.transport.TransportProtos.CalculatedFieldStateProto;
 import org.thingsboard.server.gen.transport.TransportProtos.ToCalculatedFieldMsg;
 import org.thingsboard.server.queue.common.TbProtoQueueMsg;
 import org.thingsboard.server.queue.common.consumer.PartitionedQueueConsumerManager;
+import org.thingsboard.server.queue.common.state.DefaultQueueStateService;
 import org.thingsboard.server.queue.discovery.QueueKey;
 import org.thingsboard.server.service.cf.AbstractCalculatedFieldStateService;
 import org.thingsboard.server.service.cf.CfRocksDb;

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

@@ -99,12 +99,12 @@ public class DefaultTbCalculatedFieldConsumerService extends AbstractPartitionBa
     @Override
     protected void onStartUp() {
         var queueKey = new QueueKey(ServiceType.TB_RULE_ENGINE, DataConstants.CF_QUEUE_NAME);
-        PartitionedQueueConsumerManager<TbProtoQueueMsg<ToCalculatedFieldMsg>> eventConsumer = PartitionedQueueConsumerManager.<TbProtoQueueMsg<ToCalculatedFieldMsg>>create()
+        var eventConsumer = PartitionedQueueConsumerManager.<TbProtoQueueMsg<ToCalculatedFieldMsg>>create()
                 .queueKey(queueKey)
                 .topic(partitionService.getTopic(queueKey))
                 .pollInterval(pollInterval)
                 .msgPackProcessor(this::processMsgs)
-                .consumerCreator((config, partitionId) -> queueFactory.createToCalculatedFieldMsgConsumer())
+                .consumerCreator((queueConfig, tpi) -> queueFactory.createToCalculatedFieldMsgConsumer(tpi))
                 .queueAdmin(queueFactory.getCalculatedFieldQueueAdmin())
                 .consumerExecutor(consumersExecutor)
                 .scheduler(scheduler)
@@ -127,7 +127,7 @@ public class DefaultTbCalculatedFieldConsumerService extends AbstractPartitionBa
     protected void onPartitionChangeEvent(PartitionChangeEvent event) {
         try {
             event.getNewPartitions().forEach((queueKey, partitions) -> {
-                if (queueKey.getQueueName().equals(DataConstants.CF_QUEUE_NAME)) {
+                if (DataConstants.CF_QUEUE_NAME.equals(queueKey.getQueueName())) {
                     stateService.restore(queueKey, partitions);
                 }
             });
@@ -175,9 +175,7 @@ public class DefaultTbCalculatedFieldConsumerService extends AbstractPartitionBa
                 packSubmitFuture.cancel(true);
                 log.info("Timeout to process message: {}", pendingMsgHolder.getMsg());
             }
-            if (log.isDebugEnabled()) {
+            ctx.getAckMap().forEach((id, msg) -> log.warn("[{}] Timeout to process message: {}", id, msg.getValue()));
-                ctx.getAckMap().forEach((id, msg) -> log.debug("[{}] Timeout to process message: {}", id, msg.getValue()));
-            }
             ctx.getFailedMap().forEach((id, msg) -> log.warn("[{}] Failed to process message: {}", id, msg.getValue()));
         }
         consumer.commit();

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

@@ -206,7 +206,7 @@ public class DefaultTbCoreConsumerService extends AbstractConsumerService<ToCore
                 .queueKey(new QueueKey(ServiceType.TB_CORE))
                 .config(QueueConfig.of(consumerPerPartition, pollInterval))
                 .msgPackProcessor(this::processMsgs)
-                .consumerCreator((config, partitionId) -> queueFactory.createToCoreMsgConsumer())
+                .consumerCreator((config, tpi) -> queueFactory.createToCoreMsgConsumer())
                 .consumerExecutor(consumersExecutor)
                 .scheduler(scheduler)
                 .taskExecutor(mgmtExecutor)

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

@@ -102,7 +102,7 @@ public class DefaultTbEdgeConsumerService extends AbstractConsumerService<ToEdge
                 .queueKey(new QueueKey(ServiceType.TB_CORE).withQueueName(DataConstants.EDGE_QUEUE_NAME))
                 .config(QueueConfig.of(consumerPerPartition, pollInterval))
                 .msgPackProcessor(this::processMsgs)
-                .consumerCreator((config, partitionId) -> queueFactory.createEdgeMsgConsumer())
+                .consumerCreator((config, tpi) -> queueFactory.createEdgeMsgConsumer())
                 .consumerExecutor(consumersExecutor)
                 .scheduler(scheduler)
                 .taskExecutor(mgmtExecutor)

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

@@ -72,7 +72,12 @@ public class TbRuleEngineQueueConsumerManager extends MainQueueConsumerManager<T
                                             ExecutorService consumerExecutor,
                                             ScheduledExecutorService scheduler,
                                             ExecutorService taskExecutor) {
-        super(queueKey, null, null, ctx.getQueueFactory()::createToRuleEngineMsgConsumer, consumerExecutor, scheduler, taskExecutor, null);
+        super(queueKey, null, null,
+                (queueConfig, tpi) -> {
+                    Integer partitionId = tpi != null ? tpi.getPartition().orElse(-1) : null;
+                    return ctx.getQueueFactory().createToRuleEngineMsgConsumer(queueConfig, partitionId);
+                },
+                consumerExecutor, scheduler, taskExecutor, null);
         this.ctx = ctx;
         this.stats = new TbRuleEngineConsumerStats(queueKey, ctx.getStatsFactory());
     }

application/src/main/resources/thingsboard.yml

@@ -1848,6 +1848,10 @@ queue:
     pool_size: "${TB_QUEUE_CF_POOL_SIZE:8}"
     # RocksDB path for storing CF states
     rocks_db_path: "${TB_QUEUE_CF_ROCKS_DB_PATH:${user.home}/.rocksdb/cf_states}"
+    # The fetch size specifies how many rows will be fetched from the database per request for initial fetching
+    init_fetch_pack_size: "${TB_QUEUE_CF_FETCH_PACK_SIZE:50000}"
+    # The fetch size specifies how many rows will be fetched from the database per request for per-tenant fetching
+    init_tenant_fetch_pack_size: "${TB_QUEUE_CF_TENANT_FETCH_PACK_SIZE:1000}"
   transport:
     # For high-priority notifications that require minimum latency and processing time
     notifications_topic: "${TB_QUEUE_TRANSPORT_NOTIFICATIONS_TOPIC:tb_transport.notifications}"

common/dao-api/src/main/java/org/thingsboard/server/dao/asset/AssetService.java

@@ -66,6 +66,8 @@ public interface AssetService extends EntityDaoService {
 
     PageData<ProfileEntityIdInfo> findProfileEntityIdInfos(PageLink pageLink);
 
+    PageData<ProfileEntityIdInfo> findProfileEntityIdInfosByTenantId(TenantId tenantId, PageLink pageLink);
+
     PageData<AssetId> findAssetIdsByTenantIdAndAssetProfileId(TenantId tenantId, AssetProfileId assetProfileId, PageLink pageLink);
 
     ListenableFuture<List<Asset>> findAssetsByTenantIdAndIdsAsync(TenantId tenantId, List<AssetId> assetIds);

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

@@ -39,6 +39,8 @@ public interface CalculatedFieldService extends EntityDaoService {
 
     PageData<CalculatedField> findAllCalculatedFields(PageLink pageLink);
 
+    PageData<CalculatedField> findCalculatedFieldsByTenantId(TenantId tenantId, PageLink pageLink);
+
     PageData<CalculatedField> findAllCalculatedFieldsByEntityId(TenantId tenantId, EntityId entityId, PageLink pageLink);
 
     void deleteCalculatedField(TenantId tenantId, CalculatedFieldId calculatedFieldId);
@@ -53,6 +55,8 @@ public interface CalculatedFieldService extends EntityDaoService {
 
     List<CalculatedFieldLink> findAllCalculatedFieldLinksByEntityId(TenantId tenantId, EntityId entityId);
 
+    PageData<CalculatedFieldLink> findAllCalculatedFieldLinksByTenantId(TenantId tenantId, PageLink pageLink);
+
     PageData<CalculatedFieldLink> findAllCalculatedFieldLinks(PageLink pageLink);
 
     boolean referencedInAnyCalculatedField(TenantId tenantId, EntityId referencedEntityId);

common/dao-api/src/main/java/org/thingsboard/server/dao/device/DeviceService.java

@@ -76,6 +76,8 @@ public interface DeviceService extends EntityDaoService {
 
     PageData<ProfileEntityIdInfo> findProfileEntityIdInfos(PageLink pageLink);
 
+    PageData<ProfileEntityIdInfo> findProfileEntityIdInfosByTenantId(TenantId tenantId, PageLink pageLink);
+
     PageData<Device> findDevicesByTenantIdAndType(TenantId tenantId, String type, PageLink pageLink);
 
     PageData<DeviceId> findDeviceIdsByTenantIdAndDeviceProfileId(TenantId tenantId, DeviceProfileId deviceProfileId, PageLink pageLink);

common/edqs/src/main/java/org/thingsboard/server/edqs/processor/EdqsProcessor.java

@@ -139,7 +139,7 @@ public class EdqsProcessor implements TbQueueHandler<TbProtoQueueMsg<ToEdqsMsg>,
                     }
                     consumer.commit();
                 })
-                .consumerCreator((config, partitionId) -> queueFactory.createEdqsEventsConsumer())
+                .consumerCreator((config, tpi) -> queueFactory.createEdqsEventsConsumer())
                 .queueAdmin(queueFactory.getEdqsQueueAdmin())
                 .consumerExecutor(consumersExecutor)
                 .taskExecutor(taskExecutor)

common/edqs/src/main/java/org/thingsboard/server/edqs/state/KafkaEdqsStateService.java

@@ -59,7 +59,8 @@ public class KafkaEdqsStateService implements EdqsStateService {
     private final EdqsConfig config;
     private final EdqsPartitionService partitionService;
     private final KafkaEdqsQueueFactory queueFactory;
-    @Autowired @Lazy
+    @Autowired
+    @Lazy
     private EdqsProcessor edqsProcessor;
 
     private PartitionedQueueConsumerManager<TbProtoQueueMsg<ToEdqsMsg>> stateConsumer;
@@ -93,7 +94,7 @@ public class KafkaEdqsStateService implements EdqsStateService {
                     }
                     consumer.commit();
                 })
-                .consumerCreator((config, partitionId) -> queueFactory.createEdqsStateConsumer())
+                .consumerCreator((config, tpi) -> queueFactory.createEdqsStateConsumer())
                 .queueAdmin(queueAdmin)
                 .consumerExecutor(eventConsumer.getConsumerExecutor())
                 .taskExecutor(eventConsumer.getTaskExecutor())

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

@@ -136,6 +136,8 @@ public enum MsgType {
     EDGE_SYNC_RESPONSE_FROM_EDGE_SESSION_MSG,
 
 
+    CF_CACHE_INIT_MSG, // Sent to init caches for CF actor;
+    CF_INIT_PROFILE_ENTITY_MSG, // Sent to init profile entities cache;
     CF_INIT_MSG, // Sent to init particular calculated field;
     CF_LINK_INIT_MSG, // Sent to init particular calculated field;
     CF_STATE_RESTORE_MSG, // Sent to restore particular calculated field entity state;

common/message/src/main/java/org/thingsboard/server/common/msg/cf/CalculatedFieldCacheInitMsg.java

@@ -0,0 +1,33 @@
+/**
+ * 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.cf;
+
+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;
+
+@Data
+public class CalculatedFieldCacheInitMsg implements ToCalculatedFieldSystemMsg {
+
+    private final TenantId tenantId;
+
+    @Override
+    public MsgType getMsgType() {
+        return MsgType.CF_CACHE_INIT_MSG;
+    }
+
+}

application/src/main/java/org/thingsboard/server/service/cf/cache/CalculatedFieldEntityProfileCache.java → common/message/src/main/java/org/thingsboard/server/common/msg/cf/CalculatedFieldInitProfileEntityMsg.java

@@ -13,24 +13,24 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package org.thingsboard.server.service.cf.cache;
+package org.thingsboard.server.common.msg.cf;
 
-import org.springframework.context.ApplicationListener;
+import lombok.Data;
 import org.thingsboard.server.common.data.id.EntityId;
 import org.thingsboard.server.common.data.id.TenantId;
-import org.thingsboard.server.queue.discovery.event.PartitionChangeEvent;
+import org.thingsboard.server.common.msg.MsgType;
+import org.thingsboard.server.common.msg.ToCalculatedFieldSystemMsg;
 
-import java.util.Collection;
+@Data
+public class CalculatedFieldInitProfileEntityMsg implements ToCalculatedFieldSystemMsg {
 
-public interface CalculatedFieldEntityProfileCache extends ApplicationListener<PartitionChangeEvent> {
+    private final TenantId tenantId;
+    private final EntityId profileEntityId;
+    private final EntityId entityId;
 
-    void add(TenantId tenantId, EntityId profileId, EntityId entityId);
+    @Override
+    public MsgType getMsgType() {
+        return MsgType.CF_INIT_PROFILE_ENTITY_MSG;
+    }
 
-    void update(TenantId tenantId, EntityId oldProfileId, EntityId newProfileId, EntityId entityId);
-
-    void evict(TenantId tenantId, EntityId entityId);
-
-    Collection<EntityId> getMyEntityIdsByProfileId(TenantId tenantId, EntityId profileId);
-
-    int getEntityIdPartition(TenantId tenantId, EntityId entityId);
 }

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

@@ -54,7 +54,7 @@ public class MainQueueConsumerManager<M extends TbQueueMsg, C extends QueueConfi
     @Getter
     protected C config;
     protected final MsgPackProcessor<M, C> msgPackProcessor;
-    protected final BiFunction<C, Integer, TbQueueConsumer<M>> consumerCreator;
+    protected final BiFunction<C, TopicPartitionInfo, TbQueueConsumer<M>> consumerCreator;
     @Getter
     protected final ExecutorService consumerExecutor;
     @Getter
@@ -74,7 +74,7 @@ public class MainQueueConsumerManager<M extends TbQueueMsg, C extends QueueConfi
     @Builder
     public MainQueueConsumerManager(QueueKey queueKey, C config,
                                     MsgPackProcessor<M, C> msgPackProcessor,
-                                    BiFunction<C, Integer, TbQueueConsumer<M>> consumerCreator,
+                                    BiFunction<C, TopicPartitionInfo, TbQueueConsumer<M>> consumerCreator,
                                     ExecutorService consumerExecutor,
                                     ScheduledExecutorService scheduler,
                                     ExecutorService taskExecutor,
@@ -313,7 +313,7 @@ public class MainQueueConsumerManager<M extends TbQueueMsg, C extends QueueConfi
                 Runnable callback = onStop != null ? () -> onStop.accept(tpi) : null;
 
                 TbQueueConsumerTask<M> consumer = new TbQueueConsumerTask<>(key, () -> {
-                    TbQueueConsumer<M> queueConsumer = consumerCreator.apply(config, partitionId);
+                    TbQueueConsumer<M> queueConsumer = consumerCreator.apply(config, tpi);
                     if (startOffsetProvider != null && queueConsumer instanceof TbKafkaConsumerTemplate<M> kafkaConsumer) {
                         kafkaConsumer.setStartOffsetProvider(startOffsetProvider);
                     }

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

@@ -45,7 +45,7 @@ public class PartitionedQueueConsumerManager<M extends TbQueueMsg> extends MainQ
 
     @Builder(builderMethodName = "create") // not to conflict with super.builder()
     public PartitionedQueueConsumerManager(QueueKey queueKey, String topic, long pollInterval, MsgPackProcessor<M, QueueConfig> msgPackProcessor,
-                                           BiFunction<QueueConfig, Integer, TbQueueConsumer<M>> consumerCreator, TbQueueAdmin queueAdmin,
+                                           BiFunction<QueueConfig, TopicPartitionInfo, TbQueueConsumer<M>> consumerCreator, TbQueueAdmin queueAdmin,
                                            ExecutorService consumerExecutor, ScheduledExecutorService scheduler,
                                            ExecutorService taskExecutor, Consumer<Throwable> uncaughtErrorHandler) {
         super(queueKey, QueueConfig.of(true, pollInterval), msgPackProcessor, consumerCreator, consumerExecutor, scheduler, taskExecutor, uncaughtErrorHandler);

common/queue/src/main/java/org/thingsboard/server/queue/provider/InMemoryMonolithQueueFactory.java

@@ -22,6 +22,7 @@ import org.springframework.scheduling.annotation.Scheduled;
 import org.springframework.stereotype.Component;
 import org.thingsboard.server.common.data.queue.Queue;
 import org.thingsboard.server.common.msg.queue.ServiceType;
+import org.thingsboard.server.common.msg.queue.TopicPartitionInfo;
 import org.thingsboard.server.gen.js.JsInvokeProtos;
 import org.thingsboard.server.gen.transport.TransportProtos;
 import org.thingsboard.server.gen.transport.TransportProtos.CalculatedFieldStateProto;
@@ -133,7 +134,7 @@ public class InMemoryMonolithQueueFactory implements TbCoreQueueFactory, TbRuleE
     }
 
     @Override
-    public TbQueueConsumer<TbProtoQueueMsg<TransportProtos.ToCalculatedFieldMsg>> createToCalculatedFieldMsgConsumer() {
+    public TbQueueConsumer<TbProtoQueueMsg<TransportProtos.ToCalculatedFieldMsg>> createToCalculatedFieldMsgConsumer(TopicPartitionInfo tpi) {
         return new InMemoryTbQueueConsumer<>(storage, topicService.buildTopicName(calculatedFieldSettings.getEventTopic()));
     }
 

common/queue/src/main/java/org/thingsboard/server/queue/provider/KafkaMonolithQueueFactory.java

@@ -20,10 +20,12 @@ import jakarta.annotation.PreDestroy;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression;
 import org.springframework.context.annotation.Bean;
 import org.springframework.stereotype.Component;
+import org.thingsboard.server.common.data.DataConstants;
 import org.thingsboard.server.common.data.id.EdgeId;
 import org.thingsboard.server.common.data.id.TenantId;
 import org.thingsboard.server.common.data.queue.Queue;
 import org.thingsboard.server.common.msg.queue.ServiceType;
+import org.thingsboard.server.common.msg.queue.TopicPartitionInfo;
 import org.thingsboard.server.gen.js.JsInvokeProtos;
 import org.thingsboard.server.gen.transport.TransportProtos.CalculatedFieldStateProto;
 import org.thingsboard.server.gen.transport.TransportProtos.FromEdqsMsg;
@@ -513,15 +515,24 @@ public class KafkaMonolithQueueFactory implements TbCoreQueueFactory, TbRuleEngi
     }
 
     @Override
-    public TbQueueConsumer<TbProtoQueueMsg<ToCalculatedFieldMsg>> createToCalculatedFieldMsgConsumer() {
+    public TbQueueConsumer<TbProtoQueueMsg<ToCalculatedFieldMsg>> createToCalculatedFieldMsgConsumer(TopicPartitionInfo tpi) {
+        String queueName = DataConstants.CF_QUEUE_NAME;
+        if (tpi == null) {
+            throw new IllegalArgumentException("TopicPartitionInfo is required.");
+        }
+        TenantId tenantId = tpi.getTenantId().orElse(TenantId.SYS_TENANT_ID);
+        Integer partitionId = tpi.getPartition().orElseThrow(() -> new IllegalArgumentException("PartitionId is required."));
+        String groupId = topicService.buildConsumerGroupId("cf-", tenantId, queueName, partitionId);
+
         TbKafkaConsumerTemplate.TbKafkaConsumerTemplateBuilder<TbProtoQueueMsg<ToCalculatedFieldMsg>> consumerBuilder = TbKafkaConsumerTemplate.builder();
         consumerBuilder.settings(kafkaSettings);
         consumerBuilder.topic(topicService.buildTopicName(calculatedFieldSettings.getEventTopic()));
-        consumerBuilder.clientId("monolith-calculated-field-consumer-" + serviceInfoProvider.getServiceId() + "-" + consumerCount.incrementAndGet());
+        consumerBuilder.clientId("cf-" + queueName + "-consumer-" + serviceInfoProvider.getServiceId() + "-" + consumerCount.incrementAndGet());
-        consumerBuilder.groupId(topicService.buildTopicName("monolith-calculated-field-consumer"));
+        consumerBuilder.groupId(groupId);
         consumerBuilder.decoder(msg -> new TbProtoQueueMsg<>(msg.getKey(), ToCalculatedFieldMsg.parseFrom(msg.getData()), msg.getHeaders()));
         consumerBuilder.admin(cfAdmin);
         consumerBuilder.statsService(consumerStatsService);
+
         return consumerBuilder.build();
     }
 
@@ -571,7 +582,7 @@ public class KafkaMonolithQueueFactory implements TbCoreQueueFactory, TbRuleEngi
                 .readFromBeginning(true)
                 .stopWhenRead(true)
                 .clientId("monolith-calculated-field-state-consumer-" + serviceInfoProvider.getServiceId() + "-" + consumerCount.incrementAndGet())
-                .groupId(topicService.buildTopicName("monolith-calculated-field-state-consumer"))
+                .groupId(null) // not using consumer group management
                 .decoder(msg -> new TbProtoQueueMsg<>(msg.getKey(), msg.getData() != null ? CalculatedFieldStateProto.parseFrom(msg.getData()) : null, msg.getHeaders()))
                 .admin(cfStateAdmin)
                 .statsService(consumerStatsService)

common/queue/src/main/java/org/thingsboard/server/queue/provider/KafkaTbRuleEngineQueueFactory.java

@@ -20,8 +20,11 @@ import jakarta.annotation.PreDestroy;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression;
 import org.springframework.context.annotation.Bean;
 import org.springframework.stereotype.Component;
+import org.thingsboard.server.common.data.DataConstants;
+import org.thingsboard.server.common.data.id.TenantId;
 import org.thingsboard.server.common.data.queue.Queue;
 import org.thingsboard.server.common.msg.queue.ServiceType;
+import org.thingsboard.server.common.msg.queue.TopicPartitionInfo;
 import org.thingsboard.server.gen.js.JsInvokeProtos;
 import org.thingsboard.server.gen.transport.TransportProtos.CalculatedFieldStateProto;
 import org.thingsboard.server.gen.transport.TransportProtos.FromEdqsMsg;
@@ -313,12 +316,20 @@ public class KafkaTbRuleEngineQueueFactory implements TbRuleEngineQueueFactory {
     }
 
     @Override
-    public TbQueueConsumer<TbProtoQueueMsg<ToCalculatedFieldMsg>> createToCalculatedFieldMsgConsumer() {
+    public TbQueueConsumer<TbProtoQueueMsg<ToCalculatedFieldMsg>> createToCalculatedFieldMsgConsumer(TopicPartitionInfo tpi) {
+        String queueName = DataConstants.CF_QUEUE_NAME;
+        if (tpi == null) {
+            throw new IllegalArgumentException("TopicPartitionInfo is required.");
+        }
+        TenantId tenantId = tpi.getTenantId().orElse(TenantId.SYS_TENANT_ID);
+        Integer partitionId = tpi.getPartition().orElseThrow(() -> new IllegalArgumentException("PartitionId is required."));
+        String groupId = topicService.buildConsumerGroupId("cf-", tenantId, queueName, partitionId);
+
         TbKafkaConsumerTemplate.TbKafkaConsumerTemplateBuilder<TbProtoQueueMsg<ToCalculatedFieldMsg>> consumerBuilder = TbKafkaConsumerTemplate.builder();
         consumerBuilder.settings(kafkaSettings);
         consumerBuilder.topic(topicService.buildTopicName(calculatedFieldSettings.getEventTopic()));
-        consumerBuilder.clientId("tb-rule-engine-calculated-field-consumer-" + serviceInfoProvider.getServiceId() + "-" + consumerCount.incrementAndGet());
+        consumerBuilder.clientId("cf-" + queueName + "-consumer-" + serviceInfoProvider.getServiceId() + "-" + consumerCount.incrementAndGet());
-        consumerBuilder.groupId(topicService.buildTopicName("tb-rule-engine-calculated-field-consumer"));
+        consumerBuilder.groupId(groupId);
         consumerBuilder.decoder(msg -> new TbProtoQueueMsg<>(msg.getKey(), ToCalculatedFieldMsg.parseFrom(msg.getData()), msg.getHeaders()));
         consumerBuilder.admin(cfAdmin);
         consumerBuilder.statsService(consumerStatsService);
@@ -371,7 +382,7 @@ public class KafkaTbRuleEngineQueueFactory implements TbRuleEngineQueueFactory {
                 .readFromBeginning(true)
                 .stopWhenRead(true)
                 .clientId("tb-rule-engine-calculated-field-state-consumer-" + serviceInfoProvider.getServiceId() + "-" + consumerCount.incrementAndGet())
-                .groupId(topicService.buildTopicName("tb-rule-engine-calculated-field-state-consumer"))
+                .groupId(null) // not using consumer group management
                 .decoder(msg -> new TbProtoQueueMsg<>(msg.getKey(), msg.getData() != null ? CalculatedFieldStateProto.parseFrom(msg.getData()) : null, msg.getHeaders()))
                 .admin(cfStateAdmin)
                 .statsService(consumerStatsService)

common/queue/src/main/java/org/thingsboard/server/queue/provider/TbRuleEngineQueueFactory.java

@@ -16,6 +16,7 @@
 package org.thingsboard.server.queue.provider;
 
 import org.thingsboard.server.common.data.queue.Queue;
+import org.thingsboard.server.common.msg.queue.TopicPartitionInfo;
 import org.thingsboard.server.gen.js.JsInvokeProtos;
 import org.thingsboard.server.gen.transport.TransportProtos.CalculatedFieldStateProto;
 import org.thingsboard.server.gen.transport.TransportProtos.ToCalculatedFieldMsg;
@@ -95,8 +96,8 @@ public interface TbRuleEngineQueueFactory extends TbUsageStatsClientQueueFactory
     /**
      * Used to consume messages by TB Rule Engine Service
      *
-     * @return
      * @param configuration
+     * @return
      */
     TbQueueConsumer<TbProtoQueueMsg<ToRuleEngineMsg>> createToRuleEngineMsgConsumer(Queue configuration);
 
@@ -104,9 +105,9 @@ public interface TbRuleEngineQueueFactory extends TbUsageStatsClientQueueFactory
      * Used to consume messages by TB Rule Engine Service
      * Intended usage for consumer per partition strategy
      *
-     * @return TbQueueConsumer
      * @param configuration
-     * @param partitionId as a suffix for consumer name
+     * @param partitionId   as a suffix for consumer name
+     * @return TbQueueConsumer
      */
     default TbQueueConsumer<TbProtoQueueMsg<ToRuleEngineMsg>> createToRuleEngineMsgConsumer(Queue configuration, Integer partitionId) {
         return createToRuleEngineMsgConsumer(configuration);
@@ -121,7 +122,7 @@ public interface TbRuleEngineQueueFactory extends TbUsageStatsClientQueueFactory
 
     TbQueueRequestTemplate<TbProtoJsQueueMsg<JsInvokeProtos.RemoteJsRequest>, TbProtoQueueMsg<JsInvokeProtos.RemoteJsResponse>> createRemoteJsRequestTemplate();
 
-    TbQueueConsumer<TbProtoQueueMsg<ToCalculatedFieldMsg>> createToCalculatedFieldMsgConsumer();
+    TbQueueConsumer<TbProtoQueueMsg<ToCalculatedFieldMsg>> createToCalculatedFieldMsgConsumer(TopicPartitionInfo tpi);
 
     TbQueueAdmin getCalculatedFieldQueueAdmin();
 

common/queue/src/main/java/org/thingsboard/server/queue/util/AfterStartUp.java

@@ -36,10 +36,11 @@ public @interface AfterStartUp {
 
     int STARTUP_SERVICE = 8;
     int ACTOR_SYSTEM = 9;
-    int REGULAR_SERVICE = 10;
 
     int CF_READ_PROFILE_ENTITIES_SERVICE = 10;
-    int CF_READ_CF_SERVICE = 11;
+    int CF_READ_CF_SERVICE = 10;
+
+    int REGULAR_SERVICE = 11;
 
     int BEFORE_TRANSPORT_SERVICE = Integer.MAX_VALUE - 1001;
     int TRANSPORT_SERVICE = Integer.MAX_VALUE - 1000;

dao/src/main/java/org/thingsboard/server/dao/asset/AssetDao.java

@@ -240,4 +240,6 @@ public interface AssetDao extends Dao<Asset>, TenantEntityDao<Asset>, Exportable
 
     PageData<ProfileEntityIdInfo> findProfileEntityIdInfos(PageLink pageLink);
 
+    PageData<ProfileEntityIdInfo> findProfileEntityIdInfosByTenantId(UUID tenantId, PageLink pageLink);
+
 }

dao/src/main/java/org/thingsboard/server/dao/asset/BaseAssetService.java

@@ -293,6 +293,14 @@ public class BaseAssetService extends AbstractCachedEntityService<AssetCacheKey,
         return assetDao.findProfileEntityIdInfos(pageLink);
     }
 
+    @Override
+    public PageData<ProfileEntityIdInfo> findProfileEntityIdInfosByTenantId(TenantId tenantId, PageLink pageLink) {
+        log.trace("Executing findProfileEntityIdInfosByTenantId, tenantId[{}], pageLink [{}]", tenantId, pageLink);
+        validateId(tenantId, id -> INCORRECT_TENANT_ID + id);
+        validatePageLink(pageLink);
+        return assetDao.findProfileEntityIdInfosByTenantId(tenantId.getId(), pageLink);
+    }
+
     @Override
     public PageData<AssetId> findAssetIdsByTenantIdAndAssetProfileId(TenantId tenantId, AssetProfileId assetProfileId, PageLink pageLink) {
         log.trace("Executing findAssetIdsByTenantIdAndAssetProfileId, tenantId [{}], assetProfileId [{}]", tenantId, assetProfileId);

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

@@ -104,6 +104,14 @@ public class BaseCalculatedFieldService extends AbstractEntityService implements
         return calculatedFieldDao.findAll(pageLink);
     }
 
+    @Override
+    public PageData<CalculatedField> findCalculatedFieldsByTenantId(TenantId tenantId, PageLink pageLink) {
+        log.trace("Executing findAllByTenantId, tenantId [{}], pageLink [{}]", tenantId, pageLink);
+        validateId(tenantId, id -> INCORRECT_TENANT_ID + id);
+        validatePageLink(pageLink);
+        return calculatedFieldDao.findAllByTenantId(tenantId, pageLink);
+    }
+
     @Override
     public PageData<CalculatedField> findAllCalculatedFieldsByEntityId(TenantId tenantId, EntityId entityId, PageLink pageLink) {
         log.trace("Executing findAllByEntityId, entityId [{}], pageLink [{}]", entityId, pageLink);
@@ -174,6 +182,14 @@ public class BaseCalculatedFieldService extends AbstractEntityService implements
         return calculatedFieldLinkDao.findCalculatedFieldLinksByEntityId(tenantId, entityId);
     }
 
+    @Override
+    public PageData<CalculatedFieldLink> findAllCalculatedFieldLinksByTenantId(TenantId tenantId, PageLink pageLink) {
+        log.trace("Executing findAllCalculatedFieldLinksByTenantId, tenantId[{}] pageLink [{}]", tenantId, pageLink);
+        validateId(tenantId, id -> INCORRECT_TENANT_ID + id);
+        validatePageLink(pageLink);
+        return calculatedFieldLinkDao.findAllByTenantId(tenantId, pageLink);
+    }
+
     @Override
     public PageData<CalculatedFieldLink> findAllCalculatedFieldLinks(PageLink pageLink) {
         log.trace("Executing findAllCalculatedFieldLinks, pageLink [{}]", pageLink);

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

@@ -37,6 +37,8 @@ public interface CalculatedFieldDao extends Dao<CalculatedField> {
 
     PageData<CalculatedField> findAll(PageLink pageLink);
 
+    PageData<CalculatedField> findAllByTenantId(TenantId tenantId, PageLink pageLink);
+
     PageData<CalculatedField> findAllByEntityId(TenantId tenantId, EntityId entityId, PageLink pageLink);
 
     List<CalculatedField> removeAllByEntityId(TenantId tenantId, EntityId entityId);

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

@@ -31,8 +31,12 @@ public interface CalculatedFieldLinkDao extends Dao<CalculatedFieldLink> {
 
     List<CalculatedFieldLink> findCalculatedFieldLinksByEntityId(TenantId tenantId, EntityId entityId);
 
+    List<CalculatedFieldLink> findCalculatedFieldLinksByTenantId(TenantId tenantId);
+
     List<CalculatedFieldLink> findAll();
 
     PageData<CalculatedFieldLink> findAll(PageLink pageLink);
 
+    PageData<CalculatedFieldLink> findAllByTenantId(TenantId tenantId, PageLink pageLink);
+
 }

dao/src/main/java/org/thingsboard/server/dao/device/DeviceDao.java

@@ -23,9 +23,7 @@ import org.thingsboard.server.common.data.DeviceInfoFilter;
 import org.thingsboard.server.common.data.DeviceTransportType;
 import org.thingsboard.server.common.data.EntitySubtype;
 import org.thingsboard.server.common.data.ProfileEntityIdInfo;
-import org.thingsboard.server.common.data.id.AssetId;
 import org.thingsboard.server.common.data.id.DeviceId;
-import org.thingsboard.server.common.data.id.DeviceProfileId;
 import org.thingsboard.server.common.data.id.TenantId;
 import org.thingsboard.server.common.data.ota.OtaPackageType;
 import org.thingsboard.server.common.data.page.PageData;
@@ -233,6 +231,8 @@ public interface DeviceDao extends Dao<Device>, TenantEntityDao<Device>, Exporta
 
     PageData<ProfileEntityIdInfo> findProfileEntityIdInfos(PageLink pageLink);
 
+    PageData<ProfileEntityIdInfo> findProfileEntityIdInfosByTenantId(UUID tenantId, PageLink pageLink);
+
     PageData<DeviceInfo> findDeviceInfosByFilter(DeviceInfoFilter filter, PageLink pageLink);
 
 }

dao/src/main/java/org/thingsboard/server/dao/device/DeviceServiceImpl.java

@@ -394,6 +394,14 @@ public class DeviceServiceImpl extends CachedVersionedEntityService<DeviceCacheK
         return deviceDao.findProfileEntityIdInfos(pageLink);
     }
 
+    @Override
+    public PageData<ProfileEntityIdInfo> findProfileEntityIdInfosByTenantId(TenantId tenantId, PageLink pageLink) {
+        log.trace("Executing findProfileEntityIdInfosByTenantId, tenantId[{}], pageLink [{}]", tenantId, pageLink);
+        validateId(tenantId, id -> INCORRECT_TENANT_ID + id);
+        validatePageLink(pageLink);
+        return deviceDao.findProfileEntityIdInfosByTenantId(tenantId.getId(), pageLink);
+    }
+
     @Override
     public PageData<Device> findDevicesByTenantIdAndType(TenantId tenantId, String type, PageLink pageLink) {
         log.trace("Executing findDevicesByTenantIdAndType, tenantId [{}], type [{}], pageLink [{}]", tenantId, type, pageLink);

dao/src/main/java/org/thingsboard/server/dao/sql/asset/JpaAssetDao.java

@@ -39,14 +39,12 @@ import org.thingsboard.server.dao.model.sql.AssetEntity;
 import org.thingsboard.server.dao.model.sql.AssetInfoEntity;
 import org.thingsboard.server.dao.sql.JpaAbstractDao;
 import org.thingsboard.server.dao.sql.device.NativeAssetRepository;
-import org.thingsboard.server.dao.sql.device.NativeDeviceRepository;
 import org.thingsboard.server.dao.util.SqlDao;
 
 import java.util.Arrays;
 import java.util.List;
 import java.util.Optional;
 import java.util.UUID;
-import java.util.stream.Collectors;
 
 import static org.thingsboard.server.dao.DaoUtil.convertTenantEntityInfosToDto;
 
@@ -262,10 +260,16 @@ public class JpaAssetDao extends JpaAbstractDao<AssetEntity, Asset> implements A
 
     @Override
     public PageData<ProfileEntityIdInfo> findProfileEntityIdInfos(PageLink pageLink) {
-        log.debug("Find profile device id infos by pageLink [{}]", pageLink);
+        log.debug("Find profile asset id infos by pageLink [{}]", pageLink);
         return nativeAssetRepository.findProfileEntityIdInfos(DaoUtil.toPageable(pageLink));
     }
 
+    @Override
+    public PageData<ProfileEntityIdInfo> findProfileEntityIdInfosByTenantId(UUID tenantId, PageLink pageLink) {
+        log.debug("Find profile asset id infos by pageLink [{}]", pageLink);
+        return nativeAssetRepository.findProfileEntityIdInfosByTenantId(tenantId, DaoUtil.toPageable(pageLink));
+    }
+
     @Override
     public Long countByTenantId(TenantId tenantId) {
         return assetRepository.countByTenantId(tenantId.getId());

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

@@ -15,6 +15,8 @@
  */
 package org.thingsboard.server.dao.sql.cf;
 
+import org.springframework.data.domain.Page;
+import org.springframework.data.domain.Pageable;
 import org.springframework.data.jpa.repository.JpaRepository;
 import org.thingsboard.server.dao.model.sql.CalculatedFieldLinkEntity;
 
@@ -27,4 +29,8 @@ public interface CalculatedFieldLinkRepository extends JpaRepository<CalculatedF
 
     List<CalculatedFieldLinkEntity> findAllByTenantIdAndEntityId(UUID tenantId, UUID entityId);
 
+    List<CalculatedFieldLinkEntity> findAllByTenantId(UUID tenantId);
+
+    Page<CalculatedFieldLinkEntity> findAllByTenantId(UUID tenantId, Pageable pageable);
+
 }

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

@@ -32,6 +32,8 @@ public interface CalculatedFieldRepository extends JpaRepository<CalculatedField
 
     List<CalculatedFieldEntity> findAllByTenantIdAndEntityId(UUID tenantId, UUID entityId);
 
+    Page<CalculatedFieldEntity> findAllByTenantId(UUID tenantId, Pageable pageable);
+
     Page<CalculatedFieldEntity> findAllByTenantIdAndEntityId(UUID tenantId, UUID entityId, Pageable pageable);
 
     List<CalculatedFieldEntity> findAllByTenantId(UUID tenantId);

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

@@ -71,6 +71,12 @@ public class JpaCalculatedFieldDao extends JpaAbstractDao<CalculatedFieldEntity,
         return nativeCalculatedFieldRepository.findCalculatedFields(DaoUtil.toPageable(pageLink));
     }
 
+    @Override
+    public PageData<CalculatedField> findAllByTenantId(TenantId tenantId, PageLink pageLink) {
+        log.debug("Try to find calculated fields by tenantId[{}] and pageLink [{}]", tenantId, pageLink);
+        return DaoUtil.toPageData(calculatedFieldRepository.findAllByTenantId(tenantId.getId(), DaoUtil.toPageable(pageLink)));
+    }
+
     @Override
     public PageData<CalculatedField> findAllByEntityId(TenantId tenantId, EntityId entityId, PageLink pageLink) {
         log.debug("Try to find calculated fields by entityId[{}] and pageLink [{}]", entityId, pageLink);

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

@@ -54,6 +54,11 @@ public class JpaCalculatedFieldLinkDao extends JpaAbstractDao<CalculatedFieldLin
         return DaoUtil.convertDataList(calculatedFieldLinkRepository.findAllByTenantIdAndEntityId(tenantId.getId(), entityId.getId()));
     }
 
+    @Override
+    public List<CalculatedFieldLink> findCalculatedFieldLinksByTenantId(TenantId tenantId) {
+        return DaoUtil.convertDataList(calculatedFieldLinkRepository.findAllByTenantId(tenantId.getId()));
+    }
+
     @Override
     public List<CalculatedFieldLink> findAll() {
         return DaoUtil.convertDataList(calculatedFieldLinkRepository.findAll());
@@ -65,6 +70,12 @@ public class JpaCalculatedFieldLinkDao extends JpaAbstractDao<CalculatedFieldLin
         return nativeCalculatedFieldRepository.findCalculatedFieldLinks(DaoUtil.toPageable(pageLink));
     }
 
+    @Override
+    public PageData<CalculatedFieldLink> findAllByTenantId(TenantId tenantId, PageLink pageLink) {
+        log.debug("Try to find calculated field links by tenantId [{}], pageLink [{}]", tenantId, pageLink);
+        return DaoUtil.toPageData(calculatedFieldLinkRepository.findAllByTenantId(tenantId.getId(), DaoUtil.toPageable(pageLink)));
+    }
+
     @Override
     protected Class<CalculatedFieldLinkEntity> getEntityClass() {
         return CalculatedFieldLinkEntity.class;

dao/src/main/java/org/thingsboard/server/dao/sql/device/DefaultNativeAssetRepository.java

@@ -20,12 +20,9 @@ import org.springframework.data.domain.Pageable;
 import org.springframework.jdbc.core.namedparam.NamedParameterJdbcTemplate;
 import org.springframework.stereotype.Repository;
 import org.springframework.transaction.support.TransactionTemplate;
-import org.thingsboard.server.common.data.DeviceIdInfo;
 import org.thingsboard.server.common.data.ProfileEntityIdInfo;
 import org.thingsboard.server.common.data.id.AssetId;
 import org.thingsboard.server.common.data.id.AssetProfileId;
-import org.thingsboard.server.common.data.id.DeviceId;
-import org.thingsboard.server.common.data.id.DeviceProfileId;
 import org.thingsboard.server.common.data.id.TenantId;
 import org.thingsboard.server.common.data.page.PageData;
 
@@ -43,8 +40,8 @@ public class DefaultNativeAssetRepository extends AbstractNativeRepository imple
 
     @Override
     public PageData<ProfileEntityIdInfo> findProfileEntityIdInfos(Pageable pageable) {
-        String PROFILE_DEVICE_ID_INFO_QUERY = "SELECT tenant_id as tenantId, asset_profile_id as profileId, id as id FROM asset ORDER BY created_time ASC LIMIT %s OFFSET %s";
+        String PROFILE_ASSET_ID_INFO_QUERY = "SELECT tenant_id as tenantId, asset_profile_id as profileId, id as id FROM asset ORDER BY created_time ASC LIMIT %s OFFSET %s";
-        return find(COUNT_QUERY, PROFILE_DEVICE_ID_INFO_QUERY, pageable, row -> {
+        return find(COUNT_QUERY, PROFILE_ASSET_ID_INFO_QUERY, pageable, row -> {
             AssetId id = new AssetId((UUID) row.get("id"));
             AssetProfileId profileId = new AssetProfileId((UUID) row.get("profileId"));
             var tenantIdObj = row.get("tenantId");
@@ -52,4 +49,14 @@ public class DefaultNativeAssetRepository extends AbstractNativeRepository imple
         });
     }
 
+    @Override
+    public PageData<ProfileEntityIdInfo> findProfileEntityIdInfosByTenantId(UUID tenantId, Pageable pageable) {
+        String PROFILE_ASSET_ID_INFO_QUERY = String.format("SELECT tenant_id as tenantId, asset_profile_id as profileId, id as id FROM asset WHERE tenant_id = '%s' ORDER BY created_time ASC LIMIT %%s OFFSET %%s", tenantId);
+        return find(COUNT_QUERY, PROFILE_ASSET_ID_INFO_QUERY, pageable, row -> {
+            AssetId id = new AssetId((UUID) row.get("id"));
+            AssetProfileId profileId = new AssetProfileId((UUID) row.get("profileId"));
+            var tenantIdObj = row.get("tenantId");
+            return ProfileEntityIdInfo.create(tenantIdObj != null ? (UUID) tenantIdObj : TenantId.SYS_TENANT_ID.getId(), profileId, id);
+        });
+    }
 }

dao/src/main/java/org/thingsboard/server/dao/sql/device/DefaultNativeDeviceRepository.java

@@ -61,4 +61,15 @@ public class DefaultNativeDeviceRepository extends AbstractNativeRepository impl
         });
     }
 
+    @Override
+    public PageData<ProfileEntityIdInfo> findProfileEntityIdInfosByTenantId(UUID tenantId, Pageable pageable) {
+        String PROFILE_DEVICE_ID_INFO_QUERY = String.format("SELECT tenant_id as tenantId, device_profile_id as profileId, id as id FROM device WHERE tenant_id = '%s' ORDER BY created_time ASC LIMIT %%s OFFSET %%s", tenantId);
+        return find(COUNT_QUERY, PROFILE_DEVICE_ID_INFO_QUERY, pageable, row -> {
+            DeviceId id = new DeviceId((UUID) row.get("id"));
+            DeviceProfileId profileId = new DeviceProfileId((UUID) row.get("profileId"));
+            var tenantIdObj = row.get("tenantId");
+            return ProfileEntityIdInfo.create(tenantIdObj != null ? (UUID) tenantIdObj : TenantId.SYS_TENANT_ID.getId(), profileId, id);
+        });
+    }
+
 }

dao/src/main/java/org/thingsboard/server/dao/sql/device/JpaDeviceDao.java

@@ -179,11 +179,11 @@ public class JpaDeviceDao extends JpaAbstractDao<DeviceEntity, Device> implement
     @Override
     public PageData<DeviceId> findDeviceIdsByTenantIdAndDeviceProfileId(UUID tenantId, UUID deviceProfileId, PageLink pageLink) {
         return DaoUtil.pageToPageData(
-                deviceRepository.findIdsByTenantIdAndDeviceProfileId(
+                        deviceRepository.findIdsByTenantIdAndDeviceProfileId(
-                        tenantId,
+                                tenantId,
-                        deviceProfileId,
+                                deviceProfileId,
-                        pageLink.getTextSearch(),
+                                pageLink.getTextSearch(),
-                        DaoUtil.toPageable(pageLink)))
+                                DaoUtil.toPageable(pageLink)))
                 .mapData(DeviceId::new);
     }
 
@@ -281,6 +281,12 @@ public class JpaDeviceDao extends JpaAbstractDao<DeviceEntity, Device> implement
         return nativeDeviceRepository.findProfileEntityIdInfos(DaoUtil.toPageable(pageLink));
     }
 
+    @Override
+    public PageData<ProfileEntityIdInfo> findProfileEntityIdInfosByTenantId(UUID tenantId, PageLink pageLink) {
+        log.debug("Find profile device id infos by tenantId[{}], pageLink [{}]", tenantId, pageLink);
+        return nativeDeviceRepository.findProfileEntityIdInfosByTenantId(tenantId, DaoUtil.toPageable(pageLink));
+    }
+
     @Override
     public Device findByTenantIdAndExternalId(UUID tenantId, UUID externalId) {
         return DaoUtil.getData(deviceRepository.findByTenantIdAndExternalId(tenantId, externalId));

dao/src/main/java/org/thingsboard/server/dao/sql/device/NativeProfileEntityRepository.java

@@ -19,8 +19,12 @@ import org.springframework.data.domain.Pageable;
 import org.thingsboard.server.common.data.ProfileEntityIdInfo;
 import org.thingsboard.server.common.data.page.PageData;
 
+import java.util.UUID;
+
 public interface NativeProfileEntityRepository {
 
     PageData<ProfileEntityIdInfo> findProfileEntityIdInfos(Pageable pageable);
 
+    PageData<ProfileEntityIdInfo> findProfileEntityIdInfosByTenantId(UUID tenantId, Pageable pageable);
+
 }