Merge pull request #8808 from volodymyr-babak/app-init-sequence

HotFix - fixed init of rule chains - init only on APP_INIT msg

application/src/main/java/org/thingsboard/server/actors/app/AppActor.java

@@ -73,10 +73,14 @@ public class AppActor extends ContextAwareActor {
     @Override
     protected boolean doProcess(TbActorMsg msg) {
         if (!ruleChainsInitialized) {
-            initTenantActors();
-            ruleChainsInitialized = true;
-            if (msg.getMsgType() != MsgType.APP_INIT_MSG && msg.getMsgType() != MsgType.PARTITION_CHANGE_MSG) {
-                log.warn("Rule Chains initialized by unexpected message: {}", msg);
+            if (MsgType.APP_INIT_MSG.equals(msg.getMsgType())) {
+                initTenantActors();
+                ruleChainsInitialized = true;
+            } else {
+                if (!msg.getMsgType().isIgnoreOnStart()) {
+                    log.warn("Attempt to initialize Rule Chains by unexpected message: {}", msg);
+                }
+                return true;
             }
         }
         switch (msg.getMsgType()) {

application/src/main/java/org/thingsboard/server/service/queue/DefaultTbRuleEngineConsumerService.java

@@ -259,7 +259,21 @@ public class DefaultTbRuleEngineConsumerService extends AbstractConsumerService<
     }
 
     void launchConsumer(TbQueueConsumer<TbProtoQueueMsg<ToRuleEngineMsg>> consumer, Queue configuration, TbRuleEngineConsumerStats stats, String threadSuffix) {
-        consumersExecutor.execute(() -> consumerLoop(consumer, configuration, stats, threadSuffix));
+        if (isReady) {
+            consumersExecutor.execute(() -> consumerLoop(consumer, configuration, stats, threadSuffix));
+        } else {
+            scheduleLaunchConsumer(consumer, configuration, stats, threadSuffix);
+        }
+    }
+
+    private void scheduleLaunchConsumer(TbQueueConsumer<TbProtoQueueMsg<ToRuleEngineMsg>> consumer, Queue configuration, TbRuleEngineConsumerStats stats, String threadSuffix) {
+        repartitionExecutor.schedule(() -> {
+            if (isReady) {
+                consumersExecutor.execute(() -> consumerLoop(consumer, configuration, stats, threadSuffix));
+            } else {
+                scheduleLaunchConsumer(consumer, configuration, stats, threadSuffix);
+            }
+        }, 10, TimeUnit.SECONDS);
     }
 
     void consumerLoop(TbQueueConsumer<TbProtoQueueMsg<ToRuleEngineMsg>> consumer, org.thingsboard.server.common.data.queue.Queue configuration, TbRuleEngineConsumerStats stats, String threadSuffix) {

application/src/main/java/org/thingsboard/server/service/queue/processing/AbstractConsumerService.java

@@ -68,7 +68,7 @@ public abstract class AbstractConsumerService<N extends com.google.protobuf.Gene
     protected volatile ExecutorService consumersExecutor;
     protected volatile ExecutorService notificationsConsumerExecutor;
     protected volatile boolean stopped = false;
-
+    protected volatile boolean isReady = false;
     protected final ActorSystemContext actorContext;
     protected final DataDecodingEncodingService encodingService;
     protected final TbTenantProfileCache tenantProfileCache;
@@ -108,6 +108,7 @@ public abstract class AbstractConsumerService<N extends com.google.protobuf.Gene
     public void onApplicationEvent(ApplicationReadyEvent event) {
         log.info("Subscribing to notifications: {}", nfConsumer.getTopic());
         this.nfConsumer.subscribe();
+        this.isReady = true;
         launchNotificationsConsumer();
         launchMainConsumers();
     }

common/message/src/main/java/org/thingsboard/server/common/msg/MsgType.java

@@ -15,6 +15,7 @@
  */
 package org.thingsboard.server.common.msg;
 
+import lombok.Getter;
 import org.thingsboard.server.common.msg.queue.PartitionChangeMsg;
 import org.thingsboard.server.common.msg.queue.QueueToRuleEngineMsg;
 
@@ -28,7 +29,7 @@ public enum MsgType {
      *
      * See {@link PartitionChangeMsg}
      */
-    PARTITION_CHANGE_MSG,
+    PARTITION_CHANGE_MSG(true),
 
     APP_INIT_MSG,
 
@@ -108,7 +109,7 @@ public enum MsgType {
      * Message that is sent from the Device Actor to Rule Engine. Requires acknowledgement
      */
 
-    SESSION_TIMEOUT_MSG,
+    SESSION_TIMEOUT_MSG(true),
 
     STATS_PERSIST_TICK_MSG,
 
@@ -130,4 +131,14 @@ public enum MsgType {
     EDGE_SYNC_REQUEST_TO_EDGE_SESSION_MSG,
     EDGE_SYNC_RESPONSE_FROM_EDGE_SESSION_MSG;
 
+    @Getter
+    private final boolean ignoreOnStart;
+
+    MsgType() {
+        this.ignoreOnStart = false;
+    }
+
+    MsgType(boolean ignoreOnStart) {
+        this.ignoreOnStart = ignoreOnStart;
+    }
 }

rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/profile/TbDeviceProfileNode.java

@@ -76,6 +76,10 @@ public class TbDeviceProfileNode implements TbNode {
         this.ctx = ctx;
         scheduleAlarmHarvesting(ctx, null);
         ctx.addDeviceProfileListeners(this::onProfileUpdate, this::onDeviceUpdate);
+        initAlarmRuleState(false);
+    }
+
+    private void initAlarmRuleState(boolean printNewlyAddedDeviceStates) {
         if (config.isFetchAlarmRulesStateOnStart()) {
             log.info("[{}] Fetching alarm rule state", ctx.getSelfId());
             int fetchCount = 0;
@@ -86,7 +90,7 @@ public class TbDeviceProfileNode implements TbNode {
                     for (RuleNodeState rns : states.getData()) {
                         fetchCount++;
                         if (rns.getEntityId().getEntityType().equals(EntityType.DEVICE) && ctx.isLocalEntity(rns.getEntityId())) {
-                            getOrCreateDeviceState(ctx, new DeviceId(rns.getEntityId().getId()), rns);
+                            getOrCreateDeviceState(ctx, new DeviceId(rns.getEntityId().getId()), rns, printNewlyAddedDeviceStates);
                         }
                     }
                 }
@@ -130,7 +134,7 @@ public class TbDeviceProfileNode implements TbNode {
                     removeDeviceState(deviceId);
                     ctx.tellSuccess(msg);
                 } else {
-                    DeviceState deviceState = getOrCreateDeviceState(ctx, deviceId, null);
+                    DeviceState deviceState = getOrCreateDeviceState(ctx, deviceId, null, false);
                     if (deviceState != null) {
                         deviceState.process(ctx, msg);
                     } else {
@@ -148,6 +152,7 @@ public class TbDeviceProfileNode implements TbNode {
     public void onPartitionChangeMsg(TbContext ctx, PartitionChangeMsg msg) {
         // Cleanup the cache for all entities that are no longer assigned to current server partitions
         deviceStates.entrySet().removeIf(entry -> !ctx.isLocalEntity(entry.getKey()));
+        initAlarmRuleState(true);
     }
 
     @Override
@@ -156,13 +161,16 @@ public class TbDeviceProfileNode implements TbNode {
         deviceStates.clear();
     }
 
-    protected DeviceState getOrCreateDeviceState(TbContext ctx, DeviceId deviceId, RuleNodeState rns) {
+    protected DeviceState getOrCreateDeviceState(TbContext ctx, DeviceId deviceId, RuleNodeState rns, boolean printNewlyAddedDeviceStates) {
         DeviceState deviceState = deviceStates.get(deviceId);
         if (deviceState == null) {
             DeviceProfile deviceProfile = cache.get(ctx.getTenantId(), deviceId);
             if (deviceProfile != null) {
                 deviceState = new DeviceState(ctx, config, deviceId, new ProfileState(deviceProfile), rns);
                 deviceStates.put(deviceId, deviceState);
+                if (printNewlyAddedDeviceStates) {
+                    log.info("[{}][{}] Device [{}] was added during PartitionChangeMsg", ctx.getTenantId(), ctx.getSelfId(), deviceId);
+                }
             }
         }
         return deviceState;