Merge branch 'feature/entity-data-query' of github.com:thingsboard/thingsboard into feature/entity-data-query

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

@@ -355,10 +355,9 @@ public class TelemetryController extends BaseController {
                 DataConstants.SHARED_SCOPE.equals(scope) ||
                 DataConstants.CLIENT_SCOPE.equals(scope)) {
             return accessValidator.validateEntityAndCallback(getCurrentUser(), Operation.WRITE_ATTRIBUTES, entityIdSrc, (result, tenantId, entityId) -> {
-                ListenableFuture<List<Void>> future = attributesService.removeAll(user.getTenantId(), entityId, scope, keys);
+                tsSubService.deleteAndNotify(tenantId, entityId, scope, keys, new FutureCallback<Void>() {
-                Futures.addCallback(future, new FutureCallback<List<Void>>() {
                     @Override
-                    public void onSuccess(@Nullable List<Void> tmp) {
+                    public void onSuccess(@Nullable Void tmp) {
                         logAttributesDeleted(user, entityId, scope, keys, null);
                         if (entityIdSrc.getEntityType().equals(EntityType.DEVICE)) {
                             DeviceId deviceId = new DeviceId(entityId.getId());
@@ -375,7 +374,7 @@ public class TelemetryController extends BaseController {
                         logAttributesDeleted(user, entityId, scope, keys, t);
                         result.setResult(new ResponseEntity<>(HttpStatus.INTERNAL_SERVER_ERROR));
                     }
-                }, executor);
+                });
             });
         } else {
             return getImmediateDeferredResult("Invalid attribute scope: " + scope, HttpStatus.BAD_REQUEST);

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

@@ -31,6 +31,7 @@ import org.thingsboard.server.gen.transport.TransportProtos.FromDeviceRPCRespons
 import org.thingsboard.server.gen.transport.TransportProtos.LocalSubscriptionServiceMsgProto;
 import org.thingsboard.server.gen.transport.TransportProtos.SubscriptionMgrMsgProto;
 import org.thingsboard.server.gen.transport.TransportProtos.TbAttributeUpdateProto;
+import org.thingsboard.server.gen.transport.TransportProtos.TbAttributeDeleteProto;
 import org.thingsboard.server.gen.transport.TransportProtos.TbSubscriptionCloseProto;
 import org.thingsboard.server.gen.transport.TransportProtos.TbTimeSeriesUpdateProto;
 import org.thingsboard.server.gen.transport.TransportProtos.ToCoreMsg;
@@ -54,6 +55,7 @@ import org.thingsboard.server.service.transport.msg.TransportToDeviceActorMsgWra
 
 import javax.annotation.PostConstruct;
 import javax.annotation.PreDestroy;
+import java.util.ArrayList;
 import java.util.List;
 import java.util.Optional;
 import java.util.UUID;
@@ -259,6 +261,12 @@ public class DefaultTbCoreConsumerService extends AbstractConsumerService<ToCore
                     new TenantId(new UUID(proto.getTenantIdMSB(), proto.getTenantIdLSB())),
                     TbSubscriptionUtils.toEntityId(proto.getEntityType(), proto.getEntityIdMSB(), proto.getEntityIdLSB()),
                     proto.getScope(), TbSubscriptionUtils.toAttributeKvList(proto.getDataList()), callback);
+        } else if (msg.hasAttrDelete()) {
+            TbAttributeDeleteProto proto = msg.getAttrDelete();
+            subscriptionManagerService.onAttributesDelete(
+                    new TenantId(new UUID(proto.getTenantIdMSB(), proto.getTenantIdLSB())),
+                    TbSubscriptionUtils.toEntityId(proto.getEntityType(), proto.getEntityIdMSB(), proto.getEntityIdLSB()),
+                    proto.getScope(), proto.getKeysList(), callback);
         } else {
             throwNotHandled(msg, callback);
         }

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

@@ -32,6 +32,7 @@ import org.thingsboard.server.common.data.kv.AttributeKvEntry;
 import org.thingsboard.server.common.data.kv.BaseReadTsKvQuery;
 import org.thingsboard.server.common.data.kv.BasicTsKvEntry;
 import org.thingsboard.server.common.data.kv.ReadTsKvQuery;
+import org.thingsboard.server.common.data.kv.StringDataEntry;
 import org.thingsboard.server.common.data.kv.TsKvEntry;
 import org.thingsboard.server.common.msg.queue.ServiceType;
 import org.thingsboard.server.common.msg.queue.TbCallback;
@@ -253,6 +254,32 @@ public class DefaultSubscriptionManagerService implements SubscriptionManagerSer
         callback.onSuccess();
     }
 
+    @Override
+    public void onAttributesDelete(TenantId tenantId, EntityId entityId, String scope, List<String> keys, TbCallback callback) {
+        onLocalSubUpdate(entityId,
+                s -> {
+                    if (TbSubscriptionType.ATTRIBUTES.equals(s.getType())) {
+                        return (TbAttributeSubscription) s;
+                    } else {
+                        return null;
+                    }
+                },
+                s -> (TbAttributeSubscriptionScope.ANY_SCOPE.equals(s.getScope()) || scope.equals(s.getScope().name())),
+                s -> {
+                    List<TsKvEntry> subscriptionUpdate = null;
+                    for (String key : keys) {
+                        if (s.isAllKeys() || s.getKeyStates().containsKey(key)) {
+                            if (subscriptionUpdate == null) {
+                                subscriptionUpdate = new ArrayList<>();
+                            }
+                            subscriptionUpdate.add(new BasicTsKvEntry(0, new StringDataEntry(key, null)));
+                        }
+                    }
+                    return subscriptionUpdate;
+                });
+        callback.onSuccess();
+    }
+
     private <T extends TbSubscription> void onLocalSubUpdate(EntityId entityId,
                                                              Function<TbSubscription, T> castFunction,
                                                              Predicate<T> filterFunction,

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

@@ -35,4 +35,5 @@ public interface SubscriptionManagerService extends ApplicationListener<Partitio
 
     void onAttributesUpdate(TenantId tenantId, EntityId entityId, String scope, List<AttributeKvEntry> attributes, TbCallback callback);
 
+    void onAttributesDelete(TenantId tenantId, EntityId entityId, String scope, List<String> keys, TbCallback empty);
 }

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

@@ -218,12 +218,17 @@ public class TbEntityDataSubCtx {
                 latestCtxValues.forEach((k, v) -> {
                     TsValue update = latestUpdate.get(k);
                     if (update != null) {
-                        if (update.getTs() < v.getTs()) {
+                        //Ignore notifications about deleted keys
-                            log.trace("[{}][{}][{}] Removed stale update for key: {} and ts: {}", sessionId, cmdId, subscriptionUpdate.getSubscriptionId(), k, update.getTs());
+                        if (!(update.getTs() == 0 && (update.getValue() == null || update.getValue().isEmpty()))) {
-                            latestUpdate.remove(k);
+                            if (update.getTs() < v.getTs()) {
-                        } else if ((update.getTs() == v.getTs() && update.getValue().equals(v.getValue()))) {
+                                log.trace("[{}][{}][{}] Removed stale update for key: {} and ts: {}", sessionId, cmdId, subscriptionUpdate.getSubscriptionId(), k, update.getTs());
-                            log.trace("[{}][{}][{}] Removed duplicate update for key: {} and ts: {}", sessionId, cmdId, subscriptionUpdate.getSubscriptionId(), k, update.getTs());
+                                latestUpdate.remove(k);
-                            latestUpdate.remove(k);
+                            } else if ((update.getTs() == v.getTs() && update.getValue().equals(v.getValue()))) {
+                                log.trace("[{}][{}][{}] Removed duplicate update for key: {} and ts: {}", sessionId, cmdId, subscriptionUpdate.getSubscriptionId(), k, update.getTs());
+                                latestUpdate.remove(k);
+                            }
+                        } else {
+                            log.trace("[{}][{}][{}] Received deleted notification for: {}", sessionId, cmdId, subscriptionUpdate.getSubscriptionId(), k);
                         }
                     }
                 });

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

@@ -34,6 +34,7 @@ import org.thingsboard.server.gen.transport.TransportProtos.KeyValueType;
 import org.thingsboard.server.gen.transport.TransportProtos.SubscriptionMgrMsgProto;
 import org.thingsboard.server.gen.transport.TransportProtos.TbAttributeSubscriptionProto;
 import org.thingsboard.server.gen.transport.TransportProtos.TbAttributeUpdateProto;
+import org.thingsboard.server.gen.transport.TransportProtos.TbAttributeDeleteProto;
 import org.thingsboard.server.gen.transport.TransportProtos.TbSubscriptionCloseProto;
 import org.thingsboard.server.gen.transport.TransportProtos.TbSubscriptionKetStateProto;
 import org.thingsboard.server.gen.transport.TransportProtos.TbSubscriptionProto;
@@ -182,6 +183,22 @@ public class TbSubscriptionUtils {
         return ToCoreMsg.newBuilder().setToSubscriptionMgrMsg(msgBuilder.build()).build();
     }
 
+    public static ToCoreMsg toAttributesDeleteProto(TenantId tenantId, EntityId entityId, String scope, List<String> keys) {
+        TbAttributeDeleteProto.Builder builder = TbAttributeDeleteProto.newBuilder();
+        builder.setEntityType(entityId.getEntityType().name());
+        builder.setEntityIdMSB(entityId.getId().getMostSignificantBits());
+        builder.setEntityIdLSB(entityId.getId().getLeastSignificantBits());
+        builder.setTenantIdMSB(tenantId.getId().getMostSignificantBits());
+        builder.setTenantIdLSB(tenantId.getId().getLeastSignificantBits());
+        builder.setScope(scope);
+        builder.addAllKeys(keys);
+
+        SubscriptionMgrMsgProto.Builder msgBuilder = SubscriptionMgrMsgProto.newBuilder();
+        msgBuilder.setAttrDelete(builder);
+        return ToCoreMsg.newBuilder().setToSubscriptionMgrMsg(msgBuilder.build()).build();
+    }
+
+
     private static TsKvProto.Builder toKeyValueProto(long ts, KvEntry attr) {
         KeyValueProto.Builder dataBuilder = KeyValueProto.newBuilder();
         dataBuilder.setKey(attr.getKey());

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

@@ -133,6 +133,13 @@ public class DefaultTelemetrySubscriptionService implements TelemetrySubscriptio
         addWsCallback(saveFuture, success -> onAttributesUpdate(tenantId, entityId, scope, attributes));
     }
 
+    @Override
+    public void deleteAndNotify(TenantId tenantId, EntityId entityId, String scope, List<String> keys, FutureCallback<Void> callback) {
+        ListenableFuture<List<Void>> deleteFuture = attrService.removeAll(tenantId, entityId, scope, keys);
+        addMainCallback(deleteFuture, callback);
+        addWsCallback(deleteFuture, success -> onAttributesDelete(tenantId, entityId, scope, keys));
+    }
+
     @Override
     public void saveAttrAndNotify(TenantId tenantId, EntityId entityId, String scope, String key, long value, FutureCallback<Void> callback) {
         saveAndNotify(tenantId, entityId, scope, Collections.singletonList(new BaseAttributeKvEntry(new LongDataEntry(key, value)
@@ -171,6 +178,20 @@ public class DefaultTelemetrySubscriptionService implements TelemetrySubscriptio
         }
     }
 
+    private void onAttributesDelete(TenantId tenantId, EntityId entityId, String scope, List<String> keys) {
+        TopicPartitionInfo tpi = partitionService.resolve(ServiceType.TB_CORE, tenantId, entityId);
+        if (currentPartitions.contains(tpi)) {
+            if (subscriptionManagerService.isPresent()) {
+                subscriptionManagerService.get().onAttributesDelete(tenantId, entityId, scope, keys, TbCallback.EMPTY);
+            } else {
+                log.warn("Possible misconfiguration because subscriptionManagerService is null!");
+            }
+        } else {
+            TransportProtos.ToCoreMsg toCoreMsg = TbSubscriptionUtils.toAttributesDeleteProto(tenantId, entityId, scope, keys);
+            clusterService.pushMsgToCore(tpi, entityId.getId(), toCoreMsg, null);
+        }
+    }
+
     private void onTimeSeriesUpdate(TenantId tenantId, EntityId entityId, List<TsKvEntry> ts) {
         TopicPartitionInfo tpi = partitionService.resolve(ServiceType.TB_CORE, tenantId, entityId);
         if (currentPartitions.contains(tpi)) {

application/src/main/resources/logback.xml

@@ -30,6 +30,8 @@
 <!--    <logger name="org.thingsboard.server.service.queue" level="TRACE" />-->
 <!--    <logger name="org.thingsboard.server.service.transport" level="TRACE" />-->
 
+<!--    <logger name="org.thingsboard.server.service.subscription" level="TRACE"/>-->
+<!--    <logger name="org.thingsboard.server.service.telemetry" level="TRACE"/>-->
     <logger name="com.microsoft.azure.servicebus.primitives.CoreMessageReceiver" level="OFF" />
 
     <root level="INFO">

application/src/test/java/org/thingsboard/server/controller/BaseEntityQueryControllerTest.java

@@ -102,6 +102,7 @@ public abstract class BaseEntityQueryControllerTest extends AbstractControllerTe
             device.setType("default");
             device.setLabel("testLabel" + (int) (Math.random() * 1000));
             devices.add(doPost("/api/device", device, Device.class));
+            Thread.sleep(1);
         }
         DeviceTypeFilter filter = new DeviceTypeFilter();
         filter.setDeviceType("default");
@@ -141,6 +142,7 @@ public abstract class BaseEntityQueryControllerTest extends AbstractControllerTe
             device.setType("default");
             device.setLabel("testLabel" + (int) (Math.random() * 1000));
             devices.add(doPost("/api/device", device, Device.class));
+            Thread.sleep(1);
         }
 
         DeviceTypeFilter filter = new DeviceTypeFilter();
@@ -210,6 +212,7 @@ public abstract class BaseEntityQueryControllerTest extends AbstractControllerTe
             device.setType("default");
             device.setLabel("testLabel" + (int) (Math.random() * 1000));
             devices.add(doPost("/api/device?accessToken=" + name, device, Device.class));
+            Thread.sleep(1);
             long temperature = (long) (Math.random() * 100);
             temperatures.add(temperature);
             if (temperature > 45) {

application/src/test/java/org/thingsboard/server/controller/BaseWebsocketApiTest.java

@@ -455,22 +455,18 @@ public class BaseWebsocketApiTest extends AbstractWebsocketTest {
         Assert.assertEquals(0, pageData.getData().get(0).getLatest().get(EntityKeyType.SERVER_ATTRIBUTE).get("serverAttributeKey").getTs());
         Assert.assertEquals("", pageData.getData().get(0).getLatest().get(EntityKeyType.SERVER_ATTRIBUTE).get("serverAttributeKey").getValue());
 
-        AttributeKvEntry dataPoint1 = new BaseAttributeKvEntry(now - TimeUnit.MINUTES.toMillis(1), new LongDataEntry("serverAttributeKey", 42L));
-        List<AttributeKvEntry> tsData = Arrays.asList(dataPoint1);
-        sendAttributes(device, TbAttributeSubscriptionScope.SERVER_SCOPE, tsData);
 
+        wsClient.registerWaitForUpdate();
         Thread.sleep(100);
 
-        cmd = new EntityDataCmd(1, edq, null, latestCmd, null);
+        AttributeKvEntry dataPoint1 = new BaseAttributeKvEntry(now - TimeUnit.MINUTES.toMillis(1), new LongDataEntry("serverAttributeKey", 42L));
-        wrapper = new TelemetryPluginCmdsWrapper();
+        List<AttributeKvEntry> tsData = Arrays.asList(dataPoint1);
-        wrapper.setEntityDataCmds(Collections.singletonList(cmd));
+        sendAttributes(device, TbAttributeSubscriptionScope.SERVER_SCOPE, tsData);
 
-        wsClient.send(mapper.writeValueAsString(wrapper));
+        msg = wsClient.waitForUpdate();
-        msg = wsClient.waitForReply();
         update = mapper.readValue(msg, EntityDataUpdate.class);
 
         Assert.assertEquals(1, update.getCmdId());
-
         List<EntityData> listData = update.getUpdate();
         Assert.assertNotNull(listData);
         Assert.assertEquals(1, listData.size());
@@ -483,6 +479,7 @@ public class BaseWebsocketApiTest extends AbstractWebsocketTest {
         AttributeKvEntry dataPoint2 = new BaseAttributeKvEntry(now, new LongDataEntry("serverAttributeKey", 52L));
 
         wsClient.registerWaitForUpdate();
+        Thread.sleep(100);
         sendAttributes(device, TbAttributeSubscriptionScope.SERVER_SCOPE, Arrays.asList(dataPoint2));
         msg = wsClient.waitForUpdate();
 
@@ -498,12 +495,14 @@ public class BaseWebsocketApiTest extends AbstractWebsocketTest {
 
         //Sending update from the past, while latest value has new timestamp;
         wsClient.registerWaitForUpdate();
+        Thread.sleep(100);
         sendAttributes(device, TbAttributeSubscriptionScope.SERVER_SCOPE, Arrays.asList(dataPoint1));
         msg = wsClient.waitForUpdate(TimeUnit.SECONDS.toMillis(1));
         Assert.assertNull(msg);
 
         //Sending duplicate update again
         wsClient.registerWaitForUpdate();
+        Thread.sleep(100);
         sendAttributes(device, TbAttributeSubscriptionScope.SERVER_SCOPE, Arrays.asList(dataPoint2));
         msg = wsClient.waitForUpdate(TimeUnit.SECONDS.toMillis(1));
         Assert.assertNull(msg);

common/queue/src/main/proto/queue.proto

@@ -295,6 +295,16 @@ message TbAttributeUpdateProto {
   repeated TsKvProto data = 7;
 }
 
+message TbAttributeDeleteProto {
+  string entityType = 1;
+  int64 entityIdMSB = 2;
+  int64 entityIdLSB = 3;
+  int64 tenantIdMSB = 4;
+  int64 tenantIdLSB = 5;
+  string scope = 6;
+  repeated string keys = 7;
+}
+
 message TbTimeSeriesUpdateProto {
   string entityType = 1;
   int64 entityIdMSB = 2;
@@ -340,6 +350,7 @@ message SubscriptionMgrMsgProto {
   TbSubscriptionCloseProto subClose = 3;
   TbTimeSeriesUpdateProto tsUpdate = 4;
   TbAttributeUpdateProto attrUpdate = 5;
+  TbAttributeDeleteProto attrDelete = 6;
 }
 
 message LocalSubscriptionServiceMsgProto {

dao/src/main/java/org/thingsboard/server/dao/sql/query/DefaultEntityQueryRepository.java

@@ -398,7 +398,7 @@ public class DefaultEntityQueryRepository implements EntityQueryRepository {
         //TODO: fetch last level only.
         //TODO: fetch distinct records.
         String lvlFilter = getLvlFilter(entityFilter.getMaxLevel());
-        String selectFields = "SELECT tenant_id, customer_id, id, type, name, label FROM " + entityType.name() + " WHERE id in ( SELECT entity_id";
+        String selectFields = "SELECT tenant_id, customer_id, id, created_time, type, name, label FROM " + entityType.name() + " WHERE id in ( SELECT entity_id";
         String from = getQueryTemplate(entityFilter.getDirection());
         String whereFilter = " WHERE re.relation_type = :where_relation_type AND re.to_type = :where_entity_type";
 

dao/src/test/java/org/thingsboard/server/dao/service/BaseEntityServiceTest.java

@@ -61,6 +61,7 @@ import org.thingsboard.server.dao.attributes.AttributesService;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collections;
+import java.util.Comparator;
 import java.util.List;
 import java.util.concurrent.ExecutionException;
 import java.util.stream.Collectors;
@@ -87,7 +88,7 @@ public abstract class BaseEntityServiceTest extends AbstractServiceTest {
     }
 
     @Test
-    public void testCountEntitiesByQuery() {
+    public void testCountEntitiesByQuery() throws InterruptedException {
         List<Device> devices = new ArrayList<>();
         for (int i = 0; i < 97; i++) {
             Device device = new Device();
@@ -130,7 +131,7 @@ public abstract class BaseEntityServiceTest extends AbstractServiceTest {
     }
 
     @Test
-    public void testCountHierarchicalEntitiesByQuery() {
+    public void testCountHierarchicalEntitiesByQuery() throws InterruptedException {
         List<Asset> assets = new ArrayList<>();
         List<Device> devices = new ArrayList<>();
         createTestHierarchy(assets, devices, new ArrayList<>(), new ArrayList<>(), new ArrayList<>(), new ArrayList<>());
@@ -407,7 +408,7 @@ public abstract class BaseEntityServiceTest extends AbstractServiceTest {
         deviceService.deleteDevicesByTenantId(tenantId);
     }
 
-    private void createTestHierarchy(List<Asset> assets, List<Device> devices, List<Long> consumptions, List<Long> highConsumptions, List<Long> temperatures, List<Long> highTemperatures) {
+    private void createTestHierarchy(List<Asset> assets, List<Device> devices, List<Long> consumptions, List<Long> highConsumptions, List<Long> temperatures, List<Long> highTemperatures) throws InterruptedException {
         for (int i = 0; i < 5; i++) {
             Asset asset = new Asset();
             asset.setTenantId(tenantId);
@@ -415,6 +416,8 @@ public abstract class BaseEntityServiceTest extends AbstractServiceTest {
             asset.setType("type" + i);
             asset.setLabel("AssetLabel" + i);
             asset = assetService.saveAsset(asset);
+            //TO make sure devices have different created time
+            Thread.sleep(1);
             assets.add(asset);
             EntityRelation er = new EntityRelation();
             er.setFrom(tenantId);
@@ -434,6 +437,8 @@ public abstract class BaseEntityServiceTest extends AbstractServiceTest {
                 device.setType("default" + j);
                 device.setLabel("testLabel" + (int) (Math.random() * 1000));
                 device = deviceService.saveDevice(device);
+                //TO make sure devices have different created time
+                Thread.sleep(1);
                 devices.add(device);
                 er = new EntityRelation();
                 er.setFrom(asset.getId());
@@ -451,7 +456,7 @@ public abstract class BaseEntityServiceTest extends AbstractServiceTest {
     }
 
     @Test
-    public void testSimpleFindEntityDataByQuery() {
+    public void testSimpleFindEntityDataByQuery() throws InterruptedException {
         List<Device> devices = new ArrayList<>();
         for (int i = 0; i < 97; i++) {
             Device device = new Device();
@@ -459,6 +464,8 @@ public abstract class BaseEntityServiceTest extends AbstractServiceTest {
             device.setName("Device" + i);
             device.setType("default");
             device.setLabel("testLabel" + (int) (Math.random() * 1000));
+            //TO make sure devices have different created time
+            Thread.sleep(1);
             devices.add(deviceService.saveDevice(device));
         }
 
@@ -490,13 +497,16 @@ public abstract class BaseEntityServiceTest extends AbstractServiceTest {
 
         List<EntityId> loadedIds = loadedEntities.stream().map(EntityData::getEntityId).collect(Collectors.toList());
         List<EntityId> deviceIds = devices.stream().map(Device::getId).collect(Collectors.toList());
-
+        deviceIds.sort(Comparator.comparing(EntityId::getId));
+        loadedIds.sort(Comparator.comparing(EntityId::getId));
         Assert.assertEquals(deviceIds, loadedIds);
 
         List<String> loadedNames = loadedEntities.stream().map(entityData ->
                 entityData.getLatest().get(EntityKeyType.ENTITY_FIELD).get("name").getValue()).collect(Collectors.toList());
         List<String> deviceNames = devices.stream().map(Device::getName).collect(Collectors.toList());
 
+        Collections.sort(loadedNames);
+        Collections.sort(deviceNames);
         Assert.assertEquals(deviceNames, loadedNames);
 
         sortOrder = new EntityDataSortOrder(
@@ -525,6 +535,8 @@ public abstract class BaseEntityServiceTest extends AbstractServiceTest {
             device.setType("default");
             device.setLabel("testLabel" + (int) (Math.random() * 1000));
             devices.add(deviceService.saveDevice(device));
+            //TO make sure devices have different created time
+            Thread.sleep(1);
             long temperature = (long) (Math.random() * 100);
             temperatures.add(temperature);
             if (temperature > 45) {
@@ -560,8 +572,11 @@ public abstract class BaseEntityServiceTest extends AbstractServiceTest {
             loadedEntities.addAll(data.getData());
         }
         Assert.assertEquals(67, loadedEntities.size());
-        List<String> loadedTemperatures = loadedEntities.stream().map(entityData ->
+        List<String> loadedTemperatures = new ArrayList<>();
-                entityData.getLatest().get(EntityKeyType.ATTRIBUTE).get("temperature").getValue()).collect(Collectors.toList());
+        for (Device device : devices) {
+            loadedTemperatures.add(loadedEntities.stream().filter(entityData -> entityData.getEntityId().equals(device.getId())).findFirst().orElse(null)
+                    .getLatest().get(EntityKeyType.ATTRIBUTE).get("temperature").getValue());
+        }
         List<String> deviceTemperatures = temperatures.stream().map(aLong -> Long.toString(aLong)).collect(Collectors.toList());
         Assert.assertEquals(deviceTemperatures, loadedTemperatures);
 

rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/RuleEngineTelemetryService.java

@@ -44,4 +44,7 @@ public interface RuleEngineTelemetryService {
 
     void saveAttrAndNotify(TenantId tenantId, EntityId entityId, String scope, String key, boolean value, FutureCallback<Void> callback);
 
+    void deleteAndNotify(TenantId tenantId, EntityId entityId, String scope, List<String> keys, FutureCallback<Void> callback);
+
+
 }