Merge pull request #14757 from irynamatveieva/fix/entity-agg-cf-metric-ts

Time series data aggregation: share current ts across metrics during calculation

application/src/main/java/org/thingsboard/server/service/cf/ctx/state/aggregation/single/EntityAggregationCalculatedFieldState.java

@@ -68,6 +68,8 @@ public class EntityAggregationCalculatedFieldState extends BaseCalculatedFieldSt
 
     private CalculatedFieldProcessingService cfProcessingService;
 
+    private long now;
+
     public EntityAggregationCalculatedFieldState(EntityId entityId) {
         super(entityId);
     }
@@ -100,7 +102,7 @@ public class EntityAggregationCalculatedFieldState extends BaseCalculatedFieldSt
     @Override
     public ListenableFuture<CalculatedFieldResult> performCalculation(Map<String, ArgumentEntry> updatedArgs, CalculatedFieldCtx ctx) throws Exception {
         createIntervalIfNotExist();
-        long now = System.currentTimeMillis();
+        now = System.currentTimeMillis();
 
         if (DebugModeUtil.isDebugFailuresAvailable(ctx.getCalculatedField())) {
             if (debugTracker == null) {
@@ -114,7 +116,7 @@ public class EntityAggregationCalculatedFieldState extends BaseCalculatedFieldSt
         Map<AggIntervalEntry, Map<String, ArgumentEntry>> results = new HashMap<>();
         List<AggIntervalEntry> expiredIntervals = new ArrayList<>();
         getIntervals().forEach((intervalEntry, argIntervalStatuses) -> {
-            processInterval(now, intervalEntry, argIntervalStatuses, expiredIntervals, results);
+            processInterval(intervalEntry, argIntervalStatuses, expiredIntervals, results);
         });
         removeExpiredIntervals(expiredIntervals);
 
@@ -202,8 +204,7 @@ public class EntityAggregationCalculatedFieldState extends BaseCalculatedFieldSt
         return intervals;
     }
 
-    private void processInterval(long now,
-                                 AggIntervalEntry intervalEntry,
+    private void processInterval(AggIntervalEntry intervalEntry,
                                  Map<String, AggIntervalEntryStatus> args,
                                  List<AggIntervalEntry> expiredIntervals,
                                  Map<AggIntervalEntry, Map<String, ArgumentEntry>> results) {
@@ -244,11 +245,11 @@ public class EntityAggregationCalculatedFieldState extends BaseCalculatedFieldSt
         args.forEach((argName, argEntryIntervalStatus) -> {
             if (argEntryIntervalStatus.intervalPassed(cfCheckInterval)) {
                 if (argEntryIntervalStatus.argsUpdated()) {
-                    argEntryIntervalStatus.setLastMetricsEvalTs(System.currentTimeMillis());
+                    argEntryIntervalStatus.setLastMetricsEvalTs(now);
                     argEntryIntervalStatus.setLastArgsRefreshTs(DEFAULT_LAST_UPDATE_TS);
                     processArgument(intervalEntry, argName, false, results);
                 } else if (argEntryIntervalStatus.getLastMetricsEvalTs() == DEFAULT_LAST_UPDATE_TS) {
-                    argEntryIntervalStatus.setLastMetricsEvalTs(System.currentTimeMillis());
+                    argEntryIntervalStatus.setLastMetricsEvalTs(now);
                     processArgument(intervalEntry, argName, true, results);
                 }
             }