Cluster mode fixes

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

@@ -108,6 +108,7 @@ import org.thingsboard.server.queue.discovery.TbServiceInfoProvider;
 import org.thingsboard.server.service.apiusage.TbApiUsageStateService;
 import org.thingsboard.server.service.cf.CalculatedFieldExecutionService;
 import org.thingsboard.server.service.cf.cache.CalculatedFieldEntityProfileCache;
+import org.thingsboard.server.service.cf.ctx.CalculatedFieldStateService;
 import org.thingsboard.server.service.cf.ctx.state.ArgumentEntry;
 import org.thingsboard.server.service.component.ComponentDiscoveryService;
 import org.thingsboard.server.service.edge.rpc.EdgeRpcService;
@@ -527,6 +528,11 @@ public class ActorSystemContext {
     @Getter
     private CalculatedFieldExecutionService calculatedFieldExecutionService;
 
+    @Lazy
+    @Autowired(required = false)
+    @Getter
+    private CalculatedFieldStateService calculatedFieldStateService;
+
     @Lazy
     @Autowired(required = false)
     @Getter

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

@@ -1,12 +1,12 @@
 /**
  * Copyright © 2016-2024 The Thingsboard Authors
- *
+ * <p>
  * 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
- *
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
  * 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.
@@ -38,6 +38,7 @@ import org.thingsboard.server.gen.transport.TransportProtos.TsKvProto;
 import org.thingsboard.server.service.cf.CalculatedFieldExecutionService;
 import org.thingsboard.server.service.cf.CalculatedFieldResult;
 import org.thingsboard.server.service.cf.ctx.CalculatedFieldEntityCtxId;
+import org.thingsboard.server.service.cf.ctx.CalculatedFieldStateService;
 import org.thingsboard.server.service.cf.ctx.state.ArgumentEntry;
 import org.thingsboard.server.service.cf.ctx.state.CalculatedFieldCtx;
 import org.thingsboard.server.service.cf.ctx.state.CalculatedFieldState;
@@ -67,6 +68,7 @@ public class CalculatedFieldEntityMessageProcessor extends AbstractContextAwareM
     final TenantId tenantId;
     final EntityId entityId;
     final CalculatedFieldExecutionService cfService;
+    final CalculatedFieldStateService cfStateService;
 
     TbActorCtx ctx;
     Map<CalculatedFieldId, CalculatedFieldState> states = new HashMap<>();
@@ -76,6 +78,7 @@ public class CalculatedFieldEntityMessageProcessor extends AbstractContextAwareM
         this.tenantId = tenantId;
         this.entityId = entityId;
         this.cfService = systemContext.getCalculatedFieldExecutionService();
+        this.cfStateService = systemContext.getCalculatedFieldStateService();
     }
 
     void init(TbActorCtx ctx) {
@@ -102,13 +105,13 @@ public class CalculatedFieldEntityMessageProcessor extends AbstractContextAwareM
         log.info("[{}] Processing CF entity delete msg.", msg.getEntityId());
         if (this.entityId.equals(msg.getEntityId())) {
             MultipleTbCallback multipleTbCallback = new MultipleTbCallback(states.size(), msg.getCallback());
-            states.forEach((cfId, state) -> cfService.deleteStateFromStorage(new CalculatedFieldEntityCtxId(tenantId, cfId, entityId), multipleTbCallback));
+            states.forEach((cfId, state) -> cfStateService.removeState(new CalculatedFieldEntityCtxId(tenantId, cfId, entityId), multipleTbCallback));
             ctx.stop(ctx.getSelf());
         } else {
             var cfId = new CalculatedFieldId(msg.getEntityId().getId());
             var state = states.remove(cfId);
             if (state != null) {
-                cfService.deleteStateFromStorage(new CalculatedFieldEntityCtxId(tenantId, cfId, entityId), msg.getCallback());
+                cfStateService.removeState(new CalculatedFieldEntityCtxId(tenantId, cfId, entityId), msg.getCallback());
             }
         }
     }
@@ -178,8 +181,13 @@ public class CalculatedFieldEntityMessageProcessor extends AbstractContextAwareM
             log.info("[{}] No new argument values to process for CF.", ctx.getCfId());
             callback.onSuccess(CALLBACKS_PER_CF);
         }
-        CalculatedFieldState state = getOrInitState(ctx);
-        if (state.updateState(newArgValues)) {
+        CalculatedFieldState state = states.get(ctx.getCfId());
+        boolean justRestored = false;
+        if (state == null) {
+            state = getOrInitState(ctx);
+            justRestored = true;
+        }
+        if (state.updateState(newArgValues) || justRestored) {
             cfIdList = new ArrayList<>(cfIdList);
             cfIdList.add(ctx.getCfId());
             processStateIfReady(ctx, cfIdList, state, tbMsgId, tbMsgType, callback);
@@ -222,7 +230,7 @@ public class CalculatedFieldEntityMessageProcessor extends AbstractContextAwareM
         } else {
             callback.onSuccess(); // State was updated but no calculation performed;
         }
-        cfService.pushStateToStorage(ctx, new CalculatedFieldEntityCtxId(tenantId, ctx.getCfId(), entityId), state, callback);
+        cfStateService.persistState(ctx, new CalculatedFieldEntityCtxId(tenantId, ctx.getCfId(), entityId), state, callback);
     }
 
     private Map<String, ArgumentEntry> mapToArguments(CalculatedFieldCtx ctx, List<TsKvProto> data) {

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

@@ -42,6 +42,7 @@ import org.thingsboard.server.gen.transport.TransportProtos.CalculatedFieldEntit
 import org.thingsboard.server.service.cf.CalculatedFieldExecutionService;
 import org.thingsboard.server.service.cf.cache.CalculatedFieldEntityProfileCache;
 import org.thingsboard.server.service.cf.ctx.CalculatedFieldEntityCtxId;
+import org.thingsboard.server.service.cf.ctx.CalculatedFieldStateService;
 import org.thingsboard.server.service.cf.ctx.state.CalculatedFieldCtx;
 import org.thingsboard.server.service.profile.TbAssetProfileCache;
 import org.thingsboard.server.service.profile.TbDeviceProfileCache;
@@ -68,6 +69,7 @@ public class CalculatedFieldManagerMessageProcessor extends AbstractContextAware
     private final ConcurrentMap<EntityId, List<CalculatedFieldLink>> entityIdCalculatedFieldLinks = new ConcurrentHashMap<>();
 
     private final CalculatedFieldExecutionService cfExecService;
+    private final CalculatedFieldStateService cfStateService;
     private final CalculatedFieldEntityProfileCache cfEntityCache;
     private final CalculatedFieldService cfDaoService;
     private final TbAssetProfileCache assetProfileCache;
@@ -80,6 +82,7 @@ public class CalculatedFieldManagerMessageProcessor extends AbstractContextAware
         super(systemContext);
         this.cfEntityCache = systemContext.getCalculatedFieldEntityProfileCache();
         this.cfExecService = systemContext.getCalculatedFieldExecutionService();
+        this.cfStateService = systemContext.getCalculatedFieldStateService();
         this.cfDaoService = systemContext.getCalculatedFieldService();
         this.assetProfileCache = systemContext.getAssetProfileCache();
         this.deviceProfileCache = systemContext.getDeviceProfileCache();
@@ -127,7 +130,7 @@ public class CalculatedFieldManagerMessageProcessor extends AbstractContextAware
             log.info("Pushing CF state restore msg to specific actor [{}]", msg.getId().entityId());
             getOrCreateActor(msg.getId().entityId()).tell(msg);
         } else {
-            cfExecService.deleteStateFromStorage(msg.getId(), msg.getCallback());
+            cfStateService.removeState(msg.getId(), msg.getCallback());
         }
     }
 

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

@@ -43,13 +43,10 @@ public interface CalculatedFieldExecutionService {
 
     void pushRequestToQueue(AttributesSaveRequest request, List<Long> result, FutureCallback<Void> callback);
 
-    void pushStateToStorage(CalculatedFieldCtx ctx, CalculatedFieldEntityCtxId stateId, CalculatedFieldState state, TbCallback callback);
-
     ListenableFuture<CalculatedFieldState> fetchStateFromDb(CalculatedFieldCtx ctx, EntityId entityId);
 
     void pushMsgToRuleEngine(TenantId tenantId, EntityId entityId, CalculatedFieldResult calculationResult, List<CalculatedFieldId> cfIds, TbCallback callback);
 
     void pushMsgToLinks(CalculatedFieldTelemetryMsg msg, List<CalculatedFieldEntityCtxId> linkedCalculatedFields, TbCallback callback);
 
-    void deleteStateFromStorage(CalculatedFieldEntityCtxId calculatedFieldEntityCtxId, TbCallback callback);
 }

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

@@ -141,7 +141,6 @@ public class DefaultCalculatedFieldExecutionService extends AbstractPartitionBas
     private final CalculatedFieldCache calculatedFieldCache;
     private final AttributesService attributesService;
     private final TimeseriesService timeseriesService;
-    private final CalculatedFieldStateService stateService;
     private final TbClusterService clusterService;
     private final ApiLimitService apiLimitService;
 
@@ -263,16 +262,6 @@ public class DefaultCalculatedFieldExecutionService extends AbstractPartitionBas
         }, calculatedFieldCallbackExecutor);
     }
 
-    @Override
-    public void pushStateToStorage(CalculatedFieldCtx ctx, CalculatedFieldEntityCtxId stateId, CalculatedFieldState state, TbCallback callback) {
-        stateService.persistState(ctx, stateId, state, callback);
-    }
-
-    @Override
-    public void deleteStateFromStorage(CalculatedFieldEntityCtxId calculatedFieldEntityCtxId, TbCallback callback) {
-        stateService.removeState(calculatedFieldEntityCtxId, callback);
-    }
-
     @Override
     protected Map<TopicPartitionInfo, List<ListenableFuture<?>>> onAddedPartitions(Set<TopicPartitionInfo> addedPartitions) {
         var result = new HashMap<TopicPartitionInfo, List<ListenableFuture<?>>>();

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

@@ -29,7 +29,6 @@ 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;
-import org.thingsboard.server.service.cf.ctx.CalculatedFieldStateService;
 
 @Slf4j
 @Service
@@ -38,9 +37,6 @@ import org.thingsboard.server.service.cf.ctx.CalculatedFieldStateService;
 public class DefaultCalculatedFieldInitService implements CalculatedFieldInitService {
 
     private final CalculatedFieldEntityProfileCache entityProfileCache;
-    private final CalculatedFieldStateService stateService;
-
-    private final ActorSystemContext actorSystemContext;
     private final AssetService assetService;
     private final DeviceService deviceService;
 
@@ -62,9 +58,4 @@ public class DefaultCalculatedFieldInitService implements CalculatedFieldInitSer
         }
     }
 
-    @AfterStartUp(order = AfterStartUp.CF_STATE_RESTORE_SERVICE)
-    public void initCalculatedFieldStates() {
-        stateService.restoreStates().forEach((k, v) -> actorSystemContext.tell(new CalculatedFieldStateRestoreMsg(k, v)));
-    }
-
 }

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

@@ -21,6 +21,7 @@ import org.rocksdb.RocksDBException;
 import org.rocksdb.RocksIterator;
 import org.rocksdb.WriteOptions;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
 import org.springframework.stereotype.Service;
 import org.thingsboard.server.gen.transport.TransportProtos.CalculatedFieldEntityCtxIdProto;
 import org.thingsboard.server.gen.transport.TransportProtos.CalculatedFieldStateProto;
@@ -32,7 +33,7 @@ import java.util.Map;
 
 @Service
 @Slf4j
-@ConditionalOnExpression("'${service.type:null}'=='monolith'")
+@ConditionalOnProperty(prefix = "zk", value = "enabled", havingValue = "false", matchIfMissing = true)
 public class RocksDBService {
 
     private final RocksDB db;

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

@@ -39,6 +39,7 @@ import java.util.stream.Collectors;
 @Service
 @Slf4j
 @RequiredArgsConstructor
+//TODO: remove and use TenantEntityProfileCache in each CalculatedFieldManagerMessageProcessor;
 public class DefaultCalculatedFieldEntityProfileCache extends TbApplicationEventListener<PartitionChangeEvent> implements CalculatedFieldEntityProfileCache {
 
     private static final Integer UNKNOWN = -1;

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

@@ -19,12 +19,8 @@ import org.thingsboard.server.common.msg.queue.TbCallback;
 import org.thingsboard.server.service.cf.ctx.state.CalculatedFieldCtx;
 import org.thingsboard.server.service.cf.ctx.state.CalculatedFieldState;
 
-import java.util.Map;
-
 public interface CalculatedFieldStateService {
 
-    Map<CalculatedFieldEntityCtxId, CalculatedFieldState> restoreStates();
-
     void persistState(CalculatedFieldCtx ctx, CalculatedFieldEntityCtxId stateId, CalculatedFieldState state, TbCallback callback);
 
     void removeState(CalculatedFieldEntityCtxId stateId, TbCallback callback);

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

@@ -0,0 +1,68 @@
+/**
+ * Copyright © 2016-2024 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.ctx.state;
+
+import lombok.RequiredArgsConstructor;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
+import org.springframework.stereotype.Service;
+import org.thingsboard.server.actors.ActorSystemContext;
+import org.thingsboard.server.actors.calculatedField.CalculatedFieldStateRestoreMsg;
+import org.thingsboard.server.common.data.StringUtils;
+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.EntityIdFactory;
+import org.thingsboard.server.common.data.id.TenantId;
+import org.thingsboard.server.common.data.kv.BasicKvEntry;
+import org.thingsboard.server.common.msg.queue.TbCallback;
+import org.thingsboard.server.common.util.KvProtoUtil;
+import org.thingsboard.server.gen.transport.TransportProtos.CalculatedFieldEntityCtxIdProto;
+import org.thingsboard.server.gen.transport.TransportProtos.CalculatedFieldStateProto;
+import org.thingsboard.server.gen.transport.TransportProtos.SingleValueArgumentProto;
+import org.thingsboard.server.gen.transport.TransportProtos.TsValueListProto;
+import org.thingsboard.server.gen.transport.TransportProtos.TsValueProto;
+import org.thingsboard.server.queue.util.AfterStartUp;
+import org.thingsboard.server.service.cf.RocksDBService;
+import org.thingsboard.server.service.cf.ctx.CalculatedFieldEntityCtxId;
+import org.thingsboard.server.service.cf.ctx.CalculatedFieldStateService;
+
+import java.util.Map;
+import java.util.Optional;
+import java.util.TreeMap;
+import java.util.UUID;
+import java.util.stream.Collectors;
+
+@Service
+@RequiredArgsConstructor
+@ConditionalOnExpression("'${zk.enabled:false}'=='true' && ('${service.type:null}'=='monolith' || '${service.type:null}'=='tb-rule-engine')")
+public class KafkaCalculatedFieldStateService implements CalculatedFieldStateService {
+
+    @AfterStartUp(order = AfterStartUp.CF_STATE_RESTORE_SERVICE)
+    public void initCalculatedFieldStates() {
+    }
+
+    @Override
+    public void persistState(CalculatedFieldCtx ctx, CalculatedFieldEntityCtxId stateId, CalculatedFieldState state, TbCallback callback) {
+        callback.onSuccess();
+    }
+
+    @Override
+    public void removeState(CalculatedFieldEntityCtxId ctxId, TbCallback callback) {
+        callback.onSuccess();
+    }
+
+}

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

@@ -16,8 +16,10 @@
 package org.thingsboard.server.service.cf.ctx.state;
 
 import lombok.RequiredArgsConstructor;
-import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
 import org.springframework.stereotype.Service;
+import org.thingsboard.server.actors.ActorSystemContext;
+import org.thingsboard.server.actors.calculatedField.CalculatedFieldStateRestoreMsg;
 import org.thingsboard.server.common.data.StringUtils;
 import org.thingsboard.server.common.data.cf.CalculatedFieldType;
 import org.thingsboard.server.common.data.id.CalculatedFieldId;
@@ -32,6 +34,7 @@ import org.thingsboard.server.gen.transport.TransportProtos.CalculatedFieldState
 import org.thingsboard.server.gen.transport.TransportProtos.SingleValueArgumentProto;
 import org.thingsboard.server.gen.transport.TransportProtos.TsValueListProto;
 import org.thingsboard.server.gen.transport.TransportProtos.TsValueProto;
+import org.thingsboard.server.queue.util.AfterStartUp;
 import org.thingsboard.server.service.cf.RocksDBService;
 import org.thingsboard.server.service.cf.ctx.CalculatedFieldEntityCtxId;
 import org.thingsboard.server.service.cf.ctx.CalculatedFieldStateService;
@@ -44,12 +47,12 @@ import java.util.stream.Collectors;
 
 @Service
 @RequiredArgsConstructor
-@ConditionalOnExpression("'${service.type:null}'=='monolith'")
-public class RocksDBStateService implements CalculatedFieldStateService {
+@ConditionalOnProperty(prefix = "zk", value = "enabled", havingValue = "false", matchIfMissing = true)
+public class RocksDBCalculatedFieldStateService implements CalculatedFieldStateService {
 
+    private final ActorSystemContext actorSystemContext;
     private final RocksDBService rocksDBService;
 
-    @Override
     public Map<CalculatedFieldEntityCtxId, CalculatedFieldState> restoreStates() {
         return rocksDBService.getAll().entrySet().stream()
                 .collect(Collectors.toMap(
@@ -58,6 +61,12 @@ public class RocksDBStateService implements CalculatedFieldStateService {
                 ));
     }
 
+    @AfterStartUp(order = AfterStartUp.CF_STATE_RESTORE_SERVICE)
+    public void initCalculatedFieldStates() {
+        restoreStates().forEach((k, v) -> actorSystemContext.tell(new CalculatedFieldStateRestoreMsg(k, v)));
+    }
+
+
     @Override
     public void persistState(CalculatedFieldCtx ctx, CalculatedFieldEntityCtxId stateId, CalculatedFieldState state, TbCallback callback) {
         CalculatedFieldStateProto stateProto = toProto(stateId, state);

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

@@ -674,7 +674,7 @@ public class DefaultTbClusterService implements TbClusterService {
                         .oldName(old.getName())
                         .name(entity.getName())
                         .build();
-                pushMsgToCalculatedFields(entity.getTenantId(), entity.getId(), ToCalculatedFieldMsg.newBuilder().setComponentLifecycleMsg(toProto(msg)).build(), TbQueueCallback.EMPTY);
+                broadcastToCalculatedFields(ToCalculatedFieldNotificationMsg.newBuilder().setComponentLifecycleMsg(toProto(msg)).build(), TbQueueCallback.EMPTY);
             }
             if (deviceNameChanged || deviceProfileChanged) {
                 pushMsgToCore(new DeviceNameOrTypeUpdateMsg(entity.getTenantId(), entity.getId(), entity.getName(), entity.getType()), null);
@@ -687,7 +687,7 @@ public class DefaultTbClusterService implements TbClusterService {
                     .profileId(entity.getDeviceProfileId())
                     .name(entity.getName())
                     .build();
-            pushMsgToCalculatedFields(entity.getTenantId(), entity.getId(), ToCalculatedFieldMsg.newBuilder().setComponentLifecycleMsg(toProto(msg)).build(), TbQueueCallback.EMPTY);
+            broadcastToCalculatedFields(ToCalculatedFieldNotificationMsg.newBuilder().setComponentLifecycleMsg(toProto(msg)).build(), TbQueueCallback.EMPTY);
         }
         broadcastEntityStateChangeEvent(entity.getTenantId(), entity.getId(), created ? ComponentLifecycleEvent.CREATED : ComponentLifecycleEvent.UPDATED);
         sendDeviceStateServiceEvent(entity.getTenantId(), entity.getId(), created, !created, false);
@@ -710,7 +710,7 @@ public class DefaultTbClusterService implements TbClusterService {
                         .oldName(old.getName())
                         .name(entity.getName())
                         .build();
-                pushMsgToCalculatedFields(entity.getTenantId(), entity.getId(), ToCalculatedFieldMsg.newBuilder().setComponentLifecycleMsg(toProto(msg)).build(), TbQueueCallback.EMPTY);
+                broadcastToCalculatedFields(ToCalculatedFieldNotificationMsg.newBuilder().setComponentLifecycleMsg(toProto(msg)).build(), TbQueueCallback.EMPTY);
             }
         } else {
             ComponentLifecycleMsg msg = ComponentLifecycleMsg.builder()
@@ -720,7 +720,7 @@ public class DefaultTbClusterService implements TbClusterService {
                     .profileId(entity.getAssetProfileId())
                     .name(entity.getName())
                     .build();
-            pushMsgToCalculatedFields(entity.getTenantId(), entity.getId(), ToCalculatedFieldMsg.newBuilder().setComponentLifecycleMsg(toProto(msg)).build(), TbQueueCallback.EMPTY);
+            broadcastToCalculatedFields(ToCalculatedFieldNotificationMsg.newBuilder().setComponentLifecycleMsg(toProto(msg)).build(), TbQueueCallback.EMPTY);
         }
         broadcastEntityStateChangeEvent(entity.getTenantId(), entity.getId(), created ? ComponentLifecycleEvent.CREATED : ComponentLifecycleEvent.UPDATED);
     }
@@ -872,6 +872,6 @@ public class DefaultTbClusterService implements TbClusterService {
 
     private void handleCalculatedFieldEntityDeleted(TenantId tenantId, EntityId entityId) {
         ComponentLifecycleMsg msg = new ComponentLifecycleMsg(tenantId, entityId, ComponentLifecycleEvent.DELETED);
-        pushMsgToCalculatedFields(tenantId, entityId, ToCalculatedFieldMsg.newBuilder().setComponentLifecycleMsg(toProto(msg)).build(), TbQueueCallback.EMPTY);
+        broadcastToCalculatedFields(ToCalculatedFieldNotificationMsg.newBuilder().setComponentLifecycleMsg(toProto(msg)).build(), TbQueueCallback.EMPTY);
     }
 }