Housekeeper: tests and refactoring

application/src/main/java/org/thingsboard/server/service/housekeeper/DefaultHousekeeperService.java

@@ -42,7 +42,6 @@ import org.thingsboard.server.service.housekeeper.stats.HousekeeperStatsService;
 import javax.annotation.PreDestroy;
 import java.util.List;
 import java.util.Map;
-import java.util.UUID;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
@@ -58,11 +57,12 @@ public class DefaultHousekeeperService implements HousekeeperService {
 
     private final Map<HousekeeperTaskType, HousekeeperTaskProcessor<?>> taskProcessors;
 
-    private final TbQueueConsumer<TbProtoQueueMsg<ToHousekeeperServiceMsg>> consumer;
-    private final TbQueueProducer<TbProtoQueueMsg<ToHousekeeperServiceMsg>> producer;
     private final HousekeeperReprocessingService reprocessingService;
     private final HousekeeperStatsService statsService;
     private final NotificationRuleProcessor notificationRuleProcessor;
+    private final TbQueueConsumer<TbProtoQueueMsg<ToHousekeeperServiceMsg>> consumer;
+    private final TbQueueProducer<TbProtoQueueMsg<ToHousekeeperServiceMsg>> producer;
+    private final TopicPartitionInfo submitTpi;
 
     private final ExecutorService consumerExecutor = Executors.newSingleThreadExecutor(ThingsBoardThreadFactory.forName("housekeeper-consumer"));
     private final ExecutorService executor = Executors.newSingleThreadExecutor(ThingsBoardThreadFactory.forName("housekeeper-task-processor"));
@@ -80,11 +80,12 @@ public class DefaultHousekeeperService implements HousekeeperService {
                                      HousekeeperStatsService statsService,
                                      NotificationRuleProcessor notificationRuleProcessor,
                                      @Lazy List<HousekeeperTaskProcessor<?>> taskProcessors) {
-        this.consumer = queueFactory.createHousekeeperMsgConsumer();
-        this.producer = producerProvider.getHousekeeperMsgProducer();
         this.reprocessingService = reprocessingService;
         this.statsService = statsService;
         this.notificationRuleProcessor = notificationRuleProcessor;
+        this.consumer = queueFactory.createHousekeeperMsgConsumer();
+        this.producer = producerProvider.getHousekeeperMsgProducer();
+        this.submitTpi = TopicPartitionInfo.builder().topic(producer.getDefaultTopic()).build();
         this.taskProcessors = taskProcessors.stream().collect(Collectors.toMap(HousekeeperTaskProcessor::getTaskType, p -> p));
     }
 
@@ -169,10 +170,13 @@ public class DefaultHousekeeperService implements HousekeeperService {
     }
 
     @Override
-    public void submitTask(UUID key, HousekeeperTask task) {
+    public void submitTask(HousekeeperTask task) {
         log.trace("[{}][{}][{}] Submitting task: {}", task.getTenantId(), task.getEntityId().getEntityType(), task.getEntityId(), task.getTaskType());
-        TopicPartitionInfo tpi = TopicPartitionInfo.builder().topic(producer.getDefaultTopic()).build();
+        /*
-        producer.send(tpi, new TbProtoQueueMsg<>(key, ToHousekeeperServiceMsg.newBuilder()
+         * using msg key as entity id so that msgs related to certain entity are pushed to same partition,
+         * e.g. on tenant deletion (entity id is tenant id), we need to clean up tenant entities in certain order
+         * */
+        producer.send(submitTpi, new TbProtoQueueMsg<>(task.getEntityId().getId(), ToHousekeeperServiceMsg.newBuilder()
                 .setTask(HousekeeperTaskProto.newBuilder()
                         .setValue(JacksonUtil.toString(task))
                         .setTs(task.getTs())

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

@@ -17,10 +17,10 @@ package org.thingsboard.server.service.housekeeper.processor;
 
 import lombok.RequiredArgsConstructor;
 import org.springframework.stereotype.Component;
-import org.thingsboard.server.dao.entity.EntityDaoService;
-import org.thingsboard.server.dao.entity.EntityServiceRegistry;
 import org.thingsboard.server.common.data.housekeeper.EntitiesDeletionHousekeeperTask;
 import org.thingsboard.server.common.data.housekeeper.HousekeeperTaskType;
+import org.thingsboard.server.dao.entity.EntityDaoService;
+import org.thingsboard.server.dao.entity.EntityServiceRegistry;
 
 @Component
 @RequiredArgsConstructor
@@ -31,9 +31,6 @@ public class EntitiesDeletionTaskProcessor implements HousekeeperTaskProcessor<E
     @Override
     public void process(EntitiesDeletionHousekeeperTask task) throws Exception {
         EntityDaoService entityService = entityServiceRegistry.getServiceByEntityType(task.getEntityType());
-        if (entityService == null) {
-            throw new IllegalArgumentException("Unsupported entity type " + task.getEntityType());
-        }
         entityService.deleteByTenantId(task.getTenantId());
     }
 

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

@@ -30,6 +30,8 @@ public class EntityAlarmsDeletionTaskProcessor implements HousekeeperTaskProcess
     @Override
     public void process(HousekeeperTask task) throws Exception {
         alarmService.deleteEntityAlarmRecords(task.getTenantId(), task.getEntityId());
+        // fixme: do we need to remove alarms by originator ???
+        // fixme: why alarm comments are not deleted ??
     }
 
     @Override

application/src/main/java/org/thingsboard/server/service/sync/ie/exporting/DefaultExportableEntitiesService.java

@@ -147,9 +147,6 @@ public class DefaultExportableEntitiesService implements ExportableEntitiesServi
     public <I extends EntityId> void removeById(TenantId tenantId, I id) {
         EntityType entityType = id.getEntityType();
         EntityDaoService entityService = entityServiceRegistry.getServiceByEntityType(entityType);
-        if (entityService == null) {
-            throw new IllegalArgumentException("Unsupported entity type " + entityType);
-        }
         entityService.deleteEntity(tenantId, id);
     }
 

application/src/test/java/org/thingsboard/server/service/housekeeper/HousekeeperServiceTest.java

@@ -0,0 +1,303 @@
+/**
+ * Copyright © 2016-2024 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.service.housekeeper;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.node.TextNode;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.test.mock.mockito.SpyBean;
+import org.springframework.test.context.TestPropertySource;
+import org.thingsboard.common.util.JacksonUtil;
+import org.thingsboard.rule.engine.metadata.TbGetAttributesNode;
+import org.thingsboard.rule.engine.metadata.TbGetAttributesNodeConfiguration;
+import org.thingsboard.server.common.data.DataConstants;
+import org.thingsboard.server.common.data.Device;
+import org.thingsboard.server.common.data.EventInfo;
+import org.thingsboard.server.common.data.alarm.Alarm;
+import org.thingsboard.server.common.data.alarm.AlarmSeverity;
+import org.thingsboard.server.common.data.event.EventType;
+import org.thingsboard.server.common.data.event.LifecycleEvent;
+import org.thingsboard.server.common.data.housekeeper.HousekeeperTaskType;
+import org.thingsboard.server.common.data.id.EntityId;
+import org.thingsboard.server.common.data.id.RuleChainId;
+import org.thingsboard.server.common.data.id.RuleNodeId;
+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.BaseAttributeKvEntry;
+import org.thingsboard.server.common.data.kv.BaseReadTsKvQuery;
+import org.thingsboard.server.common.data.kv.BasicTsKvEntry;
+import org.thingsboard.server.common.data.kv.StringDataEntry;
+import org.thingsboard.server.common.data.kv.TsKvEntry;
+import org.thingsboard.server.common.data.msg.TbNodeConnectionType;
+import org.thingsboard.server.common.data.page.PageLink;
+import org.thingsboard.server.common.data.page.TimePageLink;
+import org.thingsboard.server.common.data.rule.RuleChain;
+import org.thingsboard.server.common.data.rule.RuleChainMetaData;
+import org.thingsboard.server.common.data.rule.RuleChainType;
+import org.thingsboard.server.common.data.rule.RuleNode;
+import org.thingsboard.server.controller.AbstractControllerTest;
+import org.thingsboard.server.dao.alarm.AlarmService;
+import org.thingsboard.server.dao.attributes.AttributesService;
+import org.thingsboard.server.dao.event.EventService;
+import org.thingsboard.server.dao.housekeeper.HousekeeperService;
+import org.thingsboard.server.dao.rule.RuleChainService;
+import org.thingsboard.server.dao.service.DaoSqlTest;
+import org.thingsboard.server.dao.timeseries.TimeseriesService;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.Optional;
+import java.util.concurrent.TimeUnit;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.awaitility.Awaitility.await;
+import static org.mockito.ArgumentMatchers.argThat;
+import static org.mockito.Mockito.verify;
+import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.status;
+
+@DaoSqlTest
+@TestPropertySource(properties = {
+        "transport.http.enabled=true"
+})
+public class HousekeeperServiceTest extends AbstractControllerTest {
+
+    @SpyBean
+    private HousekeeperService housekeeperService;
+    @Autowired
+    private EventService eventService;
+    @Autowired
+    private TimeseriesService timeseriesService;
+    @Autowired
+    private AttributesService attributesService;
+    @Autowired
+    private RuleChainService ruleChainService;
+    @Autowired
+    private AlarmService alarmService;
+
+    private TenantId tenantId;
+
+    private static final String TELEMETRY_KEY = "telemetry1";
+    private static final String ATTRIBUTE_KEY = "_attribute1";
+    private static final String KV_VALUE = "ewfewfwef";
+
+    @Before
+    public void setUp() throws Exception {
+        loginTenantAdmin();
+        this.tenantId = super.tenantId;
+    }
+
+    @After
+    public void tearDown() throws Exception {
+    }
+
+    @Test
+    public void whenDeviceIsDeleted_thenCleanUpRelatedData() throws Exception {
+        Device device = createDevice("test", "test");
+        createRelatedData(device.getId());
+
+        doDelete("/api/device/" + device.getId()).andExpect(status().isOk());
+
+        await().atMost(10, TimeUnit.SECONDS).untilAsserted(() -> {
+            verifyNoRelatedData(device.getId());
+        });
+    }
+
+    @Test
+    public void whenRuleChainIsDeleted_thenCleanUpRelatedData() throws Exception {
+        RuleChainMetaData ruleChainMetaData = createRuleChain();
+        RuleChainId ruleChainId = ruleChainMetaData.getRuleChainId();
+        RuleNodeId ruleNode1Id = ruleChainMetaData.getNodes().get(0).getId();
+        RuleNodeId ruleNode2Id = ruleChainMetaData.getNodes().get(1).getId();
+        createRelatedData(ruleChainId);
+        createRelatedData(ruleNode1Id);
+        createRelatedData(ruleNode2Id);
+
+        doDelete("/api/ruleChain/" + ruleChainId).andExpect(status().isOk());
+
+        await().atMost(10, TimeUnit.SECONDS).untilAsserted(() -> {
+            verifyNoRelatedData(ruleNode1Id);
+            verifyNoRelatedData(ruleNode2Id);
+            verifyNoRelatedData(ruleChainId);
+        });
+    }
+
+    @Test
+    public void whenUserIsDeleted_thenCleanUpRelatedData() throws Exception {
+        Device device = createDevice("test", "test");
+        UserId userId = customerUserId;
+        createRelatedData(userId);
+        Alarm alarm = Alarm.builder()
+                .type("test")
+                .tenantId(tenantId)
+                .originator(device.getId())
+                .severity(AlarmSeverity.MAJOR)
+                .build();
+        alarm = doPost("/api/alarm", alarm, Alarm.class);
+        alarm = doPost("/api/alarm/" + alarm.getId() + "/assign/" + userId, "", Alarm.class);
+        assertThat(alarm.getAssigneeId()).isEqualTo(userId);
+        assertThat(alarmService.findAlarmIdsByAssigneeId(tenantId, userId, new PageLink(100)).getData()).isNotEmpty();
+
+        doDelete("/api/user/" + userId).andExpect(status().isOk());
+
+        await().atMost(10, TimeUnit.SECONDS).untilAsserted(() -> {
+            verifyNoRelatedData(userId);
+        });
+        assertThat(alarmService.findAlarmById(tenantId, alarm.getId()).getAssigneeId()).isNull();
+    }
+
+    @Test
+    public void whenTenantIsDeleted_thenDeleteAllEntitiesAndCleanUpRelatedData() throws Exception {
+        loginDifferentTenant();
+        tenantId = differentTenantId;
+
+        createRelatedData(tenantId);
+
+        Device device = createDevice("test", "test");
+        createRelatedData(device.getId());
+
+        RuleChainMetaData ruleChainMetaData = createRuleChain();
+        RuleChainId ruleChainId = ruleChainMetaData.getRuleChainId();
+        RuleNodeId ruleNode1Id = ruleChainMetaData.getNodes().get(0).getId();
+        RuleNodeId ruleNode2Id = ruleChainMetaData.getNodes().get(1).getId();
+        createRelatedData(ruleChainId);
+        createRelatedData(ruleNode1Id);
+        createRelatedData(ruleNode2Id);
+
+        UserId userId = savedDifferentTenantUser.getId();
+        createRelatedData(userId);
+
+        loginSysAdmin();
+        deleteDifferentTenant();
+
+        await().atMost(30, TimeUnit.SECONDS).untilAsserted(() -> {
+            verifyNoRelatedData(device.getId());
+            verifyNoRelatedData(ruleNode1Id);
+            verifyNoRelatedData(ruleNode2Id);
+            verifyNoRelatedData(ruleChainId);
+            verifyNoRelatedData(userId);
+            verifyNoRelatedData(tenantId);
+        });
+    }
+
+    private void createRelatedData(EntityId entityId) throws Exception {
+        createTelemetry(entityId);
+        for (String scope : List.of(DataConstants.SERVER_SCOPE, DataConstants.SHARED_SCOPE, DataConstants.CLIENT_SCOPE)) {
+            createAttribute(entityId, scope, scope + ATTRIBUTE_KEY);
+        }
+        createEvent(entityId);
+    }
+
+    private void verifyNoRelatedData(EntityId entityId) throws Exception {
+        List<HousekeeperTaskType> expectedTaskTypes = List.of(HousekeeperTaskType.DELETE_TELEMETRY, HousekeeperTaskType.DELETE_ATTRIBUTES, HousekeeperTaskType.DELETE_EVENTS, HousekeeperTaskType.DELETE_ENTITY_ALARMS);
+        for (HousekeeperTaskType taskType : expectedTaskTypes) {
+            verify(housekeeperService).submitTask(argThat(task -> task.getTaskType() == taskType && task.getEntityId().equals(entityId)));
+        }
+
+        assertThat(getLatestTelemetry(entityId)).isNull();
+        assertThat(getTimeseriesHistory(entityId)).isEmpty();
+        for (String scope : List.of(DataConstants.SERVER_SCOPE, DataConstants.SHARED_SCOPE, DataConstants.CLIENT_SCOPE)) {
+            assertThat(getAttribute(entityId, scope, scope + ATTRIBUTE_KEY)).isNull();
+        }
+        assertThat(getEvents(entityId)).isEmpty();
+    }
+
+    private void createAttribute(EntityId entityId, String scope, String key) throws Exception {
+        attributesService.save(tenantId, entityId, scope, new BaseAttributeKvEntry(System.currentTimeMillis(), new StringDataEntry(key, KV_VALUE))).get();
+    }
+
+    private void createTelemetry(EntityId entityId) throws Exception {
+        timeseriesService.save(tenantId, entityId, new BasicTsKvEntry(System.currentTimeMillis(), new StringDataEntry(TELEMETRY_KEY, KV_VALUE))).get();
+    }
+
+    private void createEvent(EntityId entityId) {
+        LifecycleEvent event = LifecycleEvent.builder()
+                .tenantId(tenantId)
+                .entityId(entityId.getId())
+                .serviceId("test")
+                .lcEventType("test")
+                .success(true)
+                .build();
+        eventService.saveAsync(event);
+        await().atMost(10, TimeUnit.SECONDS)
+                .until(() -> !getEvents(entityId).isEmpty());
+
+    }
+
+    private TsKvEntry getLatestTelemetry(EntityId entityId) throws Exception {
+        return timeseriesService.findLatest(tenantId, entityId, HousekeeperServiceTest.TELEMETRY_KEY).get().orElse(null);
+    }
+
+    private List<TsKvEntry> getTimeseriesHistory(EntityId entityId) throws Exception {
+        return timeseriesService.findAll(tenantId, entityId, List.of(new BaseReadTsKvQuery(HousekeeperServiceTest.TELEMETRY_KEY, 0, System.currentTimeMillis(), 10, "DESC"))).get();
+    }
+
+    private AttributeKvEntry getAttribute(EntityId entityId, String scope, String key) throws Exception {
+        return attributesService.find(tenantId, entityId, scope, key).get().orElse(null);
+    }
+
+    private List<EventInfo> getEvents(EntityId entityId) {
+        return eventService.findEvents(tenantId, entityId, EventType.LC_EVENT, new TimePageLink(100)).getData()
+                .stream().filter(event -> Optional.ofNullable(event.getBody()).map(body -> body.get("event"))
+                        .map(JsonNode::asText).orElse("").equals("test"))
+                .collect(Collectors.toList());
+    }
+
+    private RuleChainMetaData createRuleChain() {
+        RuleChain ruleChain = new RuleChain();
+        ruleChain.setTenantId(tenantId);
+        ruleChain.setName("Test");
+        ruleChain.setType(RuleChainType.CORE);
+        ruleChain.setDebugMode(true);
+        ruleChain.setConfiguration(JacksonUtil.newObjectNode().set("a", new TextNode("b")));
+        ruleChain = ruleChainService.saveRuleChain(ruleChain);
+        RuleChainId ruleChainId = ruleChain.getId();
+
+        RuleChainMetaData metaData = new RuleChainMetaData();
+        metaData.setRuleChainId(ruleChainId);
+
+        RuleNode ruleNode1 = new RuleNode();
+        ruleNode1.setName("Simple Rule Node 1");
+        ruleNode1.setType(org.thingsboard.rule.engine.metadata.TbGetAttributesNode.class.getName());
+        ruleNode1.setConfigurationVersion(TbGetAttributesNode.class.getAnnotation(org.thingsboard.rule.engine.api.RuleNode.class).version());
+        ruleNode1.setDebugMode(true);
+        TbGetAttributesNodeConfiguration configuration1 = new TbGetAttributesNodeConfiguration();
+        configuration1.setServerAttributeNames(Collections.singletonList("serverAttributeKey1"));
+        ruleNode1.setConfiguration(JacksonUtil.valueToTree(configuration1));
+
+        RuleNode ruleNode2 = new RuleNode();
+        ruleNode2.setName("Simple Rule Node 2");
+        ruleNode2.setType(org.thingsboard.rule.engine.metadata.TbGetAttributesNode.class.getName());
+        ruleNode2.setConfigurationVersion(TbGetAttributesNode.class.getAnnotation(org.thingsboard.rule.engine.api.RuleNode.class).version());
+        ruleNode2.setDebugMode(true);
+        TbGetAttributesNodeConfiguration configuration2 = new TbGetAttributesNodeConfiguration();
+        configuration2.setServerAttributeNames(Collections.singletonList("serverAttributeKey2"));
+        ruleNode2.setConfiguration(JacksonUtil.valueToTree(configuration2));
+
+        metaData.setNodes(Arrays.asList(ruleNode1, ruleNode2));
+        metaData.setFirstNodeIndex(0);
+        metaData.addConnectionInfo(0, 1, TbNodeConnectionType.SUCCESS);
+        ruleChainService.saveRuleChainMetaData(tenantId, metaData, Function.identity());
+        return ruleChainService.loadRuleChainMetaData(tenantId, ruleChainId);
+    }
+
+}

dao/src/main/java/org/thingsboard/server/dao/entity/DefaultEntityServiceRegistry.java

@@ -27,6 +27,7 @@ import org.thingsboard.server.common.data.EntityType;
 
 import java.util.HashMap;
 import java.util.Map;
+import java.util.Optional;
 
 @Service
 @RequiredArgsConstructor
@@ -52,7 +53,8 @@ public class DefaultEntityServiceRegistry implements EntityServiceRegistry {
 
     @Override
     public EntityDaoService getServiceByEntityType(EntityType entityType) {
-        return entityDaoServicesMap.get(entityType);
+        return Optional.ofNullable(entityDaoServicesMap.get(entityType))
+                .orElseThrow(() -> new IllegalArgumentException("Unsupported entity type " + entityType));
     }
 
 }

dao/src/main/java/org/thingsboard/server/dao/housekeeper/CleanUpService.java

@@ -17,6 +17,8 @@ package org.thingsboard.server.dao.housekeeper;
 
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.context.annotation.Lazy;
 import org.springframework.stereotype.Component;
 import org.springframework.transaction.event.TransactionalEventListener;
 import org.thingsboard.server.common.data.EntityType;
@@ -24,11 +26,12 @@ import org.thingsboard.server.common.data.User;
 import org.thingsboard.server.common.data.housekeeper.HousekeeperTask;
 import org.thingsboard.server.common.data.id.EntityId;
 import org.thingsboard.server.common.data.id.TenantId;
+import org.thingsboard.server.dao.entity.EntityDaoService;
+import org.thingsboard.server.dao.entity.EntityServiceRegistry;
 import org.thingsboard.server.dao.eventsourcing.DeleteEntityEvent;
 import org.thingsboard.server.dao.relation.RelationService;
 
 import java.util.Optional;
-import java.util.UUID;
 
 @Component
 @RequiredArgsConstructor
@@ -37,12 +40,15 @@ public class CleanUpService {
 
     private final Optional<HousekeeperService> housekeeperService;
     private final RelationService relationService;
+    @Autowired
+    @Lazy
+    private EntityServiceRegistry entityServiceRegistry;
 
     @TransactionalEventListener(fallbackExecution = true)
     public void handleEntityDeletionEvent(DeleteEntityEvent<?> event) {
         TenantId tenantId = event.getTenantId();
         EntityId entityId = event.getEntityId();
-        log.trace("[{}] Handling entity deletion event: {}", tenantId, event);
+        log.trace("[{}][{}][{}] Handling entity deletion event", tenantId, entityId.getEntityType(), entityId.getId());
         cleanUpRelatedData(tenantId, entityId);
         if (entityId.getEntityType() == EntityType.USER) {
             housekeeperService.ifPresent(housekeeperService -> {
@@ -52,6 +58,7 @@ public class CleanUpService {
     }
 
     public void cleanUpRelatedData(TenantId tenantId, EntityId entityId) {
+        log.trace("[{}][{}][{}] Cleaning up related data", tenantId, entityId.getEntityType(), entityId.getId());
         // todo: skipped entities list
         relationService.deleteEntityRelations(tenantId, entityId);
         housekeeperService.ifPresent(housekeeperService -> {
@@ -63,11 +70,14 @@ public class CleanUpService {
     }
 
     public void removeTenantEntities(TenantId tenantId, EntityType... entityTypes) {
-        UUID tasksKey = UUID.randomUUID(); // so that all tasks are pushed to single partition to be processed synchronously
+        housekeeperService.ifPresentOrElse(housekeeperService -> {
-        // todo: just use tenantId as key in the impl
+            for (EntityType entityType : entityTypes) {
-        housekeeperService.ifPresent(housekeeperService -> {
+                housekeeperService.submitTask(HousekeeperTask.deleteEntities(tenantId, entityType));
+            }
+        }, () -> {
             for (EntityType entityType : entityTypes) {
-                housekeeperService.submitTask(tasksKey, HousekeeperTask.deleteEntities(tenantId, entityType));
+                EntityDaoService entityService = entityServiceRegistry.getServiceByEntityType(entityType);
+                entityService.deleteByTenantId(tenantId);
             }
         });
     }

dao/src/main/java/org/thingsboard/server/dao/housekeeper/HousekeeperService.java

@@ -17,15 +17,8 @@ package org.thingsboard.server.dao.housekeeper;
 
 import org.thingsboard.server.common.data.housekeeper.HousekeeperTask;
 
-import java.util.UUID;
-
 public interface HousekeeperService {
 
-    default void submitTask(HousekeeperTask task) {
+    void submitTask(HousekeeperTask task);
-        submitTask(task.getEntityId().getId(), task);
-    }
-
-    // tasks with the same key will be pushed to the same partition and thus processed synchronously
-    void submitTask(UUID key, HousekeeperTask task);
 
 }

dao/src/main/java/org/thingsboard/server/dao/rule/BaseRuleChainService.java

@@ -438,12 +438,14 @@ public class BaseRuleChainService extends AbstractEntityService implements RuleC
         checkRuleNodesAndDelete(tenantId, ruleChainId);
     }
 
+    @Transactional
     @Override
     public void deleteRuleChainsByTenantId(TenantId tenantId) {
         Validator.validateId(tenantId, "Incorrect tenant id for delete rule chains request.");
         tenantRuleChainsRemover.removeEntities(tenantId, tenantId);
     }
 
+    @Transactional
     @Override
     public void deleteByTenantId(TenantId tenantId) {
         deleteRuleChainsByTenantId(tenantId);
@@ -776,10 +778,10 @@ public class BaseRuleChainService extends AbstractEntityService implements RuleC
 
     private void deleteRuleNodes(TenantId tenantId, List<RuleNode> ruleNodes) {
         List<RuleNodeId> ruleNodeIds = ruleNodes.stream().map(RuleNode::getId).collect(Collectors.toList());
-        for (var node : ruleNodes) {
-            cleanUpService.cleanUpRelatedData(tenantId, node.getId());
-        }
         ruleNodeDao.deleteByIdIn(ruleNodeIds);
+        for (var nodeId : ruleNodeIds) {
+            cleanUpService.cleanUpRelatedData(tenantId, nodeId);
+        }
     }
 
     @Override