Merge pull request thingsboard#10255 from thingsboard/fix/actor-system

Fix actor msg broadcast

Author: ashvayka

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

@@ -17,6 +17,7 @@
 
 import lombok.extern.slf4j.Slf4j;
 import org.thingsboard.server.actors.ActorSystemContext;
+import org.thingsboard.server.actors.ProcessFailureStrategy;
 import org.thingsboard.server.actors.TbActor;
 import org.thingsboard.server.actors.TbActorCtx;
 import org.thingsboard.server.actors.TbActorException;
@@ -88,7 +89,7 @@ protected boolean doProcess(TbActorMsg msg) {
             case APP_INIT_MSG:
                 break;
             case PARTITION_CHANGE_MSG:
-                ctx.broadcastToChildren(msg);
+                ctx.broadcastToChildren(msg, true);
                 break;
             case COMPONENT_LIFE_CYCLE_MSG:
                 onComponentLifecycleMsg((ComponentLifecycleMsg) msg);
@@ -219,6 +220,12 @@ private void onToEdgeSessionMsg(EdgeSessionMsg msg) {
         }
     }
 
+    @Override
+    public ProcessFailureStrategy onProcessFailure(TbActorMsg msg, Throwable t) {
+        log.error("Failed to process msg: {}", msg, t);
+        return doProcessFailure(t);
+    }
+
     public static class ActorCreator extends ContextBasedCreator {
 
         public ActorCreator(ActorSystemContext context) {

application/src/main/java/org/thingsboard/server/actors/ruleChain/RuleChainManagerActor.java

@@ -59,17 +59,19 @@ public RuleChainManagerActor(ActorSystemContext systemContext, TenantId tenantId
     }
 
     protected void initRuleChains() {
-        ruleChainsInitialized = true;
+        log.debug("[{}] Initializing rule chains", tenantId);
         for (RuleChain ruleChain : new PageDataIterable<>(link -> ruleChainService.findTenantRuleChainsByType(tenantId, RuleChainType.CORE, link), ContextAwareActor.ENTITY_PACK_LIMIT)) {
             RuleChainId ruleChainId = ruleChain.getId();
             log.debug("[{}|{}] Creating rule chain actor", ruleChainId.getEntityType(), ruleChain.getId());
             TbActorRef actorRef = getOrCreateActor(ruleChainId, id -> ruleChain);
             visit(ruleChain, actorRef);
             log.debug("[{}|{}] Rule Chain actor created.", ruleChainId.getEntityType(), ruleChainId.getId());
         }
+        ruleChainsInitialized = true;
     }
 
     protected void destroyRuleChains() {
+        log.debug("[{}] Destroying rule chains", tenantId);
         for (RuleChain ruleChain : new PageDataIterable<>(link -> ruleChainService.findTenantRuleChainsByType(tenantId, RuleChainType.CORE, link), ContextAwareActor.ENTITY_PACK_LIMIT)) {
             ctx.stop(new TbEntityActorId(ruleChain.getId()));
         }

application/src/main/java/org/thingsboard/server/actors/service/ContextAwareActor.java

@@ -47,8 +47,8 @@ public boolean process(TbActorMsg msg) {
     protected abstract boolean doProcess(TbActorMsg msg);
 
     @Override
-    public ProcessFailureStrategy onProcessFailure(Throwable t) {
-        log.debug("[{}] Processing failure: ", getActorRef().getActorId(), t);
+    public ProcessFailureStrategy onProcessFailure(TbActorMsg msg, Throwable t) {
+        log.debug("[{}] Processing failure for msg {}", getActorRef().getActorId(), msg, t);
         return doProcessFailure(t);
     }
 

application/src/main/java/org/thingsboard/server/actors/service/DefaultActorService.java

@@ -122,7 +122,7 @@ public void onApplicationEvent(ApplicationReadyEvent applicationReadyEvent) {
     @Override
     protected void onTbApplicationEvent(PartitionChangeEvent event) {
         log.info("Received partition change event.");
-        this.appActor.tellWithHighPriority(new PartitionChangeMsg(event.getServiceType()));
+        appActor.tellWithHighPriority(new PartitionChangeMsg(event.getServiceType()));
     }
 
     @Override

application/src/main/java/org/thingsboard/server/actors/tenant/TenantActor.java

@@ -17,6 +17,7 @@
 
 import lombok.extern.slf4j.Slf4j;
 import org.thingsboard.server.actors.ActorSystemContext;
+import org.thingsboard.server.actors.ProcessFailureStrategy;
 import org.thingsboard.server.actors.TbActor;
 import org.thingsboard.server.actors.TbActorCtx;
 import org.thingsboard.server.actors.TbActorException;
@@ -183,7 +184,7 @@ private void onQueueToRuleEngineMsg(QueueToRuleEngineMsg msg) {
             return;
         }
         TbMsg tbMsg = msg.getMsg();
-        if (getApiUsageState().isReExecEnabled() && ruleChainsInitialized) {
+        if (getApiUsageState().isReExecEnabled()) {
             if (tbMsg.getRuleChainId() == null) {
                 if (getRootChainActor() != null) {
                     getRootChainActor().tell(msg);
@@ -207,7 +208,7 @@ private void onQueueToRuleEngineMsg(QueueToRuleEngineMsg msg) {
     }
 
     private void onRuleChainMsg(RuleChainAwareMsg msg) {
-        if (getApiUsageState().isReExecEnabled() && ruleChainsInitialized) {
+        if (getApiUsageState().isReExecEnabled()) {
             getOrCreateActor(msg.getRuleChainId()).tell(msg);
         }
     }
@@ -319,6 +320,12 @@ private ApiUsageState getApiUsageState() {
         return apiUsageState;
     }
 
+    @Override
+    public ProcessFailureStrategy onProcessFailure(TbActorMsg msg, Throwable t) {
+        log.error("[{}] Failed to process msg: {}", tenantId, msg, t);
+        return doProcessFailure(t);
+    }
+
     public static class ActorCreator extends ContextBasedCreator {
 
         private final TenantId tenantId;

common/actor/src/main/java/org/thingsboard/server/actors/DefaultTbActorSystem.java

@@ -156,14 +156,29 @@ private void tell(TbActorId target, TbActorMsg actorMsg, boolean highPriority) {
 
     @Override
     public void broadcastToChildren(TbActorId parent, TbActorMsg msg) {
-        broadcastToChildren(parent, id -> true, msg);
+        broadcastToChildren(parent, msg, false);
+    }
+
+    @Override
+    public void broadcastToChildren(TbActorId parent, TbActorMsg msg, boolean highPriority) {
+        broadcastToChildren(parent, id -> true, msg, highPriority);
     }
 
     @Override
     public void broadcastToChildren(TbActorId parent, Predicate<TbActorId> childFilter, TbActorMsg msg) {
+        broadcastToChildren(parent, childFilter, msg, false);
+    }
+
+    private void broadcastToChildren(TbActorId parent, Predicate<TbActorId> childFilter, TbActorMsg msg, boolean highPriority) {
         Set<TbActorId> children = parentChildMap.get(parent);
         if (children != null) {
-            children.stream().filter(childFilter).forEach(id -> tell(id, msg));
+            children.stream().filter(childFilter).forEach(id -> {
+                try {
+                    tell(id, msg, highPriority);
+                } catch (TbActorNotRegisteredException e) {
+                    log.warn("Actor is missing for {}", id);
+                }
+            });
         }
     }
 
@@ -190,6 +205,8 @@ public void stop(TbActorId actorId) {
                 stop(child);
             }
         }
+        parentChildMap.values().forEach(parentChildren -> parentChildren.remove(actorId));
+
         TbActorMailbox mailbox = actors.remove(actorId);
         if (mailbox != null) {
             mailbox.destroy(null);

common/actor/src/main/java/org/thingsboard/server/actors/TbActor.java

@@ -34,7 +34,7 @@ default InitFailureStrategy onInitFailure(int attempt, Throwable t) {
         return InitFailureStrategy.retryWithDelay(5000L * attempt);
     }
 
-    default ProcessFailureStrategy onProcessFailure(Throwable t) {
+    default ProcessFailureStrategy onProcessFailure(TbActorMsg msg, Throwable t) {
         if (t instanceof Error) {
             return ProcessFailureStrategy.stop();
         } else {

common/actor/src/main/java/org/thingsboard/server/actors/TbActorCtx.java

@@ -36,6 +36,8 @@ public interface TbActorCtx extends TbActorRef {
 
     void broadcastToChildren(TbActorMsg msg);
 
+    void broadcastToChildren(TbActorMsg msg, boolean highPriority);
+
     void broadcastToChildrenByType(TbActorMsg msg, EntityType entityType);
 
     void broadcastToChildren(TbActorMsg msg, Predicate<TbActorId> childFilter);

common/actor/src/main/java/org/thingsboard/server/actors/TbActorMailbox.java

@@ -160,7 +160,7 @@ private void processMailbox() {
                     destroy(updateException.getCause());
                 } catch (Throwable t) {
                     log.debug("[{}] Failed to process message: {}", selfId, msg, t);
-                    ProcessFailureStrategy strategy = actor.onProcessFailure(t);
+                    ProcessFailureStrategy strategy = actor.onProcessFailure(msg, t);
                     if (strategy.isStop()) {
                         system.stop(selfId);
                     }
@@ -190,7 +190,12 @@ public void tell(TbActorId target, TbActorMsg actorMsg) {
 
     @Override
     public void broadcastToChildren(TbActorMsg msg) {
-        system.broadcastToChildren(selfId, msg);
+        broadcastToChildren(msg, false);
+    }
+
+    @Override
+    public void broadcastToChildren(TbActorMsg msg, boolean highPriority) {
+        system.broadcastToChildren(selfId, msg, highPriority);
     }
 
     @Override

common/actor/src/main/java/org/thingsboard/server/actors/TbActorSystem.java

@@ -48,6 +48,8 @@ public interface TbActorSystem {
 
     void broadcastToChildren(TbActorId parent, TbActorMsg msg);
 
+    void broadcastToChildren(TbActorId parent, TbActorMsg msg, boolean highPriority);
+
     void broadcastToChildren(TbActorId parent, Predicate<TbActorId> childFilter, TbActorMsg msg);
 
     List<TbActorId> filterChildren(TbActorId parent, Predicate<TbActorId> childFilter);