dendi-wang/thingsboard
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

No more failures on missing RE queue

Browse Source
This commit is contained in:
Andrii Shvaika 2020-04-09 13:09:54 +03:00
parent 5bebf098ab
commit ec4e2c036f
1 changed files with 8 additions and 1 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

9
common/queue/src/main/java/org/thingsboard/server/queue/discovery/ConsistentHashPartitionService.java
View File

@ -99,7 +99,14 @@ public class ConsistentHashPartitionService implements PartitionService {
int hash = hashFunction.newHasher() int hash = hashFunction.newHasher()
.putLong(entityId.getId().getMostSignificantBits()) .putLong(entityId.getId().getMostSignificantBits())
.putLong(entityId.getId().getLeastSignificantBits()).hash().asInt(); .putLong(entityId.getId().getLeastSignificantBits()).hash().asInt();
int partition = Math.abs(hash % partitionSizes.get(serviceQueue)); Integer partitionSize = partitionSizes.get(serviceQueue);
int partition;
if (partitionSize != null) {
partition = Math.abs(hash % partitionSize);
} else {
//TODO: In 2.6/3.1 this should not happen because all Rule Engine Queues will be in the DB and we always know their partition sizes.
partition = 0;
}
boolean isolatedTenant = isIsolated(serviceQueue, tenantId); boolean isolatedTenant = isIsolated(serviceQueue, tenantId);
TopicPartitionInfoKey cacheKey = new TopicPartitionInfoKey(serviceQueue, isolatedTenant ? tenantId : null, partition); TopicPartitionInfoKey cacheKey = new TopicPartitionInfoKey(serviceQueue, isolatedTenant ? tenantId : null, partition);
return tpiCache.computeIfAbsent(cacheKey, key -> buildTopicPartitionInfo(serviceQueue, tenantId, partition)); return tpiCache.computeIfAbsent(cacheKey, key -> buildTopicPartitionInfo(serviceQueue, tenantId, partition));