Browse Source

HashPartitionService: set myPartitions atomically

pull/8131/head
Sergey Matvienko 3 years ago
parent
commit
0f324d1132
  1. 10
      common/queue/src/main/java/org/thingsboard/server/queue/discovery/HashPartitionService.java

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

@ -68,7 +68,7 @@ public class HashPartitionService implements PartitionService {
private final TenantRoutingInfoService tenantRoutingInfoService;
private final QueueRoutingInfoService queueRoutingInfoService;
private ConcurrentMap<QueueKey, List<Integer>> myPartitions = new ConcurrentHashMap<>();
private volatile ConcurrentMap<QueueKey, List<Integer>> myPartitions = new ConcurrentHashMap<>();
private final ConcurrentMap<QueueKey, String> partitionTopicsMap = new ConcurrentHashMap<>();
private final ConcurrentMap<QueueKey, Integer> partitionSizesMap = new ConcurrentHashMap<>();
@ -217,17 +217,19 @@ public class HashPartitionService implements PartitionService {
}
queueServicesMap.values().forEach(list -> list.sort(Comparator.comparing(ServiceInfo::getServiceId)));
ConcurrentMap<QueueKey, List<Integer>> oldPartitions = myPartitions;
myPartitions = new ConcurrentHashMap<>();
final ConcurrentMap<QueueKey, List<Integer>> newPartitions = new ConcurrentHashMap<>();
partitionSizesMap.forEach((queueKey, size) -> {
for (int i = 0; i < size; i++) {
ServiceInfo serviceInfo = resolveByPartitionIdx(queueServicesMap.get(queueKey), queueKey, i);
if (currentService.equals(serviceInfo)) {
myPartitions.computeIfAbsent(queueKey, key -> new ArrayList<>()).add(i);
newPartitions.computeIfAbsent(queueKey, key -> new ArrayList<>()).add(i);
}
}
});
final ConcurrentMap<QueueKey, List<Integer>> oldPartitions = myPartitions;
myPartitions = newPartitions;
oldPartitions.forEach((queueKey, partitions) -> {
if (!myPartitions.containsKey(queueKey)) {
log.info("[{}] NO MORE PARTITIONS FOR CURRENT KEY", queueKey);

Loading…
Cancel
Save