Fix for race condition in the partition change events

@@ -34,6 +34,7 @@ import org.thingsboard.server.actors.app.AppInitMsg;
 import org.thingsboard.server.actors.stats.StatsActor;
 import org.thingsboard.server.common.msg.queue.PartitionChangeMsg;
 import org.thingsboard.server.queue.discovery.PartitionChangeEvent;
+import org.thingsboard.server.queue.discovery.TbApplicationEventListener;
 
 import javax.annotation.PostConstruct;
 import javax.annotation.PreDestroy;
@@ -43,7 +44,7 @@ import java.util.concurrent.ScheduledExecutorService;
 
 @Service
 @Slf4j
-public class DefaultActorService implements ActorService {
+public class DefaultActorService extends TbApplicationEventListener<PartitionChangeEvent> implements ActorService {
 
     public static final String APP_DISPATCHER_NAME = "app-dispatcher";
     public static final String TENANT_DISPATCHER_NAME = "tenant-dispatcher";
@@ -120,10 +121,10 @@ public class DefaultActorService implements ActorService {
         appActor.tellWithHighPriority(new AppInitMsg());
     }
 
-    @EventListener(PartitionChangeEvent.class)
-    public void onApplicationEvent(PartitionChangeEvent partitionChangeEvent) {
+    @Override
+    protected void onTbApplicationEvent(PartitionChangeEvent event) {
         log.info("Received partition change event.");
-        this.appActor.tellWithHighPriority(new PartitionChangeMsg(partitionChangeEvent.getServiceQueueKey(), partitionChangeEvent.getPartitions()));
+        this.appActor.tellWithHighPriority(new PartitionChangeMsg(event.getServiceQueueKey(), event.getPartitions()));
     }
 
     @PreDestroy

@@ -54,6 +54,7 @@ import org.thingsboard.server.gen.transport.TransportProtos.UsageStatsKVProto;
 import org.thingsboard.server.queue.common.TbProtoQueueMsg;
 import org.thingsboard.server.queue.discovery.PartitionChangeEvent;
 import org.thingsboard.server.queue.discovery.PartitionService;
+import org.thingsboard.server.queue.discovery.TbApplicationEventListener;
 import org.thingsboard.server.queue.scheduler.SchedulerComponent;
 import org.thingsboard.server.service.queue.TbClusterService;
 import org.thingsboard.server.service.telemetry.InternalTelemetryService;
@@ -78,7 +79,7 @@ import java.util.stream.Collectors;
 
 @Slf4j
 @Service
-public class DefaultTbApiUsageStateService implements TbApiUsageStateService {
+public class DefaultTbApiUsageStateService extends TbApplicationEventListener<PartitionChangeEvent> implements TbApiUsageStateService {
 
     public static final String HOURLY = "Hourly";
     public static final FutureCallback<Integer> VOID_CALLBACK = new FutureCallback<Integer>() {
@@ -188,7 +189,7 @@ public class DefaultTbApiUsageStateService implements TbApiUsageStateService {
     }
 
     @Override
-    public void onApplicationEvent(PartitionChangeEvent partitionChangeEvent) {
+    protected void onTbApplicationEvent(PartitionChangeEvent partitionChangeEvent) {
         if (partitionChangeEvent.getServiceType().equals(ServiceType.TB_CORE)) {
             myTenantStates.entrySet().removeIf(entry -> !partitionService.resolve(ServiceType.TB_CORE, entry.getKey(), entry.getKey()).isMyPartition());
             otherTenantStates.entrySet().removeIf(entry -> partitionService.resolve(ServiceType.TB_CORE, entry.getKey(), entry.getKey()).isMyPartition());

@@ -151,12 +151,12 @@ public class DefaultTbCoreConsumerService extends AbstractConsumerService<ToCore
     }
 
     @Override
-    public void onApplicationEvent(PartitionChangeEvent partitionChangeEvent) {
-        if (partitionChangeEvent.getServiceType().equals(getServiceType())) {
-            log.info("Subscribing to partitions: {}", partitionChangeEvent.getPartitions());
-            this.mainConsumer.subscribe(partitionChangeEvent.getPartitions());
+    protected void onTbApplicationEvent(PartitionChangeEvent event) {
+        if (event.getServiceType().equals(getServiceType())) {
+            log.info("Subscribing to partitions: {}", event.getPartitions());
+            this.mainConsumer.subscribe(event.getPartitions());
             this.usageStatsConsumer.subscribe(
-                    partitionChangeEvent
+                    event
                             .getPartitions()
                             .stream()
                             .map(tpi -> tpi.newByTopic(usageStatsConsumer.getTopic()))

@@ -140,11 +140,11 @@ public class DefaultTbRuleEngineConsumerService extends AbstractConsumerService<
     }
 
     @Override
-    public void onApplicationEvent(PartitionChangeEvent partitionChangeEvent) {
-        if (partitionChangeEvent.getServiceType().equals(getServiceType())) {
-            ServiceQueue serviceQueue = partitionChangeEvent.getServiceQueueKey().getServiceQueue();
-            log.info("[{}] Subscribing to partitions: {}", serviceQueue.getQueue(), partitionChangeEvent.getPartitions());
-            consumers.get(serviceQueue.getQueue()).subscribe(partitionChangeEvent.getPartitions());
+    protected void onTbApplicationEvent(PartitionChangeEvent event) {
+        if (event.getServiceType().equals(getServiceType())) {
+            ServiceQueue serviceQueue = event.getServiceQueueKey().getServiceQueue();
+            log.info("[{}] Subscribing to partitions: {}", serviceQueue.getQueue(), event.getPartitions());
+            consumers.get(serviceQueue.getQueue()).subscribe(event.getPartitions());
         }
     }
 

@@ -36,6 +36,7 @@ import org.thingsboard.server.queue.TbQueueConsumer;
 import org.thingsboard.server.queue.common.TbProtoQueueMsg;
 import org.thingsboard.server.queue.discovery.PartitionChangeEvent;
 import org.thingsboard.server.common.transport.util.DataDecodingEncodingService;
+import org.thingsboard.server.queue.discovery.TbApplicationEventListener;
 import org.thingsboard.server.service.apiusage.TbApiUsageStateService;
 import org.thingsboard.server.service.profile.TbDeviceProfileCache;
 import org.thingsboard.server.dao.tenant.TbTenantProfileCache;
@@ -56,7 +57,7 @@ import java.util.function.Function;
 import java.util.stream.Collectors;
 
 @Slf4j
-public abstract class AbstractConsumerService<N extends com.google.protobuf.GeneratedMessageV3> implements ApplicationListener<PartitionChangeEvent> {
+public abstract class AbstractConsumerService<N extends com.google.protobuf.GeneratedMessageV3> extends TbApplicationEventListener<PartitionChangeEvent> {
 
     protected volatile ExecutorService consumersExecutor;
     protected volatile ExecutorService notificationsConsumerExecutor;

@@ -56,6 +56,7 @@ import org.thingsboard.server.dao.util.mapping.JacksonUtil;
 import org.thingsboard.server.gen.transport.TransportProtos;
 import org.thingsboard.server.queue.discovery.PartitionChangeEvent;
 import org.thingsboard.server.queue.discovery.PartitionService;
+import org.thingsboard.server.queue.discovery.TbApplicationEventListener;
 import org.thingsboard.server.queue.util.TbCoreComponent;
 import org.thingsboard.server.service.queue.TbClusterService;
 import org.thingsboard.server.service.telemetry.TelemetrySubscriptionService;
@@ -90,7 +91,7 @@ import static org.thingsboard.server.common.data.DataConstants.SERVER_SCOPE;
 @Service
 @TbCoreComponent
 @Slf4j
-public class DefaultDeviceStateService implements DeviceStateService {
+public class DefaultDeviceStateService extends TbApplicationEventListener<PartitionChangeEvent> implements DeviceStateService {
 
     public static final String ACTIVITY_STATE = "active";
     public static final String LAST_CONNECT_TIME = "lastConnectTime";
@@ -294,7 +295,7 @@ public class DefaultDeviceStateService implements DeviceStateService {
     }
 
     @Override
-    public void onApplicationEvent(PartitionChangeEvent partitionChangeEvent) {
+    protected void onTbApplicationEvent(PartitionChangeEvent partitionChangeEvent) {
         if (ServiceType.TB_CORE.equals(partitionChangeEvent.getServiceType())) {
             deduplicationExecutor.submit(partitionChangeEvent.getPartitions());
         }

@@ -48,6 +48,7 @@ import org.thingsboard.server.queue.TbQueueProducer;
 import org.thingsboard.server.queue.common.TbProtoQueueMsg;
 import org.thingsboard.server.queue.discovery.PartitionChangeEvent;
 import org.thingsboard.server.queue.discovery.PartitionService;
+import org.thingsboard.server.queue.discovery.TbApplicationEventListener;
 import org.thingsboard.server.queue.discovery.TbServiceInfoProvider;
 import org.thingsboard.server.queue.provider.TbQueueProducerProvider;
 import org.thingsboard.server.queue.util.TbCoreComponent;
@@ -76,7 +77,7 @@ import java.util.function.Predicate;
 @Slf4j
 @TbCoreComponent
 @Service
-public class DefaultSubscriptionManagerService implements SubscriptionManagerService {
+public class DefaultSubscriptionManagerService extends TbApplicationEventListener<PartitionChangeEvent> implements SubscriptionManagerService {
 
     @Autowired
     private AttributesService attrService;
@@ -178,7 +179,7 @@ public class DefaultSubscriptionManagerService implements SubscriptionManagerSer
     }
 
     @Override
-    public void onApplicationEvent(PartitionChangeEvent partitionChangeEvent) {
+    protected void onTbApplicationEvent(PartitionChangeEvent partitionChangeEvent) {
         if (ServiceType.TB_CORE.equals(partitionChangeEvent.getServiceType())) {
             Set<TopicPartitionInfo> removedPartitions = new HashSet<>(currentPartitions);
             removedPartitions.removeAll(partitionChangeEvent.getPartitions());

@@ -28,6 +28,7 @@ import org.thingsboard.server.queue.discovery.PartitionService;
 import org.thingsboard.server.common.msg.queue.ServiceType;
 import org.thingsboard.server.common.msg.queue.TopicPartitionInfo;
 import org.thingsboard.server.common.msg.queue.TbCallback;
+import org.thingsboard.server.queue.discovery.TbApplicationEventListener;
 import org.thingsboard.server.queue.util.TbCoreComponent;
 import org.thingsboard.server.service.queue.TbClusterService;
 import org.thingsboard.server.service.telemetry.sub.AlarmSubscriptionUpdate;
@@ -62,6 +63,34 @@ public class DefaultTbLocalSubscriptionService implements TbLocalSubscriptionSer
     private SubscriptionManagerService subscriptionManagerService;
 
     private ExecutorService subscriptionUpdateExecutor;
+    
+    private TbApplicationEventListener<PartitionChangeEvent> partitionChangeListener = new TbApplicationEventListener<>() {
+        @Override
+        protected void onTbApplicationEvent(PartitionChangeEvent event) {
+            if (ServiceType.TB_CORE.equals(event.getServiceType())) {
+                currentPartitions.clear();
+                currentPartitions.addAll(event.getPartitions());
+            }
+        }
+    };
+
+    private TbApplicationEventListener<ClusterTopologyChangeEvent> clusterTopologyChangeListener = new TbApplicationEventListener<>() {
+        @Override
+        protected void onTbApplicationEvent(ClusterTopologyChangeEvent event) {
+            if (event.getServiceQueueKeys().stream().anyMatch(key -> ServiceType.TB_CORE.equals(key.getServiceType()))) {
+                /*
+                 * If the cluster topology has changed, we need to push all current subscriptions to SubscriptionManagerService again.
+                 * Otherwise, the SubscriptionManagerService may "forget" those subscriptions in case of restart.
+                 * Although this is resource consuming operation, it is cheaper than sending ping/pong commands periodically
+                 * It is also cheaper then caching the subscriptions by entity id and then lookup of those caches every time we have new telemetry in SubscriptionManagerService.
+                 * Even if we cache locally the list of active subscriptions by entity id, it is still time consuming operation to get them from cache
+                 * Since number of subscriptions is usually much less then number of devices that are pushing data.
+                 */
+                subscriptionsBySessionId.values().forEach(map -> map.values()
+                        .forEach(sub -> pushSubscriptionToManagerService(sub, true)));
+            }
+        }
+    };
 
     @PostConstruct
     public void initExecutor() {
@@ -77,28 +106,14 @@ public class DefaultTbLocalSubscriptionService implements TbLocalSubscriptionSer
 
     @Override
     @EventListener(PartitionChangeEvent.class)
-    public void onApplicationEvent(PartitionChangeEvent partitionChangeEvent) {
-        if (ServiceType.TB_CORE.equals(partitionChangeEvent.getServiceType())) {
-            currentPartitions.clear();
-            currentPartitions.addAll(partitionChangeEvent.getPartitions());
-        }
+    public void onApplicationEvent(PartitionChangeEvent event) {
+        partitionChangeListener.onApplicationEvent(event);
     }
 
     @Override
     @EventListener(ClusterTopologyChangeEvent.class)
     public void onApplicationEvent(ClusterTopologyChangeEvent event) {
-        if (event.getServiceQueueKeys().stream().anyMatch(key -> ServiceType.TB_CORE.equals(key.getServiceType()))) {
-            /*
-             * If the cluster topology has changed, we need to push all current subscriptions to SubscriptionManagerService again.
-             * Otherwise, the SubscriptionManagerService may "forget" those subscriptions in case of restart.
-             * Although this is resource consuming operation, it is cheaper than sending ping/pong commands periodically
-             * It is also cheaper then caching the subscriptions by entity id and then lookup of those caches every time we have new telemetry in SubscriptionManagerService.
-             * Even if we cache locally the list of active subscriptions by entity id, it is still time consuming operation to get them from cache
-             * Since number of subscriptions is usually much less then number of devices that are pushing data.
-             */
-            subscriptionsBySessionId.values().forEach(map -> map.values()
-                    .forEach(sub -> pushSubscriptionToManagerService(sub, true)));
-        }
+        clusterTopologyChangeListener.onApplicationEvent(event);
     }
 
     //TODO 3.1: replace null callbacks with callbacks from websocket service.

@@ -41,6 +41,7 @@ import org.thingsboard.server.dao.timeseries.TimeseriesService;
 import org.thingsboard.server.gen.transport.TransportProtos;
 import org.thingsboard.server.queue.discovery.PartitionChangeEvent;
 import org.thingsboard.server.queue.discovery.PartitionService;
+import org.thingsboard.server.queue.discovery.TbApplicationEventListener;
 import org.thingsboard.server.service.queue.TbClusterService;
 import org.thingsboard.server.service.subscription.SubscriptionManagerService;
 import org.thingsboard.server.service.subscription.TbSubscriptionUtils;
@@ -61,7 +62,7 @@ import java.util.function.Consumer;
  * Created by ashvayka on 27.03.18.
  */
 @Slf4j
-public abstract class AbstractSubscriptionService implements ApplicationListener<PartitionChangeEvent> {
+public abstract class AbstractSubscriptionService extends TbApplicationEventListener<PartitionChangeEvent>{
 
     protected final Set<TopicPartitionInfo> currentPartitions = ConcurrentHashMap.newKeySet();
 
@@ -97,8 +98,7 @@ public abstract class AbstractSubscriptionService implements ApplicationListener
     }
 
     @Override
-    @EventListener(PartitionChangeEvent.class)
-    public void onApplicationEvent(PartitionChangeEvent partitionChangeEvent) {
+    protected void onTbApplicationEvent(PartitionChangeEvent partitionChangeEvent) {
         if (ServiceType.TB_CORE.equals(partitionChangeEvent.getServiceType())) {
             currentPartitions.clear();
             currentPartitions.addAll(partitionChangeEvent.getPartitions());

@@ -22,7 +22,9 @@ import org.thingsboard.server.common.msg.queue.ServiceQueueKey;
 import java.util.Set;
 
 
-public class ClusterTopologyChangeEvent extends ApplicationEvent {
+public class ClusterTopologyChangeEvent extends TbApplicationEvent {
+
+    private static final long serialVersionUID = -2441739930040282254L;
 
     @Getter
     private final Set<ServiceQueueKey> serviceQueueKeys;

@@ -126,7 +126,7 @@ public class HashPartitionService implements PartitionService {
     }
 
     @Override
-    public void recalculatePartitions(ServiceInfo currentService, List<ServiceInfo> otherServices) {
+    public synchronized void recalculatePartitions(ServiceInfo currentService, List<ServiceInfo> otherServices) {
         logServiceInfo(currentService);
         otherServices.forEach(this::logServiceInfo);
         Map<ServiceQueueKey, List<ServiceInfo>> queueServicesMap = new HashMap<>();
@@ -134,7 +134,7 @@ public class HashPartitionService implements PartitionService {
         for (ServiceInfo other : otherServices) {
             addNode(queueServicesMap, other);
         }
-        queueServicesMap.values().forEach(list -> list.sort((a, b) -> a.getServiceId().compareTo(b.getServiceId())));
+        queueServicesMap.values().forEach(list -> list.sort(Comparator.comparing(ServiceInfo::getServiceId)));
 
         ConcurrentMap<ServiceQueueKey, List<Integer>> oldPartitions = myPartitions;
         TenantId myIsolatedOrSystemTenantId = getSystemOrIsolatedTenantId(currentService);

@@ -24,7 +24,9 @@ import org.thingsboard.server.common.msg.queue.TopicPartitionInfo;
 import java.util.Set;
 
 
-public class PartitionChangeEvent extends ApplicationEvent {
+public class PartitionChangeEvent extends TbApplicationEvent {
+
+    private static final long serialVersionUID = -8731788167026510559L;
 
     @Getter
     private final ServiceQueueKey serviceQueueKey;

+/**
+ * Copyright © 2016-2021 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
+ *
+ * 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.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.thingsboard.server.queue.discovery;
+
+import lombok.Getter;
+import org.springframework.context.ApplicationEvent;
+
+import java.util.concurrent.atomic.AtomicInteger;
+
+public class TbApplicationEvent extends ApplicationEvent {
+
+    private static final long serialVersionUID = 3884264064887765146L;
+
+    private static final AtomicInteger sequence = new AtomicInteger();
+
+    @Getter
+    private final int sequenceNumber;
+
+    public TbApplicationEvent(Object source) {
+        super(source);
+        sequenceNumber = sequence.incrementAndGet();
+    }
+
+}

+/**
+ * Copyright © 2016-2021 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
+ *
+ * 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.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.thingsboard.server.queue.discovery;
+
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.context.ApplicationListener;
+
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReentrantLock;
+
+@Slf4j
+public abstract class TbApplicationEventListener<T extends TbApplicationEvent> implements ApplicationListener<T> {
+
+    private int lastProcessedSequenceNumber = Integer.MIN_VALUE;
+    private final Lock seqNumberLock = new ReentrantLock();
+
+    @Override
+    public void onApplicationEvent(T event) {
+        boolean validUpdate = false;
+        seqNumberLock.lock();
+        try {
+            if (event.getSequenceNumber() > lastProcessedSequenceNumber) {
+                validUpdate = true;
+                lastProcessedSequenceNumber = event.getSequenceNumber();
+            }
+        } finally {
+            seqNumberLock.unlock();
+        }
+        if (validUpdate) {
+            onTbApplicationEvent(event);
+        } else {
+            log.info("Application event ignored due to invalid sequence number ({} > {}). Event: {}", lastProcessedSequenceNumber, event.getSequenceNumber(), event);
+        }
+    }
+
+    protected abstract void onTbApplicationEvent(T event);
+
+
+}