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

Improve partition service logs

Browse Source
This commit is contained in:
ViacheslavKlimov 2024-01-25 12:04:18 +02:00
parent a571153b7c
commit 27f448a543
1 changed files with 6 additions and 11 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -325,13 +325,11 @@ public class HashPartitionService implements PartitionService {
.forEach(removed::add); .forEach(removed::add);
} }
removed.forEach(queueKey -> { removed.forEach(queueKey -> {
log.info("[{}] NO MORE PARTITIONS FOR CURRENT KEY", queueKey);
changedPartitionsMap.put(queueKey, Collections.emptySet()); changedPartitionsMap.put(queueKey, Collections.emptySet());
}); });
myPartitions.forEach((queueKey, partitions) -> { myPartitions.forEach((queueKey, partitions) -> {
if (!partitions.equals(oldPartitions.get(queueKey))) { if (!partitions.equals(oldPartitions.get(queueKey))) {
log.info("[{}] NEW PARTITIONS: {}", queueKey, partitions);
Set<TopicPartitionInfo> tpiList = partitions.stream() Set<TopicPartitionInfo> tpiList = partitions.stream()
.map(partition -> buildTopicPartitionInfo(queueKey, partition)) .map(partition -> buildTopicPartitionInfo(queueKey, partition))
.collect(Collectors.toSet()); .collect(Collectors.toSet());
@ -377,14 +375,11 @@ public class HashPartitionService implements PartitionService {
} }
private void publishPartitionChangeEvent(ServiceType serviceType, Map<QueueKey, Set<TopicPartitionInfo>> partitionsMap) { private void publishPartitionChangeEvent(ServiceType serviceType, Map<QueueKey, Set<TopicPartitionInfo>> partitionsMap) {
if (log.isDebugEnabled()) { log.info("Partitions changed: {}", System.lineSeparator() + partitionsMap.entrySet().stream()
log.debug("Publishing partition change event for service type " + serviceType + ":" + System.lineSeparator() + .map(entry -> "[" + entry.getKey() + "] - [" + entry.getValue().stream()
partitionsMap.entrySet().stream() .map(tpi -> tpi.getPartition().orElse(-1).toString()).sorted()
.map(entry -> entry.getKey() + " - " + entry.getValue().stream() .collect(Collectors.joining(", ")) + "]")
.map(TopicPartitionInfo::getFullTopicName).sorted()
.collect(Collectors.toList()))
.collect(Collectors.joining(System.lineSeparator()))); .collect(Collectors.joining(System.lineSeparator())));
}
applicationEventPublisher.publishEvent(new PartitionChangeEvent(this, serviceType, partitionsMap)); applicationEventPublisher.publishEvent(new PartitionChangeEvent(this, serviceType, partitionsMap));
} }
@ -489,7 +484,7 @@ public class HashPartitionService implements PartitionService {
} }
private void logServiceInfo(TransportProtos.ServiceInfo server) { private void logServiceInfo(TransportProtos.ServiceInfo server) {
log.info("[{}] Found common server: [{}]", server.getServiceId(), server.getServiceTypesList()); log.info("[{}] Found common server: {}", server.getServiceId(), server.getServiceTypesList());
} }
private void addNode(Map<QueueKey, List<ServiceInfo>> queueServiceList, ServiceInfo instance) { private void addNode(Map<QueueKey, List<ServiceInfo>> queueServiceList, ServiceInfo instance) {