Attributes deletion task processor

application/src/main/java/org/thingsboard/server/service/housekeeper/processor/AttributesDeletionTaskProcessor.java

@@ -16,6 +16,7 @@
 package org.thingsboard.server.service.housekeeper.processor;
 
 import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
 import org.springframework.stereotype.Component;
 import org.thingsboard.server.dao.attributes.AttributesService;
 import org.thingsboard.server.dao.housekeeper.data.HousekeeperTask;
@@ -23,13 +24,15 @@ import org.thingsboard.server.dao.housekeeper.data.HousekeeperTaskType;
 
 @Component
 @RequiredArgsConstructor
+@Slf4j
 public class AttributesDeletionTaskProcessor implements HousekeeperTaskProcessor<HousekeeperTask> {
 
     private final AttributesService attributesService;
 
     @Override
     public void process(HousekeeperTask task) throws Exception {
-//        attributesService.removeAll(task.getTenantId(), task.getEntityId(), DataConstants.CLIENT_SCOPE);
+        int deletedCount = attributesService.removeAllByEntityId(task.getTenantId(), task.getEntityId());
+        log.trace("[{}][{}][{}] Deleted {} attributes", task.getTenantId(), task.getEntityId().getEntityType(), task.getEntityId(), deletedCount);
     }
 
     @Override

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

@@ -49,4 +49,6 @@ public interface AttributesService {
 
     List<String> findAllKeysByEntityIds(TenantId tenantId, EntityType entityType, List<EntityId> entityIds, String scope);
 
+    int removeAllByEntityId(TenantId tenantId, EntityId entityId);
+
 }

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

@@ -16,6 +16,7 @@
 package org.thingsboard.server.dao.attributes;
 
 import com.google.common.util.concurrent.ListenableFuture;
+import org.apache.commons.lang3.tuple.Pair;
 import org.thingsboard.server.common.data.EntityType;
 import org.thingsboard.server.common.data.id.DeviceProfileId;
 import org.thingsboard.server.common.data.id.EntityId;
@@ -46,4 +47,7 @@ public interface AttributesDao {
     List<String> findAllKeysByEntityIds(TenantId tenantId, EntityType entityType, List<EntityId> entityIds);
 
     List<String> findAllKeysByEntityIdsAndAttributeType(TenantId tenantId, EntityType entityType, List<EntityId> entityIds, String attributeType);
+
+    List<Pair<String, String>> removeAllByEntityId(TenantId tenantId, EntityId entityId);
+
 }

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

@@ -18,6 +18,7 @@ package org.thingsboard.server.dao.attributes;
 import com.google.common.util.concurrent.Futures;
 import com.google.common.util.concurrent.ListenableFuture;
 import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.lang3.tuple.Pair;
 import org.springframework.beans.factory.annotation.Value;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
 import org.springframework.context.annotation.Primary;
@@ -113,4 +114,11 @@ public class BaseAttributesService implements AttributesService {
         validate(entityId, scope);
         return Futures.allAsList(attributesDao.removeAll(tenantId, entityId, scope, attributeKeys));
     }
+
+    @Override
+    public int removeAllByEntityId(TenantId tenantId, EntityId entityId) {
+        List<Pair<String, String>> deleted = attributesDao.removeAllByEntityId(tenantId, entityId);
+        return deleted.size();
+    }
+
 }

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

@@ -20,6 +20,7 @@ import com.google.common.util.concurrent.ListenableFuture;
 import com.google.common.util.concurrent.ListeningExecutorService;
 import com.google.common.util.concurrent.MoreExecutors;
 import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.lang3.tuple.Pair;
 import org.springframework.beans.factory.annotation.Value;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
 import org.springframework.context.annotation.Primary;
@@ -147,47 +148,47 @@ public class CachedAttributesService implements AttributesService {
         return Futures.transformAsync(cacheExecutor.submit(() -> findCachedAttributes(entityId, scope, attributeKeys)),
                 wrappedCachedAttributes -> {
 
-        List<AttributeKvEntry> cachedAttributes = wrappedCachedAttributes.values().stream()
-                .map(TbCacheValueWrapper::get)
-                .filter(Objects::nonNull)
-                .collect(Collectors.toList());
-        if (wrappedCachedAttributes.size() == attributeKeys.size()) {
-            log.trace("[{}][{}] Found all attributes from cache: {}", entityId, scope, attributeKeys);
-            return Futures.immediateFuture(cachedAttributes);
-        }
-
-        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(() -> {
-            var cacheTransaction = cache.newTransactionForKeys(notFoundKeys);
-            try {
-                log.trace("[{}][{}] Lookup attributes from db: {}", entityId, scope, notFoundAttributeKeys);
-                List<AttributeKvEntry> result = attributesDao.find(tenantId, entityId, scope, notFoundAttributeKeys);
-                for (AttributeKvEntry foundInDbAttribute : result) {
-                    AttributeCacheKey attributeCacheKey = new AttributeCacheKey(scope, entityId, foundInDbAttribute.getKey());
-                    cacheTransaction.putIfAbsent(attributeCacheKey, foundInDbAttribute);
-                    notFoundAttributeKeys.remove(foundInDbAttribute.getKey());
-                }
-                for (String key : notFoundAttributeKeys) {
-                    cacheTransaction.putIfAbsent(new AttributeCacheKey(scope, entityId, key), null);
-                }
-                List<AttributeKvEntry> mergedAttributes = new ArrayList<>(cachedAttributes);
-                mergedAttributes.addAll(result);
-                cacheTransaction.commit();
-                log.trace("[{}][{}] Commit cache transaction: {}", entityId, scope, notFoundAttributeKeys);
-                return mergedAttributes;
-            } catch (Throwable e) {
-                cacheTransaction.rollback();
-                log.debug("Could not find attributes from cache: [{}] [{}] [{}]", entityId, scope, notFoundAttributeKeys, e);
-                throw e;
-            }
-        });
-
-        }, MoreExecutors.directExecutor()); // cacheExecutor analyse and returns results or submit to DB executor
+                    List<AttributeKvEntry> cachedAttributes = wrappedCachedAttributes.values().stream()
+                            .map(TbCacheValueWrapper::get)
+                            .filter(Objects::nonNull)
+                            .collect(Collectors.toList());
+                    if (wrappedCachedAttributes.size() == attributeKeys.size()) {
+                        log.trace("[{}][{}] Found all attributes from cache: {}", entityId, scope, attributeKeys);
+                        return Futures.immediateFuture(cachedAttributes);
+                    }
+
+                    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(() -> {
+                        var cacheTransaction = cache.newTransactionForKeys(notFoundKeys);
+                        try {
+                            log.trace("[{}][{}] Lookup attributes from db: {}", entityId, scope, notFoundAttributeKeys);
+                            List<AttributeKvEntry> result = attributesDao.find(tenantId, entityId, scope, notFoundAttributeKeys);
+                            for (AttributeKvEntry foundInDbAttribute : result) {
+                                AttributeCacheKey attributeCacheKey = new AttributeCacheKey(scope, entityId, foundInDbAttribute.getKey());
+                                cacheTransaction.putIfAbsent(attributeCacheKey, foundInDbAttribute);
+                                notFoundAttributeKeys.remove(foundInDbAttribute.getKey());
+                            }
+                            for (String key : notFoundAttributeKeys) {
+                                cacheTransaction.putIfAbsent(new AttributeCacheKey(scope, entityId, key), null);
+                            }
+                            List<AttributeKvEntry> mergedAttributes = new ArrayList<>(cachedAttributes);
+                            mergedAttributes.addAll(result);
+                            cacheTransaction.commit();
+                            log.trace("[{}][{}] Commit cache transaction: {}", entityId, scope, notFoundAttributeKeys);
+                            return mergedAttributes;
+                        } catch (Throwable e) {
+                            cacheTransaction.rollback();
+                            log.debug("Could not find attributes from cache: [{}] [{}] [{}]", entityId, scope, notFoundAttributeKeys, e);
+                            throw e;
+                        }
+                    });
+
+                }, MoreExecutors.directExecutor()); // cacheExecutor analyse and returns results or submit to DB executor
     }
 
     private Map<String, TbCacheValueWrapper<AttributeKvEntry>> findCachedAttributes(EntityId entityId, String scope, Collection<String> attributeKeys) {
@@ -268,4 +269,15 @@ public class CachedAttributesService implements AttributesService {
         }, cacheExecutor)).collect(Collectors.toList()));
     }
 
+    @Override
+    public int removeAllByEntityId(TenantId tenantId, EntityId entityId) {
+        List<Pair<String, String>> result = attributesDao.removeAllByEntityId(tenantId, entityId);
+        result.forEach(deleted -> {
+            String scope = deleted.getKey();
+            String key = deleted.getValue();
+            cache.evict(new AttributeCacheKey(scope, entityId, key));
+        });
+        return result.size();
+    }
+
 }