Browse Source

added system partitons handling

pull/13040/head
IrynaMatveieva 2 years ago
parent
commit
2d95b7ad58
  1. 32
      application/src/main/java/org/thingsboard/server/service/cf/cache/DefaultCalculatedFieldEntityProfileCache.java

32
application/src/main/java/org/thingsboard/server/service/cf/cache/DefaultCalculatedFieldEntityProfileCache.java

@ -28,10 +28,14 @@ import org.thingsboard.server.queue.discovery.TbApplicationEventListener;
import org.thingsboard.server.queue.discovery.event.PartitionChangeEvent; import org.thingsboard.server.queue.discovery.event.PartitionChangeEvent;
import org.thingsboard.server.queue.util.TbRuleEngineComponent; import org.thingsboard.server.queue.util.TbRuleEngineComponent;
import java.util.ArrayList;
import java.util.Collection; import java.util.Collection;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap; import java.util.concurrent.ConcurrentMap;
import java.util.stream.Collectors;
@TbRuleEngineComponent @TbRuleEngineComponent
@Service @Service
@ -46,18 +50,32 @@ public class DefaultCalculatedFieldEntityProfileCache extends TbApplicationEvent
@Override @Override
protected void onTbApplicationEvent(PartitionChangeEvent event) { protected void onTbApplicationEvent(PartitionChangeEvent event) {
var tenantPartitions = event.getCfPartitions().stream() Map<TenantId, List<Integer>> tenantPartitions = new HashMap<>();
List<Integer> systemPartitions = new ArrayList<>();
event.getCfPartitions().stream()
.filter(TopicPartitionInfo::isMyPartition) .filter(TopicPartitionInfo::isMyPartition)
.filter(tpi -> tpi.getTenantId().isPresent()) .forEach(tpi -> {
.collect(Collectors.groupingBy( Integer partition = tpi.getPartition().orElse(UNKNOWN);
tpi -> tpi.getTenantId().get(), Optional<TenantId> tenantIdOpt = tpi.getTenantId();
Collectors.mapping(tpi -> tpi.getPartition().orElse(UNKNOWN), Collectors.toList()) if (tenantIdOpt.isPresent()) {
)); tenantPartitions.computeIfAbsent(tenantIdOpt.get(), id -> new ArrayList<>()).add(partition);
} else {
systemPartitions.add(partition);
}
});
tenantPartitions.forEach((tenantId, partitions) -> { tenantPartitions.forEach((tenantId, partitions) -> {
var cache = tenantCache.computeIfAbsent(tenantId, id -> new TenantEntityProfileCache()); var cache = tenantCache.computeIfAbsent(tenantId, id -> new TenantEntityProfileCache());
cache.setMyPartitions(partitions); cache.setMyPartitions(partitions);
}); });
tenantCache.keySet().stream()
.filter(tenantId -> !tenantPartitions.containsKey(tenantId))
.forEach(tenantId -> {
var cache = tenantCache.get(tenantId);
cache.setMyPartitions(systemPartitions);
});
} }
@Override @Override

Loading…
Cancel
Save