diff --git a/application/src/main/java/org/thingsboard/server/service/entity/queue/DefaultTbQueueService.java b/application/src/main/java/org/thingsboard/server/service/entity/queue/DefaultTbQueueService.java index e58c4450fb..655d4beb88 100644 --- a/application/src/main/java/org/thingsboard/server/service/entity/queue/DefaultTbQueueService.java +++ b/application/src/main/java/org/thingsboard/server/service/entity/queue/DefaultTbQueueService.java @@ -253,7 +253,7 @@ public class DefaultTbQueueService implements TbQueueService { QueueId queueIdForRemove = queue.getId(); if (deviceProfileQueues.containsKey(queueIdForRemove)) { Queue foundQueue = queueService.findQueueByTenantIdAndName(tenantId, q); - if (foundQueue == null) { + if (foundQueue == null || queue.equals(foundQueue)) { foundQueue = queueService.findQueueByTenantIdAndName(tenantId, MAIN); } QueueId newQueueId = foundQueue.getId(); diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/discovery/HashPartitionService.java b/common/queue/src/main/java/org/thingsboard/server/queue/discovery/HashPartitionService.java index 8e83af2600..246fac00af 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/discovery/HashPartitionService.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/discovery/HashPartitionService.java @@ -141,20 +141,20 @@ public class HashPartitionService implements PartitionService { public void updateQueue(TransportProtos.QueueUpdateMsg queueUpdateMsg) { TenantId tenantId = new TenantId(new UUID(queueUpdateMsg.getTenantIdMSB(), queueUpdateMsg.getTenantIdLSB())); QueueKey queueKey = new QueueKey(ServiceType.TB_RULE_ENGINE, queueUpdateMsg.getQueueName(), tenantId); - partitionTopicsMap.put(queueKey, queueUpdateMsg.getQueueTopic()); - partitionSizesMap.put(queueKey, queueUpdateMsg.getPartitions()); QueueRoutingInfo queue = new QueueRoutingInfo(queueUpdateMsg); queuesById.put(queue.getQueueId(), queue); + partitionTopicsMap.put(queueKey, queueUpdateMsg.getQueueTopic()); + partitionSizesMap.put(queueKey, queueUpdateMsg.getPartitions()); } @Override public void removeQueue(TransportProtos.QueueDeleteMsg queueDeleteMsg) { TenantId tenantId = new TenantId(new UUID(queueDeleteMsg.getTenantIdMSB(), queueDeleteMsg.getTenantIdLSB())); QueueKey queueKey = new QueueKey(ServiceType.TB_RULE_ENGINE, queueDeleteMsg.getQueueName(), tenantId); - partitionTopicsMap.remove(queueKey); - partitionSizesMap.remove(queueKey); queuesById.remove(new QueueId(new UUID(queueDeleteMsg.getQueueIdMSB(), queueDeleteMsg.getQueueIdLSB()))); myPartitions.remove(queueKey); + partitionTopicsMap.remove(queueKey); + partitionSizesMap.remove(queueKey); } @Override @@ -166,14 +166,29 @@ public class HashPartitionService implements PartitionService { public TopicPartitionInfo resolve(ServiceType serviceType, QueueId queueId, TenantId tenantId, EntityId entityId) { QueueKey queueKey; if (queueId == null) { - queueKey = isIsolated(serviceType, tenantId) ? new QueueKey(serviceType, tenantId) : new QueueKey(serviceType); + queueKey = getMainQueueKey(serviceType, tenantId); } else { - queueKey = new QueueKey(serviceType, queuesById.get(queueId)); + QueueRoutingInfo queueRoutingInfo = queuesById.get(queueId); + + //TODO: replace if we can notify CheckPoint rule nodes about queue changes + if (queueRoutingInfo == null) { + log.warn("Queue was removed but still used in CheckPoint rule node. [{}][{}]", tenantId, entityId); + queueKey = getMainQueueKey(serviceType, tenantId); + } else if (!queueRoutingInfo.getTenantId().equals(getIsolatedOrSystemTenantId(serviceType, tenantId))) { + log.warn("Tenant profile was changed but CheckPoint rule node still uses the queue from system level. [{}][{}]", tenantId, entityId); + queueKey = getMainQueueKey(serviceType, tenantId); + } else { + queueKey = new QueueKey(serviceType, queueRoutingInfo); + } } return resolve(queueKey, entityId); } + private QueueKey getMainQueueKey(ServiceType serviceType, TenantId tenantId) { + return new QueueKey(serviceType, getIsolatedOrSystemTenantId(serviceType, tenantId)); + } + private TopicPartitionInfo resolve(QueueKey queueKey, EntityId entityId) { int hash = hashFunction.newHasher() .putLong(entityId.getId().getMostSignificantBits()) @@ -357,6 +372,10 @@ public class HashPartitionService implements PartitionService { } } + private TenantId getIsolatedOrSystemTenantId(ServiceType serviceType, TenantId tenantId) { + return isIsolated(serviceType, tenantId) ? tenantId : TenantId.SYS_TENANT_ID; + } + private void logServiceInfo(TransportProtos.ServiceInfo server) { log.info("[{}] Found common server: [{}]", server.getServiceId(), server.getServiceTypesList()); }