From f342bd989ef891be8a56e8ea35ba2ca4c0e024e1 Mon Sep 17 00:00:00 2001 From: ViacheslavKlimov Date: Thu, 13 Feb 2025 15:31:09 +0200 Subject: [PATCH] Minor improvements --- .../auth/rest/RestAuthenticationProvider.java | 1 - .../queue/discovery/HashPartitionService.java | 30 ++++++------------- .../server/queue/kafka/KafkaTbQueueMsg.java | 2 +- .../server/queue/util/AfterStartUp.java | 8 ++--- 4 files changed, 14 insertions(+), 27 deletions(-) diff --git a/application/src/main/java/org/thingsboard/server/service/security/auth/rest/RestAuthenticationProvider.java b/application/src/main/java/org/thingsboard/server/service/security/auth/rest/RestAuthenticationProvider.java index 56215b82ac..b9fe54deec 100644 --- a/application/src/main/java/org/thingsboard/server/service/security/auth/rest/RestAuthenticationProvider.java +++ b/application/src/main/java/org/thingsboard/server/service/security/auth/rest/RestAuthenticationProvider.java @@ -106,7 +106,6 @@ public class RestAuthenticationProvider implements AuthenticationProvider { if (twoFactorAuthService.isTwoFaEnabled(securityUser.getTenantId(), securityUser.getId())) { return new MfaAuthenticationToken(securityUser); } else { - systemSecurityService.logLoginAction(securityUser, authentication.getDetails(), ActionType.LOGIN, null); } } else { 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 249dfa859f..751e235adc 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 @@ -228,7 +228,7 @@ public class HashPartitionService implements PartitionService { }); if (serviceInfoProvider.isService(ServiceType.TB_RULE_ENGINE)) { publishPartitionChangeEvent(ServiceType.TB_RULE_ENGINE, queueKeys.stream() - .collect(Collectors.toMap(k -> k, k -> Collections.emptySet())), Collections.emptyMap()); + .collect(Collectors.toMap(k -> k, k -> Collections.emptySet()))); } } @@ -408,7 +408,6 @@ public class HashPartitionService implements PartitionService { myPartitions = newPartitions; Map> changedPartitionsMap = new HashMap<>(); - Map> oldPartitionsMap = new HashMap<>(); Set removed = new HashSet<>(); oldPartitions.forEach((queueKey, partitions) -> { @@ -429,16 +428,16 @@ public class HashPartitionService implements PartitionService { myPartitions.forEach((queueKey, partitions) -> { if (!partitions.equals(oldPartitions.get(queueKey))) { - changedPartitionsMap.put(queueKey, toTpiList(queueKey, partitions)); - oldPartitionsMap.put(queueKey, toTpiList(queueKey, oldPartitions.get(queueKey))); + Set tpiList = partitions.stream() + .map(partition -> buildTopicPartitionInfo(queueKey, partition)) + .collect(Collectors.toSet()); + changedPartitionsMap.put(queueKey, tpiList); } }); if (!changedPartitionsMap.isEmpty()) { changedPartitionsMap.entrySet().stream() .collect(Collectors.groupingBy(entry -> entry.getKey().getType(), Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue))) - .forEach((serviceType, partitionsMap) -> { - publishPartitionChangeEvent(serviceType, partitionsMap, oldPartitionsMap); - }); + .forEach(this::publishPartitionChangeEvent); } if (currentOtherServices == null) { @@ -470,15 +469,13 @@ public class HashPartitionService implements PartitionService { applicationEventPublisher.publishEvent(new ServiceListChangedEvent(otherServices, currentService)); } - private void publishPartitionChangeEvent(ServiceType serviceType, - Map> newPartitions, - Map> oldPartitions) { - log.info("Partitions changed: {}", System.lineSeparator() + newPartitions.entrySet().stream() + private void publishPartitionChangeEvent(ServiceType serviceType, Map> partitionsMap) { + log.info("Partitions changed: {}", System.lineSeparator() + partitionsMap.entrySet().stream() .map(entry -> "[" + entry.getKey() + "] - [" + entry.getValue().stream() .map(tpi -> tpi.getPartition().orElse(-1).toString()).sorted() .collect(Collectors.joining(", ")) + "]") .collect(Collectors.joining(System.lineSeparator()))); - PartitionChangeEvent event = new PartitionChangeEvent(this, serviceType, newPartitions); + PartitionChangeEvent event = new PartitionChangeEvent(this, serviceType, partitionsMap); try { applicationEventPublisher.publishEvent(event); } catch (Exception e) { @@ -486,15 +483,6 @@ public class HashPartitionService implements PartitionService { } } - private Set toTpiList(QueueKey queueKey, List partitions) { - if (partitions == null) { - return Collections.emptySet(); - } - return partitions.stream() - .map(partition -> buildTopicPartitionInfo(queueKey, partition)) - .collect(Collectors.toSet()); - } - @Override public Set getAllServiceIds(ServiceType serviceType) { return getAllServices(serviceType).stream().map(ServiceInfo::getServiceId).collect(Collectors.toSet()); diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/kafka/KafkaTbQueueMsg.java b/common/queue/src/main/java/org/thingsboard/server/queue/kafka/KafkaTbQueueMsg.java index fe2c8b8a90..302f36ff1a 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/kafka/KafkaTbQueueMsg.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/kafka/KafkaTbQueueMsg.java @@ -31,7 +31,7 @@ public class KafkaTbQueueMsg implements TbQueueMsg { private final byte[] data; public KafkaTbQueueMsg(ConsumerRecord record) { - if (record.key().length() == UUID_LENGTH) { + if (record.key().length() <= UUID_LENGTH) { this.key = UUID.fromString(record.key()); } else { this.key = UUID.randomUUID(); diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/util/AfterStartUp.java b/common/queue/src/main/java/org/thingsboard/server/queue/util/AfterStartUp.java index a24f0663b5..6943fa6753 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/util/AfterStartUp.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/util/AfterStartUp.java @@ -1,12 +1,12 @@ /** * Copyright © 2016-2024 The Thingsboard Authors - *

+ * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. * You may obtain a copy of the License at - *

- * http://www.apache.org/licenses/LICENSE-2.0 - *

+ * + * http://www.apache.org/licenses/LICENSE-2.0 + * * Unless required by applicable law or agreed to in writing, software * distributed under the License is distributed on an "AS IS" BASIS, * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.