Set correct latest ts for key states in case of timeseries aggregation query

application/src/main/java/org/thingsboard/server/service/subscription/TbAbstractDataSubCtx.java

@@ -18,6 +18,7 @@ package org.thingsboard.server.service.subscription;
 import lombok.Getter;
 import lombok.extern.slf4j.Slf4j;
 import org.thingsboard.server.common.data.id.EntityId;
+import org.thingsboard.server.common.data.kv.Aggregation;
 import org.thingsboard.server.common.data.page.PageData;
 import org.thingsboard.server.common.data.query.AbstractDataQuery;
 import org.thingsboard.server.common.data.query.EntityData;
@@ -197,6 +198,9 @@ public abstract class TbAbstractDataSubCtx<T extends AbstractDataQuery<? extends
             entityData.getTimeseries().forEach((k, v) -> {
                 long ts = Arrays.stream(v).map(TsValue::getTs).max(Long::compareTo).orElse(0L);
                 log.trace("[{}][{}] Updating key: {} with ts: {}", serviceId, cmdId, k, ts);
+                if (!Aggregation.NONE.equals(getCurrentAggregation()) && ts < endTs) {
+                    ts = endTs;
+                }
                 keyStates.put(k, ts);
             });
         }
@@ -247,4 +251,5 @@ public abstract class TbAbstractDataSubCtx<T extends AbstractDataQuery<? extends
 
     abstract void sendWsMsg(String sessionId, TelemetrySubscriptionUpdate subscriptionUpdate, EntityKeyType keyType, boolean resultToLatestValues);
 
+    protected abstract Aggregation getCurrentAggregation();
 }

application/src/main/java/org/thingsboard/server/service/subscription/TbAlarmDataSubCtx.java

@@ -23,6 +23,7 @@ import org.thingsboard.server.common.data.alarm.Alarm;
 import org.thingsboard.server.common.data.alarm.AlarmSearchStatus;
 import org.thingsboard.server.common.data.id.AlarmId;
 import org.thingsboard.server.common.data.id.EntityId;
+import org.thingsboard.server.common.data.kv.Aggregation;
 import org.thingsboard.server.common.data.page.PageData;
 import org.thingsboard.server.common.data.query.AlarmData;
 import org.thingsboard.server.common.data.query.AlarmDataPageLink;
@@ -205,6 +206,11 @@ public class TbAlarmDataSubCtx extends TbAbstractDataSubCtx<AlarmDataQuery> {
         }
     }
 
+    @Override
+    protected Aggregation getCurrentAggregation() {
+        return Aggregation.NONE;
+    }
+
     private void sendWsMsg(String sessionId, AlarmSubscriptionUpdate subscriptionUpdate) {
         Alarm alarm = subscriptionUpdate.getAlarm();
         AlarmId alarmId = alarm.getId();

application/src/main/java/org/thingsboard/server/service/subscription/TbEntityDataSubCtx.java

@@ -19,6 +19,7 @@ import lombok.Getter;
 import lombok.Setter;
 import lombok.extern.slf4j.Slf4j;
 import org.thingsboard.server.common.data.id.EntityId;
+import org.thingsboard.server.common.data.kv.Aggregation;
 import org.thingsboard.server.common.data.page.PageData;
 import org.thingsboard.server.common.data.query.EntityData;
 import org.thingsboard.server.common.data.query.EntityDataQuery;
@@ -86,6 +87,11 @@ public class TbEntityDataSubCtx extends TbAbstractDataSubCtx<EntityDataQuery> {
         }
     }
 
+    @Override
+    protected Aggregation getCurrentAggregation() {
+        return this.curTsCmd.getAgg();
+    }
+
     private void sendLatestWsMsg(EntityId entityId, String sessionId, TelemetrySubscriptionUpdate subscriptionUpdate, EntityKeyType keyType) {
         Map<String, TsValue> latestUpdate = new HashMap<>();
         subscriptionUpdate.getData().forEach((k, v) -> {