Merge branch 'rc' into features/add_tooltip_option_to_show_stack_mode_total_value_on_timeseries_chart_widgets

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

@@ -244,7 +244,7 @@ public class CalculatedFieldController extends BaseController {
             );
 
             Object[] args = new Object[ctxAndArgNames.size()];
-            args[0] = new TbelCfCtx(arguments, getLastUpdateTimestamp(arguments));
+            args[0] = new TbelCfCtx(arguments, getLatestTimestamp(arguments));
             for (int i = 1; i < ctxAndArgNames.size(); i++) {
                 var arg = arguments.get(ctxAndArgNames.get(i));
                 if (arg instanceof TbelCfSingleValueArg svArg) {
@@ -267,7 +267,7 @@ public class CalculatedFieldController extends BaseController {
         return result;
     }
 
-    private long getLastUpdateTimestamp(Map<String, TbelCfArg> arguments) {
+    private long getLatestTimestamp(Map<String, TbelCfArg> arguments) {
         long lastUpdateTimestamp = -1;
         for (TbelCfArg entry : arguments.values()) {
             if (entry instanceof TbelCfSingleValueArg singleValueArg) {

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

@@ -21,6 +21,7 @@ import org.thingsboard.rule.engine.api.AttributesSaveRequest;
 import org.thingsboard.rule.engine.api.RuleEngineCalculatedFieldQueueService;
 import org.thingsboard.rule.engine.api.TimeseriesDeleteRequest;
 import org.thingsboard.rule.engine.api.TimeseriesSaveRequest;
+import org.thingsboard.server.common.data.kv.AttributesSaveResult;
 import org.thingsboard.server.common.data.kv.TimeseriesSaveResult;
 
 import java.util.List;
@@ -35,7 +36,7 @@ public interface CalculatedFieldQueueService extends RuleEngineCalculatedFieldQu
      */
     void pushRequestToQueue(TimeseriesSaveRequest request, TimeseriesSaveResult result, FutureCallback<Void> callback);
 
-    void pushRequestToQueue(AttributesSaveRequest request, List<Long> result, FutureCallback<Void> callback);
+    void pushRequestToQueue(AttributesSaveRequest request, AttributesSaveResult result, FutureCallback<Void> callback);
 
     void pushRequestToQueue(AttributesDeleteRequest request, List<String> result, FutureCallback<Void> callback);
 

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

@@ -32,6 +32,7 @@ import org.thingsboard.server.common.data.id.DeviceId;
 import org.thingsboard.server.common.data.id.EntityId;
 import org.thingsboard.server.common.data.id.TenantId;
 import org.thingsboard.server.common.data.kv.AttributeKvEntry;
+import org.thingsboard.server.common.data.kv.AttributesSaveResult;
 import org.thingsboard.server.common.data.kv.TimeseriesSaveResult;
 import org.thingsboard.server.common.data.kv.TsKvEntry;
 import org.thingsboard.server.common.data.msg.TbMsgType;
@@ -96,7 +97,7 @@ public class DefaultCalculatedFieldQueueService implements CalculatedFieldQueueS
     }
 
     @Override
-    public void pushRequestToQueue(AttributesSaveRequest request, List<Long> result, FutureCallback<Void> callback) {
+    public void pushRequestToQueue(AttributesSaveRequest request, AttributesSaveResult result, FutureCallback<Void> callback) {
         var tenantId = request.getTenantId();
         var entityId = request.getEntityId();
         checkEntityAndPushToQueue(tenantId, entityId, cf -> cf.matches(request.getEntries(), request.getScope()), cf -> cf.linkMatches(entityId, request.getEntries(), request.getScope()),
@@ -176,7 +177,7 @@ public class DefaultCalculatedFieldQueueService implements CalculatedFieldQueueS
 
         for (int i = 0; i < entries.size(); i++) {
             TsKvProto.Builder tsProtoBuilder = toTsKvProto(entries.get(i)).toBuilder();
-            if (result != null) {
+            if (versions != null && !versions.isEmpty() && versions.get(i) != null) {
                 tsProtoBuilder.setVersion(versions.get(i));
             }
             telemetryMsg.addTsData(tsProtoBuilder.build());
@@ -186,17 +187,18 @@ public class DefaultCalculatedFieldQueueService implements CalculatedFieldQueueS
         return msg.build();
     }
 
-    private ToCalculatedFieldMsg toCalculatedFieldTelemetryMsgProto(AttributesSaveRequest request, List<Long> versions) {
+    private ToCalculatedFieldMsg toCalculatedFieldTelemetryMsgProto(AttributesSaveRequest request, AttributesSaveResult result) {
         ToCalculatedFieldMsg.Builder msg = ToCalculatedFieldMsg.newBuilder();
 
         CalculatedFieldTelemetryMsgProto.Builder telemetryMsg = buildTelemetryMsgProto(request.getTenantId(), request.getEntityId(), request.getPreviousCalculatedFieldIds(), request.getTbMsgId(), request.getTbMsgType());
         telemetryMsg.setScope(AttributeScopeProto.valueOf(request.getScope().name()));
+
         List<AttributeKvEntry> entries = request.getEntries();
+        List<Long> versions = result.versions();
+
         for (int i = 0; i < entries.size(); i++) {
             AttributeValueProto.Builder attrProtoBuilder = ProtoUtils.toProto(entries.get(i)).toBuilder();
-            if (versions != null) {
-                attrProtoBuilder.setVersion(versions.get(i));
-            }
+            attrProtoBuilder.setVersion(versions.get(i));
             telemetryMsg.addAttrData(attrProtoBuilder.build());
         }
         msg.setTelemetryMsg(telemetryMsg.build());

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

@@ -35,7 +35,7 @@ public abstract class BaseCalculatedFieldState implements CalculatedFieldState {
     protected Map<String, ArgumentEntry> arguments;
     protected boolean sizeExceedsLimit;
 
-    protected long lastUpdateTimestamp = -1;
+    protected long latestTimestamp = -1;
 
     public BaseCalculatedFieldState(List<String> requiredArguments) {
         this.requiredArguments = requiredArguments;
@@ -110,12 +110,14 @@ public abstract class BaseCalculatedFieldState implements CalculatedFieldState {
     protected abstract void validateNewEntry(ArgumentEntry newEntry);
 
     private void updateLastUpdateTimestamp(ArgumentEntry entry) {
+        long newTs = this.latestTimestamp;
         if (entry instanceof SingleValueArgumentEntry singleValueArgumentEntry) {
-            this.lastUpdateTimestamp = singleValueArgumentEntry.getTs();
+            newTs = singleValueArgumentEntry.getTs();
         } else if (entry instanceof TsRollingArgumentEntry tsRollingArgumentEntry) {
             Map.Entry<Long, Double> lastEntry = tsRollingArgumentEntry.getTsRecords().lastEntry();
-            this.lastUpdateTimestamp = (lastEntry != null) ? lastEntry.getKey() : System.currentTimeMillis();
+            newTs = (lastEntry != null) ? lastEntry.getKey() : System.currentTimeMillis();
         }
+        this.latestTimestamp = Math.max(this.latestTimestamp, newTs);
     }
 
 }

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

@@ -62,7 +62,7 @@ public class CalculatedFieldCtx {
     private final List<String> argNames;
     private Output output;
     private String expression;
-    private boolean preserveMsgTs;
+    private boolean useLatestTs;
     private TbelInvokeService tbelInvokeService;
     private CalculatedFieldScriptEngine calculatedFieldScriptEngine;
     private ThreadLocal<Expression> customExpression;
@@ -96,7 +96,7 @@ public class CalculatedFieldCtx {
         this.argNames = new ArrayList<>(arguments.keySet());
         this.output = configuration.getOutput();
         this.expression = configuration.getExpression();
-        this.preserveMsgTs = CalculatedFieldType.SIMPLE.equals(calculatedField.getType()) && ((SimpleCalculatedFieldConfiguration) configuration).isPreserveMsgTs();
+        this.useLatestTs = CalculatedFieldType.SIMPLE.equals(calculatedField.getType()) && ((SimpleCalculatedFieldConfiguration) configuration).isUseLatestTs();
         this.tbelInvokeService = tbelInvokeService;
 
         this.maxDataPointsPerRollingArg = apiLimitService.getLimit(tenantId, DefaultTenantProfileConfiguration::getMaxDataPointsPerRollingArg);

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

@@ -42,7 +42,7 @@ public interface CalculatedFieldState {
 
     Map<String, ArgumentEntry> getArguments();
 
-    long getLastUpdateTimestamp();
+    long getLatestTimestamp();
 
     void setRequiredArguments(List<String> requiredArguments);
 

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

@@ -66,7 +66,7 @@ public class ScriptCalculatedFieldState extends BaseCalculatedFieldState {
                 args.add(arg);
             }
         }
-        args.set(0, new TbelCfCtx(arguments, getLastUpdateTimestamp()));
+        args.set(0, new TbelCfCtx(arguments, getLatestTimestamp()));
         ListenableFuture<JsonNode> resultFuture = ctx.getCalculatedFieldScriptEngine().executeJsonAsync(args.toArray());
         Output output = ctx.getOutput();
         return Futures.transform(resultFuture,

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

@@ -68,7 +68,7 @@ public class SimpleCalculatedFieldState extends BaseCalculatedFieldState {
 
         Output output = ctx.getOutput();
         Object result = formatResult(expressionResult, output.getDecimalsByDefault());
-        JsonNode outputResult = createResultJson(ctx.isPreserveMsgTs(), output.getName(), result);
+        JsonNode outputResult = createResultJson(ctx.isUseLatestTs(), output.getName(), result);
 
         return Futures.immediateFuture(new CalculatedFieldResult(output.getType(), output.getScope(), outputResult));
     }
@@ -83,14 +83,14 @@ public class SimpleCalculatedFieldState extends BaseCalculatedFieldState {
         return TbUtils.toFixed(expressionResult, decimals);
     }
 
-    private JsonNode createResultJson(boolean preserveMsgTs, String outputName, Object result) {
+    private JsonNode createResultJson(boolean useLatestTs, String outputName, Object result) {
         ObjectNode valuesNode = JacksonUtil.newObjectNode();
         valuesNode.set(outputName, JacksonUtil.valueToTree(result));
 
-        long lastTimestamp = getLastUpdateTimestamp();
-        if (preserveMsgTs && lastTimestamp != -1) {
+        long latestTs = getLatestTimestamp();
+        if (useLatestTs && latestTs != -1) {
             ObjectNode resultNode = JacksonUtil.newObjectNode();
-            resultNode.put("ts", lastTimestamp);
+            resultNode.put("ts", latestTs);
             resultNode.set("values", valuesNode);
             return resultNode;
         } else {

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

@@ -107,7 +107,7 @@ public class SingleValueArgumentEntry implements ArgumentEntry {
     @Override
     public boolean updateEntry(ArgumentEntry entry) {
         if (entry instanceof SingleValueArgumentEntry singleValueEntry) {
-            if (singleValueEntry.getTs() == this.ts) {
+            if (singleValueEntry.getTs() <= this.ts) {
                 return false;
             }
 

application/src/main/java/org/thingsboard/server/service/device/DeviceProvisionServiceImpl.java

@@ -33,6 +33,7 @@ import org.thingsboard.server.common.data.id.CustomerId;
 import org.thingsboard.server.common.data.id.TenantId;
 import org.thingsboard.server.common.data.id.UserId;
 import org.thingsboard.server.common.data.kv.AttributeKvEntry;
+import org.thingsboard.server.common.data.kv.AttributesSaveResult;
 import org.thingsboard.server.common.data.kv.BaseAttributeKvEntry;
 import org.thingsboard.server.common.data.kv.StringDataEntry;
 import org.thingsboard.server.common.data.msg.TbMsgType;
@@ -62,8 +63,6 @@ import org.thingsboard.server.queue.discovery.PartitionService;
 import org.thingsboard.server.queue.provider.TbQueueProducerProvider;
 import org.thingsboard.server.queue.util.TbCoreComponent;
 
-import java.util.Collections;
-import java.util.List;
 import java.util.Optional;
 import java.util.concurrent.ExecutionException;
 import java.util.regex.Matcher;
@@ -240,10 +239,11 @@ public class DeviceProvisionServiceImpl implements DeviceProvisionService {
         return deviceCredentialsService.updateDeviceCredentials(tenantId, deviceCredentials);
     }
 
-    private ListenableFuture<List<Long>> saveProvisionStateAttribute(Device device) {
-        return attributesService.save(device.getTenantId(), device.getId(), AttributeScope.SERVER_SCOPE,
-                Collections.singletonList(new BaseAttributeKvEntry(new StringDataEntry(DEVICE_PROVISION_STATE, PROVISIONED_STATE),
-                        System.currentTimeMillis())));
+    private ListenableFuture<AttributesSaveResult> saveProvisionStateAttribute(Device device) {
+        return attributesService.save(
+                device.getTenantId(), device.getId(), AttributeScope.SERVER_SCOPE,
+                new BaseAttributeKvEntry(new StringDataEntry(DEVICE_PROVISION_STATE, PROVISIONED_STATE), System.currentTimeMillis())
+        );
     }
 
     private DeviceCredentials getDeviceCredentials(Device device) {

application/src/main/java/org/thingsboard/server/service/edge/rpc/EdgeGrpcSession.java

@@ -35,6 +35,7 @@ import org.thingsboard.server.common.data.edge.EdgeEventType;
 import org.thingsboard.server.common.data.id.EdgeId;
 import org.thingsboard.server.common.data.id.TenantId;
 import org.thingsboard.server.common.data.kv.AttributeKvEntry;
+import org.thingsboard.server.common.data.kv.AttributesSaveResult;
 import org.thingsboard.server.common.data.kv.BaseAttributeKvEntry;
 import org.thingsboard.server.common.data.kv.LongDataEntry;
 import org.thingsboard.server.common.data.kv.StringDataEntry;
@@ -581,10 +582,10 @@ public abstract class EdgeGrpcSession implements Closeable {
                 @Override
                 public void onSuccess(@Nullable Pair<Long, Long> newStartTsAndSeqId) {
                     if (newStartTsAndSeqId != null) {
-                        ListenableFuture<List<Long>> updateFuture = updateQueueStartTsAndSeqId(newStartTsAndSeqId);
+                        ListenableFuture<AttributesSaveResult> updateFuture = updateQueueStartTsAndSeqId(newStartTsAndSeqId);
                         Futures.addCallback(updateFuture, new FutureCallback<>() {
                             @Override
-                            public void onSuccess(@Nullable List<Long> list) {
+                            public void onSuccess(@Nullable AttributesSaveResult saveResult) {
                                 log.debug("[{}][{}] queue offset was updated [{}]", tenantId, edge.getId(), newStartTsAndSeqId);
                                 boolean newEventsAvailable;
                                 if (fetcher.isSeqIdNewCycleStarted()) {
@@ -645,8 +646,7 @@ public abstract class EdgeGrpcSession implements Closeable {
                             log.trace("[{}][{}] entity message processed [{}]", tenantId, edge.getId(), downlinkMsg);
                         }
                     }
-                    case ATTRIBUTES_UPDATED, POST_ATTRIBUTES, ATTRIBUTES_DELETED, TIMESERIES_UPDATED ->
-                            downlinkMsg = ctx.getTelemetryProcessor().convertTelemetryEventToDownlink(edge, edgeEvent);
+                    case ATTRIBUTES_UPDATED, POST_ATTRIBUTES, ATTRIBUTES_DELETED, TIMESERIES_UPDATED -> downlinkMsg = ctx.getTelemetryProcessor().convertTelemetryEventToDownlink(edge, edgeEvent);
                     default -> log.warn("[{}][{}] Unsupported action type [{}]", tenantId, edge.getId(), edgeEvent.getAction());
                 }
             } catch (Exception e) {
@@ -722,13 +722,14 @@ public abstract class EdgeGrpcSession implements Closeable {
         return startSeqId;
     }
 
-    private ListenableFuture<List<Long>> updateQueueStartTsAndSeqId(Pair<Long, Long> pair) {
+    private ListenableFuture<AttributesSaveResult> updateQueueStartTsAndSeqId(Pair<Long, Long> pair) {
         newStartTs = pair.getFirst();
         newStartSeqId = pair.getSecond();
         log.trace("[{}] updateQueueStartTsAndSeqId [{}][{}][{}]", sessionId, edge.getId(), newStartTs, newStartSeqId);
-        List<AttributeKvEntry> attributes = Arrays.asList(
+        List<AttributeKvEntry> attributes = List.of(
                 new BaseAttributeKvEntry(new LongDataEntry(QUEUE_START_TS_ATTR_KEY, newStartTs), System.currentTimeMillis()),
-                new BaseAttributeKvEntry(new LongDataEntry(QUEUE_START_SEQ_ID_ATTR_KEY, newStartSeqId), System.currentTimeMillis()));
+                new BaseAttributeKvEntry(new LongDataEntry(QUEUE_START_SEQ_ID_ATTR_KEY, newStartSeqId), System.currentTimeMillis())
+        );
         return ctx.getAttributesService().save(edge.getTenantId(), edge.getId(), AttributeScope.SERVER_SCOPE, attributes);
     }
 

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

@@ -64,6 +64,7 @@ import org.thingsboard.server.common.data.id.CustomerId;
 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.kv.AttributesSaveResult;
 import org.thingsboard.server.common.data.kv.BaseAttributeKvEntry;
 import org.thingsboard.server.common.data.kv.BasicTsKvEntry;
 import org.thingsboard.server.common.data.kv.BooleanDataEntry;
@@ -581,9 +582,9 @@ public class DefaultSystemDataLoaderService implements SystemDataLoaderService {
                     Collections.singletonList(new BasicTsKvEntry(System.currentTimeMillis(), new BooleanDataEntry(key, value))), 0L);
             addTsCallback(saveFuture, new TelemetrySaveCallback<>(deviceId, key, value));
         } else {
-            ListenableFuture<List<Long>> saveFuture = attributesService.save(TenantId.SYS_TENANT_ID, deviceId, AttributeScope.SERVER_SCOPE,
-                    Collections.singletonList(new BaseAttributeKvEntry(new BooleanDataEntry(key, value)
-                            , System.currentTimeMillis())));
+            ListenableFuture<AttributesSaveResult> saveFuture = attributesService.save(
+                    TenantId.SYS_TENANT_ID, deviceId, AttributeScope.SERVER_SCOPE, new BaseAttributeKvEntry(new BooleanDataEntry(key, value), System.currentTimeMillis())
+            );
             addTsCallback(saveFuture, new TelemetrySaveCallback<>(deviceId, key, value));
         }
     }
@@ -611,7 +612,7 @@ public class DefaultSystemDataLoaderService implements SystemDataLoaderService {
     }
 
     private <S> void addTsCallback(ListenableFuture<S> saveFuture, final FutureCallback<S> callback) {
-        Futures.addCallback(saveFuture, new FutureCallback<S>() {
+        Futures.addCallback(saveFuture, new FutureCallback<>() {
             @Override
             public void onSuccess(@Nullable S result) {
                 callback.onSuccess(result);

application/src/main/java/org/thingsboard/server/service/state/DefaultDeviceStateService.java

@@ -27,11 +27,10 @@ import jakarta.annotation.Nonnull;
 import jakarta.annotation.Nullable;
 import jakarta.annotation.PostConstruct;
 import jakarta.annotation.PreDestroy;
-import lombok.Getter;
 import lombok.RequiredArgsConstructor;
-import lombok.Setter;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.commons.lang3.tuple.Pair;
+import org.checkerframework.checker.nullness.qual.NonNull;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.annotation.Value;
 import org.springframework.context.annotation.Lazy;
@@ -170,35 +169,22 @@ public class DefaultDeviceStateService extends AbstractPartitionBasedService<Dev
     @Lazy
     private TelemetrySubscriptionService tsSubService;
 
-    @Value("${state.defaultInactivityTimeoutInSec}")
-    @Getter
-    @Setter
-    private long defaultInactivityTimeoutInSec;
-
     @Value("#{${state.defaultInactivityTimeoutInSec} * 1000}")
-    @Getter
-    @Setter
     private long defaultInactivityTimeoutMs;
 
     @Value("${state.defaultStateCheckIntervalInSec}")
-    @Getter
     private int defaultStateCheckIntervalInSec;
 
     @Value("${usage.stats.devices.report_interval:60}")
-    @Getter
     private int defaultActivityStatsIntervalInSec;
 
     @Value("${state.persistToTelemetry:false}")
-    @Getter
-    @Setter
     private boolean persistToTelemetry;
 
     @Value("${state.initFetchPackSize:50000}")
-    @Getter
     private int initFetchPackSize;
 
     @Value("${state.telemetryTtl:0}")
-    @Getter
     private int telemetryTtl;
 
     private ListeningExecutorService deviceStateExecutor;
@@ -281,12 +267,11 @@ public class DefaultDeviceStateService extends AbstractPartitionBasedService<Dev
             DeviceState state = stateData.getState();
             state.setLastActivityTime(lastReportedActivity);
             if (!state.isActive()) {
-                state.setActive(true);
                 if (lastReportedActivity <= state.getLastInactivityAlarmTime()) {
                     state.setLastInactivityAlarmTime(0);
                     save(stateData.getTenantId(), deviceId, INACTIVITY_ALARM_TIME, 0);
                 }
-                onDeviceActivityStatusChange(deviceId, true, stateData);
+                onDeviceActivityStatusChange(true, stateData);
             }
         } else {
             log.debug("updateActivityState - fetched state IS NULL for device {}, lastReportedActivity {}", deviceId, lastReportedActivity);
@@ -355,7 +340,7 @@ public class DefaultDeviceStateService extends AbstractPartitionBasedService<Dev
             return;
         }
         log.trace("[{}][{}] On device inactivity: processing inactivity event with ts [{}].", tenantId.getId(), deviceId.getId(), lastInactivityTime);
-        reportInactivity(lastInactivityTime, deviceId, stateData);
+        reportInactivity(lastInactivityTime, stateData);
     }
 
     @Override
@@ -387,7 +372,7 @@ public class DefaultDeviceStateService extends AbstractPartitionBasedService<Dev
                             }
 
                             @Override
-                            public void onFailure(Throwable t) {
+                            public void onFailure(@NonNull Throwable t) {
                                 log.warn("Failed to register device to the state service", t);
                                 callback.onFailure(t);
                             }
@@ -539,7 +524,7 @@ public class DefaultDeviceStateService extends AbstractPartitionBasedService<Dev
         }
     }
 
-    void reportActivityStats() {
+    private void reportActivityStats() {
         try {
             Map<TenantId, Pair<AtomicInteger, AtomicInteger>> stats = new HashMap<>();
             for (DeviceStateData stateData : deviceStates.values()) {
@@ -574,7 +559,7 @@ public class DefaultDeviceStateService extends AbstractPartitionBasedService<Dev
                     && (state.getLastInactivityAlarmTime() == 0L || state.getLastInactivityAlarmTime() <= state.getLastActivityTime())
                     && stateData.getDeviceCreationTime() + state.getInactivityTimeout() <= ts) {
                 if (partitionService.resolve(ServiceType.TB_CORE, stateData.getTenantId(), deviceId).isMyPartition()) {
-                    reportInactivity(ts, deviceId, stateData);
+                    reportInactivity(ts, stateData);
                 } else {
                     cleanupEntity(deviceId);
                 }
@@ -585,15 +570,25 @@ public class DefaultDeviceStateService extends AbstractPartitionBasedService<Dev
         }
     }
 
-    private void reportInactivity(long ts, DeviceId deviceId, DeviceStateData stateData) {
-        DeviceState state = stateData.getState();
-        state.setActive(false);
-        state.setLastInactivityAlarmTime(ts);
-        save(stateData.getTenantId(), deviceId, INACTIVITY_ALARM_TIME, ts);
-        onDeviceActivityStatusChange(deviceId, false, stateData);
+    private void reportInactivity(long ts, DeviceStateData stateData) {
+        var tenantId = stateData.getTenantId();
+        var deviceId = stateData.getDeviceId();
+
+        Futures.addCallback(save(stateData.getTenantId(), deviceId, INACTIVITY_ALARM_TIME, ts), new FutureCallback<>() {
+            @Override
+            public void onSuccess(Void success) {
+                stateData.getState().setLastInactivityAlarmTime(ts);
+                onDeviceActivityStatusChange(false, stateData);
+            }
+
+            @Override
+            public void onFailure(@NonNull Throwable t) {
+                log.error("[{}][{}] Failed to update device last inactivity alarm time to '{}'. Device state data: {}", tenantId, deviceId, ts, stateData, t);
+            }
+        }, deviceStateCallbackExecutor);
     }
 
-    boolean isActive(long ts, DeviceState state) {
+    private static boolean isActive(long ts, DeviceState state) {
         return ts < state.getLastActivityTime() + state.getInactivityTimeout();
     }
 
@@ -616,17 +611,32 @@ public class DefaultDeviceStateService extends AbstractPartitionBasedService<Dev
         }
     }
 
-    private void onDeviceActivityStatusChange(DeviceId deviceId, boolean active, DeviceStateData stateData) {
-        save(stateData.getTenantId(), deviceId, ACTIVITY_STATE, active);
-        pushRuleEngineMessage(stateData, active ? TbMsgType.ACTIVITY_EVENT : TbMsgType.INACTIVITY_EVENT);
-        TbMsgMetaData metaData = stateData.getMetaData();
-        notificationRuleProcessor.process(DeviceActivityTrigger.builder()
-                .tenantId(stateData.getTenantId()).customerId(stateData.getCustomerId())
-                .deviceId(deviceId).active(active)
-                .deviceName(metaData.getValue("deviceName"))
-                .deviceType(metaData.getValue("deviceType"))
-                .deviceLabel(metaData.getValue("deviceLabel"))
-                .build());
+    private void onDeviceActivityStatusChange(boolean active, DeviceStateData stateData) {
+        var tenantId = stateData.getTenantId();
+        var deviceId = stateData.getDeviceId();
+
+        Futures.addCallback(save(tenantId, deviceId, ACTIVITY_STATE, active), new FutureCallback<>() {
+            @Override
+            public void onSuccess(Void success) {
+                stateData.getState().setActive(active);
+                pushRuleEngineMessage(stateData, active ? TbMsgType.ACTIVITY_EVENT : TbMsgType.INACTIVITY_EVENT);
+                TbMsgMetaData metaData = stateData.getMetaData();
+                notificationRuleProcessor.process(DeviceActivityTrigger.builder()
+                        .tenantId(tenantId)
+                        .customerId(stateData.getCustomerId())
+                        .deviceId(deviceId)
+                        .active(active)
+                        .deviceName(metaData.getValue("deviceName"))
+                        .deviceType(metaData.getValue("deviceType"))
+                        .deviceLabel(metaData.getValue("deviceLabel"))
+                        .build());
+            }
+
+            @Override
+            public void onFailure(@NonNull Throwable t) {
+                log.error("[{}][{}] Failed to change device activity status to '{}'. Device state data: {}", tenantId, deviceId, active, stateData, t);
+            }
+        }, deviceStateCallbackExecutor);
     }
 
     boolean cleanDeviceStateIfBelongsToExternalPartition(TenantId tenantId, final DeviceId deviceId) {
@@ -634,8 +644,7 @@ public class DefaultDeviceStateService extends AbstractPartitionBasedService<Dev
         boolean cleanup = !partitionedEntities.containsKey(tpi);
         if (cleanup) {
             cleanupEntity(deviceId);
-            log.debug("[{}][{}] device belongs to external partition. Probably rebalancing is in progress. Topic: {}"
-                    , tenantId, deviceId, tpi.getFullTopicName());
+            log.debug("[{}][{}] device belongs to external partition. Probably rebalancing is in progress. Topic: {}", tenantId, deviceId, tpi.getFullTopicName());
         }
         return cleanup;
     }
@@ -766,7 +775,7 @@ public class DefaultDeviceStateService extends AbstractPartitionBasedService<Dev
 
     }
 
-    DeviceStateData toDeviceStateData(EntityData ed, DeviceIdInfo deviceIdInfo) {
+    private DeviceStateData toDeviceStateData(EntityData ed, DeviceIdInfo deviceIdInfo) {
         long lastActivityTime = getEntryValue(ed, getKeyType(), LAST_ACTIVITY_TIME, 0L);
         long inactivityAlarmTime = getEntryValue(ed, getKeyType(), INACTIVITY_ALARM_TIME, 0L);
         long inactivityTimeout = getEntryValue(ed, EntityKeyType.SERVER_ATTRIBUTE, INACTIVITY_TIMEOUT, defaultInactivityTimeoutMs);
@@ -849,6 +858,9 @@ public class DefaultDeviceStateService extends AbstractPartitionBasedService<Dev
     }
 
     private void pushRuleEngineMessage(DeviceStateData stateData, TbMsgType msgType) {
+        var tenantId = stateData.getTenantId();
+        var deviceId = stateData.getDeviceId();
+
         DeviceState state = stateData.getState();
         try {
             String data;
@@ -865,7 +877,7 @@ public class DefaultDeviceStateService extends AbstractPartitionBasedService<Dev
             }
             TbMsg tbMsg = TbMsg.newMsg()
                     .type(msgType)
-                    .originator(stateData.getDeviceId())
+                    .originator(deviceId)
                     .customerId(stateData.getCustomerId())
                     .copyMetaData(md)
                     .dataType(TbMsgDataType.JSON)
@@ -873,21 +885,22 @@ public class DefaultDeviceStateService extends AbstractPartitionBasedService<Dev
                     .build();
             clusterService.pushMsgToRuleEngine(stateData.getTenantId(), stateData.getDeviceId(), tbMsg, null);
         } catch (Exception e) {
-            log.warn("[{}] Failed to push inactivity alarm: {}", stateData.getDeviceId(), state, e);
+            log.warn("[{}][{}] Failed to push '{}' message to the rule engine due to {}. Device state: {}", tenantId, deviceId, msgType, e.getMessage(), state);
         }
     }
 
-    private void save(TenantId tenantId, DeviceId deviceId, String key, long value) {
-        save(tenantId, deviceId, new LongDataEntry(key, value), getCurrentTimeMillis());
+    private ListenableFuture<Void> save(TenantId tenantId, DeviceId deviceId, String key, long value) {
+        return save(tenantId, deviceId, new LongDataEntry(key, value), getCurrentTimeMillis());
     }
 
-    private void save(TenantId tenantId, DeviceId deviceId, String key, boolean value) {
-        save(tenantId, deviceId, new BooleanDataEntry(key, value), getCurrentTimeMillis());
+    private ListenableFuture<Void> save(TenantId tenantId, DeviceId deviceId, String key, boolean value) {
+        return save(tenantId, deviceId, new BooleanDataEntry(key, value), getCurrentTimeMillis());
     }
 
-    private void save(TenantId tenantId, DeviceId deviceId, KvEntry kvEntry, long ts) {
+    private ListenableFuture<Void> save(TenantId tenantId, DeviceId deviceId, KvEntry kvEntry, long ts) {
+        ListenableFuture<?> future;
         if (persistToTelemetry) {
-            tsSubService.saveTimeseriesInternal(TimeseriesSaveRequest.builder()
+            future = tsSubService.saveTimeseriesInternal(TimeseriesSaveRequest.builder()
                     .tenantId(tenantId)
                     .entityId(deviceId)
                     .entry(new BasicTsKvEntry(ts, kvEntry))
@@ -895,7 +908,7 @@ public class DefaultDeviceStateService extends AbstractPartitionBasedService<Dev
                     .callback(new TelemetrySaveCallback<>(deviceId, kvEntry))
                     .build());
         } else {
-            tsSubService.saveAttributes(AttributesSaveRequest.builder()
+            future = tsSubService.saveAttributesInternal(AttributesSaveRequest.builder()
                     .tenantId(tenantId)
                     .entityId(deviceId)
                     .scope(AttributeScope.SERVER_SCOPE)
@@ -903,20 +916,14 @@ public class DefaultDeviceStateService extends AbstractPartitionBasedService<Dev
                     .callback(new TelemetrySaveCallback<>(deviceId, kvEntry))
                     .build());
         }
+        return Futures.transform(future, __ -> null, MoreExecutors.directExecutor());
     }
 
     long getCurrentTimeMillis() {
         return System.currentTimeMillis();
     }
 
-    private static class TelemetrySaveCallback<T> implements FutureCallback<T> {
-        private final DeviceId deviceId;
-        private final KvEntry kvEntry;
-
-        TelemetrySaveCallback(DeviceId deviceId, KvEntry kvEntry) {
-            this.deviceId = deviceId;
-            this.kvEntry = kvEntry;
-        }
+    private record TelemetrySaveCallback<T>(DeviceId deviceId, KvEntry kvEntry) implements FutureCallback<T> {
 
         @Override
         public void onSuccess(@Nullable T result) {
@@ -924,9 +931,10 @@ public class DefaultDeviceStateService extends AbstractPartitionBasedService<Dev
         }
 
         @Override
-        public void onFailure(Throwable t) {
+        public void onFailure(@NonNull Throwable t) {
             log.warn("[{}] Failed to update entry {}", deviceId, kvEntry, t);
         }
+
     }
 
 }

application/src/main/java/org/thingsboard/server/service/sync/ie/importing/impl/ResourceImportService.java

@@ -66,7 +66,7 @@ public class ResourceImportService extends BaseEntityImportService<TbResourceId,
     protected void cleanupForComparison(TbResource resource) {
         super.cleanupForComparison(resource);
         resource.setSearchText(null);
-        if (resource.getDescriptor().isNull()) {
+        if (resource.getDescriptor() != null && resource.getDescriptor().isNull()) {
             resource.setDescriptor(null);
         }
     }

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

@@ -45,6 +45,7 @@ import org.thingsboard.server.common.data.id.DeviceId;
 import org.thingsboard.server.common.data.id.EntityId;
 import org.thingsboard.server.common.data.id.TenantId;
 import org.thingsboard.server.common.data.kv.AttributeKvEntry;
+import org.thingsboard.server.common.data.kv.AttributesSaveResult;
 import org.thingsboard.server.common.data.kv.KvEntry;
 import org.thingsboard.server.common.data.kv.TimeseriesSaveResult;
 import org.thingsboard.server.common.data.kv.TsKvEntry;
@@ -62,7 +63,6 @@ import org.thingsboard.server.service.state.DefaultDeviceStateService;
 import org.thingsboard.server.service.subscription.TbSubscriptionUtils;
 
 import java.util.ArrayList;
-import java.util.Collections;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
@@ -190,17 +190,16 @@ public class DefaultTelemetrySubscriptionService extends AbstractSubscriptionSer
     }
 
     @Override
-    public void saveAttributesInternal(AttributesSaveRequest request) {
-        log.trace("Executing saveInternal [{}]", request);
+    public ListenableFuture<AttributesSaveResult> saveAttributesInternal(AttributesSaveRequest request) {
         TenantId tenantId = request.getTenantId();
         EntityId entityId = request.getEntityId();
         AttributesSaveRequest.Strategy strategy = request.getStrategy();
-        ListenableFuture<List<Long>> resultFuture;
+        ListenableFuture<AttributesSaveResult> resultFuture;
 
         if (strategy.saveAttributes()) {
             resultFuture = attrService.save(tenantId, entityId, request.getScope(), request.getEntries());
         } else {
-            resultFuture = Futures.immediateFuture(Collections.emptyList());
+            resultFuture = Futures.immediateFuture(AttributesSaveResult.EMPTY);
         }
 
         addMainCallback(resultFuture, result -> {
@@ -228,6 +227,7 @@ public class DefaultTelemetrySubscriptionService extends AbstractSubscriptionSer
         if (strategy.sendWsUpdate()) {
             addWsCallback(resultFuture, success -> onAttributesUpdate(tenantId, entityId, request.getScope().name(), request.getEntries()));
         }
+        return resultFuture;
     }
 
     private static boolean shouldSendSharedAttributesUpdatedNotification(AttributesSaveRequest request) {

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

@@ -21,6 +21,7 @@ import org.thingsboard.rule.engine.api.AttributesSaveRequest;
 import org.thingsboard.rule.engine.api.RuleEngineTelemetryService;
 import org.thingsboard.rule.engine.api.TimeseriesDeleteRequest;
 import org.thingsboard.rule.engine.api.TimeseriesSaveRequest;
+import org.thingsboard.server.common.data.kv.AttributesSaveResult;
 import org.thingsboard.server.common.data.kv.TimeseriesSaveResult;
 
 /**
@@ -30,7 +31,7 @@ public interface InternalTelemetryService extends RuleEngineTelemetryService {
 
     ListenableFuture<TimeseriesSaveResult> saveTimeseriesInternal(TimeseriesSaveRequest request);
 
-    void saveAttributesInternal(AttributesSaveRequest request);
+    ListenableFuture<AttributesSaveResult> saveAttributesInternal(AttributesSaveRequest request);
 
     void deleteTimeseriesInternal(TimeseriesDeleteRequest request);
 

application/src/main/resources/thingsboard.yml

@@ -898,6 +898,8 @@ state:
   # Used only when state.persistToTelemetry is set to 'true' and Cassandra is used for timeseries data.
   # 0 means time-to-live mechanism is disabled.
   telemetryTtl: "${STATE_TELEMETRY_TTL:0}"
+  # Number of device records to fetch per batch when initializing device activity states
+  initFetchPackSize: "${TB_DEVICE_STATE_INIT_FETCH_PACK_SIZE:50000}"
   # Configuration properties for rule nodes related to device activity state
   rule:
     node:

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

@@ -464,7 +464,7 @@ public class CalculatedFieldIntegrationTest extends CalculatedFieldControllerTes
     }
 
     @Test
-    public void testSimpleCalculatedFieldWhenPreserveMsgTsIsTrue() throws Exception {
+    public void testSimpleCalculatedFieldWhenUseLatestTsIsTrue() throws Exception {
         Device testDevice = createDevice("Test device", "1234567890");
         long ts = System.currentTimeMillis() - 300000L;
         doPost("/api/plugins/telemetry/DEVICE/" + testDevice.getUuidId() + "/timeseries/" + DataConstants.SERVER_SCOPE, JacksonUtil.toJsonNode(String.format("{\"ts\": %s, \"values\": {\"temperature\":30}}", ts)));
@@ -489,7 +489,7 @@ public class CalculatedFieldIntegrationTest extends CalculatedFieldControllerTes
         output.setType(OutputType.TIME_SERIES);
         config.setOutput(output);
 
-        config.setPreserveMsgTs(true);
+        config.setUseLatestTs(true);
 
         calculatedField.setConfiguration(config);
 
@@ -506,7 +506,69 @@ public class CalculatedFieldIntegrationTest extends CalculatedFieldControllerTes
     }
 
     @Test
-    public void testScriptCalculatedFieldWhenUsedMsgTsInScript() throws Exception {
+    public void testSimpleCalculatedFieldWhenUseLatestTsIsTrueAndTelemetryBeforeLatest() throws Exception {
+        Device testDevice = createDevice("Test device", "1234567890");
+        long ts = System.currentTimeMillis();
+
+        long tsA = ts - 300000L;
+        doPost("/api/plugins/telemetry/DEVICE/" + testDevice.getUuidId() + "/timeseries/" + DataConstants.SERVER_SCOPE, JacksonUtil.toJsonNode(String.format("{\"ts\": %s, \"values\": {\"a\":1}}", tsA)));
+
+        long tsB = ts - 300L;
+        doPost("/api/plugins/telemetry/DEVICE/" + testDevice.getUuidId() + "/timeseries/" + DataConstants.SERVER_SCOPE, JacksonUtil.toJsonNode(String.format("{\"ts\": %s, \"values\": {\"b\":5}}", tsB)));
+
+        CalculatedField calculatedField = new CalculatedField();
+        calculatedField.setEntityId(testDevice.getId());
+        calculatedField.setType(CalculatedFieldType.SIMPLE);
+        calculatedField.setName("a + b");
+        calculatedField.setDebugSettings(DebugSettings.all());
+        calculatedField.setConfigurationVersion(1);
+
+        SimpleCalculatedFieldConfiguration config = new SimpleCalculatedFieldConfiguration();
+
+        Argument argument1 = new Argument();
+        ReferencedEntityKey refEntityKey1 = new ReferencedEntityKey("a", ArgumentType.TS_LATEST, null);
+        argument1.setRefEntityKey(refEntityKey1);
+        Argument argument2 = new Argument();
+        ReferencedEntityKey refEntityKey2 = new ReferencedEntityKey("b", ArgumentType.TS_LATEST, null);
+        argument2.setRefEntityKey(refEntityKey2);
+        config.setArguments(Map.of("a", argument1, "b", argument2));
+        config.setExpression("a + b");
+
+        Output output = new Output();
+        output.setName("c");
+        output.setType(OutputType.TIME_SERIES);
+        config.setOutput(output);
+
+        config.setUseLatestTs(true);
+
+        calculatedField.setConfiguration(config);
+
+        CalculatedField savedCalculatedField = doPost("/api/calculatedField", calculatedField, CalculatedField.class);
+
+        await().alias("create CF -> perform initial calculation").atMost(TIMEOUT, TimeUnit.SECONDS)
+                .pollInterval(POLL_INTERVAL, TimeUnit.SECONDS)
+                .untilAsserted(() -> {
+                    ObjectNode c = getLatestTelemetry(testDevice.getId(), "c");
+                    assertThat(c).isNotNull();
+                    assertThat(c.get("c").get(0).get("ts").asText()).isEqualTo(Long.toString(tsB));
+                    assertThat(c.get("c").get(0).get("value").asText()).isEqualTo("6.0");
+                });
+
+        long tsABeforeTsB = tsB - 300L;
+        doPost("/api/plugins/telemetry/DEVICE/" + testDevice.getUuidId() + "/timeseries/" + DataConstants.SERVER_SCOPE, JacksonUtil.toJsonNode(String.format("{\"ts\": %s, \"values\": {\"a\":10}}", tsABeforeTsB)));
+
+        await().alias("update telemetry with ts less than latest -> save result with latest ts").atMost(TIMEOUT, TimeUnit.SECONDS)
+                .pollInterval(POLL_INTERVAL, TimeUnit.SECONDS)
+                .untilAsserted(() -> {
+                    ObjectNode c = getLatestTelemetry(testDevice.getId(), "c");
+                    assertThat(c).isNotNull();
+                    assertThat(c.get("c").get(0).get("ts").asText()).isEqualTo(Long.toString(tsB));// also tsB, since this is the latest timestamp
+                    assertThat(c.get("c").get(0).get("value").asText()).isEqualTo("15.0");
+                });
+    }
+
+    @Test
+    public void testScriptCalculatedFieldWhenUsedLatestTsInScript() throws Exception {
         Device testDevice = createDevice("Test device", "1234567890");
         long ts = System.currentTimeMillis() - 300000L;
         doPost("/api/plugins/telemetry/DEVICE/" + testDevice.getUuidId() + "/timeseries/" + DataConstants.SERVER_SCOPE, JacksonUtil.toJsonNode(String.format("{\"ts\": %s, \"values\": {\"temperature\":30}}", ts)));
@@ -524,7 +586,7 @@ public class CalculatedFieldIntegrationTest extends CalculatedFieldControllerTes
         ReferencedEntityKey refEntityKey = new ReferencedEntityKey("temperature", ArgumentType.TS_LATEST, null);
         argument.setRefEntityKey(refEntityKey);
         config.setArguments(Map.of("T", argument));
-        config.setExpression("return {\"ts\": ctx.msgTs, \"values\": {\"fahrenheitTemp\": (T * 1.8) + 32}};");
+        config.setExpression("return {\"ts\": ctx.latestTs, \"values\": {\"fahrenheitTemp\": (T * 1.8) + 32}};");
 
         Output output = new Output();
         output.setType(OutputType.TIME_SERIES);

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

@@ -53,7 +53,7 @@ public class SingleValueArgumentEntryTest {
     }
 
     @Test
-    void testUpdateEntryWithThaSameTs() {
+    void testUpdateEntryWithTheSameTs() {
         assertThat(entry.updateEntry(new SingleValueArgumentEntry(ts, new LongDataEntry("key", 13L), 363L))).isFalse();
     }
 
@@ -81,6 +81,11 @@ public class SingleValueArgumentEntryTest {
         assertThat(entry.updateEntry(new SingleValueArgumentEntry(ts + 18, new LongDataEntry("key", 11L), 364L))).isTrue();
     }
 
+    @Test
+    void testUpdateEntryWithOldTs() {
+        assertThat(entry.updateEntry(new SingleValueArgumentEntry(ts - 10, new LongDataEntry("key", 14L), 365L))).isFalse();
+    }
+
     @Test
     void testToTbelCfArgWhenJsonIsObject() {
         entry = new SingleValueArgumentEntry(ts, new JsonDataEntry("key", "{\"test\": 10}"), 370L);

application/src/test/java/org/thingsboard/server/service/entitiy/EntityServiceTest.java

@@ -44,6 +44,7 @@ import org.thingsboard.server.common.data.id.EntityViewId;
 import org.thingsboard.server.common.data.id.IdBased;
 import org.thingsboard.server.common.data.id.TenantId;
 import org.thingsboard.server.common.data.kv.AttributeKvEntry;
+import org.thingsboard.server.common.data.kv.AttributesSaveResult;
 import org.thingsboard.server.common.data.kv.BaseAttributeKvEntry;
 import org.thingsboard.server.common.data.kv.BasicTsKvEntry;
 import org.thingsboard.server.common.data.kv.DoubleDataEntry;
@@ -395,7 +396,7 @@ public class EntityServiceTest extends AbstractControllerTest {
         List<Long> highTemperatures = new ArrayList<>();
         createTestHierarchy(tenantId, assets, devices, new ArrayList<>(), new ArrayList<>(), temperatures, highTemperatures);
 
-        List<ListenableFuture<List<Long>>> attributeFutures = new ArrayList<>();
+        List<ListenableFuture<AttributesSaveResult>> attributeFutures = new ArrayList<>();
         for (int i = 0; i < devices.size(); i++) {
             Device device = devices.get(i);
             attributeFutures.add(saveLongAttribute(device.getId(), "temperature", temperatures.get(i), AttributeScope.CLIENT_SCOPE));
@@ -545,7 +546,7 @@ public class EntityServiceTest extends AbstractControllerTest {
         List<Long> highTemperatures = new ArrayList<>();
         createTestHierarchy(tenantId, assets, devices, new ArrayList<>(), new ArrayList<>(), temperatures, highTemperatures);
 
-        List<ListenableFuture<List<Long>>> attributeFutures = new ArrayList<>();
+        List<ListenableFuture<AttributesSaveResult>> attributeFutures = new ArrayList<>();
         for (int i = 0; i < devices.size(); i++) {
             Device device = devices.get(i);
             attributeFutures.add(saveLongAttribute(device.getId(), "temperature", temperatures.get(i), AttributeScope.CLIENT_SCOPE));
@@ -599,7 +600,7 @@ public class EntityServiceTest extends AbstractControllerTest {
         List<Long> highConsumptions = new ArrayList<>();
         createTestHierarchy(tenantId, assets, devices, consumptions, highConsumptions, new ArrayList<>(), new ArrayList<>());
 
-        List<ListenableFuture<List<Long>>> attributeFutures = new ArrayList<>();
+        List<ListenableFuture<AttributesSaveResult>> attributeFutures = new ArrayList<>();
         for (int i = 0; i < assets.size(); i++) {
             Asset asset = assets.get(i);
             attributeFutures.add(saveLongAttribute(asset.getId(), "consumption", consumptions.get(i), AttributeScope.SERVER_SCOPE));
@@ -1506,7 +1507,7 @@ public class EntityServiceTest extends AbstractControllerTest {
             }
         }
 
-        List<ListenableFuture<List<Long>>> attributeFutures = new ArrayList<>();
+        List<ListenableFuture<AttributesSaveResult>> attributeFutures = new ArrayList<>();
         for (int i = 0; i < devices.size(); i++) {
             Device device = devices.get(i);
             for (AttributeScope currentScope : AttributeScope.values()) {
@@ -1578,7 +1579,7 @@ public class EntityServiceTest extends AbstractControllerTest {
             }
         }
 
-        List<ListenableFuture<List<Long>>> attributeFutures = new ArrayList<>();
+        List<ListenableFuture<AttributesSaveResult>> attributeFutures = new ArrayList<>();
         for (int i = 0; i < devices.size(); i++) {
             Device device = devices.get(i);
             attributeFutures.add(saveLongAttribute(device.getId(), "temperature", temperatures.get(i), AttributeScope.CLIENT_SCOPE));
@@ -1808,7 +1809,7 @@ public class EntityServiceTest extends AbstractControllerTest {
             }
         }
 
-        List<ListenableFuture<List<Long>>> attributeFutures = new ArrayList<>();
+        List<ListenableFuture<AttributesSaveResult>> attributeFutures = new ArrayList<>();
         for (int i = 0; i < devices.size(); i++) {
             Device device = devices.get(i);
             attributeFutures.add(saveStringAttribute(device.getId(), "attributeString", attributeStrings.get(i), AttributeScope.CLIENT_SCOPE));
@@ -2269,16 +2270,16 @@ public class EntityServiceTest extends AbstractControllerTest {
         return filter;
     }
 
-    private ListenableFuture<List<Long>> saveLongAttribute(EntityId entityId, String key, long value, AttributeScope scope) {
+    private ListenableFuture<AttributesSaveResult> saveLongAttribute(EntityId entityId, String key, long value, AttributeScope scope) {
         KvEntry attrValue = new LongDataEntry(key, value);
         AttributeKvEntry attr = new BaseAttributeKvEntry(attrValue, 42L);
-        return attributesService.save(tenantId, entityId, scope, Collections.singletonList(attr));
+        return attributesService.save(tenantId, entityId, scope, List.of(attr));
     }
 
-    private ListenableFuture<List<Long>> saveStringAttribute(EntityId entityId, String key, String value, AttributeScope scope) {
+    private ListenableFuture<AttributesSaveResult> saveStringAttribute(EntityId entityId, String key, String value, AttributeScope scope) {
         KvEntry attrValue = new StringDataEntry(key, value);
         AttributeKvEntry attr = new BaseAttributeKvEntry(attrValue, 42L);
-        return attributesService.save(tenantId, entityId, scope, Collections.singletonList(attr));
+        return attributesService.save(tenantId, entityId, scope, List.of(attr));
     }
 
     private ListenableFuture<TimeseriesSaveResult> saveTimeseries(EntityId entityId, String key, Double value) {
@@ -2294,8 +2295,8 @@ public class EntityServiceTest extends AbstractControllerTest {
     }
 
     protected void createMultiRootHierarchy(List<Asset> buildings, List<Asset> apartments,
-                                          Map<String, Map<UUID, String>> entityNameByTypeMap,
-                                          Map<UUID, UUID> childParentRelationMap) throws InterruptedException {
+                                            Map<String, Map<UUID, String>> entityNameByTypeMap,
+                                            Map<UUID, UUID> childParentRelationMap) throws InterruptedException {
         for (int k = 0; k < 3; k++) {
             Asset building = new Asset();
             building.setTenantId(tenantId);

application/src/test/java/org/thingsboard/server/service/state/DefaultDeviceStateServiceTest.java

@@ -16,6 +16,9 @@
 package org.thingsboard.server.service.state;
 
 import com.google.common.util.concurrent.Futures;
+import com.google.common.util.concurrent.ListeningExecutorService;
+import com.google.common.util.concurrent.MoreExecutors;
+import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.extension.ExtendWith;
@@ -31,13 +34,13 @@ import org.thingsboard.rule.engine.api.AttributesSaveRequest;
 import org.thingsboard.server.cluster.TbClusterService;
 import org.thingsboard.server.common.data.AttributeScope;
 import org.thingsboard.server.common.data.Device;
-import org.thingsboard.server.common.data.DeviceIdInfo;
 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.kv.AttributeKvEntry;
+import org.thingsboard.server.common.data.kv.AttributesSaveResult;
 import org.thingsboard.server.common.data.msg.TbMsgType;
 import org.thingsboard.server.common.data.notification.rule.trigger.DeviceActivityTrigger;
-import org.thingsboard.server.common.data.page.PageData;
 import org.thingsboard.server.common.msg.TbMsg;
 import org.thingsboard.server.common.msg.TbMsgMetaData;
 import org.thingsboard.server.common.msg.notification.NotificationRuleProcessor;
@@ -50,20 +53,22 @@ import org.thingsboard.server.dao.sql.query.EntityQueryRepository;
 import org.thingsboard.server.dao.timeseries.TimeseriesService;
 import org.thingsboard.server.gen.transport.TransportProtos;
 import org.thingsboard.server.queue.discovery.PartitionService;
-import org.thingsboard.server.queue.discovery.QueueKey;
-import org.thingsboard.server.queue.discovery.event.PartitionChangeEvent;
 import org.thingsboard.server.queue.usagestats.DefaultTbApiUsageReportClient;
 import org.thingsboard.server.service.telemetry.TelemetrySubscriptionService;
 
+import java.time.Duration;
 import java.util.Collections;
+import java.util.HashSet;
 import java.util.List;
-import java.util.Map;
+import java.util.Set;
 import java.util.UUID;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
+import java.util.concurrent.ThreadLocalRandom;
 import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicLong;
 import java.util.stream.Stream;
 
 import static org.assertj.core.api.Assertions.assertThat;
@@ -77,8 +82,8 @@ import static org.mockito.BDDMockito.then;
 import static org.mockito.BDDMockito.willReturn;
 import static org.mockito.Mockito.doAnswer;
 import static org.mockito.Mockito.doReturn;
+import static org.mockito.Mockito.lenient;
 import static org.mockito.Mockito.never;
-import static org.mockito.Mockito.reset;
 import static org.mockito.Mockito.spy;
 import static org.mockito.Mockito.times;
 import static org.mockito.Mockito.verify;
@@ -90,7 +95,10 @@ import static org.thingsboard.server.service.state.DefaultDeviceStateService.LAS
 import static org.thingsboard.server.service.state.DefaultDeviceStateService.LAST_DISCONNECT_TIME;
 
 @ExtendWith(MockitoExtension.class)
-public class DefaultDeviceStateServiceTest {
+class DefaultDeviceStateServiceTest {
+
+    ListeningExecutorService deviceStateExecutor;
+    ListeningExecutorService deviceStateCallbackExecutor;
 
     @Mock
     DeviceService deviceService;
@@ -113,25 +121,48 @@ public class DefaultDeviceStateServiceTest {
     @Mock
     DefaultTbApiUsageReportClient defaultTbApiUsageReportClient;
 
-    TenantId tenantId = new TenantId(UUID.fromString("00797a3b-7aeb-4b5b-b57a-c2a810d0f112"));
-    DeviceId deviceId = DeviceId.fromString("00797a3b-7aeb-4b5b-b57a-c2a810d0f112");
-    TopicPartitionInfo tpi;
+    long defaultInactivityTimeoutMs = Duration.ofMinutes(10L).toMillis();
+
+    TenantId tenantId = TenantId.fromUUID(UUID.fromString("00797a3b-7aeb-4b5b-b57a-c2a810d0f112"));
+    DeviceId deviceId = DeviceId.fromString("c209f718-42e5-11f0-9fe2-0242ac120002");
+    TopicPartitionInfo tpi = TopicPartitionInfo.builder()
+            .topic("tb_core")
+            .partition(0)
+            .myPartition(true)
+            .build();
 
     DefaultDeviceStateService service;
 
     @BeforeEach
-    public void setUp() {
+    void setUp() {
         service = spy(new DefaultDeviceStateService(deviceService, attributesService, tsService, clusterService, partitionService, entityQueryRepository, null, defaultTbApiUsageReportClient, notificationRuleProcessor));
         ReflectionTestUtils.setField(service, "tsSubService", telemetrySubscriptionService);
+        ReflectionTestUtils.setField(service, "defaultInactivityTimeoutMs", defaultInactivityTimeoutMs);
         ReflectionTestUtils.setField(service, "defaultStateCheckIntervalInSec", 60);
         ReflectionTestUtils.setField(service, "defaultActivityStatsIntervalInSec", 60);
-        ReflectionTestUtils.setField(service, "initFetchPackSize", 10);
+        ReflectionTestUtils.setField(service, "initFetchPackSize", 50000);
+
+        deviceStateExecutor = MoreExecutors.newDirectExecutorService();
+        ReflectionTestUtils.setField(service, "deviceStateExecutor", deviceStateExecutor);
+
+        deviceStateCallbackExecutor = MoreExecutors.newDirectExecutorService();
+        ReflectionTestUtils.setField(service, "deviceStateCallbackExecutor", deviceStateCallbackExecutor);
+
+        lenient().when(partitionService.resolve(ServiceType.TB_CORE, tenantId, deviceId)).thenReturn(tpi);
+
+        ConcurrentMap<TopicPartitionInfo, Set<DeviceId>> partitionedEntities = new ConcurrentHashMap<>();
+        partitionedEntities.put(tpi, new HashSet<>());
+        ReflectionTestUtils.setField(service, "partitionedEntities", partitionedEntities);
+    }
 
-        tpi = TopicPartitionInfo.builder().myPartition(true).build();
+    @AfterEach
+    void cleanup() {
+        deviceStateExecutor.shutdownNow();
+        deviceStateCallbackExecutor.shutdownNow();
     }
 
     @Test
-    public void givenDeviceBelongsToExternalPartition_whenOnDeviceConnect_thenCleansStateAndDoesNotReportConnect() {
+    void givenDeviceBelongsToExternalPartition_whenOnDeviceConnect_thenCleansStateAndDoesNotReportConnect() {
         // GIVEN
         doReturn(true).when(service).cleanDeviceStateIfBelongsToExternalPartition(tenantId, deviceId);
 
@@ -149,7 +180,7 @@ public class DefaultDeviceStateServiceTest {
 
     @ParameterizedTest
     @ValueSource(longs = {Long.MIN_VALUE, -100, -1})
-    public void givenNegativeLastConnectTime_whenOnDeviceConnect_thenSkipsThisEvent(long negativeLastConnectTime) {
+    void givenNegativeLastConnectTime_whenOnDeviceConnect_thenSkipsThisEvent(long negativeLastConnectTime) {
         // GIVEN
         doReturn(false).when(service).cleanDeviceStateIfBelongsToExternalPartition(tenantId, deviceId);
 
@@ -166,7 +197,7 @@ public class DefaultDeviceStateServiceTest {
 
     @ParameterizedTest
     @MethodSource("provideOutdatedTimestamps")
-    public void givenOutdatedLastConnectTime_whenOnDeviceDisconnect_thenSkipsThisEvent(long outdatedLastConnectTime, long currentLastConnectTime) {
+    void givenOutdatedLastConnectTime_whenOnDeviceDisconnect_thenSkipsThisEvent(long outdatedLastConnectTime, long currentLastConnectTime) {
         // GIVEN
         doReturn(false).when(service).cleanDeviceStateIfBelongsToExternalPartition(tenantId, deviceId);
 
@@ -188,7 +219,7 @@ public class DefaultDeviceStateServiceTest {
     }
 
     @Test
-    public void givenDeviceBelongsToMyPartition_whenOnDeviceConnect_thenReportsConnect() {
+    void givenDeviceBelongsToMyPartition_whenOnDeviceConnect_thenReportsConnect() {
         // GIVEN
         var deviceStateData = DeviceStateData.builder()
                 .tenantId(tenantId)
@@ -202,11 +233,13 @@ public class DefaultDeviceStateServiceTest {
         service.deviceStates.put(deviceId, deviceStateData);
         long lastConnectTime = System.currentTimeMillis();
 
+        mockSuccessfulSaveAttributes();
+
         // WHEN
         service.onDeviceConnect(tenantId, deviceId, lastConnectTime);
 
         // THEN
-        then(telemetrySubscriptionService).should().saveAttributes(argThat(request ->
+        then(telemetrySubscriptionService).should().saveAttributesInternal(argThat(request ->
                 request.getTenantId().equals(tenantId) && request.getEntityId().equals(deviceId) &&
                         request.getScope().equals(AttributeScope.SERVER_SCOPE) &&
                         request.getEntries().get(0).getKey().equals(LAST_CONNECT_TIME) &&
@@ -221,7 +254,7 @@ public class DefaultDeviceStateServiceTest {
     }
 
     @Test
-    public void givenDeviceBelongsToExternalPartition_whenOnDeviceDisconnect_thenCleansStateAndDoesNotReportDisconnect() {
+    void givenDeviceBelongsToExternalPartition_whenOnDeviceDisconnect_thenCleansStateAndDoesNotReportDisconnect() {
         // GIVEN
         doReturn(true).when(service).cleanDeviceStateIfBelongsToExternalPartition(tenantId, deviceId);
 
@@ -238,7 +271,7 @@ public class DefaultDeviceStateServiceTest {
 
     @ParameterizedTest
     @ValueSource(longs = {Long.MIN_VALUE, -100, -1})
-    public void givenNegativeLastDisconnectTime_whenOnDeviceDisconnect_thenSkipsThisEvent(long negativeLastDisconnectTime) {
+    void givenNegativeLastDisconnectTime_whenOnDeviceDisconnect_thenSkipsThisEvent(long negativeLastDisconnectTime) {
         // GIVEN
         doReturn(false).when(service).cleanDeviceStateIfBelongsToExternalPartition(tenantId, deviceId);
 
@@ -254,7 +287,7 @@ public class DefaultDeviceStateServiceTest {
 
     @ParameterizedTest
     @MethodSource("provideOutdatedTimestamps")
-    public void givenOutdatedLastDisconnectTime_whenOnDeviceDisconnect_thenSkipsThisEvent(long outdatedLastDisconnectTime, long currentLastDisconnectTime) {
+    void givenOutdatedLastDisconnectTime_whenOnDeviceDisconnect_thenSkipsThisEvent(long outdatedLastDisconnectTime, long currentLastDisconnectTime) {
         // GIVEN
         doReturn(false).when(service).cleanDeviceStateIfBelongsToExternalPartition(tenantId, deviceId);
 
@@ -275,7 +308,7 @@ public class DefaultDeviceStateServiceTest {
     }
 
     @Test
-    public void givenDeviceBelongsToMyPartition_whenOnDeviceDisconnect_thenReportsDisconnect() {
+    void givenDeviceBelongsToMyPartition_whenOnDeviceDisconnect_thenReportsDisconnect() {
         // GIVEN
         var deviceStateData = DeviceStateData.builder()
                 .tenantId(tenantId)
@@ -289,11 +322,13 @@ public class DefaultDeviceStateServiceTest {
         service.deviceStates.put(deviceId, deviceStateData);
         long lastDisconnectTime = System.currentTimeMillis();
 
+        mockSuccessfulSaveAttributes();
+
         // WHEN
         service.onDeviceDisconnect(tenantId, deviceId, lastDisconnectTime);
 
         // THEN
-        then(telemetrySubscriptionService).should().saveAttributes(argThat(request ->
+        then(telemetrySubscriptionService).should().saveAttributesInternal(argThat(request ->
                 request.getTenantId().equals(tenantId) && request.getEntityId().equals(deviceId) &&
                         request.getScope().equals(AttributeScope.SERVER_SCOPE) &&
                         request.getEntries().get(0).getKey().equals(LAST_DISCONNECT_TIME) &&
@@ -308,7 +343,7 @@ public class DefaultDeviceStateServiceTest {
     }
 
     @Test
-    public void givenDeviceBelongsToExternalPartition_whenOnDeviceInactivity_thenCleansStateAndDoesNotReportInactivity() {
+    void givenDeviceBelongsToExternalPartition_whenOnDeviceInactivity_thenCleansStateAndDoesNotReportInactivity() {
         // GIVEN
         doReturn(true).when(service).cleanDeviceStateIfBelongsToExternalPartition(tenantId, deviceId);
 
@@ -325,7 +360,7 @@ public class DefaultDeviceStateServiceTest {
 
     @ParameterizedTest
     @ValueSource(longs = {Long.MIN_VALUE, -100, -1})
-    public void givenNegativeLastInactivityTime_whenOnDeviceInactivity_thenSkipsThisEvent(long negativeLastInactivityTime) {
+    void givenNegativeLastInactivityTime_whenOnDeviceInactivity_thenSkipsThisEvent(long negativeLastInactivityTime) {
         // GIVEN
         doReturn(false).when(service).cleanDeviceStateIfBelongsToExternalPartition(tenantId, deviceId);
 
@@ -341,7 +376,7 @@ public class DefaultDeviceStateServiceTest {
 
     @ParameterizedTest
     @MethodSource("provideOutdatedTimestamps")
-    public void givenReceivedInactivityTimeIsLessThanOrEqualToCurrentInactivityTime_whenOnDeviceInactivity_thenSkipsThisEvent(
+    void givenReceivedInactivityTimeIsLessThanOrEqualToCurrentInactivityTime_whenOnDeviceInactivity_thenSkipsThisEvent(
             long outdatedLastInactivityTime, long currentLastInactivityTime
     ) {
         // GIVEN
@@ -365,7 +400,7 @@ public class DefaultDeviceStateServiceTest {
 
     @ParameterizedTest
     @MethodSource("provideOutdatedTimestamps")
-    public void givenReceivedInactivityTimeIsLessThanOrEqualToCurrentActivityTime_whenOnDeviceInactivity_thenSkipsThisEvent(
+    void givenReceivedInactivityTimeIsLessThanOrEqualToCurrentActivityTime_whenOnDeviceInactivity_thenSkipsThisEvent(
             long outdatedLastInactivityTime, long currentLastActivityTime
     ) {
         // GIVEN
@@ -398,7 +433,7 @@ public class DefaultDeviceStateServiceTest {
     }
 
     @Test
-    public void givenDeviceBelongsToMyPartition_whenOnDeviceInactivity_thenReportsInactivity() {
+    void givenDeviceBelongsToMyPartition_whenOnDeviceInactivity_thenReportsInactivity() {
         // GIVEN
         var deviceStateData = DeviceStateData.builder()
                 .tenantId(tenantId)
@@ -412,17 +447,19 @@ public class DefaultDeviceStateServiceTest {
         service.deviceStates.put(deviceId, deviceStateData);
         long lastInactivityTime = System.currentTimeMillis();
 
+        mockSuccessfulSaveAttributes();
+
         // WHEN
         service.onDeviceInactivity(tenantId, deviceId, lastInactivityTime);
 
         // THEN
-        then(telemetrySubscriptionService).should().saveAttributes(argThat(request ->
+        then(telemetrySubscriptionService).should().saveAttributesInternal(argThat(request ->
                 request.getTenantId().equals(tenantId) && request.getEntityId().equals(deviceId) &&
                         request.getScope().equals(AttributeScope.SERVER_SCOPE) &&
                         request.getEntries().get(0).getKey().equals(INACTIVITY_ALARM_TIME) &&
                         request.getEntries().get(0).getValue().equals(lastInactivityTime)
         ));
-        then(telemetrySubscriptionService).should().saveAttributes(argThat(request ->
+        then(telemetrySubscriptionService).should().saveAttributesInternal(argThat(request ->
                 request.getTenantId().equals(tenantId) && request.getEntityId().equals(deviceId) &&
                         request.getScope().equals(AttributeScope.SERVER_SCOPE) &&
                         request.getEntries().get(0).getKey().equals(ACTIVITY_STATE) &&
@@ -445,7 +482,7 @@ public class DefaultDeviceStateServiceTest {
     }
 
     @Test
-    public void givenInactivityTimeoutReached_whenUpdateInactivityStateIfExpired_thenReportsInactivity() {
+    void givenInactivityTimeoutReached_whenUpdateInactivityStateIfExpired_thenReportsInactivity() {
         // GIVEN
         var deviceStateData = DeviceStateData.builder()
                 .tenantId(tenantId)
@@ -456,16 +493,18 @@ public class DefaultDeviceStateServiceTest {
 
         given(partitionService.resolve(ServiceType.TB_CORE, tenantId, deviceId)).willReturn(tpi);
 
+        mockSuccessfulSaveAttributes();
+
         // WHEN
         service.updateInactivityStateIfExpired(System.currentTimeMillis(), deviceId, deviceStateData);
 
         // THEN
-        then(telemetrySubscriptionService).should().saveAttributes(argThat(request ->
+        then(telemetrySubscriptionService).should().saveAttributesInternal(argThat(request ->
                 request.getTenantId().equals(tenantId) && request.getEntityId().equals(deviceId) &&
                         request.getScope().equals(AttributeScope.SERVER_SCOPE) &&
                         request.getEntries().get(0).getKey().equals(INACTIVITY_ALARM_TIME)
         ));
-        then(telemetrySubscriptionService).should().saveAttributes(argThat(request ->
+        then(telemetrySubscriptionService).should().saveAttributesInternal(argThat(request ->
                 request.getTenantId().equals(tenantId) && request.getEntityId().equals(deviceId) &&
                         request.getScope().equals(AttributeScope.SERVER_SCOPE) &&
                         request.getEntries().get(0).getKey().equals(ACTIVITY_STATE) &&
@@ -488,7 +527,7 @@ public class DefaultDeviceStateServiceTest {
     }
 
     @Test
-    public void givenDeviceIdFromDeviceStatesMap_whenGetOrFetchDeviceStateData_thenNoStackOverflow() {
+    void givenDeviceIdFromDeviceStatesMap_whenGetOrFetchDeviceStateData_thenNoStackOverflow() {
         service.deviceStates.put(deviceId, deviceStateDataMock);
         DeviceStateData deviceStateData = service.getOrFetchDeviceStateData(deviceId);
         assertThat(deviceStateData).isEqualTo(deviceStateDataMock);
@@ -496,7 +535,7 @@ public class DefaultDeviceStateServiceTest {
     }
 
     @Test
-    public void givenDeviceIdWithoutDeviceStateInMap_whenGetOrFetchDeviceStateData_thenFetchDeviceStateData() {
+    void givenDeviceIdWithoutDeviceStateInMap_whenGetOrFetchDeviceStateData_thenFetchDeviceStateData() {
         service.deviceStates.clear();
         willReturn(deviceStateDataMock).given(service).fetchDeviceStateDataUsingSeparateRequests(deviceId);
         DeviceStateData deviceStateData = service.getOrFetchDeviceStateData(deviceId);
@@ -504,172 +543,18 @@ public class DefaultDeviceStateServiceTest {
         verify(service).fetchDeviceStateDataUsingSeparateRequests(deviceId);
     }
 
-    private void initStateService(long timeout) throws InterruptedException {
-        service.stop();
-        reset(service, telemetrySubscriptionService);
-        service.setDefaultInactivityTimeoutMs(timeout);
-        service.init();
-        when(partitionService.resolve(ServiceType.TB_CORE, tenantId, deviceId)).thenReturn(tpi);
-        when(entityQueryRepository.findEntityDataByQueryInternal(any())).thenReturn(new PageData<>());
-        var deviceIdInfo = new DeviceIdInfo(tenantId.getId(), null, deviceId.getId());
-        when(deviceService.findDeviceIdInfos(any()))
-                .thenReturn(new PageData<>(List.of(deviceIdInfo), 0, 1, false));
-        PartitionChangeEvent event = new PartitionChangeEvent(this, ServiceType.TB_CORE, Map.of(
-                new QueueKey(ServiceType.TB_CORE), Collections.singleton(tpi)
-        ), Collections.emptyMap());
-        service.onApplicationEvent(event);
-        Thread.sleep(100);
-    }
-
-    @Test
-    public void increaseInactivityForInactiveDeviceTest() throws Exception {
-        final long defaultTimeout = 1;
-        initStateService(defaultTimeout);
-        DeviceState deviceState = DeviceState.builder().build();
-        DeviceStateData deviceStateData = DeviceStateData.builder()
-                .tenantId(tenantId)
-                .deviceId(deviceId)
-                .state(deviceState)
-                .metaData(new TbMsgMetaData())
-                .build();
-
-        service.deviceStates.put(deviceId, deviceStateData);
-        service.getPartitionedEntities(tpi).add(deviceId);
-
-        service.onDeviceActivity(tenantId, deviceId, System.currentTimeMillis());
-        activityVerify(true);
-        Thread.sleep(defaultTimeout);
-        service.checkStates();
-        activityVerify(false);
-
-        reset(telemetrySubscriptionService);
-
-        long increase = 100;
-        long newTimeout = System.currentTimeMillis() - deviceState.getLastActivityTime() + increase;
-
-        service.onDeviceInactivityTimeoutUpdate(tenantId, deviceId, newTimeout);
-        activityVerify(true);
-        Thread.sleep(increase);
-        service.checkStates();
-        activityVerify(false);
-
-        reset(telemetrySubscriptionService);
-
-        service.onDeviceActivity(tenantId, deviceId, System.currentTimeMillis());
-        activityVerify(true);
-        Thread.sleep(newTimeout + 5);
-        service.checkStates();
-        activityVerify(false);
-    }
-
-    @Test
-    public void increaseInactivityForActiveDeviceTest() throws Exception {
-        final long defaultTimeout = 1000;
-        initStateService(defaultTimeout);
-        DeviceState deviceState = DeviceState.builder().build();
-        DeviceStateData deviceStateData = DeviceStateData.builder()
-                .tenantId(tenantId)
-                .deviceId(deviceId)
-                .state(deviceState)
-                .metaData(new TbMsgMetaData())
-                .build();
-
-        service.deviceStates.put(deviceId, deviceStateData);
-        service.getPartitionedEntities(tpi).add(deviceId);
-
-        service.onDeviceActivity(tenantId, deviceId, System.currentTimeMillis());
-        activityVerify(true);
-
-        reset(telemetrySubscriptionService);
-
-        long increase = 100;
-        long newTimeout = System.currentTimeMillis() - deviceState.getLastActivityTime() + increase;
-
-        service.onDeviceInactivityTimeoutUpdate(tenantId, deviceId, newTimeout);
-        verify(telemetrySubscriptionService, never()).saveAttributes(argThat(request ->
-                request.getEntityId().equals(deviceId) && request.getEntries().get(0).getKey().equals(ACTIVITY_STATE)
-        ));
-        Thread.sleep(defaultTimeout + increase);
-        service.checkStates();
-        activityVerify(false);
-
-        reset(telemetrySubscriptionService);
-
-        service.onDeviceActivity(tenantId, deviceId, System.currentTimeMillis());
-        activityVerify(true);
-        Thread.sleep(newTimeout);
-        service.checkStates();
-        activityVerify(false);
-    }
-
-    @Test
-    public void increaseSmallInactivityForInactiveDeviceTest() throws Exception {
-        final long defaultTimeout = 1;
-        initStateService(defaultTimeout);
-        DeviceState deviceState = DeviceState.builder().build();
-        DeviceStateData deviceStateData = DeviceStateData.builder()
-                .tenantId(tenantId)
-                .deviceId(deviceId)
-                .state(deviceState)
-                .metaData(new TbMsgMetaData())
-                .build();
-
-        service.deviceStates.put(deviceId, deviceStateData);
-        service.getPartitionedEntities(tpi).add(deviceId);
-
-        service.onDeviceActivity(tenantId, deviceId, System.currentTimeMillis());
-        activityVerify(true);
-        Thread.sleep(defaultTimeout);
-        service.checkStates();
-        activityVerify(false);
-
-        reset(telemetrySubscriptionService);
-
-        long newTimeout = 1;
-        Thread.sleep(newTimeout);
-        verify(telemetrySubscriptionService, never()).saveAttributes(argThat(request ->
-                request.getEntityId().equals(deviceId) && request.getEntries().get(0).getKey().equals(ACTIVITY_STATE)
-        ));
-    }
+    @MethodSource
+    @ParameterizedTest
+    void testOnDeviceInactivityTimeoutUpdate(boolean initialActivityStatus, long newInactivityTimeout, boolean expectedActivityStatus) {
+        // GIVEN
+        doReturn(200L).when(service).getCurrentTimeMillis();
 
-    @Test
-    public void decreaseInactivityForActiveDeviceTest() throws Exception {
-        final long defaultTimeout = 1000;
-        initStateService(defaultTimeout);
-        DeviceState deviceState = DeviceState.builder().build();
-        DeviceStateData deviceStateData = DeviceStateData.builder()
-                .tenantId(tenantId)
-                .deviceId(deviceId)
-                .state(deviceState)
-                .metaData(new TbMsgMetaData())
+        var deviceState = DeviceState.builder()
+                .active(initialActivityStatus)
+                .lastActivityTime(100L)
                 .build();
 
-        service.deviceStates.put(deviceId, deviceStateData);
-        service.getPartitionedEntities(tpi).add(deviceId);
-
-        service.onDeviceActivity(tenantId, deviceId, System.currentTimeMillis());
-        activityVerify(true);
-
-        long newTimeout = 1;
-        Thread.sleep(newTimeout);
-
-        service.onDeviceInactivityTimeoutUpdate(tenantId, deviceId, newTimeout);
-        activityVerify(false);
-        reset(telemetrySubscriptionService);
-
-        service.onDeviceInactivityTimeoutUpdate(tenantId, deviceId, defaultTimeout);
-        activityVerify(true);
-        Thread.sleep(defaultTimeout);
-        service.checkStates();
-        activityVerify(false);
-    }
-
-    @Test
-    public void decreaseInactivityForInactiveDeviceTest() throws Exception {
-        final long defaultTimeout = 1000;
-        initStateService(defaultTimeout);
-        DeviceState deviceState = DeviceState.builder().build();
-        DeviceStateData deviceStateData = DeviceStateData.builder()
+        var deviceStateData = DeviceStateData.builder()
                 .tenantId(tenantId)
                 .deviceId(deviceId)
                 .state(deviceState)
@@ -679,31 +564,44 @@ public class DefaultDeviceStateServiceTest {
         service.deviceStates.put(deviceId, deviceStateData);
         service.getPartitionedEntities(tpi).add(deviceId);
 
-        service.onDeviceActivity(tenantId, deviceId, System.currentTimeMillis());
-        activityVerify(true);
-        Thread.sleep(defaultTimeout);
-        service.checkStates();
-        activityVerify(false);
-        reset(telemetrySubscriptionService);
+        mockSuccessfulSaveAttributes();
 
-        long newTimeout = 1;
+        // WHEN
+        service.onDeviceInactivityTimeoutUpdate(tenantId, deviceId, newInactivityTimeout);
 
-        service.onDeviceInactivityTimeoutUpdate(tenantId, deviceId, newTimeout);
-        verify(telemetrySubscriptionService, never()).saveAttributes(argThat(request ->
-                request.getEntityId().equals(deviceId) && request.getEntries().get(0).getKey().equals(ACTIVITY_STATE)
-        ));
+        // THEN
+        long expectedInactivityTimeout = newInactivityTimeout != 0 ? newInactivityTimeout : defaultInactivityTimeoutMs;
+        assertThat(deviceState.getInactivityTimeout()).isEqualTo(expectedInactivityTimeout);
+
+        assertThat(deviceState.isActive()).isEqualTo(expectedActivityStatus);
+        if (initialActivityStatus != expectedActivityStatus) {
+            then(telemetrySubscriptionService).should().saveAttributesInternal(argThat(request -> {
+                AttributeKvEntry entry = request.getEntries().get(0);
+                return request.getEntityId().equals(deviceId) && entry.getKey().equals(ACTIVITY_STATE) && entry.getValue().equals(expectedActivityStatus);
+            }));
+        }
     }
 
-    private void activityVerify(boolean isActive) {
-        verify(telemetrySubscriptionService).saveAttributes(argThat(request ->
-                request.getEntityId().equals(deviceId) &&
-                        request.getEntries().get(0).getKey().equals(ACTIVITY_STATE) &&
-                        request.getEntries().get(0).getValue().equals(isActive)
-        ));
+    // to simplify test, these arguments assume that the current time is 200 and the last activity time is 100
+    private static Stream<Arguments> testOnDeviceInactivityTimeoutUpdate() {
+        return Stream.of(
+                Arguments.of(true, 1L, false),
+                Arguments.of(true, 50L, false),
+                Arguments.of(true, 99L, false),
+                Arguments.of(true, 100L, false),
+                Arguments.of(true, 101L, true),
+                Arguments.of(true, 0L, true), // should use default inactivity timeout of 10 minutes
+                Arguments.of(false, 1L, false),
+                Arguments.of(false, 50L, false),
+                Arguments.of(false, 99L, false),
+                Arguments.of(false, 100L, false),
+                Arguments.of(false, 101L, true),
+                Arguments.of(false, 0L, true) // should use default inactivity timeout of 10 minutes
+        );
     }
 
     @Test
-    public void givenStateDataIsNull_whenUpdateActivityState_thenShouldCleanupDevice() {
+    void givenStateDataIsNull_whenUpdateActivityState_thenShouldCleanupDevice() {
         // GIVEN
         service.deviceStates.put(deviceId, deviceStateDataMock);
 
@@ -719,7 +617,7 @@ public class DefaultDeviceStateServiceTest {
 
     @ParameterizedTest
     @MethodSource("provideParametersForUpdateActivityState")
-    public void givenTestParameters_whenUpdateActivityState_thenShouldBeInTheExpectedStateAndPerformExpectedActions(
+    void givenTestParameters_whenUpdateActivityState_thenShouldBeInTheExpectedStateAndPerformExpectedActions(
             boolean activityState, long previousActivityTime, long lastReportedActivity, long inactivityAlarmTime,
             long expectedInactivityAlarmTime, boolean shouldSetInactivityAlarmTimeToZero,
             boolean shouldUpdateActivityStateToActive
@@ -739,13 +637,15 @@ public class DefaultDeviceStateServiceTest {
                 .metaData(new TbMsgMetaData())
                 .build();
 
+        mockSuccessfulSaveAttributes();
+
         // WHEN
         service.updateActivityState(deviceId, deviceStateData, lastReportedActivity);
 
         // THEN
         assertThat(deviceState.isActive()).isEqualTo(true);
         assertThat(deviceState.getLastActivityTime()).isEqualTo(lastReportedActivity);
-        then(telemetrySubscriptionService).should().saveAttributes(argThat(request ->
+        then(telemetrySubscriptionService).should().saveAttributesInternal(argThat(request ->
                 request.getEntityId().equals(deviceId) &&
                         request.getEntries().get(0).getKey().equals(LAST_ACTIVITY_TIME) &&
                         request.getEntries().get(0).getValue().equals(lastReportedActivity)
@@ -753,7 +653,7 @@ public class DefaultDeviceStateServiceTest {
 
         assertThat(deviceState.getLastInactivityAlarmTime()).isEqualTo(expectedInactivityAlarmTime);
         if (shouldSetInactivityAlarmTimeToZero) {
-            then(telemetrySubscriptionService).should().saveAttributes(argThat(request ->
+            then(telemetrySubscriptionService).should().saveAttributesInternal(argThat(request ->
                     request.getEntityId().equals(deviceId) &&
                             request.getEntries().get(0).getKey().equals(INACTIVITY_ALARM_TIME) &&
                             request.getEntries().get(0).getValue().equals(0L)
@@ -761,7 +661,7 @@ public class DefaultDeviceStateServiceTest {
         }
 
         if (shouldUpdateActivityStateToActive) {
-            then(telemetrySubscriptionService).should().saveAttributes(argThat(request ->
+            then(telemetrySubscriptionService).should().saveAttributesInternal(argThat(request ->
                     request.getEntityId().equals(deviceId) &&
                             request.getEntries().get(0).getKey().equals(ACTIVITY_STATE) &&
                             request.getEntries().get(0).getValue().equals(true)
@@ -809,59 +709,8 @@ public class DefaultDeviceStateServiceTest {
         );
     }
 
-    @ParameterizedTest
-    @MethodSource("provideParametersForDecreaseInactivityTimeout")
-    public void givenTestParameters_whenOnDeviceInactivityTimeout_thenShouldBeInTheExpectedStateAndPerformExpectedActions(
-            boolean activityState, long newInactivityTimeout, long timeIncrement, boolean expectedActivityState
-    ) throws Exception {
-        // GIVEN
-        long defaultInactivityTimeout = 10000;
-        initStateService(defaultInactivityTimeout);
-
-        var currentTime = new AtomicLong(System.currentTimeMillis());
-
-        DeviceState deviceState = DeviceState.builder()
-                .active(activityState)
-                .lastActivityTime(currentTime.get())
-                .inactivityTimeout(defaultInactivityTimeout)
-                .build();
-
-        DeviceStateData deviceStateData = DeviceStateData.builder()
-                .tenantId(tenantId)
-                .deviceId(deviceId)
-                .state(deviceState)
-                .metaData(new TbMsgMetaData())
-                .build();
-
-        service.deviceStates.put(deviceId, deviceStateData);
-        service.getPartitionedEntities(tpi).add(deviceId);
-
-        given(service.getCurrentTimeMillis()).willReturn(currentTime.addAndGet(timeIncrement));
-
-        // WHEN
-        service.onDeviceInactivityTimeoutUpdate(tenantId, deviceId, newInactivityTimeout);
-
-        // THEN
-        assertThat(deviceState.getInactivityTimeout()).isEqualTo(newInactivityTimeout);
-        assertThat(deviceState.isActive()).isEqualTo(expectedActivityState);
-        if (activityState && !expectedActivityState) {
-            then(telemetrySubscriptionService).should().saveAttributes(argThat(request ->
-                    request.getEntityId().equals(deviceId) && request.getEntries().get(0).getKey().equals(ACTIVITY_STATE) &&
-                            request.getEntries().get(0).getValue().equals(false)
-            ));
-        }
-    }
-
-    private static Stream<Arguments> provideParametersForDecreaseInactivityTimeout() {
-        return Stream.of(
-                Arguments.of(true, 1, 0, true),
-
-                Arguments.of(true, 1, 1, false)
-        );
-    }
-
     @Test
-    public void givenStateDataIsNull_whenUpdateInactivityTimeoutIfExpired_thenShouldCleanupDevice() {
+    void givenStateDataIsNull_whenUpdateInactivityTimeoutIfExpired_thenShouldCleanupDevice() {
         // GIVEN
         service.deviceStates.put(deviceId, deviceStateDataMock);
 
@@ -875,7 +724,7 @@ public class DefaultDeviceStateServiceTest {
     }
 
     @Test
-    public void givenNotMyPartition_whenUpdateInactivityTimeoutIfExpired_thenShouldCleanupDevice() {
+    void givenNotMyPartition_whenUpdateInactivityTimeoutIfExpired_thenShouldCleanupDevice() {
         // GIVEN
         long currentTime = System.currentTimeMillis();
 
@@ -911,7 +760,7 @@ public class DefaultDeviceStateServiceTest {
 
     @ParameterizedTest
     @MethodSource("provideParametersForUpdateInactivityStateIfExpired")
-    public void givenTestParameters_whenUpdateInactivityStateIfExpired_thenShouldBeInTheExpectedStateAndPerformExpectedActions(
+    void givenTestParameters_whenUpdateInactivityStateIfExpired_thenShouldBeInTheExpectedStateAndPerformExpectedActions(
             boolean activityState, long ts, long lastActivityTime, long lastInactivityAlarmTime, long inactivityTimeout, long deviceCreationTime,
             boolean expectedActivityState, long expectedLastInactivityAlarmTime, boolean shouldUpdateActivityStateToInactive
     ) {
@@ -933,6 +782,7 @@ public class DefaultDeviceStateServiceTest {
 
         if (shouldUpdateActivityStateToInactive) {
             given(partitionService.resolve(ServiceType.TB_CORE, tenantId, deviceId)).willReturn(tpi);
+            mockSuccessfulSaveAttributes();
         }
 
         // WHEN
@@ -943,7 +793,7 @@ public class DefaultDeviceStateServiceTest {
         assertThat(state.getLastInactivityAlarmTime()).isEqualTo(expectedLastInactivityAlarmTime);
 
         if (shouldUpdateActivityStateToInactive) {
-            then(telemetrySubscriptionService).should().saveAttributes(argThat(request ->
+            then(telemetrySubscriptionService).should().saveAttributesInternal(argThat(request ->
                     request.getEntityId().equals(deviceId) && request.getEntries().get(0).getKey().equals(ACTIVITY_STATE) &&
                             request.getEntries().get(0).getValue().equals(false)
             ));
@@ -961,7 +811,7 @@ public class DefaultDeviceStateServiceTest {
             assertThat(actualNotification.getDeviceId()).isEqualTo(deviceId);
             assertThat(actualNotification.isActive()).isFalse();
 
-            then(telemetrySubscriptionService).should().saveAttributes(argThat(request ->
+            then(telemetrySubscriptionService).should().saveAttributesInternal(argThat(request ->
                     request.getTenantId().equals(tenantId) && request.getEntityId().equals(deviceId) &&
                             request.getScope().equals(AttributeScope.SERVER_SCOPE) &&
                             request.getEntries().get(0).getKey().equals(INACTIVITY_ALARM_TIME) &&
@@ -1033,7 +883,79 @@ public class DefaultDeviceStateServiceTest {
     }
 
     @Test
-    public void givenConcurrentAccess_whenGetOrFetchDeviceStateData_thenFetchDeviceStateDataInvokedOnce() {
+    void givenInactiveDevice_whenActivityStatusChangesToActiveButFailedToSaveUpdatedActivityStatus_thenShouldNotUpdateCache2() {
+        // GIVEN
+        var deviceState = DeviceState.builder()
+                .active(false)
+                .lastActivityTime(100L)
+                .inactivityTimeout(50L)
+                .build();
+
+        var deviceStateData = DeviceStateData.builder()
+                .tenantId(tenantId)
+                .deviceId(deviceId)
+                .state(deviceState)
+                .metaData(TbMsgMetaData.EMPTY)
+                .build();
+
+        service.deviceStates.put(deviceId, deviceStateData);
+        service.getPartitionedEntities(tpi).add(deviceId);
+
+        // WHEN-THEN
+
+        // simulating short DB outage
+        given(telemetrySubscriptionService.saveAttributesInternal(any())).willReturn(Futures.immediateFailedFuture(new RuntimeException("failed to save")));
+        doReturn(200L).when(service).getCurrentTimeMillis();
+        service.onDeviceActivity(tenantId, deviceId, 180L);
+        assertThat(deviceState.isActive()).isFalse(); // still inactive
+
+        // 10 millis pass... and new activity message it received
+
+        // this time DB save is successful
+        when(telemetrySubscriptionService.saveAttributesInternal(any())).thenReturn(Futures.immediateFuture(AttributesSaveResult.of(generateRandomVersions(1))));
+        doReturn(210L).when(service).getCurrentTimeMillis();
+        service.onDeviceActivity(tenantId, deviceId, 190L);
+        assertThat(deviceState.isActive()).isTrue();
+    }
+
+    @Test
+    void givenActiveDevice_whenActivityStatusChangesToInactiveButFailedToSaveUpdatedActivityStatus_thenShouldNotUpdateCache() {
+        // GIVEN
+        var deviceState = DeviceState.builder()
+                .active(true)
+                .lastActivityTime(100L)
+                .inactivityTimeout(50L)
+                .build();
+
+        var deviceStateData = DeviceStateData.builder()
+                .tenantId(tenantId)
+                .deviceId(deviceId)
+                .state(deviceState)
+                .metaData(TbMsgMetaData.EMPTY)
+                .build();
+
+        service.deviceStates.put(deviceId, deviceStateData);
+        service.getPartitionedEntities(tpi).add(deviceId);
+
+        // WHEN-THEN (assuming periodic activity states check is done every 100 millis)
+
+        // simulating short DB outage
+        given(telemetrySubscriptionService.saveAttributesInternal(any())).willReturn(Futures.immediateFailedFuture(new RuntimeException("failed to save")));
+        doReturn(200L).when(service).getCurrentTimeMillis();
+        service.checkStates();
+        assertThat(deviceState.isActive()).isTrue(); // still active
+
+        // waiting 100 millis... periodic activity states check is triggered again
+
+        // this time DB save is successful
+        when(telemetrySubscriptionService.saveAttributesInternal(any())).thenReturn(Futures.immediateFuture(AttributesSaveResult.of(generateRandomVersions(1))));
+        doReturn(300L).when(service).getCurrentTimeMillis();
+        service.checkStates();
+        assertThat(deviceState.isActive()).isFalse();
+    }
+
+    @Test
+    void givenConcurrentAccess_whenGetOrFetchDeviceStateData_thenFetchDeviceStateDataInvokedOnce() {
         doAnswer(invocation -> {
             Thread.sleep(100);
             return deviceStateDataMock;
@@ -1069,10 +991,8 @@ public class DefaultDeviceStateServiceTest {
     }
 
     @Test
-    public void givenDeviceAdded_whenOnQueueMsg_thenShouldCacheAndSaveActivityToFalse() throws InterruptedException {
+    void givenDeviceAdded_whenOnQueueMsg_thenShouldCacheAndSaveActivityToFalse() {
         // GIVEN
-        final long defaultTimeout = 1000;
-        initStateService(defaultTimeout);
         given(deviceService.findDeviceById(any(TenantId.class), any(DeviceId.class))).willReturn(new Device(deviceId));
         given(attributesService.find(any(TenantId.class), any(EntityId.class), any(AttributeScope.class), anyCollection())).willReturn(Futures.immediateFuture(Collections.emptyList()));
 
@@ -1086,13 +1006,15 @@ public class DefaultDeviceStateServiceTest {
                 .setDeleted(false)
                 .build();
 
+        mockSuccessfulSaveAttributes();
+
         // WHEN
         service.onQueueMsg(proto, TbCallback.EMPTY);
 
         // THEN
         await().atMost(1, TimeUnit.SECONDS).untilAsserted(() -> {
             assertThat(service.deviceStates.get(deviceId).getState().isActive()).isEqualTo(false);
-            then(telemetrySubscriptionService).should().saveAttributes(argThat(request ->
+            then(telemetrySubscriptionService).should().saveAttributesInternal(argThat(request ->
                     request.getEntityId().equals(deviceId) && request.getEntries().get(0).getKey().equals(ACTIVITY_STATE) &&
                             request.getEntries().get(0).getValue().equals(false)
             ));
@@ -1100,14 +1022,12 @@ public class DefaultDeviceStateServiceTest {
     }
 
     @Test
-    public void givenDeviceActivityEventHappenedAfterAdded_whenOnDeviceActivity_thenShouldCacheAndSaveActivityToTrue() throws InterruptedException {
+    void givenDeviceActivityEventHappenedAfterAdded_whenOnDeviceActivity_thenShouldCacheAndSaveActivityToTrue() {
         // GIVEN
-        final long defaultTimeout = 1000;
-        initStateService(defaultTimeout);
         long currentTime = System.currentTimeMillis();
         DeviceState deviceState = DeviceState.builder()
                 .active(false)
-                .inactivityTimeout(service.getDefaultInactivityTimeoutInSec())
+                .inactivityTimeout(defaultInactivityTimeoutMs)
                 .build();
         DeviceStateData stateData = DeviceStateData.builder()
                 .tenantId(tenantId)
@@ -1118,12 +1038,14 @@ public class DefaultDeviceStateServiceTest {
                 .build();
         service.deviceStates.put(deviceId, stateData);
 
+        mockSuccessfulSaveAttributes();
+
         // WHEN
         service.onDeviceActivity(tenantId, deviceId, currentTime);
 
         // THEN
         ArgumentCaptor<AttributesSaveRequest> attributeRequestCaptor = ArgumentCaptor.forClass(AttributesSaveRequest.class);
-        then(telemetrySubscriptionService).should(times(2)).saveAttributes(attributeRequestCaptor.capture());
+        then(telemetrySubscriptionService).should(times(2)).saveAttributesInternal(attributeRequestCaptor.capture());
 
         await().atMost(1, TimeUnit.SECONDS).untilAsserted(() -> {
             assertThat(service.deviceStates.get(deviceId).getState().isActive()).isEqualTo(true);
@@ -1151,15 +1073,14 @@ public class DefaultDeviceStateServiceTest {
     }
 
     @Test
-    public void givenDeviceActivityEventHappenedBeforeAdded_whenOnQueueMsg_thenShouldSaveActivityStateUsingValueFromCache() throws InterruptedException {
+    void givenDeviceActivityEventHappenedBeforeAdded_whenOnQueueMsg_thenShouldSaveActivityStateUsingValueFromCache() {
         // GIVEN
-        final long defaultTimeout = 1000;
-        initStateService(defaultTimeout);
         given(deviceService.findDeviceById(any(TenantId.class), any(DeviceId.class))).willReturn(new Device(deviceId));
         given(attributesService.find(any(TenantId.class), any(EntityId.class), any(AttributeScope.class), anyCollection())).willReturn(Futures.immediateFuture(Collections.emptyList()));
 
         long currentTime = System.currentTimeMillis();
-        DeviceState deviceState = DeviceState.builder()
+
+        var deviceState = DeviceState.builder()
                 .active(true)
                 .lastConnectTime(currentTime - 8000)
                 .lastActivityTime(currentTime - 4000)
@@ -1167,16 +1088,20 @@ public class DefaultDeviceStateServiceTest {
                 .lastInactivityAlarmTime(0)
                 .inactivityTimeout(3000)
                 .build();
-        DeviceStateData stateData = DeviceStateData.builder()
+
+        var stateData = DeviceStateData.builder()
                 .tenantId(tenantId)
                 .deviceId(deviceId)
                 .deviceCreationTime(currentTime - 10000)
                 .state(deviceState)
                 .build();
+
         service.deviceStates.put(deviceId, stateData);
 
+        mockSuccessfulSaveAttributes();
+
         // WHEN
-        TransportProtos.DeviceStateServiceMsgProto proto = TransportProtos.DeviceStateServiceMsgProto.newBuilder()
+        var proto = TransportProtos.DeviceStateServiceMsgProto.newBuilder()
                 .setTenantIdMSB(tenantId.getId().getMostSignificantBits())
                 .setTenantIdLSB(tenantId.getId().getLeastSignificantBits())
                 .setDeviceIdMSB(deviceId.getId().getMostSignificantBits())
@@ -1190,11 +1115,25 @@ public class DefaultDeviceStateServiceTest {
         // THEN
         await().atMost(1, TimeUnit.SECONDS).untilAsserted(() -> {
             assertThat(service.deviceStates.get(deviceId).getState().isActive()).isEqualTo(true);
-            then(telemetrySubscriptionService).should().saveAttributes(argThat(request ->
+            then(telemetrySubscriptionService).should().saveAttributesInternal(argThat(request ->
                     request.getEntityId().equals(deviceId) && request.getEntries().get(0).getKey().equals(ACTIVITY_STATE) &&
                             request.getEntries().get(0).getValue().equals(true)
             ));
         });
     }
 
+    private void mockSuccessfulSaveAttributes() {
+        lenient().when(telemetrySubscriptionService.saveAttributesInternal(any())).thenAnswer(invocation -> {
+            AttributesSaveRequest request = invocation.getArgument(0);
+            return Futures.immediateFuture(generateRandomVersions(request.getEntries().size()));
+        });
+    }
+
+    private static List<Long> generateRandomVersions(int n) {
+        return ThreadLocalRandom.current()
+                .longs(n)
+                .boxed()
+                .toList();
+    }
+
 }

application/src/test/java/org/thingsboard/server/service/telemetry/DefaultTelemetrySubscriptionServiceTest.java

@@ -48,6 +48,7 @@ import org.thingsboard.server.common.data.id.EntityIdFactory;
 import org.thingsboard.server.common.data.id.EntityViewId;
 import org.thingsboard.server.common.data.id.TenantId;
 import org.thingsboard.server.common.data.kv.AttributeKvEntry;
+import org.thingsboard.server.common.data.kv.AttributesSaveResult;
 import org.thingsboard.server.common.data.kv.BaseAttributeKvEntry;
 import org.thingsboard.server.common.data.kv.BasicTsKvEntry;
 import org.thingsboard.server.common.data.kv.DoubleDataEntry;
@@ -472,7 +473,8 @@ class DefaultTelemetrySubscriptionServiceTest {
                 .strategy(new AttributesSaveRequest.Strategy(saveAttributes, sendWsUpdate, processCalculatedFields))
                 .build();
 
-        lenient().when(attrService.save(tenantId, entityId, request.getScope(), request.getEntries())).thenReturn(immediateFuture(listOfNNumbers(request.getEntries().size())));
+        lenient().when(attrService.save(tenantId, entityId, request.getScope(), request.getEntries()))
+                .thenReturn(immediateFuture(AttributesSaveResult.of(listOfNNumbers(request.getEntries().size()))));
 
         // WHEN
         telemetryService.saveAttributes(request);
@@ -547,7 +549,8 @@ class DefaultTelemetrySubscriptionServiceTest {
                 .strategy(new AttributesSaveRequest.Strategy(true, false, false))
                 .build();
 
-        given(attrService.save(tenantId, deviceId, request.getScope(), entries)).willReturn(immediateFuture(listOfNNumbers(entries.size())));
+        given(attrService.save(tenantId, deviceId, request.getScope(), entries))
+                .willReturn(immediateFuture(AttributesSaveResult.of(listOfNNumbers(entries.size()))));
 
         // WHEN
         telemetryService.saveAttributes(request);
@@ -581,7 +584,8 @@ class DefaultTelemetrySubscriptionServiceTest {
                 .strategy(new AttributesSaveRequest.Strategy(true, false, false))
                 .build();
 
-        given(attrService.save(tenantId, nonDeviceId, request.getScope(), entries)).willReturn(immediateFuture(listOfNNumbers(entries.size())));
+        given(attrService.save(tenantId, nonDeviceId, request.getScope(), entries))
+                .willReturn(immediateFuture(AttributesSaveResult.of(listOfNNumbers(entries.size()))));
 
         // WHEN
         telemetryService.saveAttributes(request);
@@ -613,7 +617,8 @@ class DefaultTelemetrySubscriptionServiceTest {
                 .strategy(new AttributesSaveRequest.Strategy(true, false, false))
                 .build();
 
-        given(attrService.save(tenantId, deviceId, request.getScope(), entries)).willReturn(immediateFuture(listOfNNumbers(entries.size())));
+        given(attrService.save(tenantId, deviceId, request.getScope(), entries))
+                .willReturn(immediateFuture(AttributesSaveResult.of(listOfNNumbers(entries.size()))));
 
         // WHEN
         telemetryService.saveAttributes(request);
@@ -640,7 +645,8 @@ class DefaultTelemetrySubscriptionServiceTest {
                 .strategy(new AttributesSaveRequest.Strategy(true, false, false))
                 .build();
 
-        given(attrService.save(tenantId, deviceId, request.getScope(), entries)).willReturn(immediateFuture(listOfNNumbers(entries.size())));
+        given(attrService.save(tenantId, deviceId, request.getScope(), entries))
+                .willReturn(immediateFuture(AttributesSaveResult.of(listOfNNumbers(entries.size()))));
 
         // WHEN
         telemetryService.saveAttributes(request);
@@ -715,7 +721,8 @@ class DefaultTelemetrySubscriptionServiceTest {
                 .strategy(new AttributesSaveRequest.Strategy(true, false, false))
                 .build();
 
-        given(attrService.save(tenantId, deviceId, request.getScope(), request.getEntries())).willReturn(immediateFuture(listOfNNumbers(request.getEntries().size())));
+        given(attrService.save(tenantId, deviceId, request.getScope(), request.getEntries()))
+                .willReturn(immediateFuture(AttributesSaveResult.of(listOfNNumbers(request.getEntries().size()))));
 
         // WHEN
         telemetryService.saveAttributes(request);
@@ -764,7 +771,8 @@ class DefaultTelemetrySubscriptionServiceTest {
                 .strategy(new AttributesSaveRequest.Strategy(true, false, false))
                 .build();
 
-        given(attrService.save(tenantId, nonDeviceId, request.getScope(), request.getEntries())).willReturn(immediateFuture(listOfNNumbers(request.getEntries().size())));
+        given(attrService.save(tenantId, nonDeviceId, request.getScope(), request.getEntries()))
+                .willReturn(immediateFuture(AttributesSaveResult.of(listOfNNumbers(request.getEntries().size()))));
 
         // WHEN
         telemetryService.saveAttributes(request);
@@ -792,7 +800,8 @@ class DefaultTelemetrySubscriptionServiceTest {
                 .strategy(new AttributesSaveRequest.Strategy(true, false, false))
                 .build();
 
-        given(attrService.save(tenantId, deviceId, request.getScope(), request.getEntries())).willReturn(immediateFuture(listOfNNumbers(request.getEntries().size())));
+        given(attrService.save(tenantId, deviceId, request.getScope(), request.getEntries()))
+                .willReturn(immediateFuture(AttributesSaveResult.of(listOfNNumbers(request.getEntries().size()))));
 
         // WHEN
         telemetryService.saveAttributes(request);
@@ -815,7 +824,8 @@ class DefaultTelemetrySubscriptionServiceTest {
                 .strategy(new AttributesSaveRequest.Strategy(true, false, false))
                 .build();
 
-        given(attrService.save(tenantId, deviceId, request.getScope(), request.getEntries())).willReturn(immediateFuture(listOfNNumbers(request.getEntries().size())));
+        given(attrService.save(tenantId, deviceId, request.getScope(), request.getEntries()))
+                .willReturn(immediateFuture(AttributesSaveResult.of(listOfNNumbers(request.getEntries().size()))));
 
         // WHEN
         telemetryService.saveAttributes(request);
@@ -843,7 +853,8 @@ class DefaultTelemetrySubscriptionServiceTest {
                 .strategy(new AttributesSaveRequest.Strategy(true, false, false))
                 .build();
 
-        given(attrService.save(tenantId, deviceId, request.getScope(), request.getEntries())).willReturn(immediateFuture(listOfNNumbers(request.getEntries().size())));
+        given(attrService.save(tenantId, deviceId, request.getScope(), request.getEntries()))
+                .willReturn(immediateFuture(AttributesSaveResult.of(listOfNNumbers(request.getEntries().size()))));
 
         // WHEN
         telemetryService.saveAttributes(request);
@@ -870,7 +881,8 @@ class DefaultTelemetrySubscriptionServiceTest {
                 .strategy(new AttributesSaveRequest.Strategy(true, false, false))
                 .build();
 
-        given(attrService.save(tenantId, deviceId, request.getScope(), request.getEntries())).willReturn(immediateFuture(listOfNNumbers(request.getEntries().size())));
+        given(attrService.save(tenantId, deviceId, request.getScope(), request.getEntries()))
+                .willReturn(immediateFuture(AttributesSaveResult.of(listOfNNumbers(request.getEntries().size()))));
 
         // WHEN
         telemetryService.saveAttributes(request);

common/dao-api/src/main/java/org/thingsboard/server/dao/attributes/AttributesService.java

@@ -21,6 +21,7 @@ import org.thingsboard.server.common.data.id.DeviceProfileId;
 import org.thingsboard.server.common.data.id.EntityId;
 import org.thingsboard.server.common.data.id.TenantId;
 import org.thingsboard.server.common.data.kv.AttributeKvEntry;
+import org.thingsboard.server.common.data.kv.AttributesSaveResult;
 
 import java.util.Collection;
 import java.util.List;
@@ -37,9 +38,9 @@ public interface AttributesService {
 
     ListenableFuture<List<AttributeKvEntry>> findAll(TenantId tenantId, EntityId entityId, AttributeScope scope);
 
-    ListenableFuture<List<Long>> save(TenantId tenantId, EntityId entityId, AttributeScope scope, List<AttributeKvEntry> attributes);
+    ListenableFuture<AttributesSaveResult> save(TenantId tenantId, EntityId entityId, AttributeScope scope, List<AttributeKvEntry> attributes);
 
-    ListenableFuture<Long> save(TenantId tenantId, EntityId entityId, AttributeScope scope, AttributeKvEntry attribute);
+    ListenableFuture<AttributesSaveResult> save(TenantId tenantId, EntityId entityId, AttributeScope scope, AttributeKvEntry attribute);
 
     ListenableFuture<List<String>> removeAll(TenantId tenantId, EntityId entityId, AttributeScope scope, List<String> attributeKeys);
 

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

@@ -23,7 +23,7 @@ import org.thingsboard.server.common.data.cf.CalculatedFieldType;
 @EqualsAndHashCode(callSuper = true)
 public class SimpleCalculatedFieldConfiguration extends BaseCalculatedFieldConfiguration implements CalculatedFieldConfiguration {
 
-    private boolean preserveMsgTs;
+    private boolean useLatestTs;
 
     @Override
     public CalculatedFieldType getType() {

common/data/src/main/java/org/thingsboard/server/common/data/kv/AttributesSaveResult.java

@@ -0,0 +1,32 @@
+/**
+ * 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.data.kv;
+
+import java.util.Collections;
+import java.util.List;
+
+public record AttributesSaveResult(List<Long> versions) {
+
+    public static final AttributesSaveResult EMPTY = new AttributesSaveResult(Collections.emptyList());
+
+    public static AttributesSaveResult of(List<Long> versions) {
+        if (versions == null) {
+            return EMPTY;
+        }
+        return new AttributesSaveResult(versions);
+    }
+
+}

common/script/script-api/src/main/java/org/thingsboard/script/api/tbel/TbelCfCtx.java

@@ -25,11 +25,11 @@ public class TbelCfCtx implements TbelCfObject {
     @Getter
     private final Map<String, TbelCfArg> args;
     @Getter
-    private final long msgTs;
+    private final long latestTs;
 
-    public TbelCfCtx(Map<String, TbelCfArg> args, long lastUpdateTs) {
+    public TbelCfCtx(Map<String, TbelCfArg> args, long latestTs) {
         this.args = Collections.unmodifiableMap(args);
-        this.msgTs = lastUpdateTs != -1 ? lastUpdateTs : System.currentTimeMillis();
+        this.latestTs = latestTs != -1 ? latestTs : System.currentTimeMillis();
     }
 
     @Override

dao/src/main/java/org/thingsboard/server/dao/attributes/BaseAttributesService.java

@@ -33,6 +33,7 @@ import org.thingsboard.server.common.data.id.DeviceProfileId;
 import org.thingsboard.server.common.data.id.EntityId;
 import org.thingsboard.server.common.data.id.TenantId;
 import org.thingsboard.server.common.data.kv.AttributeKvEntry;
+import org.thingsboard.server.common.data.kv.AttributesSaveResult;
 import org.thingsboard.server.common.data.util.TbPair;
 import org.thingsboard.server.common.msg.edqs.EdqsService;
 import org.thingsboard.server.dao.service.Validator;
@@ -41,7 +42,6 @@ import java.util.ArrayList;
 import java.util.Collection;
 import java.util.List;
 import java.util.Optional;
-import java.util.stream.Collectors;
 
 import static org.thingsboard.server.dao.attributes.AttributeUtils.validate;
 
@@ -101,26 +101,29 @@ public class BaseAttributesService implements AttributesService {
     }
 
     @Override
-    public ListenableFuture<Long> save(TenantId tenantId, EntityId entityId, AttributeScope scope, AttributeKvEntry attribute) {
+    public ListenableFuture<AttributesSaveResult> save(TenantId tenantId, EntityId entityId, AttributeScope scope, AttributeKvEntry attribute) {
         validate(entityId, scope);
         AttributeUtils.validate(attribute, valueNoXssValidation);
-        return doSave(tenantId, entityId, scope, attribute);
+        return doSave(tenantId, entityId, scope, List.of(attribute));
     }
 
     @Override
-    public ListenableFuture<List<Long>> save(TenantId tenantId, EntityId entityId, AttributeScope scope, List<AttributeKvEntry> attributes) {
+    public ListenableFuture<AttributesSaveResult> save(TenantId tenantId, EntityId entityId, AttributeScope scope, List<AttributeKvEntry> attributes) {
         validate(entityId, scope);
         AttributeUtils.validate(attributes, valueNoXssValidation);
-        List<ListenableFuture<Long>> saveFutures = attributes.stream().map(attribute -> doSave(tenantId, entityId, scope, attribute)).collect(Collectors.toList());
-        return Futures.allAsList(saveFutures);
+        return doSave(tenantId, entityId, scope, attributes);
     }
 
-    private ListenableFuture<Long> doSave(TenantId tenantId, EntityId entityId, AttributeScope scope, AttributeKvEntry attribute) {
-        ListenableFuture<Long> future = attributesDao.save(tenantId, entityId, scope, attribute);
-        return Futures.transform(future, version -> {
-            edqsService.onUpdate(tenantId, ObjectType.ATTRIBUTE_KV, new AttributeKv(entityId, scope, attribute, version));
-            return version;
-        }, MoreExecutors.directExecutor());
+    private ListenableFuture<AttributesSaveResult> doSave(TenantId tenantId, EntityId entityId, AttributeScope scope, List<AttributeKvEntry> attributes) {
+        List<ListenableFuture<Long>> futures = new ArrayList<>(attributes.size());
+        for (AttributeKvEntry attribute : attributes) {
+            ListenableFuture<Long> future = Futures.transform(attributesDao.save(tenantId, entityId, scope, attribute), version -> {
+                edqsService.onUpdate(tenantId, ObjectType.ATTRIBUTE_KV, new AttributeKv(entityId, scope, attribute, version));
+                return version;
+            }, MoreExecutors.directExecutor());
+            futures.add(future);
+        }
+        return Futures.transform(Futures.allAsList(futures), AttributesSaveResult::of, MoreExecutors.directExecutor());
     }
 
     @Override

dao/src/main/java/org/thingsboard/server/dao/attributes/CachedAttributesService.java

@@ -37,6 +37,7 @@ import org.thingsboard.server.common.data.id.DeviceProfileId;
 import org.thingsboard.server.common.data.id.EntityId;
 import org.thingsboard.server.common.data.id.TenantId;
 import org.thingsboard.server.common.data.kv.AttributeKvEntry;
+import org.thingsboard.server.common.data.kv.AttributesSaveResult;
 import org.thingsboard.server.common.data.kv.BaseAttributeKvEntry;
 import org.thingsboard.server.common.data.util.TbPair;
 import org.thingsboard.server.common.msg.edqs.EdqsService;
@@ -56,7 +57,6 @@ import java.util.Map;
 import java.util.Objects;
 import java.util.Optional;
 import java.util.Set;
-import java.util.stream.Collectors;
 
 import static org.thingsboard.server.dao.attributes.AttributeUtils.validate;
 
@@ -150,7 +150,7 @@ public class CachedAttributesService implements AttributesService {
                     List<AttributeKvEntry> cachedAttributes = wrappedCachedAttributes.values().stream()
                             .map(TbCacheValueWrapper::get)
                             .filter(Objects::nonNull)
-                            .collect(Collectors.toList());
+                            .toList();
                     if (wrappedCachedAttributes.size() == attributeKeys.size()) {
                         log.trace("[{}][{}] Found all attributes from cache: {}", entityId, scope, attributeKeys);
                         return Futures.immediateFuture(cachedAttributes);
@@ -159,8 +159,6 @@ public class CachedAttributesService implements AttributesService {
                     Set<String> notFoundAttributeKeys = new HashSet<>(attributeKeys);
                     notFoundAttributeKeys.removeAll(wrappedCachedAttributes.keySet());
 
-                    List<AttributeCacheKey> notFoundKeys = notFoundAttributeKeys.stream().map(k -> new AttributeCacheKey(scope, entityId, k)).collect(Collectors.toList());
-
                     // DB call should run in DB executor, not in cache-related executor
                     return jpaExecutorService.submit(() -> {
                         log.trace("[{}][{}] Lookup attributes from db: {}", entityId, scope, notFoundAttributeKeys);
@@ -222,33 +220,31 @@ public class CachedAttributesService implements AttributesService {
     }
 
     @Override
-    public ListenableFuture<Long> save(TenantId tenantId, EntityId entityId, AttributeScope scope, AttributeKvEntry attribute) {
+    public ListenableFuture<AttributesSaveResult> save(TenantId tenantId, EntityId entityId, AttributeScope scope, AttributeKvEntry attribute) {
         validate(entityId, scope);
         AttributeUtils.validate(attribute, valueNoXssValidation);
-        return doSave(tenantId, entityId, scope, attribute);
+        return doSave(tenantId, entityId, scope, List.of(attribute));
     }
 
     @Override
-    public ListenableFuture<List<Long>> save(TenantId tenantId, EntityId entityId, AttributeScope scope, List<AttributeKvEntry> attributes) {
+    public ListenableFuture<AttributesSaveResult> save(TenantId tenantId, EntityId entityId, AttributeScope scope, List<AttributeKvEntry> attributes) {
         validate(entityId, scope);
         AttributeUtils.validate(attributes, valueNoXssValidation);
+        return doSave(tenantId, entityId, scope, attributes);
+    }
 
+    private ListenableFuture<AttributesSaveResult> doSave(TenantId tenantId, EntityId entityId, AttributeScope scope, List<AttributeKvEntry> attributes) {
         List<ListenableFuture<Long>> futures = new ArrayList<>(attributes.size());
         for (var attribute : attributes) {
-            futures.add(doSave(tenantId, entityId, scope, attribute));
+            ListenableFuture<Long> future = Futures.transform(attributesDao.save(tenantId, entityId, scope, attribute), version -> {
+                BaseAttributeKvEntry attributeKvEntry = new BaseAttributeKvEntry(((BaseAttributeKvEntry) attribute).getKv(), attribute.getLastUpdateTs(), version);
+                put(entityId, scope, attributeKvEntry);
+                edqsService.onUpdate(tenantId, ObjectType.ATTRIBUTE_KV, new AttributeKv(entityId, scope, attributeKvEntry, version));
+                return version;
+            }, cacheExecutor);
+            futures.add(future);
         }
-
-        return Futures.allAsList(futures);
-    }
-
-    private ListenableFuture<Long> doSave(TenantId tenantId, EntityId entityId, AttributeScope scope, AttributeKvEntry attribute) {
-        ListenableFuture<Long> future = attributesDao.save(tenantId, entityId, scope, attribute);
-        return Futures.transform(future, version -> {
-            BaseAttributeKvEntry attributeKvEntry = new BaseAttributeKvEntry(((BaseAttributeKvEntry) attribute).getKv(), attribute.getLastUpdateTs(), version);
-            put(entityId, scope, attributeKvEntry);
-            edqsService.onUpdate(tenantId, ObjectType.ATTRIBUTE_KV, new AttributeKv(entityId, scope, attributeKvEntry, version));
-            return version;
-        }, cacheExecutor);
+        return Futures.transform(Futures.allAsList(futures), AttributesSaveResult::of, MoreExecutors.directExecutor());
     }
 
     private void put(EntityId entityId, AttributeScope scope, AttributeKvEntry attribute) {
@@ -270,7 +266,7 @@ public class CachedAttributesService implements AttributesService {
                 edqsService.onDelete(tenantId, ObjectType.ATTRIBUTE_KV, new AttributeKv(entityId, scope, key, version));
             }
             return key;
-        }, cacheExecutor)).collect(Collectors.toList()));
+        }, cacheExecutor)).toList());
     }
 
     @Override

ui-ngx/src/app/modules/home/components/calculated-fields/components/dialog/calculated-field-dialog.component.html

@@ -190,9 +190,9 @@
               </mat-form-field>
             </div>
             <div class="tb-form-row" [formGroup]="configFormGroup" *ngIf="outputFormGroup.get('type').value === OutputType.Timeseries">
-              <mat-slide-toggle class="mat-slide" formControlName="preserveMsgTs">
-                <div tb-hint-tooltip-icon="{{ 'calculated-fields.hint.use-message-timestamp' | translate }}" translate>
-                  calculated-fields.use-message-timestamp
+              <mat-slide-toggle class="mat-slide" formControlName="useLatestTs">
+                <div tb-hint-tooltip-icon="{{ 'calculated-fields.hint.use-latest-timestamp' | translate }}" translate>
+                  calculated-fields.use-latest-timestamp
                 </div>
               </mat-slide-toggle>
             </div>

ui-ngx/src/app/modules/home/components/calculated-fields/components/dialog/calculated-field-dialog.component.scss

@@ -45,7 +45,7 @@
       &-key {
         color: #c24c1a;
       }
-      &-time-window, &-values, &-func, &-value, &-ts, &-msgTs {
+      &-time-window, &-values, &-func, &-value, &-ts, &-latestTs {
         color: #7214D0;
       }
       &-start-ts, &-end-ts {

ui-ngx/src/app/modules/home/components/calculated-fields/components/dialog/calculated-field-dialog.component.ts

@@ -77,7 +77,7 @@ export class CalculatedFieldDialogComponent extends DialogComponent<CalculatedFi
         type: [OutputType.Timeseries],
         decimalsByDefault: [null as number, [Validators.min(0), Validators.max(15), Validators.pattern(digitsRegex)]],
       }),
-      preserveMsgTs: [false]
+      useLatestTs: [false]
     }),
   });
 
@@ -212,12 +212,12 @@ export class CalculatedFieldDialogComponent extends DialogComponent<CalculatedFi
     }
     if (this.fieldFormGroup.get('type').value === CalculatedFieldType.SIMPLE) {
       if (type === OutputType.Attribute) {
-        this.configFormGroup.get('preserveMsgTs').disable({emitEvent: false});
+        this.configFormGroup.get('useLatestTs').disable({emitEvent: false});
       } else {
-        this.configFormGroup.get('preserveMsgTs').enable({emitEvent: false});
+        this.configFormGroup.get('useLatestTs').enable({emitEvent: false});
       }
     } else {
-      this.configFormGroup.get('preserveMsgTs').disable({emitEvent: false});
+      this.configFormGroup.get('useLatestTs').disable({emitEvent: false});
     }
   }
 
@@ -227,13 +227,13 @@ export class CalculatedFieldDialogComponent extends DialogComponent<CalculatedFi
       this.configFormGroup.get('expressionSIMPLE').enable({emitEvent: false});
       this.configFormGroup.get('expressionSCRIPT').disable({emitEvent: false});
       if (this.outputFormGroup.get('type').value === OutputType.Attribute) {
-        this.configFormGroup.get('preserveMsgTs').disable({emitEvent: false});
+        this.configFormGroup.get('useLatestTs').disable({emitEvent: false});
       } else {
-        this.configFormGroup.get('preserveMsgTs').enable({emitEvent: false});
+        this.configFormGroup.get('useLatestTs').enable({emitEvent: false});
       }
     } else {
       this.outputFormGroup.get('name').disable({emitEvent: false});
-      this.configFormGroup.get('preserveMsgTs').disable({emitEvent: false});
+      this.configFormGroup.get('useLatestTs').disable({emitEvent: false});
       this.configFormGroup.get('expressionSIMPLE').disable({emitEvent: false});
       this.configFormGroup.get('expressionSCRIPT').enable({emitEvent: false});
     }

ui-ngx/src/app/shared/models/calculated-field.models.ts

@@ -526,10 +526,10 @@ export const getCalculatedFieldArgumentsEditorCompleter = (argumentsObj: Record<
           description: 'Calculated field context arguments.',
           children: {}
         },
-        msgTs: {
+        latestTs: {
           meta: 'constant',
           type: 'number',
-          description: 'Timestamp (ms) of the telemetry message that triggered the calculated field execution.'
+          description: 'Latest timestamp (ms) of the arguments telemetry.'
         }
       }
     }
@@ -582,8 +582,8 @@ const calculatedFieldArgumentsContextValueHighlightRules: AceHighlightRules = {
       next: 'calculatedFieldCtxArgs'
     },
     {
-      token: 'tb.calculated-field-msgTs',
-      regex: /msgTs/,
+      token: 'tb.calculated-field-latestTs',
+      regex: /latestTs/,
       next: 'no_regex'
     },
     endGroupHighlightRule

ui-ngx/src/assets/help/en_US/calculated-field/expression_fn.md

@@ -1,7 +1,7 @@
 ## Calculated Field TBEL Script Function
 
 The **calculate()** function is a user-defined script that enables custom calculations using [TBEL](${siteBaseUrl}/docs${docPlatformPrefix}/user-guide/tbel/) on telemetry and attribute data.
-It receives arguments configured in the calculated field setup, along with an additional `ctx` object that stores `msgTs` and provides access to all arguments.
+It receives arguments configured in the calculated field setup, along with an additional `ctx` object that stores `latestTs` and provides access to all arguments.
 
 ### Function Signature
 
@@ -216,14 +216,14 @@ The return format depends on the output type configured in the calculated field
 
 ### Message timestamp
 
-The `ctx` object also includes property `msgTs`, which represents the timestamp of the incoming telemetry message that triggered the calculated field execution in milliseconds.
+The `ctx` object also includes property `latestTs`, which represents the latest timestamp of the arguments telemetry in milliseconds.
 
-You can use `ctx.msgTs` to set the timestamp of the resulting output explicitly when returning a time series object.
+You can use `ctx.latestTs` to set the timestamp of the resulting output explicitly when returning a time series object.
 
 ```javascript
 var temperatureC = (temperatureF - 32) / 1.8;
 return {
-  ts: ctx.msgTs,
+  ts: ctx.latestTs,
   values: {
     "temperatureC": toFixed(temperatureC, 2)
   }

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

@@ -1069,7 +1069,7 @@
         "delete-multiple-title": "Are you sure you want to delete { count, plural, =1 {1 calculated field} other {# calculated fields} }?",
         "delete-multiple-text": "Be careful, after the confirmation all selected calculated fields will be removed and all related data will become unrecoverable.",
         "test-with-this-message": "Test with this message",
-        "use-message-timestamp": "Use message timestamp",
+        "use-latest-timestamp": "Use latest timestamp",
         "hint": {
             "arguments-simple-with-rolling": "Simple type calculated field should not contain keys with time series rolling type.",
             "arguments-empty": "Arguments should not be empty.",
@@ -1086,7 +1086,7 @@
             "decimals-range": "Decimals by default should be a number between 0 and 15.",
             "expression": "Default expression demonstrates how to transform a temperature from Fahrenheit to Celsius.",
             "arguments-entity-not-found": "Argument target entity not found.",
-            "use-message-timestamp": "If enabled, the calculated value will be persisted using the timestamp of the telemetry that triggered the calculation, instead of the server time."
+            "use-latest-timestamp": "If enabled, the calculated value will be persisted using the most recent timestamp from the arguments telemetry, instead of the server time."
         }
     },
     "confirm-on-exit": {