From ffdb16766ce033a4002bad6a4675693178230178 Mon Sep 17 00:00:00 2001 From: YevhenBondarenko Date: Thu, 6 Jul 2023 13:31:25 +0200 Subject: [PATCH] improvements --- .../queue/discovery/ZkDiscoveryService.java | 17 ++++++++++------- 1 file changed, 10 insertions(+), 7 deletions(-) diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/discovery/ZkDiscoveryService.java b/common/queue/src/main/java/org/thingsboard/server/queue/discovery/ZkDiscoveryService.java index 17d046a4cb..24a7863b24 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/discovery/ZkDiscoveryService.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/discovery/ZkDiscoveryService.java @@ -299,16 +299,16 @@ public class ZkDiscoveryService implements DiscoveryService, PathChildrenCacheLi case CHILD_ADDED: ScheduledFuture task = delayedTasks.remove(instance.getServiceId()); if (task != null) { - if (!task.cancel(false)) { - log.debug("[{}] Going to recalculate partitions due to adding new node [{}]", + if (task.cancel(false)) { + log.debug("[{}] Recalculate partitions ignored. Service was restarted in time [{}].", instance.getServiceId(), instance.getServiceTypesList()); - recalculatePartitions(); } else { - log.debug("[{}] Recalculate partitions ignored. Service restarted in time [{}]", + log.debug("[{}] Going to recalculate partitions. Service was not restarted in time [{}]!", instance.getServiceId(), instance.getServiceTypesList()); + recalculatePartitions(); } } else { - log.debug("[{}] Going to recalculate partitions due to adding new node [{}]", + log.debug("[{}] Going to recalculate partitions due to adding new node [{}].", instance.getServiceId(), instance.getServiceTypesList()); recalculatePartitions(); } @@ -317,8 +317,10 @@ public class ZkDiscoveryService implements DiscoveryService, PathChildrenCacheLi ScheduledFuture future = zkExecutorService.schedule(() -> { log.debug("[{}] Going to recalculate partitions due to removed node [{}]", instance.getServiceId(), instance.getServiceTypesList()); - delayedTasks.remove(instance.getServiceId()); - recalculatePartitions(); + ScheduledFuture removedTask = delayedTasks.remove(instance.getServiceId()); + if (removedTask != null) { + recalculatePartitions(); + } }, recalculateDelay, TimeUnit.MILLISECONDS); delayedTasks.put(instance.getServiceId(), future); break; @@ -332,6 +334,7 @@ public class ZkDiscoveryService implements DiscoveryService, PathChildrenCacheLi * Synchronized to ensure that other servers info is up to date * */ synchronized void recalculatePartitions() { + delayedTasks.clear(); partitionService.recalculatePartitions(serviceInfoProvider.getServiceInfo(), getOtherServers()); }