Merge pull request #10157 from thingsboard/fix/rule-node-repartition

Improve actors init and repartitioning

application/src/main/java/org/thingsboard/server/actors/app/AppActor.java

@@ -202,8 +202,7 @@ public class AppActor extends ContextAwareActor {
         return Optional.ofNullable(ctx.getOrCreateChildActor(new TbEntityActorId(tenantId),
                 () -> DefaultActorService.TENANT_DISPATCHER_NAME,
                 () -> new TenantActor.ActorCreator(systemContext, tenantId),
-                () -> systemContext.getServiceInfoProvider().isService(ServiceType.TB_CORE) ||
+                () -> true));
-                        systemContext.getPartitionService().isManagedByCurrentService(tenantId)));
     }
 
     private void onToEdgeSessionMsg(EdgeSessionMsg msg) {

application/src/main/java/org/thingsboard/server/actors/ruleChain/RuleChainActorMessageProcessor.java

@@ -161,6 +161,7 @@ public class RuleChainActorMessageProcessor extends ComponentMsgProcessor<RuleCh
 
     @Override
     public void onPartitionChangeMsg(PartitionChangeMsg msg) {
+        log.debug("[{}][{}] onPartitionChangeMsg: [{}]", tenantId, entityId, msg);
         nodeActors.values().stream().map(RuleNodeCtx::getSelfActor).forEach(actorRef -> actorRef.tellWithHighPriority(msg));
     }
 

application/src/main/java/org/thingsboard/server/actors/ruleChain/RuleChainManagerActor.java

@@ -50,6 +50,8 @@ public abstract class RuleChainManagerActor extends ContextAwareActor {
     @Getter
     protected TbActorRef rootChainActor;
 
+    protected boolean ruleChainsInitialized;
+
     public RuleChainManagerActor(ActorSystemContext systemContext, TenantId tenantId) {
         super(systemContext);
         this.tenantId = tenantId;
@@ -57,6 +59,7 @@ public abstract class RuleChainManagerActor extends ContextAwareActor {
     }
 
     protected void initRuleChains() {
+        ruleChainsInitialized = true;
         for (RuleChain ruleChain : new PageDataIterable<>(link -> ruleChainService.findTenantRuleChainsByType(tenantId, RuleChainType.CORE, link), ContextAwareActor.ENTITY_PACK_LIMIT)) {
             RuleChainId ruleChainId = ruleChain.getId();
             log.debug("[{}|{}] Creating rule chain actor", ruleChainId.getEntityType(), ruleChain.getId());
@@ -70,6 +73,7 @@ public abstract class RuleChainManagerActor extends ContextAwareActor {
         for (RuleChain ruleChain : new PageDataIterable<>(link -> ruleChainService.findTenantRuleChainsByType(tenantId, RuleChainType.CORE, link), ContextAwareActor.ENTITY_PACK_LIMIT)) {
             ctx.stop(new TbEntityActorId(ruleChain.getId()));
         }
+        ruleChainsInitialized = false;
     }
 
     protected void visit(RuleChain entity, TbActorRef actorRef) {

application/src/main/java/org/thingsboard/server/actors/ruleChain/RuleNodeActorMessageProcessor.java

@@ -15,6 +15,7 @@
  */
 package org.thingsboard.server.actors.ruleChain;
 
+import lombok.extern.slf4j.Slf4j;
 import org.thingsboard.rule.engine.api.TbNode;
 import org.thingsboard.rule.engine.api.TbNodeConfiguration;
 import org.thingsboard.server.actors.ActorSystemContext;
@@ -39,6 +40,7 @@ import org.thingsboard.server.gen.transport.TransportProtos;
 /**
  * @author Andrew Shvayka
  */
+@Slf4j
 public class RuleNodeActorMessageProcessor extends ComponentMsgProcessor<RuleNodeId> {
 
     private final String ruleChainName;
@@ -61,6 +63,7 @@ public class RuleNodeActorMessageProcessor extends ComponentMsgProcessor<RuleNod
     @Override
     public void start(TbActorCtx context) throws Exception {
         if (isMyNodePartition()) {
+            log.debug("[{}][{}] Starting", tenantId, entityId);
             tbNode = initComponent(ruleNode);
             if (tbNode != null) {
                 state = ComponentLifecycleState.ACTIVE;
@@ -95,6 +98,7 @@ public class RuleNodeActorMessageProcessor extends ComponentMsgProcessor<RuleNod
 
     @Override
     public void stop(TbActorCtx context) {
+        log.debug("[{}][{}] Stopping", tenantId, entityId);
         if (tbNode != null) {
             tbNode.destroy();
             state = ComponentLifecycleState.SUSPENDED;
@@ -103,6 +107,7 @@ public class RuleNodeActorMessageProcessor extends ComponentMsgProcessor<RuleNod
 
     @Override
     public void onPartitionChangeMsg(PartitionChangeMsg msg) throws Exception {
+        log.debug("[{}][{}] onPartitionChangeMsg: [{}]", tenantId, entityId, msg);
         if (tbNode != null) {
             if (!isMyNodePartition()) {
                 stop(null);
@@ -185,9 +190,13 @@ public class RuleNodeActorMessageProcessor extends ComponentMsgProcessor<RuleNod
     }
 
     private boolean isMyNodePartition(RuleNode ruleNode) {
-        return ruleNode == null || !ruleNode.isSingletonMode()
+        boolean result = ruleNode == null || !ruleNode.isSingletonMode()
                 || systemContext.getDiscoveryService().isMonolith()
                 || defaultCtx.isLocalEntity(ruleNode.getId());
+        if (!result) {
+            log.trace("[{}][{}] Is not my node partition", tenantId, entityId);
+        }
+        return result;
     }
 
     //Message will return after processing. See RuleChainActorMessageProcessor.pushToTarget.

application/src/main/java/org/thingsboard/server/actors/service/DefaultActorService.java

@@ -31,6 +31,7 @@ import org.thingsboard.server.actors.app.AppActor;
 import org.thingsboard.server.actors.app.AppInitMsg;
 import org.thingsboard.server.actors.stats.StatsActor;
 import org.thingsboard.server.common.msg.queue.PartitionChangeMsg;
+import org.thingsboard.server.common.msg.queue.ServiceType;
 import org.thingsboard.server.queue.discovery.TbApplicationEventListener;
 import org.thingsboard.server.queue.discovery.event.PartitionChangeEvent;
 import org.thingsboard.server.queue.util.AfterStartUp;
@@ -124,6 +125,11 @@ public class DefaultActorService extends TbApplicationEventListener<PartitionCha
         this.appActor.tellWithHighPriority(new PartitionChangeMsg(event.getServiceType()));
     }
 
+    @Override
+    protected boolean filterTbApplicationEvent(PartitionChangeEvent event) {
+        return event.getServiceType() == ServiceType.TB_RULE_ENGINE || event.getServiceType() == ServiceType.TB_CORE;
+    }
+
     @PreDestroy
     public void stopActorSystem() {
         if (system != null) {

application/src/main/java/org/thingsboard/server/actors/tenant/TenantActor.java

@@ -102,6 +102,8 @@ public class TenantActor extends RuleChainManagerActor {
                             log.info("Failed to check ApiUsage \"ReExecEnabled\"!!!", e);
                             cantFindTenant = true;
                         }
+                    } else {
+                        log.info("Tenant {} is not managed by current service, skipping rule chains init", tenantId);
                     }
                 }
                 log.debug("[{}] Tenant actor started.", tenantId);
@@ -131,20 +133,7 @@ public class TenantActor extends RuleChainManagerActor {
         }
         switch (msg.getMsgType()) {
             case PARTITION_CHANGE_MSG:
-                PartitionChangeMsg partitionChangeMsg = (PartitionChangeMsg) msg;
+                onPartitionChangeMsg((PartitionChangeMsg) msg);
-                ServiceType serviceType = partitionChangeMsg.getServiceType();
-                if (ServiceType.TB_RULE_ENGINE.equals(serviceType)) {
-                    //To Rule Chain Actors
-                    broadcast(msg);
-                } else if (ServiceType.TB_CORE.equals(serviceType)) {
-                    List<TbActorId> deviceActorIds = ctx.filterChildren(new TbEntityTypeActorIdPredicate(EntityType.DEVICE) {
-                        @Override
-                        protected boolean testEntityId(EntityId entityId) {
-                            return super.testEntityId(entityId) && !isMyPartition(entityId);
-                        }
-                    });
-                    deviceActorIds.forEach(id -> ctx.stop(id));
-                }
                 break;
             case COMPONENT_LIFE_CYCLE_MSG:
                 onComponentLifecycleMsg((ComponentLifecycleMsg) msg);
@@ -194,7 +183,7 @@ public class TenantActor extends RuleChainManagerActor {
             return;
         }
         TbMsg tbMsg = msg.getMsg();
-        if (getApiUsageState().isReExecEnabled()) {
+        if (getApiUsageState().isReExecEnabled() && ruleChainsInitialized) {
             if (tbMsg.getRuleChainId() == null) {
                 if (getRootChainActor() != null) {
                     getRootChainActor().tell(msg);
@@ -218,7 +207,7 @@ public class TenantActor extends RuleChainManagerActor {
     }
 
     private void onRuleChainMsg(RuleChainAwareMsg msg) {
-        if (getApiUsageState().isReExecEnabled()) {
+        if (getApiUsageState().isReExecEnabled() && ruleChainsInitialized) {
             getOrCreateActor(msg.getRuleChainId()).tell(msg);
         }
     }
@@ -239,6 +228,35 @@ public class TenantActor extends RuleChainManagerActor {
         }
     }
 
+    private void onPartitionChangeMsg(PartitionChangeMsg msg) {
+        ServiceType serviceType = msg.getServiceType();
+        if (ServiceType.TB_RULE_ENGINE.equals(serviceType)) {
+            if (systemContext.getPartitionService().isManagedByCurrentService(tenantId)) {
+                if (!ruleChainsInitialized) {
+                    log.info("Tenant {} is now managed by this service, initializing rule chains", tenantId);
+                    initRuleChains();
+                }
+            } else {
+                if (ruleChainsInitialized) {
+                    log.info("Tenant {} is no longer managed by this service, stopping rule chains", tenantId);
+                    destroyRuleChains();
+                }
+                return;
+            }
+
+            //To Rule Chain Actors
+            broadcast(msg);
+        } else if (ServiceType.TB_CORE.equals(serviceType)) {
+            List<TbActorId> deviceActorIds = ctx.filterChildren(new TbEntityTypeActorIdPredicate(EntityType.DEVICE) {
+                @Override
+                protected boolean testEntityId(EntityId entityId) {
+                    return super.testEntityId(entityId) && !isMyPartition(entityId);
+                }
+            });
+            deviceActorIds.forEach(id -> ctx.stop(id));
+        }
+    }
+
     private void onComponentLifecycleMsg(ComponentLifecycleMsg msg) {
         if (msg.getEntityId().getEntityType().equals(EntityType.API_USAGE_STATE)) {
             ApiUsageState old = getApiUsageState();
@@ -266,7 +284,7 @@ public class TenantActor extends RuleChainManagerActor {
             onToDeviceActorMsg(new DeviceDeleteMsg(tenantId, deviceId), true);
             deletedDevices.add(deviceId);
         }
-        if (isRuleEngine) {
+        if (isRuleEngine && ruleChainsInitialized) {
             TbActorRef target = getEntityActorRef(msg.getEntityId());
             if (target != null) {
                 if (msg.getEntityId().getEntityType() == EntityType.RULE_CHAIN) {

application/src/main/java/org/thingsboard/server/service/partition/AbstractPartitionBasedService.java

@@ -89,11 +89,14 @@ public abstract class AbstractPartitionBasedService<T extends EntityId> extends
      */
     @Override
     protected void onTbApplicationEvent(PartitionChangeEvent partitionChangeEvent) {
-        if (getServiceType().equals(partitionChangeEvent.getServiceType())) {
+        log.debug("onTbApplicationEvent, processing event: {}", partitionChangeEvent);
-            log.debug("onTbApplicationEvent, processing event: {}", partitionChangeEvent);
+        subscribeQueue.add(partitionChangeEvent.getPartitions());
-            subscribeQueue.add(partitionChangeEvent.getPartitions());
+        scheduledExecutor.submit(this::pollInitStateFromDB);
-            scheduledExecutor.submit(this::pollInitStateFromDB);
+    }
-        }
+
+    @Override
+    protected boolean filterTbApplicationEvent(PartitionChangeEvent event) {
+        return getServiceType().equals(event.getServiceType());
     }
 
     protected void pollInitStateFromDB() {

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

@@ -228,16 +228,14 @@ public class DefaultTbCoreConsumerService extends AbstractConsumerService<ToCore
 
     @Override
     protected void onTbApplicationEvent(PartitionChangeEvent event) {
-        if (event.getServiceType().equals(getServiceType())) {
+        log.info("Subscribing to partitions: {}", event.getPartitions());
-            log.info("Subscribing to partitions: {}", event.getPartitions());
+        this.mainConsumer.subscribe(event.getPartitions());
-            this.mainConsumer.subscribe(event.getPartitions());
+        this.usageStatsConsumer.subscribe(
-            this.usageStatsConsumer.subscribe(
+                event
-                    event
+                        .getPartitions()
-                            .getPartitions()
+                        .stream()
-                            .stream()
+                        .map(tpi -> tpi.newByTopic(usageStatsConsumer.getTopic()))
-                            .map(tpi -> tpi.newByTopic(usageStatsConsumer.getTopic()))
+                        .collect(Collectors.toSet()));
-                            .collect(Collectors.toSet()));
-        }
         this.firmwareStatesConsumer.subscribe();
     }
 

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

@@ -99,27 +99,32 @@ public class DefaultTbRuleEngineConsumerService extends AbstractConsumerService<
         List<Queue> queues = queueService.findAllQueues();
         for (Queue configuration : queues) {
             if (partitionService.isManagedByCurrentService(configuration.getTenantId())) {
-                initConsumer(configuration);
+                QueueKey queueKey = new QueueKey(ServiceType.TB_RULE_ENGINE, configuration);
+                createConsumer(queueKey, configuration);
             }
         }
     }
 
-    private void initConsumer(Queue configuration) {
-        getOrCreateConsumer(new QueueKey(ServiceType.TB_RULE_ENGINE, configuration)).init(configuration);
-    }
-
     @Override
     protected void onTbApplicationEvent(PartitionChangeEvent event) {
-        if (event.getServiceType().equals(getServiceType())) {
+        event.getPartitionsMap().forEach((queueKey, partitions) -> {
-            event.getPartitionsMap().forEach((queueKey, partitions) -> {
+            if (partitionService.isManagedByCurrentService(queueKey.getTenantId())) {
-                var consumer = consumers.get(queueKey);
+                var consumer = getConsumer(queueKey).orElseGet(() -> {
-                if (consumer != null) {
+                    Queue config = queueService.findQueueByTenantIdAndName(queueKey.getTenantId(), queueKey.getQueueName());
-                    consumer.update(partitions);
+                    return createConsumer(queueKey, config);
-                } else {
+                });
-                    log.warn("Received invalid partition change event for {} that is not managed by this service", queueKey);
+                consumer.update(partitions);
-                }
+            }
-            });
+        });
-        }
+        consumers.keySet().stream()
+                .collect(Collectors.groupingBy(QueueKey::getTenantId))
+                .forEach((tenantId, queueKeys) -> {
+                    if (!partitionService.isManagedByCurrentService(tenantId)) {
+                        queueKeys.forEach(queueKey -> {
+                            removeConsumer(queueKey).ifPresent(TbRuleEngineQueueConsumerManager::stop);
+                        });
+                    }
+                });
     }
 
     @AfterStartUp(order = AfterStartUp.REGULAR_SERVICE)
@@ -179,7 +184,6 @@ public class DefaultTbRuleEngineConsumerService extends AbstractConsumerService<
     }
 
     private void updateQueues(List<QueueUpdateMsg> queueUpdateMsgs) {
-        boolean partitionsChanged = false;
         for (QueueUpdateMsg queueUpdateMsg : queueUpdateMsgs) {
             log.info("Received queue update msg: [{}]", queueUpdateMsg);
             TenantId tenantId = new TenantId(new UUID(queueUpdateMsg.getTenantIdMSB(), queueUpdateMsg.getTenantIdLSB()));
@@ -189,23 +193,14 @@ public class DefaultTbRuleEngineConsumerService extends AbstractConsumerService<
                 QueueKey queueKey = new QueueKey(ServiceType.TB_RULE_ENGINE, queueName, tenantId);
                 Queue queue = queueService.findQueueById(tenantId, queueId);
 
-                TbRuleEngineQueueConsumerManager consumerManager = getOrCreateConsumer(queueKey);
+                getConsumer(queueKey).ifPresentOrElse(consumer -> consumer.update(queue),
-                Queue oldQueue = consumerManager.getQueue();
+                        () -> createConsumer(queueKey, queue));
-                consumerManager.update(queue);
-
-                if (oldQueue == null || queue.getPartitions() != oldQueue.getPartitions()) {
-                    partitionsChanged = true;
-                }
-            } else {
-                partitionsChanged = true;
             }
         }
 
-        if (partitionsChanged) {
+        partitionService.updateQueues(queueUpdateMsgs);
-            partitionService.updateQueues(queueUpdateMsgs);
+        partitionService.recalculatePartitions(ctx.getServiceInfoProvider().getServiceInfo(),
-            partitionService.recalculatePartitions(ctx.getServiceInfoProvider().getServiceInfo(),
+                new ArrayList<>(partitionService.getOtherServices(ServiceType.TB_RULE_ENGINE)));
-                    new ArrayList<>(partitionService.getOtherServices(ServiceType.TB_RULE_ENGINE)));
-        }
     }
 
     private void deleteQueues(List<QueueDeleteMsg> queueDeleteMsgs) {
@@ -213,10 +208,7 @@ public class DefaultTbRuleEngineConsumerService extends AbstractConsumerService<
             log.info("Received queue delete msg: [{}]", queueDeleteMsg);
             TenantId tenantId = new TenantId(new UUID(queueDeleteMsg.getTenantIdMSB(), queueDeleteMsg.getTenantIdLSB()));
             QueueKey queueKey = new QueueKey(ServiceType.TB_RULE_ENGINE, queueDeleteMsg.getQueueName(), tenantId);
-            var consumerManager = consumers.remove(queueKey);
+            removeConsumer(queueKey).ifPresent(consumer -> consumer.delete(true));
-            if (consumerManager != null) {
-                consumerManager.delete(true);
-            }
         }
 
         partitionService.removeQueues(queueDeleteMsgs);
@@ -231,17 +223,25 @@ public class DefaultTbRuleEngineConsumerService extends AbstractConsumerService<
                         .filter(queueKey -> queueKey.getTenantId().equals(event.getTenantId()))
                         .collect(Collectors.toList());
                 toRemove.forEach(queueKey -> {
-                    var consumerManager = consumers.remove(queueKey);
+                    removeConsumer(queueKey).ifPresent(consumer -> consumer.delete(false));
-                    if (consumerManager != null) {
-                        consumerManager.delete(false);
-                    }
                 });
             }
         }
     }
 
-    private TbRuleEngineQueueConsumerManager getOrCreateConsumer(QueueKey queueKey) {
+    private Optional<TbRuleEngineQueueConsumerManager> getConsumer(QueueKey queueKey) {
-        return consumers.computeIfAbsent(queueKey, key -> new TbRuleEngineQueueConsumerManager(ctx, key));
+        return Optional.ofNullable(consumers.get(queueKey));
+    }
+
+    private TbRuleEngineQueueConsumerManager createConsumer(QueueKey queueKey, Queue queue) {
+        var consumer = new TbRuleEngineQueueConsumerManager(ctx, queueKey);
+        consumers.put(queueKey, consumer);
+        consumer.init(queue);
+        return consumer;
+    }
+
+    private Optional<TbRuleEngineQueueConsumerManager> removeConsumer(QueueKey queueKey) {
+        return Optional.ofNullable(consumers.remove(queueKey));
     }
 
     @Scheduled(fixedDelayString = "${queue.rule-engine.stats.print-interval-ms}")

application/src/main/java/org/thingsboard/server/service/queue/processing/AbstractConsumerService.java

@@ -108,6 +108,11 @@ public abstract class AbstractConsumerService<N extends com.google.protobuf.Gene
         launchMainConsumers();
     }
 
+    @Override
+    protected boolean filterTbApplicationEvent(PartitionChangeEvent event) {
+        return event.getServiceType() == getServiceType();
+    }
+
     protected abstract ServiceType getServiceType();
 
     protected abstract void launchMainConsumers();

application/src/test/java/org/thingsboard/server/queue/discovery/HashPartitionServiceTest.java

@@ -63,7 +63,6 @@ import static org.mockito.ArgumentMatchers.argThat;
 import static org.mockito.ArgumentMatchers.eq;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.never;
-import static org.mockito.Mockito.spy;
 import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.when;
 
@@ -73,9 +72,9 @@ public class HashPartitionServiceTest {
 
     public static final int ITERATIONS = 1000000;
     public static final int SERVER_COUNT = 3;
-    private HashPartitionService clusterRoutingService;
+    private HashPartitionService partitionService;
 
-    private TbServiceInfoProvider discoveryService;
+    private TbServiceInfoProvider serviceInfoProvider;
     private TenantRoutingInfoService routingInfoService;
     private ApplicationEventPublisher applicationEventPublisher;
     private QueueRoutingInfoService queueRoutingInfoService;
@@ -85,19 +84,17 @@ public class HashPartitionServiceTest {
 
     @Before
     public void setup() throws Exception {
-        discoveryService = mock(TbServiceInfoProvider.class);
+        serviceInfoProvider = mock(TbServiceInfoProvider.class);
         applicationEventPublisher = mock(ApplicationEventPublisher.class);
         routingInfoService = mock(TenantRoutingInfoService.class);
         queueRoutingInfoService = mock(QueueRoutingInfoService.class);
         topicService = mock(TopicService.class);
         when(topicService.buildTopicName(Mockito.any())).thenAnswer(i -> i.getArguments()[0]);
-        clusterRoutingService = createPartitionService();
+        partitionService = createPartitionService();
         ServiceInfo currentServer = ServiceInfo.newBuilder()
                 .setServiceId("tb-core-0")
                 .addAllServiceTypes(Collections.singletonList(ServiceType.TB_CORE.name()))
                 .build();
-//        when(queueService.resolve(Mockito.any(), Mockito.anyString())).thenAnswer(i -> i.getArguments()[1]);
-//        when(discoveryService.getServiceInfo()).thenReturn(currentServer);
         List<ServiceInfo> otherServers = new ArrayList<>();
         for (int i = 1; i < SERVER_COUNT; i++) {
             otherServers.add(ServiceInfo.newBuilder()
@@ -106,7 +103,7 @@ public class HashPartitionServiceTest {
                     .build());
         }
 
-        clusterRoutingService.recalculatePartitions(currentServer, otherServers);
+        partitionService.recalculatePartitions(currentServer, otherServers);
     }
 
     @Test
@@ -122,7 +119,7 @@ public class HashPartitionServiceTest {
         long start = System.currentTimeMillis();
         Map<Integer, Integer> map = new HashMap<>();
         for (DeviceId deviceId : devices) {
-            TopicPartitionInfo address = clusterRoutingService.resolve(ServiceType.TB_CORE, TenantId.SYS_TENANT_ID, deviceId);
+            TopicPartitionInfo address = partitionService.resolve(ServiceType.TB_CORE, TenantId.SYS_TENANT_ID, deviceId);
             Integer partition = address.getPartition().get();
             map.put(partition, map.getOrDefault(partition, 0) + 1);
         }
@@ -156,7 +153,7 @@ public class HashPartitionServiceTest {
             for (int queueIndex = 0; queueIndex < queueCount; queueIndex++) {
                 QueueKey queueKey = new QueueKey(ServiceType.TB_RULE_ENGINE, "queue" + queueIndex, tenantId);
                 for (int partition = 0; partition < partitionCount; partition++) {
-                    ServiceInfo serviceInfo = clusterRoutingService.resolveByPartitionIdx(services, queueKey, partition);
+                    ServiceInfo serviceInfo = partitionService.resolveByPartitionIdx(services, queueKey, partition, Collections.emptyMap());
                     String serviceId = serviceInfo.getServiceId();
                     map.put(serviceId, map.get(serviceId) + 1);
                 }
@@ -233,15 +230,34 @@ public class HashPartitionServiceTest {
         }
 
         Map<QueueKey, Map<ServiceInfo, List<Integer>>> serversPartitions = new HashMap<>();
-        clusterRoutingService.init();
+        when(serviceInfoProvider.isService(eq(ServiceType.TB_RULE_ENGINE))).thenReturn(true);
+        partitionService.init();
         for (ServiceInfo ruleEngine : ruleEngines) {
             List<ServiceInfo> other = new ArrayList<>(ruleEngines);
             other.removeIf(serviceInfo -> serviceInfo.getServiceId().equals(ruleEngine.getServiceId()));
 
-            clusterRoutingService.recalculatePartitions(ruleEngine, other);
+            partitionService.recalculatePartitions(ruleEngine, other);
-            clusterRoutingService.myPartitions.forEach((queueKey, partitions) -> {
+            partitionService.myPartitions.forEach((queueKey, partitions) -> {
                 serversPartitions.computeIfAbsent(queueKey, k -> new HashMap<>()).put(ruleEngine, partitions);
             });
+
+            Set<UUID> assignedTenantProfiles = ruleEngine.getAssignedTenantProfilesList().stream().map(UUID::fromString).collect(Collectors.toSet());
+            when(serviceInfoProvider.getAssignedTenantProfiles()).thenReturn(assignedTenantProfiles);
+            if (assignedTenantProfiles.isEmpty()) {
+                assertThat(partitionService.isManagedByCurrentService(TenantId.SYS_TENANT_ID)).isTrue();
+                tenants.forEach((tenantId, tenantProfileId) -> {
+                    assertThat(partitionService.isManagedByCurrentService(tenantId)).isFalse();
+                });
+            } else {
+                assertThat(partitionService.isManagedByCurrentService(TenantId.SYS_TENANT_ID)).isFalse();
+                tenants.forEach((tenantId, tenantProfileId) -> {
+                    if (assignedTenantProfiles.contains(tenantProfileId.getId())) {
+                        assertThat(partitionService.isManagedByCurrentService(tenantId)).isTrue();
+                    } else {
+                        assertThat(partitionService.isManagedByCurrentService(tenantId)).isFalse();
+                    }
+                });
+            }
         }
         assertThat(serversPartitions.keySet()).containsAll(queues.stream().map(queue -> new QueueKey(ServiceType.TB_RULE_ENGINE, queue)).collect(Collectors.toList()));
 
@@ -286,7 +302,7 @@ public class HashPartitionServiceTest {
         mockRoutingInfo(tenantId, tenantProfileId, false); // not isolated yet
         mockQueues(queues);
 
-        when(discoveryService.isService(eq(ServiceType.TB_RULE_ENGINE))).thenReturn(true);
+        when(serviceInfoProvider.isService(eq(ServiceType.TB_RULE_ENGINE))).thenReturn(true);
         Mockito.reset(applicationEventPublisher);
         HashPartitionService partitionService_common = createPartitionService();
         partitionService_common.recalculatePartitions(commonRuleEngine, List.of(dedicatedRuleEngine));
@@ -349,27 +365,6 @@ public class HashPartitionServiceTest {
         });
     }
 
-    @Test
-    public void testIsManagedByCurrentServiceCheck() {
-        TenantProfileId isolatedProfileId = new TenantProfileId(UUID.randomUUID());
-        when(discoveryService.getAssignedTenantProfiles()).thenReturn(Set.of(isolatedProfileId.getId())); // dedicated server
-        TenantProfileId regularProfileId = new TenantProfileId(UUID.randomUUID());
-
-        TenantId isolatedTenantId = new TenantId(UUID.randomUUID());
-        mockRoutingInfo(isolatedTenantId, isolatedProfileId, true);
-        TenantId regularTenantId = new TenantId(UUID.randomUUID());
-        mockRoutingInfo(regularTenantId, regularProfileId, false);
-
-        assertThat(clusterRoutingService.isManagedByCurrentService(isolatedTenantId)).isTrue();
-        assertThat(clusterRoutingService.isManagedByCurrentService(regularTenantId)).isFalse();
-
-
-        when(discoveryService.getAssignedTenantProfiles()).thenReturn(Collections.emptySet()); // common server
-
-        assertThat(clusterRoutingService.isManagedByCurrentService(isolatedTenantId)).isTrue();
-        assertThat(clusterRoutingService.isManagedByCurrentService(regularTenantId)).isTrue();
-    }
-
     @Test
     public void testPartitionsDistribution_sameTenantDifferentQueues() {
         List<ServiceInfo> ruleEngines = new ArrayList<>();
@@ -389,9 +384,9 @@ public class HashPartitionServiceTest {
                     .limit(100).collect(Collectors.toList());
 
             for (int partition = 0; partition < 10; partition++) {
-                ServiceInfo expectedAssignedRuleEngine = clusterRoutingService.resolveByPartitionIdx(ruleEngines, new QueueKey(ServiceType.TB_RULE_ENGINE, tenantId), partition);
+                ServiceInfo expectedAssignedRuleEngine = partitionService.resolveByPartitionIdx(ruleEngines, new QueueKey(ServiceType.TB_RULE_ENGINE, tenantId), partition, Collections.emptyMap());
                 for (QueueKey queueKey : queues) {
-                    ServiceInfo assignedRuleEngine = clusterRoutingService.resolveByPartitionIdx(ruleEngines, queueKey, partition);
+                    ServiceInfo assignedRuleEngine = partitionService.resolveByPartitionIdx(ruleEngines, queueKey, partition, Collections.emptyMap());
                     assertThat(assignedRuleEngine).as(queueKey + "[" + partition + "] should be assigned to " + expectedAssignedRuleEngine.getServiceId())
                             .isEqualTo(expectedAssignedRuleEngine);
                 }
@@ -403,9 +398,9 @@ public class HashPartitionServiceTest {
         verify(applicationEventPublisher).publishEvent(argThat(event -> event instanceof PartitionChangeEvent && predicate.test((PartitionChangeEvent) event)));
     }
 
-    private void mockRoutingInfo(TenantId tenantId, TenantProfileId tenantProfileId, boolean isolatedTbRuleEngine) {
+    private void mockRoutingInfo(TenantId tenantId, TenantProfileId tenantProfileId, boolean isolated) {
         when(routingInfoService.getRoutingInfo(eq(tenantId)))
-                .thenReturn(new TenantRoutingInfo(tenantId, tenantProfileId, isolatedTbRuleEngine));
+                .thenReturn(new TenantRoutingInfo(tenantId, tenantProfileId, isolated));
     }
 
     private void mockQueues(List<Queue> queues) {
@@ -424,7 +419,7 @@ public class HashPartitionServiceTest {
     }
 
     private HashPartitionService createPartitionService() {
-        HashPartitionService partitionService = new HashPartitionService(discoveryService,
+        HashPartitionService partitionService = new HashPartitionService(serviceInfoProvider,
                 routingInfoService,
                 applicationEventPublisher,
                 queueRoutingInfoService,

common/queue/src/main/java/org/thingsboard/server/queue/discovery/DefaultTbServiceInfoProvider.java

@@ -81,7 +81,7 @@ public class DefaultTbServiceInfoProvider implements TbServiceInfoProvider {
         }
         log.info("Current Service ID: {}", serviceId);
         if (serviceType.equalsIgnoreCase("monolith")) {
-            serviceTypes = Collections.unmodifiableList(Arrays.asList(ServiceType.values()));
+            serviceTypes = List.of(ServiceType.values());
         } else {
             serviceTypes = Collections.singletonList(ServiceType.of(serviceType));
         }

common/queue/src/main/java/org/thingsboard/server/queue/discovery/HashPartitionService.java

@@ -25,6 +25,7 @@ import org.thingsboard.server.common.data.exception.TenantNotFoundException;
 import org.thingsboard.server.common.data.id.EntityId;
 import org.thingsboard.server.common.data.id.TenantId;
 import org.thingsboard.server.common.data.id.TenantProfileId;
+import org.thingsboard.server.common.data.util.CollectionsUtil;
 import org.thingsboard.server.common.msg.queue.ServiceType;
 import org.thingsboard.server.common.msg.queue.TopicPartitionInfo;
 import org.thingsboard.server.gen.transport.TransportProtos;
@@ -81,7 +82,7 @@ public class HashPartitionService implements PartitionService {
 
     private List<ServiceInfo> currentOtherServices;
     private final Map<String, List<ServiceInfo>> tbTransportServicesByType = new HashMap<>();
-    private final Map<TenantProfileId, List<ServiceInfo>> responsibleServices = new HashMap<>();
+    private volatile Map<TenantProfileId, List<ServiceInfo>> responsibleServices = Collections.emptyMap();
 
     private HashFunction hashFunction;
 
@@ -218,17 +219,37 @@ public class HashPartitionService implements PartitionService {
 
     @Override
     public boolean isManagedByCurrentService(TenantId tenantId) {
-        Set<UUID> assignedTenantProfiles = serviceInfoProvider.getAssignedTenantProfiles();
+        if (serviceInfoProvider.isService(ServiceType.TB_CORE) || !serviceInfoProvider.isService(ServiceType.TB_RULE_ENGINE)) {
-        if (assignedTenantProfiles.isEmpty()) {
-            // TODO: refactor this for common servers
             return true;
+        }
+
+        boolean isManaged;
+        Set<UUID> assignedTenantProfiles = serviceInfoProvider.getAssignedTenantProfiles();
+        boolean isRegular = assignedTenantProfiles.isEmpty();
+        if (tenantId.isSysTenantId()) {
+            // All system queues are always processed on regular rule engines.
+            return isRegular;
+        }
+        TenantRoutingInfo routingInfo = getRoutingInfo(tenantId);
+        if (isRegular) {
+            if (routingInfo.isIsolated()) {
+                isManaged = hasDedicatedService(routingInfo.getProfileId());
+            } else {
+                isManaged = true;
+            }
         } else {
-            if (tenantId.isSysTenantId()) {
+            if (routingInfo.isIsolated()) {
-                return false;
+                isManaged = assignedTenantProfiles.contains(routingInfo.getProfileId().getId());
+            } else {
+                isManaged = false;
             }
-            TenantProfileId profileId = tenantRoutingInfoService.getRoutingInfo(tenantId).getProfileId();
-            return assignedTenantProfiles.contains(profileId.getId());
         }
+        log.trace("[{}] Tenant {} managed by this service", tenantId, isManaged ? "is" : "is not");
+        return isManaged;
+    }
+
+    private boolean hasDedicatedService(TenantProfileId profileId) {
+        return CollectionsUtil.isEmpty(responsibleServices.get(profileId));
     }
 
     @Override
@@ -283,14 +304,14 @@ public class HashPartitionService implements PartitionService {
     public synchronized void recalculatePartitions(ServiceInfo currentService, List<ServiceInfo> otherServices) {
         log.info("Recalculating partitions");
         tbTransportServicesByType.clear();
-        responsibleServices.clear();
         logServiceInfo(currentService);
         otherServices.forEach(this::logServiceInfo);
 
         Map<QueueKey, List<ServiceInfo>> queueServicesMap = new HashMap<>();
-        addNode(queueServicesMap, currentService);
+        Map<TenantProfileId, List<ServiceInfo>> responsibleServices = new HashMap<>();
+        addNode(currentService, queueServicesMap, responsibleServices);
         for (ServiceInfo other : otherServices) {
-            addNode(queueServicesMap, other);
+            addNode(other, queueServicesMap, responsibleServices);
         }
         queueServicesMap.values().forEach(list -> list.sort(Comparator.comparing(ServiceInfo::getServiceId)));
         responsibleServices.values().forEach(list -> list.sort(Comparator.comparing(ServiceInfo::getServiceId)));
@@ -299,7 +320,7 @@ public class HashPartitionService implements PartitionService {
         partitionSizesMap.forEach((queueKey, size) -> {
             for (int i = 0; i < size; i++) {
                 try {
-                    ServiceInfo serviceInfo = resolveByPartitionIdx(queueServicesMap.get(queueKey), queueKey, i);
+                    ServiceInfo serviceInfo = resolveByPartitionIdx(queueServicesMap.get(queueKey), queueKey, i, responsibleServices);
                     log.trace("Server responsible for {}[{}] - {}", queueKey, i, serviceInfo != null ? serviceInfo.getServiceId() : "none");
                     if (currentService.equals(serviceInfo)) {
                         newPartitions.computeIfAbsent(queueKey, key -> new ArrayList<>()).add(i);
@@ -309,6 +330,7 @@ public class HashPartitionService implements PartitionService {
                 }
             }
         });
+        this.responsibleServices = responsibleServices;
 
         final ConcurrentMap<QueueKey, List<Integer>> oldPartitions = myPartitions;
         myPartitions = newPartitions;
@@ -474,20 +496,22 @@ public class HashPartitionService implements PartitionService {
         if (TenantId.SYS_TENANT_ID.equals(tenantId)) {
             return false;
         }
-        TenantRoutingInfo routingInfo = tenantRoutingInfoMap.computeIfAbsent(tenantId, k -> {
+        TenantRoutingInfo routingInfo = getRoutingInfo(tenantId);
-            return tenantRoutingInfoService.getRoutingInfo(tenantId);
-        });
         if (routingInfo == null) {
             throw new TenantNotFoundException(tenantId);
         }
         switch (serviceType) {
             case TB_RULE_ENGINE:
-                return routingInfo.isIsolatedTbRuleEngine();
+                return routingInfo.isIsolated();
             default:
                 return false;
         }
     }
 
+    private TenantRoutingInfo getRoutingInfo(TenantId tenantId) {
+        return tenantRoutingInfoMap.computeIfAbsent(tenantId, tenantRoutingInfoService::getRoutingInfo);
+    }
+
     private TenantId getIsolatedOrSystemTenantId(ServiceType serviceType, TenantId tenantId) {
         return isIsolated(serviceType, tenantId) ? tenantId : TenantId.SYS_TENANT_ID;
     }
@@ -496,7 +520,7 @@ public class HashPartitionService implements PartitionService {
         log.info("[{}] Found common server: {}", server.getServiceId(), server.getServiceTypesList());
     }
 
-    private void addNode(Map<QueueKey, List<ServiceInfo>> queueServiceList, ServiceInfo instance) {
+    private void addNode(ServiceInfo instance, Map<QueueKey, List<ServiceInfo>> queueServiceList, Map<TenantProfileId, List<ServiceInfo>> responsibleServices) {
         for (String serviceTypeStr : instance.getServiceTypesList()) {
             ServiceType serviceType = ServiceType.of(serviceTypeStr);
             if (ServiceType.TB_RULE_ENGINE.equals(serviceType)) {
@@ -528,7 +552,8 @@ public class HashPartitionService implements PartitionService {
         }
     }
 
-    protected ServiceInfo resolveByPartitionIdx(List<ServiceInfo> servers, QueueKey queueKey, int partition) {
+    protected ServiceInfo resolveByPartitionIdx(List<ServiceInfo> servers, QueueKey queueKey, int partition,
+                                                Map<TenantProfileId, List<ServiceInfo>> responsibleServices) {
         if (servers == null || servers.isEmpty()) {
             return null;
         }

common/queue/src/main/java/org/thingsboard/server/queue/discovery/TbApplicationEventListener.java

@@ -15,21 +15,27 @@
  */
 package org.thingsboard.server.queue.discovery;
 
-import lombok.extern.slf4j.Slf4j;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 import org.springframework.context.ApplicationListener;
 import org.thingsboard.server.queue.discovery.event.TbApplicationEvent;
 
 import java.util.concurrent.locks.Lock;
 import java.util.concurrent.locks.ReentrantLock;
 
-@Slf4j
 public abstract class TbApplicationEventListener<T extends TbApplicationEvent> implements ApplicationListener<T> {
 
     private int lastProcessedSequenceNumber = Integer.MIN_VALUE;
     private final Lock seqNumberLock = new ReentrantLock();
 
+    private final Logger log = LoggerFactory.getLogger(getClass());
+
     @Override
     public void onApplicationEvent(T event) {
+        if (!filterTbApplicationEvent(event)) {
+            log.trace("Skipping event due to filter: {}", event);
+            return;
+        }
         boolean validUpdate = false;
         seqNumberLock.lock();
         try {
@@ -40,7 +46,7 @@ public abstract class TbApplicationEventListener<T extends TbApplicationEvent> i
         } finally {
             seqNumberLock.unlock();
         }
-        if (validUpdate && filterTbApplicationEvent(event)) {
+        if (validUpdate) {
             try {
                 onTbApplicationEvent(event);
             } catch (Exception e) {

common/queue/src/main/java/org/thingsboard/server/queue/discovery/TenantRoutingInfo.java

@@ -23,5 +23,5 @@ import org.thingsboard.server.common.data.id.TenantProfileId;
 public class TenantRoutingInfo {
     private final TenantId tenantId;
     private final TenantProfileId profileId;
-    private final boolean isolatedTbRuleEngine;
+    private final boolean isolated;
 }

common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/limits/DefaultTransportRateLimitService.java

@@ -93,7 +93,7 @@ public class DefaultTransportRateLimitService implements TransportRateLimitServi
         EntityTransportRateLimits deviceRateLimitPrototype = createRateLimits(update.getProfile(), false);
         for (TenantId tenantId : update.getAffectedTenants()) {
             mergeLimits(tenantId, tenantRateLimitPrototype, perTenantLimits::get, perTenantLimits::put);
-            tenantDevices.get(tenantId).forEach(deviceId -> {
+            getTenantDevices(tenantId).forEach(deviceId -> {
                 mergeLimits(deviceId, deviceRateLimitPrototype, perDeviceLimits::get, perDeviceLimits::put);
             });
         }
@@ -104,7 +104,7 @@ public class DefaultTransportRateLimitService implements TransportRateLimitServi
         EntityTransportRateLimits tenantRateLimitPrototype = createRateLimits(tenantProfileCache.get(tenantId), true);
         EntityTransportRateLimits deviceRateLimitPrototype = createRateLimits(tenantProfileCache.get(tenantId), false);
         mergeLimits(tenantId, tenantRateLimitPrototype, perTenantLimits::get, perTenantLimits::put);
-        tenantDevices.get(tenantId).forEach(deviceId -> {
+        getTenantDevices(tenantId).forEach(deviceId -> {
             mergeLimits(deviceId, deviceRateLimitPrototype, perDeviceLimits::get, perDeviceLimits::put);
         });
     }
@@ -259,8 +259,13 @@ public class DefaultTransportRateLimitService implements TransportRateLimitServi
     private EntityTransportRateLimits getDeviceRateLimits(TenantId tenantId, DeviceId deviceId) {
         return perDeviceLimits.computeIfAbsent(deviceId, k -> {
             EntityTransportRateLimits limits = createRateLimits(tenantProfileCache.get(tenantId), false);
-            tenantDevices.computeIfAbsent(tenantId, id -> ConcurrentHashMap.newKeySet()).add(deviceId);
+            getTenantDevices(tenantId).add(deviceId);
             return limits;
         });
     }
+
+    private Set<DeviceId> getTenantDevices(TenantId tenantId) {
+        return tenantDevices.computeIfAbsent(tenantId, id -> ConcurrentHashMap.newKeySet());
+    }
+
 }

rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/debug/TbMsgGeneratorNode.java

@@ -79,7 +79,6 @@ public class TbMsgGeneratorNode implements TbNode {
 
     @Override
     public void init(TbContext ctx, TbNodeConfiguration configuration) throws TbNodeException {
-        log.trace("init generator with config {}", configuration);
         this.config = TbNodeUtils.convert(configuration, TbMsgGeneratorNodeConfiguration.class);
         this.delay = TimeUnit.SECONDS.toMillis(config.getPeriodInSeconds());
         this.currentMsgCount = 0;
@@ -90,17 +89,18 @@ public class TbMsgGeneratorNode implements TbNode {
         } else {
             originatorId = ctx.getSelfId();
         }
+        log.debug("[{}] Initializing generator with config {}", originatorId, configuration);
         updateGeneratorState(ctx);
     }
 
     @Override
     public void onPartitionChangeMsg(TbContext ctx, PartitionChangeMsg msg) {
-        log.trace("onPartitionChangeMsg, PartitionChangeMsg {}, config {}", msg, config);
+        log.debug("[{}] Handling partition change msg: {}", originatorId, msg);
         updateGeneratorState(ctx);
     }
 
     private void updateGeneratorState(TbContext ctx) {
-        log.trace("updateGeneratorState, config {}", config);
+        log.trace("[{}] Updating generator state, config {}", originatorId, config);
         if (ctx.isLocalEntity(originatorId)) {
             if (initialized.compareAndSet(false, true)) {
                 this.scriptEngine = ctx.createScriptEngine(config.getScriptLang(),
@@ -114,7 +114,7 @@ public class TbMsgGeneratorNode implements TbNode {
 
     @Override
     public void onMsg(TbContext ctx, TbMsg msg) {
-        log.trace("onMsg, config {}, msg {}", config, msg);
+        log.trace("[{}] onMsg. Expected msg id: {}, msg: {}, config: {}", originatorId, nextTickId, msg, config);
         if (initialized.get() && msg.isTypeOf(TbMsgType.GENERATOR_NODE_SELF_MSG) && msg.getId().equals(nextTickId)) {
             TbStopWatch sw = TbStopWatch.create();
             withCallback(generate(ctx, msg),
@@ -138,7 +138,6 @@ public class TbMsgGeneratorNode implements TbNode {
     }
 
     private void scheduleTickMsg(TbContext ctx, TbMsg msg) {
-        log.trace("scheduleTickMsg, config {}", config);
         long curTs = System.currentTimeMillis();
         if (lastScheduledTs == 0L) {
             lastScheduledTs = curTs;
@@ -149,6 +148,7 @@ public class TbMsgGeneratorNode implements TbNode {
                 getCustomerIdFromMsg(msg), TbMsgMetaData.EMPTY, TbMsg.EMPTY_STRING);
         nextTickId = tickMsg.getId();
         ctx.tellSelf(tickMsg, curDelay);
+        log.trace("[{}] Scheduled tick msg with delay {}, msg: {}, config: {}", originatorId, curDelay, tickMsg, config);
     }
 
     private ListenableFuture<TbMsg> generate(TbContext ctx, TbMsg msg) {
@@ -175,8 +175,11 @@ public class TbMsgGeneratorNode implements TbNode {
 
     @Override
     public void destroy() {
-        log.trace("destroy, config {}", config);
+        log.debug("[{}] Stopping generator", originatorId);
+        initialized.set(false);
         prevMsg = null;
+        nextTickId = null;
+        lastScheduledTs = 0;
         if (scriptEngine != null) {
             scriptEngine.destroy();
             scriptEngine = null;