WIP: Cluster mode implementation

2 years ago · 4c71b9d5f6
20 changed files with 298 additions and 54 deletions
--- a/application/src/main/java/org/thingsboard/server/service/cf/CalculatedFieldCache.java
+++ b/application/src/main/java/org/thingsboard/server/service/cf/CalculatedFieldCache.java
@ -36,9 +36,9 @@ public interface CalculatedFieldCache {

    List<CalculatedFieldLink> getCalculatedFieldLinksByEntityId(EntityId entityId);

-    CalculatedFieldCtx getCalculatedFieldCtx(CalculatedFieldId calculatedFieldId, TbelInvokeService tbelInvokeService);
+    CalculatedFieldCtx getCalculatedFieldCtx(CalculatedFieldId calculatedFieldId);

-    List<CalculatedFieldCtx> getCalculatedFieldCtxsByEntityId(EntityId entityId, TbelInvokeService tbelInvokeService);
+    List<CalculatedFieldCtx> getCalculatedFieldCtxsByEntityId(EntityId entityId);

    Set<EntityId> getEntitiesByProfile(TenantId tenantId, EntityId entityId);

--- a/application/src/main/java/org/thingsboard/server/service/cf/CalculatedFieldExecutionService.java
+++ b/application/src/main/java/org/thingsboard/server/service/cf/CalculatedFieldExecutionService.java
@ -15,6 +15,8 @@
 */
 package org.thingsboard.server.service.cf;

+import org.thingsboard.rule.engine.api.TimeseriesSaveRequest;
+import org.thingsboard.server.common.data.kv.TimeseriesSaveResult;
 import org.thingsboard.server.common.msg.queue.TbCallback;
 import org.thingsboard.server.gen.transport.TransportProtos;
 import org.thingsboard.server.service.cf.telemetry.CalculatedFieldTelemetryUpdateRequest;
@ -22,13 +24,13 @@ import org.thingsboard.server.service.cf.telemetry.CalculatedFieldTelemetryUpdat
 public interface CalculatedFieldExecutionService {

    /**
-     * Push incoming telemetry to the CF processing queue for async processing.
-     * @param request - telemetry request;
-     * @param callback - callback to be executed when the message is ack by the queue.
+     * Filter CFs based on the request entity. Push to the queue if any matching CF exist;
+     * @param request - telemetry save request;
+     * @param request - telemetry save result;
     */
-    void pushRequestToQueue(CalculatedFieldTelemetryUpdateRequest request, TbCallback callback);
+    void pushRequestToQueue(TimeseriesSaveRequest request, TimeseriesSaveResult result);

-    void pushEntityUpdateMsg(TransportProtos.CalculatedFieldEntityUpdateMsgProto proto, TbCallback callback);
+//    void pushEntityUpdateMsg(TransportProtos.CalculatedFieldEntityUpdateMsgProto proto, TbCallback callback);

    /*  ===================================================== */

--- a/application/src/main/java/org/thingsboard/server/service/cf/DefaultCalculatedFieldCache.java
+++ b/application/src/main/java/org/thingsboard/server/service/cf/DefaultCalculatedFieldCache.java
@ -36,6 +36,7 @@ import org.thingsboard.server.dao.cf.CalculatedFieldService;
 import org.thingsboard.server.dao.device.DeviceService;
 import org.thingsboard.server.service.cf.ctx.state.CalculatedFieldCtx;

+import java.util.Collections;
 import java.util.HashSet;
 import java.util.List;
 import java.util.Set;
@ -55,6 +56,7 @@ public class DefaultCalculatedFieldCache implements CalculatedFieldCache {
    private final CalculatedFieldService calculatedFieldService;
    private final AssetService assetService;
    private final DeviceService deviceService;
+    private final TbelInvokeService tbelInvokeService;

    private final ConcurrentMap<CalculatedFieldId, CalculatedField> calculatedFields = new ConcurrentHashMap<>();
    private final ConcurrentMap<EntityId, List<CalculatedField>> entityIdCalculatedFields = new ConcurrentHashMap<>();
@ -105,7 +107,7 @@ public class DefaultCalculatedFieldCache implements CalculatedFieldCache {
    }

    @Override
-    public CalculatedFieldCtx getCalculatedFieldCtx(CalculatedFieldId calculatedFieldId, TbelInvokeService tbelInvokeService) {
+    public CalculatedFieldCtx getCalculatedFieldCtx(CalculatedFieldId calculatedFieldId) {
        CalculatedFieldCtx ctx = calculatedFieldsCtx.get(calculatedFieldId);
        if (ctx == null) {
            calculatedFieldFetchLock.lock();
@ -128,9 +130,12 @@ public class DefaultCalculatedFieldCache implements CalculatedFieldCache {
    }

    @Override
-    public List<CalculatedFieldCtx> getCalculatedFieldCtxsByEntityId(EntityId entityId, TbelInvokeService tbelInvokeService) {
+    public List<CalculatedFieldCtx> getCalculatedFieldCtxsByEntityId(EntityId entityId) {
+        if (entityId == null) {
+            return Collections.emptyList();
+        }
        return getCalculatedFieldsByEntityId(entityId).stream()
-                .map(cf -> getCalculatedFieldCtx(cf.getId(), tbelInvokeService))
+                .map(cf -> getCalculatedFieldCtx(cf.getId()))
                .toList();
    }

--- a/application/src/main/java/org/thingsboard/server/service/cf/DefaultCalculatedFieldExecutionService.java
+++ b/application/src/main/java/org/thingsboard/server/service/cf/DefaultCalculatedFieldExecutionService.java
@ -33,12 +33,13 @@ import org.springframework.beans.factory.annotation.Value;
 import org.springframework.stereotype.Service;
 import org.thingsboard.common.util.JacksonUtil;
 import org.thingsboard.common.util.ThingsBoardExecutors;
-import org.thingsboard.script.api.tbel.TbelInvokeService;
+import org.thingsboard.rule.engine.api.TimeseriesSaveRequest;
 import org.thingsboard.server.cluster.TbClusterService;
 import org.thingsboard.server.common.data.AttributeScope;
 import org.thingsboard.server.common.data.EntityType;
 import org.thingsboard.server.common.data.StringUtils;
 import org.thingsboard.server.common.data.cf.CalculatedField;
+import org.thingsboard.server.common.data.cf.CalculatedFieldLink;
 import org.thingsboard.server.common.data.cf.CalculatedFieldType;
 import org.thingsboard.server.common.data.cf.configuration.Argument;
 import org.thingsboard.server.common.data.cf.configuration.ArgumentType;
@ -60,6 +61,7 @@ import org.thingsboard.server.common.data.kv.DoubleDataEntry;
 import org.thingsboard.server.common.data.kv.KvEntry;
 import org.thingsboard.server.common.data.kv.ReadTsKvQuery;
 import org.thingsboard.server.common.data.kv.StringDataEntry;
+import org.thingsboard.server.common.data.kv.TimeseriesSaveResult;
 import org.thingsboard.server.common.data.kv.TsKvEntry;
 import org.thingsboard.server.common.data.msg.TbMsgType;
 import org.thingsboard.server.common.data.page.PageDataIterable;
@ -72,7 +74,9 @@ import org.thingsboard.server.common.util.ProtoUtils;
 import org.thingsboard.server.dao.attributes.AttributesService;
 import org.thingsboard.server.dao.cf.CalculatedFieldService;
 import org.thingsboard.server.dao.timeseries.TimeseriesService;
-import org.thingsboard.server.gen.transport.TransportProtos;
+import org.thingsboard.server.gen.transport.TransportProtos.ToCalculatedFieldMsg;
+import org.thingsboard.server.queue.TbQueueCallback;
+import org.thingsboard.server.queue.TbQueueMsgMetadata;
 import org.thingsboard.server.service.cf.ctx.CalculatedFieldEntityCtx;
 import org.thingsboard.server.service.cf.ctx.CalculatedFieldEntityCtxId;
 import org.thingsboard.server.service.cf.ctx.CalculatedFieldStateService;
@ -103,6 +107,8 @@ import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentMap;
 import java.util.function.Consumer;
+import java.util.function.Predicate;
+import java.util.function.Supplier;
 import java.util.stream.Collectors;

 import static org.thingsboard.server.common.data.DataConstants.SCOPE;
@ -113,6 +119,16 @@ import static org.thingsboard.server.common.util.ProtoUtils.toTsKvProto;
@RequiredArgsConstructor
 public class DefaultCalculatedFieldExecutionService extends AbstractPartitionBasedService<CalculatedFieldId> implements CalculatedFieldExecutionService {

+    public static final TbQueueCallback DUMMY_TB_QUEUE_CALLBACK = new TbQueueCallback() {
+        @Override
+        public void onSuccess(TbQueueMsgMetadata metadata) {
+        }
+
+        @Override
+        public void onFailure(Throwable t) {
+        }
+    };
+
    private final CalculatedFieldService calculatedFieldService;
    private final TbAssetProfileCache assetProfileCache;
    private final TbDeviceProfileCache deviceProfileCache;
@ -121,7 +137,6 @@ public class DefaultCalculatedFieldExecutionService extends AbstractPartitionBas
    private final TimeseriesService timeseriesService;
    private final CalculatedFieldStateService stateService;
    private final TbClusterService clusterService;
-    private final TbelInvokeService tbelInvokeService;

    private ListeningExecutorService calculatedFieldExecutor;
    private ListeningExecutorService calculatedFieldCallbackExecutor;
@ -169,6 +184,70 @@ public class DefaultCalculatedFieldExecutionService extends AbstractPartitionBas
        return "calculated-field-scheduled";
    }

+    @Override
+    public void pushRequestToQueue(TimeseriesSaveRequest request, TimeseriesSaveResult result) {
+        var tenantId = request.getTenantId();
+        var entityId = request.getEntityId();
+        //TODO: 1. check that request entity has calculated fields for entity or profile. If yes - push to corresponding partitions;
+        //TODO: 2. check that request entity has calculated field links. If yes - push to corresponding partitions;
+        //TODO: in 1 and 2 we should do the check as quick as possible. Should we also check the field/link keys?;
+        checkEntityAndPushToQueue(tenantId, entityId, cf -> cf.matches(request.getEntries()), cf -> cf.linkMatches(entityId, request.getEntries()),
+                () -> toCalculatedFieldTelemetryMsgProto(request, result), request.getCallback());
+    }
+
+    private void checkEntityAndPushToQueue(TenantId tenantId, EntityId entityId, Predicate<CalculatedFieldCtx> mainEntityFilter,  Predicate<CalculatedFieldCtx> linkedEntityFilter, Supplier<ToCalculatedFieldMsg> msg, FutureCallback<Void> callback) {
+        boolean send = checkEntityForCalculatedFields(tenantId, entityId, mainEntityFilter, linkedEntityFilter);
+        if (send) {
+            clusterService.pushMsgToCalculatedFields(tenantId, entityId, msg.get(), wrap(callback));
+        } else {
+            if (callback != null) {
+                callback.onSuccess(null);
+            }
+        }
+    }
+
+    private boolean checkEntityForCalculatedFields(TenantId tenantId, EntityId entityId, Predicate<CalculatedFieldCtx> filter, Predicate<CalculatedFieldCtx> linkedEntityFilter) {
+        boolean send = false;
+        if (supportedReferencedEntities.contains(entityId.getEntityType())) {
+            send = calculatedFieldCache.getCalculatedFieldCtxsByEntityId(entityId).stream().anyMatch(filter);
+            if (!send) {
+                send = calculatedFieldCache.getCalculatedFieldCtxsByEntityId(getProfileId(tenantId, entityId)).stream().anyMatch(filter);
+            }
+            if (!send) {
+                send = calculatedFieldCache.getCalculatedFieldLinksByEntityId(entityId).stream()
+                        .map(CalculatedFieldLink::getCalculatedFieldId)
+                        .map(calculatedFieldCache::getCalculatedFieldCtx)
+                        .anyMatch(linkedEntityFilter);
+            }
+        }
+        return send;
+    }
+
+    private ToCalculatedFieldMsg toCalculatedFieldTelemetryMsgProto(TimeseriesSaveRequest request, TimeseriesSaveResult result) {
+        ////TODO: IM to push to CF queue
+        return null;
+    }
+
+    private void processCalculatedFieldLinks(CalculatedFieldTelemetryUpdateRequest request, Map<TopicPartitionInfo, List<CalculatedFieldEntityCtxId>> tpiStates) {
+        TenantId tenantId = request.getTenantId();
+        EntityId entityId = request.getEntityId();
+
+        calculatedFieldCache.getCalculatedFieldLinksByEntityId(entityId)
+                .forEach(link -> {
+                    CalculatedFieldId calculatedFieldId = link.getCalculatedFieldId();
+                    CalculatedFieldCtx ctx = calculatedFieldCache.getCalculatedFieldCtx(calculatedFieldId, tbelInvokeService);
+                    EntityId targetEntityId = ctx.getEntityId();
+
+                    if (isProfileEntity(targetEntityId)) {
+                        calculatedFieldCache.getEntitiesByProfile(tenantId, targetEntityId).forEach(entityByProfile -> {
+                            processCalculatedFieldLink(request, entityByProfile, ctx, tpiStates);
+                        });
+                    } else {
+                        processCalculatedFieldLink(request, targetEntityId, ctx, tpiStates);
+                    }
+                });
+    }
+
    @Override
    protected Map<TopicPartitionInfo, List<ListenableFuture<?>>> onAddedPartitions(Set<TopicPartitionInfo> addedPartitions) {
        var result = new HashMap<TopicPartitionInfo, List<ListenableFuture<?>>>();
@ -374,7 +453,7 @@ public class DefaultCalculatedFieldExecutionService extends AbstractPartitionBas

    private void processCalculatedFields(CalculatedFieldTelemetryUpdateRequest request, EntityId cfTargetEntityId) {
        if (cfTargetEntityId != null) {
-            calculatedFieldCache.getCalculatedFieldCtxsByEntityId(cfTargetEntityId, tbelInvokeService).forEach(ctx -> {
+            calculatedFieldCache.getCalculatedFieldCtxsByEntityId(cfTargetEntityId).forEach(ctx -> {
                Map<String, KvEntry> updatedTelemetry = request.getMappedTelemetry(ctx, cfTargetEntityId);
                if (!updatedTelemetry.isEmpty()) {
                    EntityId targetEntityId = isProfileEntity(cfTargetEntityId) ? request.getEntityId() : cfTargetEntityId;
@ -391,7 +470,7 @@ public class DefaultCalculatedFieldExecutionService extends AbstractPartitionBas
        calculatedFieldCache.getCalculatedFieldLinksByEntityId(entityId)
                .forEach(link -> {
                    CalculatedFieldId calculatedFieldId = link.getCalculatedFieldId();
-                    CalculatedFieldCtx ctx = calculatedFieldCache.getCalculatedFieldCtx(calculatedFieldId, tbelInvokeService);
+                    CalculatedFieldCtx ctx = calculatedFieldCache.getCalculatedFieldCtx(calculatedFieldId);
                    EntityId targetEntityId = ctx.getEntityId();

                    if (isProfileEntity(targetEntityId)) {
@ -830,4 +909,30 @@ public class DefaultCalculatedFieldExecutionService extends AbstractPartitionBas
        };
    }

+    private static TbQueueCallback wrap(FutureCallback<Void> callback) {
+        if (callback != null) {
+            return new FutureCallbackWrapper(callback);
+        } else {
+            return DUMMY_TB_QUEUE_CALLBACK;
+        }
+    }
+
+    private static class FutureCallbackWrapper implements TbQueueCallback {
+        private final FutureCallback<Void> callback;
+
+        public FutureCallbackWrapper(FutureCallback<Void> callback) {
+            this.callback = callback;
+        }
+
+        @Override
+        public void onSuccess(TbQueueMsgMetadata metadata) {
+            callback.onSuccess(null);
+        }
+
+        @Override
+        public void onFailure(Throwable t) {
+            callback.onFailure(t);
+        }
+    }
+
 }
--- a/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/CalculatedFieldCtx.java
+++ b/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/CalculatedFieldCtx.java
@ -20,15 +20,18 @@ import org.thingsboard.script.api.tbel.TbelInvokeService;
 import org.thingsboard.server.common.data.cf.CalculatedField;
 import org.thingsboard.server.common.data.cf.CalculatedFieldType;
 import org.thingsboard.server.common.data.cf.configuration.Argument;
+import org.thingsboard.server.common.data.cf.configuration.ArgumentType;
 import org.thingsboard.server.common.data.cf.configuration.CalculatedFieldConfiguration;
 import org.thingsboard.server.common.data.cf.configuration.Output;
 import org.thingsboard.server.common.data.cf.configuration.ReferencedEntityKey;
 import org.thingsboard.server.common.data.id.CalculatedFieldId;
 import org.thingsboard.server.common.data.id.EntityId;
 import org.thingsboard.server.common.data.id.TenantId;
+import org.thingsboard.server.common.data.kv.TsKvEntry;
 import org.thingsboard.server.common.data.util.TbPair;

 import java.util.ArrayList;
+import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.stream.Collectors;
@ -41,6 +44,9 @@ public class CalculatedFieldCtx {
    private EntityId entityId;
    private CalculatedFieldType cfType;
    private final Map<String, Argument> arguments;
+    private final Map<ReferencedEntityKey, String> mainEntityArguments;
+    private final Map<EntityId, Map<ReferencedEntityKey, String>> linkedEntityArguments;
+
    private final Map<TbPair<EntityId, ReferencedEntityKey>, String> referencedEntityKeys;
    private final List<String> argNames;
    private Output output;
@ -55,6 +61,17 @@ public class CalculatedFieldCtx {
        this.cfType = calculatedField.getType();
        CalculatedFieldConfiguration configuration = calculatedField.getConfiguration();
        this.arguments = configuration.getArguments();
+        this.mainEntityArguments = new HashMap<>();
+        this.linkedEntityArguments = new HashMap<>();
+        for (Map.Entry<String, Argument> entry : arguments.entrySet()) {
+            var refId = entry.getValue().getRefEntityId();
+            var refKey = entry.getValue().getRefEntityKey();
+            if (refId == null) {
+                mainEntityArguments.put(refKey, entry.getKey());
+            } else {
+                linkedEntityArguments.computeIfAbsent(refId, key -> new HashMap<>()).put(refKey, entry.getKey());
+            }
+        }
        this.referencedEntityKeys = arguments.entrySet().stream()
                .collect(Collectors.toMap(
                        entry -> new TbPair<>(entry.getValue().getRefEntityId() == null ? entityId : entry.getValue().getRefEntityId(), entry.getValue().getRefEntityKey()),
@ -82,4 +99,30 @@ public class CalculatedFieldCtx {
        );
    }

+    public boolean matches(List<TsKvEntry> values) {
+        return matches(mainEntityArguments, values);
+    }
+
+    public boolean linkMatches(EntityId entityId, List<TsKvEntry> values) {
+        var map = linkedEntityArguments.get(entityId);
+        if (map == null) {
+            return false;
+        } else {
+            return matches(map, values);
+        }
+    }
+
+    private static boolean matches(Map<ReferencedEntityKey, String> argMap, List<TsKvEntry> values) {
+        for (TsKvEntry tsKv : values) {
+            ReferencedEntityKey latestKey = new ReferencedEntityKey(tsKv.getKey(), ArgumentType.TS_LATEST, null);
+            if (argMap.containsKey(latestKey)) {
+                return true;
+            }
+            ReferencedEntityKey rollingKey = new ReferencedEntityKey(tsKv.getKey(), ArgumentType.TS_ROLLING, null);
+            if (argMap.containsKey(rollingKey)) {
+                return true;
+            }
+        }
+        return false;
+    }
 }
--- a/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbCalculatedFieldConsumerService.java
+++ b/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbCalculatedFieldConsumerService.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
- * <p>
- * http://www.apache.org/licenses/LICENSE-2.0
- * <p>
+ *
+ *     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.
--- a/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbClusterService.java
+++ b/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbClusterService.java
@ -69,7 +69,8 @@ import org.thingsboard.server.common.msg.rule.engine.DeviceEdgeUpdateMsg;
 import org.thingsboard.server.common.msg.rule.engine.DeviceNameOrTypeUpdateMsg;
 import org.thingsboard.server.common.util.ProtoUtils;
 import org.thingsboard.server.dao.edge.EdgeService;
-import org.thingsboard.server.gen.transport.TransportProtos;
+import org.thingsboard.server.gen.transport.TransportProtos.ToCalculatedFieldMsg;
+import org.thingsboard.server.gen.transport.TransportProtos.ToCalculatedFieldNotificationMsg;
 import org.thingsboard.server.gen.transport.TransportProtos.ComponentLifecycleMsgProto;
 import org.thingsboard.server.gen.transport.TransportProtos.DeviceStateServiceMsgProto;
 import org.thingsboard.server.gen.transport.TransportProtos.EdgeNotificationMsgProto;
@ -108,7 +109,9 @@ import java.util.UUID;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.stream.Collectors;

+import static org.thingsboard.server.common.data.DataConstants.CF_QUEUE_NAME;
 import static org.thingsboard.server.common.util.ProtoUtils.toProto;
+import static org.thingsboard.server.queue.discovery.HashPartitionService.CALCULATED_FIELD_QUEUE_KEY;

@Service
@Slf4j
@ -336,6 +339,13 @@ public class DefaultTbClusterService implements TbClusterService {
        toTransportNfs.incrementAndGet();
    }

+    @Override
+    public void pushMsgToCalculatedFields(TenantId tenantId, EntityId entityId, ToCalculatedFieldMsg msg, TbQueueCallback callback) {
+        TopicPartitionInfo tpi = partitionService.resolve(CALCULATED_FIELD_QUEUE_KEY, entityId);
+        producerProvider.getCalculatedFieldsMsgProducer().send(tpi, new TbProtoQueueMsg<>(UUID.randomUUID(), msg), callback);
+        toCoreMsgs.incrementAndGet();
+    }
+
    @Override
    public void broadcastEntityStateChangeEvent(TenantId tenantId, EntityId entityId, ComponentLifecycleEvent state) {
        log.trace("[{}] Processing {} state change event: {}", tenantId, entityId.getEntityType(), state);
--- a/application/src/main/java/org/thingsboard/server/service/queue/TbCalculatedFieldConsumerService.java
+++ b/application/src/main/java/org/thingsboard/server/service/queue/TbCalculatedFieldConsumerService.java
@ -1,3 +1,18 @@
+/**
+ * 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.queue;

 import org.springframework.context.ApplicationListener;
--- a/application/src/main/java/org/thingsboard/server/service/telemetry/AbstractSubscriptionService.java
+++ b/application/src/main/java/org/thingsboard/server/service/telemetry/AbstractSubscriptionService.java
@ -116,4 +116,12 @@ public abstract class AbstractSubscriptionService extends TbApplicationEventList
        }, executor);
    }

+    protected static Consumer<Throwable> safeCallback(FutureCallback<Void> callback) {
+        if (callback != null) {
+            return callback::onFailure;
+        } else {
+            return throwable -> {};
+        }
+    }
+
 }
--- a/application/src/main/java/org/thingsboard/server/service/telemetry/DefaultTelemetrySubscriptionService.java
+++ b/application/src/main/java/org/thingsboard/server/service/telemetry/DefaultTelemetrySubscriptionService.java
@ -15,7 +15,6 @@
 */
 package org.thingsboard.server.service.telemetry;

-import com.google.common.util.concurrent.AsyncFunction;
 import com.google.common.util.concurrent.FutureCallback;
 import com.google.common.util.concurrent.Futures;
 import com.google.common.util.concurrent.ListenableFuture;
@ -53,7 +52,6 @@ import org.thingsboard.server.dao.util.KvUtils;
 import org.thingsboard.server.service.apiusage.TbApiUsageStateService;
 import org.thingsboard.server.service.cf.CalculatedFieldExecutionService;
 import org.thingsboard.server.service.cf.telemetry.CalculatedFieldAttributeUpdateRequest;
-import org.thingsboard.server.service.cf.telemetry.CalculatedFieldTimeSeriesUpdateRequest;
 import org.thingsboard.server.service.entitiy.entityview.TbEntityViewService;
 import org.thingsboard.server.service.subscription.TbSubscriptionUtils;

@ -149,14 +147,13 @@ public class DefaultTelemetrySubscriptionService extends AbstractSubscriptionSer
        } else {
            resultFuture = tsService.saveWithoutLatest(tenantId, entityId, request.getEntries(), request.getTtl());
        }
-        addMainCallback(resultFuture, request.getCallback());
+        DonAsynchron.withCallback(resultFuture, result -> {
+            calculatedFieldExecutionService.pushRequestToQueue(request, result);
+        }, safeCallback(request.getCallback()), tsCallBackExecutor);
        addWsCallback(resultFuture, success -> onTimeSeriesUpdate(tenantId, entityId, request.getEntries()));
        if (request.isSaveLatest() && !request.isOnlyLatest()) {
            addEntityViewCallback(tenantId, entityId, request.getEntries());
        }
-        // Use something very similar to addMainCallback. don't forget about tsCallBackExecutor.
-        //CalculatedFieldTimeSeriesUpdateRequest - add constructor that accepts the TimeseriesSaveRequest
-        addCallback(resultFuture, success -> calculatedFieldExecutionService.onTelemetryUpdate(new CalculatedFieldTimeSeriesUpdateRequest(tenantId, entityId, request.getEntries(), request.getPreviousCalculatedFieldIds())), tsCallBackExecutor);
        return resultFuture;
    }

@ -171,6 +168,7 @@ public class DefaultTelemetrySubscriptionService extends AbstractSubscriptionSer
        log.trace("Executing saveInternal [{}]", request);
        ListenableFuture<List<Long>> saveFuture = attrService.save(request.getTenantId(), request.getEntityId(), request.getScope(), request.getEntries());
        addMainCallback(saveFuture, request.getCallback());
+        //TODO: IM to push to CF queue
        addWsCallback(saveFuture, success -> onAttributesUpdate(request.getTenantId(), request.getEntityId(), request.getScope().name(), request.getEntries(), request.isNotifyDevice()));
        addCallback(saveFuture, success -> calculatedFieldExecutionService.onTelemetryUpdate(new CalculatedFieldAttributeUpdateRequest(request)), tsCallBackExecutor);
    }
@ -268,27 +266,21 @@ public class DefaultTelemetrySubscriptionService extends AbstractSubscriptionSer
    }

    private void onAttributesUpdate(TenantId tenantId, EntityId entityId, String scope, List<AttributeKvEntry> attributes, boolean notifyDevice) {
-        forwardToSubscriptionManagerService(tenantId, entityId, subscriptionManagerService -> {
-            subscriptionManagerService.onAttributesUpdate(tenantId, entityId, scope, attributes, notifyDevice, TbCallback.EMPTY);
-        }, () -> {
-            return TbSubscriptionUtils.toAttributesUpdateProto(tenantId, entityId, scope, attributes);
-        });
+        forwardToSubscriptionManagerService(tenantId, entityId,
+                subscriptionManagerService -> subscriptionManagerService.onAttributesUpdate(tenantId, entityId, scope, attributes, notifyDevice, TbCallback.EMPTY),
+                () -> TbSubscriptionUtils.toAttributesUpdateProto(tenantId, entityId, scope, attributes));
    }

    private void onAttributesDelete(TenantId tenantId, EntityId entityId, String scope, List<String> keys, boolean notifyDevice) {
-        forwardToSubscriptionManagerService(tenantId, entityId, subscriptionManagerService -> {
-            subscriptionManagerService.onAttributesDelete(tenantId, entityId, scope, keys, notifyDevice, TbCallback.EMPTY);
-        }, () -> {
-            return TbSubscriptionUtils.toAttributesDeleteProto(tenantId, entityId, scope, keys, notifyDevice);
-        });
+        forwardToSubscriptionManagerService(tenantId, entityId,
+                subscriptionManagerService -> subscriptionManagerService.onAttributesDelete(tenantId, entityId, scope, keys, notifyDevice, TbCallback.EMPTY),
+                () -> TbSubscriptionUtils.toAttributesDeleteProto(tenantId, entityId, scope, keys, notifyDevice));
    }

    private void onTimeSeriesUpdate(TenantId tenantId, EntityId entityId, List<TsKvEntry> ts) {
-        forwardToSubscriptionManagerService(tenantId, entityId, subscriptionManagerService -> {
-            subscriptionManagerService.onTimeSeriesUpdate(tenantId, entityId, ts, TbCallback.EMPTY);
-        }, () -> {
-            return TbSubscriptionUtils.toTimeseriesUpdateProto(tenantId, entityId, ts);
-        });
+        forwardToSubscriptionManagerService(tenantId, entityId,
+                subscriptionManagerService -> subscriptionManagerService.onTimeSeriesUpdate(tenantId, entityId, ts, TbCallback.EMPTY),
+                () -> TbSubscriptionUtils.toTimeseriesUpdateProto(tenantId, entityId, ts));
    }

    private void onTimeSeriesDelete(TenantId tenantId, EntityId entityId, List<String> keys, List<TsKvLatestRemovingResult> ts) {
@ -308,9 +300,7 @@ public class DefaultTelemetrySubscriptionService extends AbstractSubscriptionSer

            subscriptionManagerService.onTimeSeriesUpdate(tenantId, entityId, updated, TbCallback.EMPTY);
            subscriptionManagerService.onTimeSeriesDelete(tenantId, entityId, deleted, TbCallback.EMPTY);
-        }, () -> {
-            return TbSubscriptionUtils.toTimeseriesDeleteProto(tenantId, entityId, keys);
-        });
+        }, () -> TbSubscriptionUtils.toTimeseriesDeleteProto(tenantId, entityId, keys));
    }

    private <S> void addMainCallback(ListenableFuture<S> saveFuture, final FutureCallback<Void> callback) {
--- a/common/cluster-api/src/main/java/org/thingsboard/server/cluster/TbClusterService.java
+++ b/common/cluster-api/src/main/java/org/thingsboard/server/cluster/TbClusterService.java
@ -38,6 +38,7 @@ import org.thingsboard.server.common.msg.edge.ToEdgeSyncRequest;
 import org.thingsboard.server.common.msg.plugin.ComponentLifecycleMsg;
 import org.thingsboard.server.common.msg.queue.TopicPartitionInfo;
 import org.thingsboard.server.common.msg.rpc.FromDeviceRpcResponse;
+import org.thingsboard.server.gen.transport.TransportProtos;
 import org.thingsboard.server.gen.transport.TransportProtos.RestApiCallResponseMsgProto;
 import org.thingsboard.server.gen.transport.TransportProtos.ToCoreMsg;
 import org.thingsboard.server.gen.transport.TransportProtos.ToCoreNotificationMsg;
@ -76,6 +77,8 @@ public interface TbClusterService extends TbQueueClusterService {

    void pushNotificationToTransport(String targetServiceId, ToTransportMsg response, TbQueueCallback callback);

+    void pushMsgToCalculatedFields(TenantId tenantId, EntityId entityId, TransportProtos.ToCalculatedFieldMsg msg, TbQueueCallback callback);
+
    void broadcastEntityStateChangeEvent(TenantId tenantId, EntityId entityId, ComponentLifecycleEvent state);

    void onDeviceProfileChange(DeviceProfile deviceProfile, DeviceProfile oldDeviceProfile, TbQueueCallback callback);
--- a/common/queue/src/main/java/org/thingsboard/server/queue/discovery/HashPartitionService.java
+++ b/common/queue/src/main/java/org/thingsboard/server/queue/discovery/HashPartitionService.java
@ -323,7 +323,8 @@ public class HashPartitionService implements PartitionService {
        }
    }

-    private TopicPartitionInfo resolve(QueueKey queueKey, EntityId entityId) {
+    @Override
+    public TopicPartitionInfo resolve(QueueKey queueKey, EntityId entityId) {
        Integer partitionSize = partitionSizesMap.get(queueKey);
        if (partitionSize == null) {
            throw new IllegalStateException("Partitions info for queue " + queueKey + " is missing");
--- a/common/queue/src/main/java/org/thingsboard/server/queue/discovery/PartitionService.java
+++ b/common/queue/src/main/java/org/thingsboard/server/queue/discovery/PartitionService.java
@ -37,6 +37,8 @@ public interface PartitionService {

    TopicPartitionInfo resolve(ServiceType serviceType, TenantId tenantId, EntityId entityId);

+    TopicPartitionInfo resolve(QueueKey queueKey, EntityId entityId);
+
    List<TopicPartitionInfo> resolveAll(ServiceType serviceType, String queueName, TenantId tenantId, EntityId entityId);

    boolean isMyPartition(ServiceType serviceType, TenantId tenantId, EntityId entityId);
--- a/common/queue/src/main/java/org/thingsboard/server/queue/provider/TbCoreQueueFactory.java
+++ b/common/queue/src/main/java/org/thingsboard/server/queue/provider/TbCoreQueueFactory.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
- * <p>
- * http://www.apache.org/licenses/LICENSE-2.0
- * <p>
+ *
+ *     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.
@ -18,6 +18,7 @@ package org.thingsboard.server.queue.provider;
 import org.thingsboard.server.common.data.id.EdgeId;
 import org.thingsboard.server.common.data.id.TenantId;
 import org.thingsboard.server.gen.js.JsInvokeProtos;
+import org.thingsboard.server.gen.transport.TransportProtos;
 import org.thingsboard.server.gen.transport.TransportProtos.ToCalculatedFieldNotificationMsg;
 import org.thingsboard.server.gen.transport.TransportProtos.ToCoreMsg;
 import org.thingsboard.server.gen.transport.TransportProtos.ToCoreNotificationMsg;
@ -160,6 +161,8 @@ public interface TbCoreQueueFactory extends TbUsageStatsClientQueueFactory, Hous
        return null;
    }

+    TbQueueProducer<TbProtoQueueMsg<TransportProtos.ToCalculatedFieldMsg>> createToCalculatedFieldMsgProducer();
+
    TbQueueProducer<TbProtoQueueMsg<ToCalculatedFieldNotificationMsg>> createToCalculatedFieldNotificationMsgProducer();

 }
--- a/common/queue/src/main/java/org/thingsboard/server/queue/provider/TbCoreQueueProducerProvider.java
+++ b/common/queue/src/main/java/org/thingsboard/server/queue/provider/TbCoreQueueProducerProvider.java
@ -17,6 +17,8 @@ package org.thingsboard.server.queue.provider;

 import jakarta.annotation.PostConstruct;
 import org.springframework.stereotype.Service;
+import org.thingsboard.server.gen.transport.TransportProtos;
+import org.thingsboard.server.gen.transport.TransportProtos.ToCalculatedFieldNotificationMsg;
 import org.thingsboard.server.gen.transport.TransportProtos.ToCoreMsg;
 import org.thingsboard.server.gen.transport.TransportProtos.ToCoreNotificationMsg;
 import org.thingsboard.server.gen.transport.TransportProtos.ToEdgeEventNotificationMsg;
@ -48,6 +50,8 @@ public class TbCoreQueueProducerProvider implements TbQueueProducerProvider {
    private TbQueueProducer<TbProtoQueueMsg<ToUsageStatsServiceMsg>> toUsageStats;
    private TbQueueProducer<TbProtoQueueMsg<ToVersionControlServiceMsg>> toVersionControl;
    private TbQueueProducer<TbProtoQueueMsg<ToHousekeeperServiceMsg>> toHousekeeper;
+    private TbQueueProducer<TbProtoQueueMsg<TransportProtos.ToCalculatedFieldMsg>> toCalculatedFields;
+    private TbQueueProducer<TbProtoQueueMsg<ToCalculatedFieldNotificationMsg>> toCalculatedFieldNotifications;

    public TbCoreQueueProducerProvider(TbCoreQueueFactory tbQueueProvider) {
        this.tbQueueProvider = tbQueueProvider;
@ -66,6 +70,8 @@ public class TbCoreQueueProducerProvider implements TbQueueProducerProvider {
        this.toEdge = tbQueueProvider.createEdgeMsgProducer();
        this.toEdgeNotifications = tbQueueProvider.createEdgeNotificationsMsgProducer();
        this.toEdgeEvents = tbQueueProvider.createEdgeEventMsgProducer();
+        this.toCalculatedFields = tbQueueProvider.createToCalculatedFieldMsgProducer();
+        this.toCalculatedFieldNotifications = tbQueueProvider.createToCalculatedFieldNotificationMsgProducer();
    }

    @Override
@ -124,4 +130,14 @@ public class TbCoreQueueProducerProvider implements TbQueueProducerProvider {
        return toEdgeEvents;
    }

+    @Override
+    public TbQueueProducer<TbProtoQueueMsg<TransportProtos.ToCalculatedFieldMsg>> getCalculatedFieldsMsgProducer() {
+        return toCalculatedFields;
+    }
+
+    @Override
+    public TbQueueProducer<TbProtoQueueMsg<ToCalculatedFieldNotificationMsg>> getCalculatedFieldsNotificationsMsgProducer() {
+        return toCalculatedFieldNotifications;
+    }
+
 }
--- a/common/queue/src/main/java/org/thingsboard/server/queue/provider/TbQueueProducerProvider.java
+++ b/common/queue/src/main/java/org/thingsboard/server/queue/provider/TbQueueProducerProvider.java
@ -15,6 +15,8 @@
 */
 package org.thingsboard.server.queue.provider;

+import org.thingsboard.server.gen.transport.TransportProtos.ToCalculatedFieldNotificationMsg;
+import org.thingsboard.server.gen.transport.TransportProtos.ToCalculatedFieldMsg;
 import org.thingsboard.server.gen.transport.TransportProtos.ToCoreMsg;
 import org.thingsboard.server.gen.transport.TransportProtos.ToCoreNotificationMsg;
 import org.thingsboard.server.gen.transport.TransportProtos.ToEdgeEventNotificationMsg;
@ -76,7 +78,7 @@ public interface TbQueueProducerProvider {
     */
    TbQueueProducer<TbProtoQueueMsg<ToUsageStatsServiceMsg>> getTbUsageStatsMsgProducer();

-     /**
+    /**
     * Used to push messages to other instances of TB Core Service
     *
     * @return
@ -91,4 +93,8 @@ public interface TbQueueProducerProvider {

    TbQueueProducer<TbProtoQueueMsg<ToEdgeEventNotificationMsg>> getTbEdgeEventsMsgProducer();

+    TbQueueProducer<TbProtoQueueMsg<ToCalculatedFieldMsg>> getCalculatedFieldsMsgProducer();
+
+    TbQueueProducer<TbProtoQueueMsg<ToCalculatedFieldNotificationMsg>> getCalculatedFieldsNotificationsMsgProducer();
+
 }
--- a/common/queue/src/main/java/org/thingsboard/server/queue/provider/TbRuleEngineProducerProvider.java
+++ b/common/queue/src/main/java/org/thingsboard/server/queue/provider/TbRuleEngineProducerProvider.java
@ -18,6 +18,8 @@ package org.thingsboard.server.queue.provider;
 import jakarta.annotation.PostConstruct;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression;
 import org.springframework.stereotype.Service;
+import org.thingsboard.server.gen.transport.TransportProtos.ToCalculatedFieldMsg;
+import org.thingsboard.server.gen.transport.TransportProtos.ToCalculatedFieldNotificationMsg;
 import org.thingsboard.server.gen.transport.TransportProtos.ToCoreMsg;
 import org.thingsboard.server.gen.transport.TransportProtos.ToCoreNotificationMsg;
 import org.thingsboard.server.gen.transport.TransportProtos.ToEdgeEventNotificationMsg;
@ -47,6 +49,7 @@ public class TbRuleEngineProducerProvider implements TbQueueProducerProvider {
    private TbQueueProducer<TbProtoQueueMsg<ToEdgeMsg>> toEdge;
    private TbQueueProducer<TbProtoQueueMsg<ToEdgeNotificationMsg>> toEdgeNotifications;
    private TbQueueProducer<TbProtoQueueMsg<ToEdgeEventNotificationMsg>> toEdgeEvents;
+    private TbQueueProducer<TbProtoQueueMsg<ToCalculatedFieldMsg>> toCalculatedFields;

    public TbRuleEngineProducerProvider(TbRuleEngineQueueFactory tbQueueProvider) {
        this.tbQueueProvider = tbQueueProvider;
@ -64,6 +67,7 @@ public class TbRuleEngineProducerProvider implements TbQueueProducerProvider {
        this.toEdge = tbQueueProvider.createEdgeMsgProducer();
        this.toEdgeNotifications = tbQueueProvider.createEdgeNotificationsMsgProducer();
        this.toEdgeEvents = tbQueueProvider.createEdgeEventMsgProducer();
+        this.toCalculatedFields = tbQueueProvider.createToCalculatedFieldMsgProducer();
    }

    @Override
@ -121,4 +125,14 @@ public class TbRuleEngineProducerProvider implements TbQueueProducerProvider {
        return toHousekeeper;
    }

+    @Override
+    public TbQueueProducer<TbProtoQueueMsg<ToCalculatedFieldMsg>> getCalculatedFieldsMsgProducer() {
+        return toCalculatedFields;
+    }
+
+    @Override
+    public TbQueueProducer<TbProtoQueueMsg<ToCalculatedFieldNotificationMsg>> getCalculatedFieldsNotificationsMsgProducer() {
+        throw new RuntimeException("Not Implemented! Should not be used by Rule Engine Service!");
+    }
+
 }
--- a/common/queue/src/main/java/org/thingsboard/server/queue/provider/TbRuleEngineQueueFactory.java
+++ b/common/queue/src/main/java/org/thingsboard/server/queue/provider/TbRuleEngineQueueFactory.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
- * <p>
- * http://www.apache.org/licenses/LICENSE-2.0
- * <p>
+ *
+ *     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.
--- a/common/queue/src/main/java/org/thingsboard/server/queue/provider/TbTransportQueueProducerProvider.java
+++ b/common/queue/src/main/java/org/thingsboard/server/queue/provider/TbTransportQueueProducerProvider.java
@ -18,6 +18,7 @@ package org.thingsboard.server.queue.provider;
 import jakarta.annotation.PostConstruct;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression;
 import org.springframework.stereotype.Service;
+import org.thingsboard.server.gen.transport.TransportProtos;
 import org.thingsboard.server.gen.transport.TransportProtos.ToCoreMsg;
 import org.thingsboard.server.gen.transport.TransportProtos.ToCoreNotificationMsg;
 import org.thingsboard.server.gen.transport.TransportProtos.ToEdgeEventNotificationMsg;
@ -111,4 +112,13 @@ public class TbTransportQueueProducerProvider implements TbQueueProducerProvider
        return toHousekeeper;
    }

+    @Override
+    public TbQueueProducer<TbProtoQueueMsg<TransportProtos.ToCalculatedFieldMsg>> getCalculatedFieldsMsgProducer() {
+        throw new RuntimeException("Not Implemented! Should not be used by Transport!");
+    }
+
+    @Override
+    public TbQueueProducer<TbProtoQueueMsg<TransportProtos.ToCalculatedFieldNotificationMsg>> getCalculatedFieldsNotificationsMsgProducer() {
+        throw new RuntimeException("Not Implemented! Should not be used by Transport!");
+    }
 }
--- a/common/queue/src/main/java/org/thingsboard/server/queue/provider/TbVersionControlProducerProvider.java
+++ b/common/queue/src/main/java/org/thingsboard/server/queue/provider/TbVersionControlProducerProvider.java
@ -18,6 +18,7 @@ package org.thingsboard.server.queue.provider;
 import jakarta.annotation.PostConstruct;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression;
 import org.springframework.stereotype.Service;
+import org.thingsboard.server.gen.transport.TransportProtos;
 import org.thingsboard.server.gen.transport.TransportProtos.ToCoreMsg;
 import org.thingsboard.server.gen.transport.TransportProtos.ToCoreNotificationMsg;
 import org.thingsboard.server.gen.transport.TransportProtos.ToEdgeEventNotificationMsg;
@ -107,4 +108,14 @@ public class TbVersionControlProducerProvider implements TbQueueProducerProvider
        return toHousekeeper;
    }

+    @Override
+    public TbQueueProducer<TbProtoQueueMsg<TransportProtos.ToCalculatedFieldMsg>> getCalculatedFieldsMsgProducer() {
+        throw new RuntimeException("Not Implemented! Should not be used by Version Control Service!");
+    }
+
+    @Override
+    public TbQueueProducer<TbProtoQueueMsg<TransportProtos.ToCalculatedFieldNotificationMsg>> getCalculatedFieldsNotificationsMsgProducer() {
+        throw new RuntimeException("Not Implemented! Should not be used by Version Control Service!");
+    }
+
 }