Main Logic of RuleChainActor to handle queue messages

application/src/main/java/org/thingsboard/server/actors/ActorSystemContext.java

@@ -63,7 +63,9 @@ import org.thingsboard.server.dao.rule.RuleChainService;
 import org.thingsboard.server.dao.tenant.TenantService;
 import org.thingsboard.server.dao.timeseries.TimeseriesService;
 import org.thingsboard.server.dao.user.UserService;
+import org.thingsboard.server.queue.discovery.PartitionService;
 import org.thingsboard.server.queue.discovery.TbServiceInfoProvider;
+import org.thingsboard.server.queue.provider.TbRuleEngineQueueProvider;
 import org.thingsboard.server.service.component.ComponentDiscoveryService;
 import org.thingsboard.server.service.encoding.DataDecodingEncodingService;
 import org.thingsboard.server.service.executors.ClusterRpcCallbackExecutorService;
@@ -151,6 +153,14 @@ public class ActorSystemContext {
     @Getter
     private RuleChainService ruleChainService;
 
+    @Autowired
+    @Getter
+    private PartitionService partitionService;
+
+    @Autowired
+    @Getter
+    private TbRuleEngineQueueProvider ruleEngineQueueProvider;
+
     @Autowired
     @Getter
     private TimeseriesService tsService;

application/src/main/java/org/thingsboard/server/actors/device/DeviceActorMessageProcessor.java

@@ -331,18 +331,18 @@ class DeviceActorMessageProcessor extends AbstractContextAwareMsgProcessor {
     }
 
     private void handleClientSideRPCRequest(ActorContext context, SessionInfoProto sessionInfo, TransportProtos.ToServerRpcRequestMsg request) {
-        UUID sessionId = getSessionId(sessionInfo);
-        JsonObject json = new JsonObject();
-        json.addProperty("method", request.getMethodName());
-        json.add("params", JsonUtils.parse(request.getParams()));
-
-        TbMsgMetaData requestMetaData = defaultMetaData.copy();
-        requestMetaData.putValue("requestId", Integer.toString(request.getRequestId()));
-        TbMsg tbMsg = new TbMsg(UUIDs.timeBased(), SessionMsgType.TO_SERVER_RPC_REQUEST.name(), deviceId, requestMetaData, TbMsgDataType.JSON, gson.toJson(json), null, null, null);
-        context.parent().tell(new DeviceActorToRuleEngineMsg(context.self(), tbMsg), context.self());
-
-        scheduleMsgWithDelay(context, new DeviceActorClientSideRpcTimeoutMsg(request.getRequestId(), systemContext.getClientSideRpcTimeout()), systemContext.getClientSideRpcTimeout());
-        toServerRpcPendingMap.put(request.getRequestId(), new ToServerRpcRequestMetadata(sessionId, getSessionType(sessionId), sessionInfo.getNodeId()));
+//        UUID sessionId = getSessionId(sessionInfo);
+//        JsonObject json = new JsonObject();
+//        json.addProperty("method", request.getMethodName());
+//        json.add("params", JsonUtils.parse(request.getParams()));
+//
+//        TbMsgMetaData requestMetaData = defaultMetaData.copy();
+//        requestMetaData.putValue("requestId", Integer.toString(request.getRequestId()));
+//        TbMsg tbMsg = new TbMsg(UUIDs.timeBased(), SessionMsgType.TO_SERVER_RPC_REQUEST.name(), deviceId, requestMetaData, TbMsgDataType.JSON, gson.toJson(json), null, null, null);
+//        context.parent().tell(new DeviceActorToRuleEngineMsg(context.self(), tbMsg), context.self());
+//
+//        scheduleMsgWithDelay(context, new DeviceActorClientSideRpcTimeoutMsg(request.getRequestId(), systemContext.getClientSideRpcTimeout()), systemContext.getClientSideRpcTimeout());
+//        toServerRpcPendingMap.put(request.getRequestId(), new ToServerRpcRequestMetadata(sessionId, getSessionType(sessionId), sessionInfo.getNodeId()));
     }
 
     private TransportProtos.SessionType getSessionType(UUID sessionId) {
@@ -372,10 +372,6 @@ class DeviceActorMessageProcessor extends AbstractContextAwareMsgProcessor {
         }
     }
 
-    private void pushToRuleEngine(ActorContext context, TbMsg tbMsg) {
-        context.parent().tell(new DeviceActorToRuleEngineMsg(context.self(), tbMsg), context.self());
-    }
-
     void processAttributesUpdate(ActorContext context, DeviceAttributesEventNotificationMsg msg) {
         if (attributeSubscriptions.size() > 0) {
             boolean hasNotificationData = false;

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

@@ -65,6 +65,7 @@ import org.thingsboard.server.dao.rule.RuleChainService;
 import org.thingsboard.server.dao.tenant.TenantService;
 import org.thingsboard.server.dao.timeseries.TimeseriesService;
 import org.thingsboard.server.dao.user.UserService;
+import org.thingsboard.server.queue.discovery.ServiceType;
 import org.thingsboard.server.service.script.RuleNodeJsScriptEngine;
 import scala.concurrent.duration.Duration;
 
@@ -118,10 +119,7 @@ class DefaultTbContext implements TbContext {
 
     @Override
     public boolean isLocalEntity(EntityId entityId) {
-        //TODO 2.5
-//        Optional<ServerAddress> address = mainCtx.getRoutingService().resolveById(entityId);
-//        return !address.isPresent();
-        return true;
+        return mainCtx.getPartitionService().resolve(ServiceType.TB_RULE_ENGINE, getTenantId(), entityId).isMyPartition();
     }
 
     private void scheduleMsgWithDelay(Object msg, long delayInMs, ActorRef target) {
@@ -143,7 +141,7 @@ class DefaultTbContext implements TbContext {
 
     @Override
     public TbMsg newMsg(String type, EntityId originator, TbMsgMetaData metaData, String data) {
-        return new TbMsg(UUIDs.timeBased(), type, originator, metaData.copy(), TbMsgDataType.JSON, data, nodeCtx.getSelf().getRuleChainId(), nodeCtx.getSelf().getId(),null);
+        return new TbMsg(UUIDs.timeBased(), type, originator, metaData.copy(), TbMsgDataType.JSON, data, nodeCtx.getSelf().getRuleChainId(), nodeCtx.getSelf().getId(), null);
     }
 
     @Override

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

@@ -18,7 +18,6 @@ package org.thingsboard.server.actors.ruleChain;
 import akka.actor.OneForOneStrategy;
 import akka.actor.SupervisorStrategy;
 import org.thingsboard.server.actors.ActorSystemContext;
-import org.thingsboard.server.actors.device.DeviceActorToRuleEngineMsg;
 import org.thingsboard.server.actors.service.ComponentActor;
 import org.thingsboard.server.actors.service.ContextBasedCreator;
 import org.thingsboard.server.common.data.id.RuleChainId;
@@ -43,10 +42,7 @@ public class RuleChainActor extends ComponentActor<RuleChainId, RuleChainActorMe
                 onComponentLifecycleMsg((ComponentLifecycleMsg) msg);
                 break;
             case QUEUE_TO_RULE_ENGINE_MSG:
-                processor.onServiceToRuleEngineMsg((QueueToRuleEngineMsg) msg);
-                break;
-            case DEVICE_ACTOR_TO_RULE_ENGINE_MSG:
-                processor.onDeviceActorToRuleEngineMsg((DeviceActorToRuleEngineMsg) msg);
+                processor.onQueueToRuleEngineMsg((QueueToRuleEngineMsg) msg);
                 break;
             case RULE_TO_RULE_CHAIN_TELL_NEXT_MSG:
             case REMOTE_TO_RULE_CHAIN_TELL_NEXT_MSG:

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

@@ -18,11 +18,10 @@ package org.thingsboard.server.actors.ruleChain;
 import akka.actor.ActorContext;
 import akka.actor.ActorRef;
 import akka.actor.Props;
-import com.datastax.driver.core.utils.UUIDs;
-
+import com.google.protobuf.ByteString;
 import lombok.extern.slf4j.Slf4j;
+import org.thingsboard.rule.engine.api.TbRelationTypes;
 import org.thingsboard.server.actors.ActorSystemContext;
-import org.thingsboard.server.actors.device.DeviceActorToRuleEngineMsg;
 import org.thingsboard.server.actors.service.DefaultActorService;
 import org.thingsboard.server.actors.shared.ComponentMsgProcessor;
 import org.thingsboard.server.common.data.EntityType;
@@ -37,10 +36,18 @@ import org.thingsboard.server.common.data.rule.RuleChain;
 import org.thingsboard.server.common.data.rule.RuleNode;
 import org.thingsboard.server.common.msg.TbMsg;
 import org.thingsboard.server.common.msg.cluster.ClusterEventMsg;
-import org.thingsboard.server.common.msg.cluster.ServerAddress;
 import org.thingsboard.server.common.msg.plugin.ComponentLifecycleMsg;
 import org.thingsboard.server.common.msg.queue.QueueToRuleEngineMsg;
 import org.thingsboard.server.dao.rule.RuleChainService;
+import org.thingsboard.server.gen.transport.TransportProtos.ToRuleEngineMsg;
+import org.thingsboard.server.queue.MultipleTbQueueTbMsgCallbackWrapper;
+import org.thingsboard.server.queue.TbQueueCallback;
+import org.thingsboard.server.queue.TbQueueProducer;
+import org.thingsboard.server.queue.TbQueueTbMsgCallbackWrapper;
+import org.thingsboard.server.queue.common.TbProtoQueueMsg;
+import org.thingsboard.server.queue.discovery.PartitionService;
+import org.thingsboard.server.queue.discovery.ServiceType;
+import org.thingsboard.server.queue.discovery.TopicPartitionInfo;
 
 import java.util.ArrayList;
 import java.util.Collections;
@@ -56,17 +63,17 @@ import java.util.stream.Collectors;
 @Slf4j
 public class RuleChainActorMessageProcessor extends ComponentMsgProcessor<RuleChainId> {
 
-    private static final long DEFAULT_CLUSTER_PARTITION = 0L;
     private final ActorRef parent;
     private final ActorRef self;
     private final Map<RuleNodeId, RuleNodeCtx> nodeActors;
     private final Map<RuleNodeId, List<RuleNodeRelation>> nodeRoutes;
     private final RuleChainService service;
+    private final PartitionService partitionService;
+    private final TbQueueProducer<TbProtoQueueMsg<ToRuleEngineMsg>> producer;
 
     private RuleNodeId firstId;
     private RuleNodeCtx firstNode;
     private boolean started;
-    private String ruleChainName;
 
     RuleChainActorMessageProcessor(TenantId tenantId, RuleChainId ruleChainId, ActorSystemContext systemContext
             , ActorRef parent, ActorRef self) {
@@ -76,7 +83,8 @@ public class RuleChainActorMessageProcessor extends ComponentMsgProcessor<RuleCh
         this.nodeActors = new HashMap<>();
         this.nodeRoutes = new HashMap<>();
         this.service = systemContext.getRuleChainService();
-        this.ruleChainName = ruleChainId.toString();
+        this.partitionService = systemContext.getPartitionService();
+        this.producer = systemContext.getRuleEngineQueueProvider().getRuleEngineMsgProducer();
     }
 
     @Override
@@ -89,7 +97,6 @@ public class RuleChainActorMessageProcessor extends ComponentMsgProcessor<RuleCh
         if (!started) {
             RuleChain ruleChain = service.findRuleChainById(tenantId, entityId);
             if (ruleChain != null) {
-                ruleChainName = ruleChain.getName();
                 List<RuleNode> ruleNodeList = service.getRuleChainNodes(tenantId, entityId);
                 log.trace("[{}][{}] Starting rule chain with {} nodes", tenantId, entityId, ruleNodeList.size());
                 // Creating and starting the actors;
@@ -110,7 +117,6 @@ public class RuleChainActorMessageProcessor extends ComponentMsgProcessor<RuleCh
     public void onUpdate(ActorContext context) {
         RuleChain ruleChain = service.findRuleChainById(tenantId, entityId);
         if (ruleChain != null) {
-            ruleChainName = ruleChain.getName();
             List<RuleNode> ruleNodeList = service.getRuleChainNodes(tenantId, entityId);
             log.trace("[{}][{}] Updating rule chain with {} nodes", tenantId, entityId, ruleNodeList.size());
             for (RuleNode ruleNode : ruleNodeList) {
@@ -189,101 +195,113 @@ public class RuleChainActorMessageProcessor extends ComponentMsgProcessor<RuleCh
         state = ComponentLifecycleState.ACTIVE;
     }
 
-    void onServiceToRuleEngineMsg(QueueToRuleEngineMsg envelope) {
-        log.trace("[{}][{}] Processing message [{}]: {}", entityId, firstId, envelope.getTbMsg().getId(), envelope.getTbMsg());
-        checkActive();
-        if (firstNode != null) {
-            log.trace("[{}][{}] Pushing message to first rule node", entityId, firstId);
-            pushMsgToNode(firstNode, enrichWithRuleChainId(envelope.getTbMsg()), "");
-        }
-    }
-
-    void onDeviceActorToRuleEngineMsg(DeviceActorToRuleEngineMsg envelope) {
-        checkActive();
-        if (firstNode != null) {
-            pushMsgToNode(firstNode, enrichWithRuleChainId(envelope.getTbMsg()), "");
+    void onQueueToRuleEngineMsg(QueueToRuleEngineMsg envelope) {
+        TbMsg msg = envelope.getTbMsg();
+        log.trace("[{}][{}] Processing message [{}]: {}", entityId, firstId, msg.getId(), msg);
+        try {
+            checkActive();
+            RuleNodeId targetId = msg.getRuleNodeId();
+            RuleNodeCtx targetCtx;
+            if (targetId == null) {
+                targetCtx = firstNode;
+                msg = msg.copyWithRuleChainId(entityId);
+            } else {
+                targetCtx = nodeActors.get(targetId);
+            }
+            if (targetCtx != null) {
+                log.trace("[{}][{}] Pushing message to target rule node", entityId, targetId);
+                pushMsgToNode(firstNode, msg, "");
+            } else {
+                log.trace("[{}][{}] Rule node does not exist. Probably old message", entityId, targetId);
+                msg.getCallback().onSuccess();
+            }
+        } catch (Exception e) {
+            envelope.getTbMsg().getCallback().onFailure(e);
         }
     }
 
     void onRuleChainToRuleChainMsg(RuleChainToRuleChainMsg envelope) {
         checkActive();
-        if (envelope.isEnqueue()) {
-            if (firstNode != null) {
-                pushMsgToNode(firstNode, enrichWithRuleChainId(envelope.getMsg()), envelope.getFromRelationType());
-            }
+        if (firstNode != null) {
+            pushMsgToNode(firstNode, envelope.getMsg(), envelope.getFromRelationType());
         } else {
-            if (firstNode != null) {
-                pushMsgToNode(firstNode, envelope.getMsg(), envelope.getFromRelationType());
-            } else {
-//                TODO: Ack this message in Kafka
-//                TbMsg msg = envelope.getMsg();
-//                EntityId ackId = msg.getRuleNodeId() != null ? msg.getRuleNodeId() : msg.getRuleChainId();
-//                queue.ack(tenantId, envelope.getMsg(), ackId.getId(), msg.getClusterPartition());
-            }
+            envelope.getMsg().getCallback().onSuccess();
         }
     }
 
     void onTellNext(RuleNodeToRuleChainTellNextMsg envelope) {
-        checkActive();
         TbMsg msg = envelope.getMsg();
-        EntityId originatorEntityId = msg.getOriginator();
-        //TODO 2.5
-//        Optional<ServerAddress> address = systemContext.getRoutingService().resolveById(originatorEntityId);
-//        if (address.isPresent()) {
-//            onRemoteTellNext(address.get(), envelope);
-//        } else {
-            onLocalTellNext(envelope);
-//        }
+        try {
+            checkActive();
+            EntityId entityId = msg.getOriginator();
+            TopicPartitionInfo tpi = partitionService.resolve(ServiceType.TB_RULE_ENGINE, tenantId, entityId);
+            RuleNodeId originatorNodeId = envelope.getOriginator();
+            List<RuleNodeRelation> relations = nodeRoutes.get(originatorNodeId).stream()
+                    .filter(r -> contains(envelope.getRelationTypes(), r.getType()))
+                    .collect(Collectors.toList());
+            int relationsCount = relations.size();
+            if (relationsCount == 0) {
+                log.trace("[{}][{}][{}] No outbound relations to process", tenantId, entityId, msg.getId());
+                //TODO 2.5: Maybe let's check that the output relation is not a Failure?
+                if (envelope.getRelationTypes().contains(TbRelationTypes.FAILURE)) {
+                    log.debug("[{}] Failure during message processing by Rule Node [{}]. Enable and see debug events for more info", entityId, envelope.getOriginator().getId());
+                    msg.getCallback().onFailure(new RuntimeException("Failure during message processing by Rule Node [" + envelope.getOriginator().getId().toString() + "]"));
+                } else {
+                    msg.getCallback().onSuccess();
+                }
+            } else if (relationsCount == 1) {
+                for (RuleNodeRelation relation : relations) {
+                    log.trace("[{}][{}][{}] Pushing message to single target: [{}]", tenantId, entityId, msg.getId(), relation.getOut());
+                    pushToTarget(tpi, msg, relation.getOut(), relation.getType());
+                }
+            } else {
+                MultipleTbQueueTbMsgCallbackWrapper callbackWrapper = new MultipleTbQueueTbMsgCallbackWrapper(relationsCount, msg.getCallback());
+                log.trace("[{}][{}][{}] Pushing message to multiple targets: [{}]", tenantId, entityId, msg.getId(), relations);
+                for (RuleNodeRelation relation : relations) {
+                    EntityId target = relation.getOut();
+                    putToQueue(tpi, msg, callbackWrapper, target);
+                }
+            }
+        } catch (Exception e) {
+            msg.getCallback().onFailure(e);
+        }
     }
 
-    private void onRemoteTellNext(ServerAddress serverAddress, RuleNodeToRuleChainTellNextMsg envelope) {
-        TbMsg msg = envelope.getMsg();
-        log.debug("Forwarding [{}] msg to remote server [{}] due to changed originator id: [{}]", msg.getId(), serverAddress, msg.getOriginator());
-        envelope = new RemoteToRuleChainTellNextMsg(envelope, tenantId, entityId);
-        //TODO 2.5
-//        systemContext.getRpcService().tell(systemContext.getEncodingService().convertToProtoDataMessage(serverAddress, envelope));
+    private void putToQueue(TopicPartitionInfo tpi, TbMsg msg, TbQueueCallback callbackWrapper, EntityId target) {
+        switch (target.getEntityType()) {
+            case RULE_NODE:
+                putToQueue(tpi, msg.copyWithRuleNodeId(entityId, new RuleNodeId(target.getId())), callbackWrapper);
+                break;
+            case RULE_CHAIN:
+                putToQueue(tpi, msg.copyWithRuleChainId(new RuleChainId(target.getId())), callbackWrapper);
+                break;
+        }
     }
 
-    private void onLocalTellNext(RuleNodeToRuleChainTellNextMsg envelope) {
-        TbMsg msg = envelope.getMsg();
-        RuleNodeId originatorNodeId = envelope.getOriginator();
-        List<RuleNodeRelation> relations = nodeRoutes.get(originatorNodeId).stream()
-                .filter(r -> contains(envelope.getRelationTypes(), r.getType()))
-                .collect(Collectors.toList());
-        int relationsCount = relations.size();
-        EntityId ackId = msg.getRuleNodeId() != null ? msg.getRuleNodeId() : msg.getRuleChainId();
-        if (relationsCount == 0) {
-            log.trace("[{}][{}][{}] No outbound relations to process", tenantId, entityId, msg.getId());
-            if (ackId != null) {
-//                TODO: Ack this message in Kafka
-//                queue.ack(tenantId, msg, ackId.getId(), msg.getClusterPartition());
-            }
-        } else if (relationsCount == 1) {
-            for (RuleNodeRelation relation : relations) {
-                log.trace("[{}][{}][{}] Pushing message to single target: [{}]", tenantId, entityId, msg.getId(), relation.getOut());
-                pushToTarget(msg, relation.getOut(), relation.getType());
+    private void pushToTarget(TopicPartitionInfo tpi, TbMsg msg, EntityId target, String fromRelationType) {
+        if (tpi.isMyPartition()) {
+            switch (target.getEntityType()) {
+                case RULE_NODE:
+                    pushMsgToNode(nodeActors.get(new RuleNodeId(target.getId())), msg, fromRelationType);
+                    break;
+                case RULE_CHAIN:
+                    parent.tell(new RuleChainToRuleChainMsg(new RuleChainId(target.getId()), entityId, msg, fromRelationType), self);
+                    break;
             }
         } else {
-            for (RuleNodeRelation relation : relations) {
-                EntityId target = relation.getOut();
-                log.trace("[{}][{}][{}] Pushing message to multiple targets: [{}]", tenantId, entityId, msg.getId(), relation.getOut());
-                switch (target.getEntityType()) {
-                    case RULE_NODE:
-                        enqueueAndForwardMsgCopyToNode(msg, target, relation.getType());
-                        break;
-                    case RULE_CHAIN:
-                        enqueueAndForwardMsgCopyToChain(msg, target, relation.getType());
-                        break;
-                }
-            }
-            //TODO: Ideally this should happen in async way when all targets confirm that the copied messages are successfully written to corresponding target queues.
-            if (ackId != null) {
-//                TODO: Ack this message in Kafka
-//                queue.ack(tenantId, msg, ackId.getId(), msg.getClusterPartition());
-            }
+            putToQueue(tpi, msg, new TbQueueTbMsgCallbackWrapper(msg.getCallback()), target);
         }
     }
 
+    private void putToQueue(TopicPartitionInfo tpi, TbMsg newMsg, TbQueueCallback callbackWrapper) {
+        ToRuleEngineMsg toQueueMsg = ToRuleEngineMsg.newBuilder()
+                .setTenantIdMSB(tenantId.getId().getMostSignificantBits())
+                .setTenantIdLSB(tenantId.getId().getLeastSignificantBits())
+                .setTbMsg(ByteString.copyFrom(TbMsg.toByteArray(newMsg)))
+                .build();
+        producer.send(tpi, new TbProtoQueueMsg<>(newMsg.getId(), toQueueMsg), callbackWrapper);
+    }
+
     private boolean contains(Set<String> relationTypes, String type) {
         if (relationTypes == null) {
             return true;
@@ -296,38 +314,10 @@ public class RuleChainActorMessageProcessor extends ComponentMsgProcessor<RuleCh
         return false;
     }
 
-    private void enqueueAndForwardMsgCopyToChain(TbMsg msg, EntityId target, String fromRelationType) {
-        RuleChainId targetRCId = new RuleChainId(target.getId());
-        TbMsg copyMsg = msg.copy(UUIDs.timeBased(), targetRCId, null, null);
-        parent.tell(new RuleChainToRuleChainMsg(new RuleChainId(target.getId()), entityId, copyMsg, fromRelationType, true), self);
-    }
-
-    private void enqueueAndForwardMsgCopyToNode(TbMsg msg, EntityId target, String fromRelationType) {
-        RuleNodeId targetId = new RuleNodeId(target.getId());
-        RuleNodeCtx targetNodeCtx = nodeActors.get(targetId);
-        TbMsg copy = msg.copy(UUIDs.timeBased(), entityId, targetId, null);
-        pushMsgToNode(targetNodeCtx, copy, fromRelationType);
-    }
-
-    private void pushToTarget(TbMsg msg, EntityId target, String fromRelationType) {
-        switch (target.getEntityType()) {
-            case RULE_NODE:
-                pushMsgToNode(nodeActors.get(new RuleNodeId(target.getId())), msg, fromRelationType);
-                break;
-            case RULE_CHAIN:
-                parent.tell(new RuleChainToRuleChainMsg(new RuleChainId(target.getId()), entityId, msg, fromRelationType, false), self);
-                break;
-        }
-    }
-
     private void pushMsgToNode(RuleNodeCtx nodeCtx, TbMsg msg, String fromRelationType) {
         if (nodeCtx != null) {
             nodeCtx.getSelfActor().tell(new RuleChainToRuleNodeMsg(new DefaultTbContext(systemContext, nodeCtx), msg, fromRelationType), self);
         }
     }
 
-    private TbMsg enrichWithRuleChainId(TbMsg tbMsg) {
-        // We don't put firstNodeId because it may change over time;
-        return new TbMsg(tbMsg.getId(), tbMsg.getType(), tbMsg.getOriginator(), tbMsg.getMetaData().copy(), tbMsg.getDataType(), tbMsg.getData(), entityId, null, null);
-    }
 }

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

@@ -32,7 +32,6 @@ public final class RuleChainToRuleChainMsg implements TbActorMsg, RuleChainAware
     private final RuleChainId source;
     private final TbMsg msg;
     private final String fromRelationType;
-    private final boolean enqueue;
 
     @Override
     public RuleChainId getRuleChainId() {

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

@@ -27,7 +27,6 @@ import com.google.common.collect.BiMap;
 import com.google.common.collect.HashBiMap;
 import org.thingsboard.server.actors.ActorSystemContext;
 import org.thingsboard.server.actors.device.DeviceActorCreator;
-import org.thingsboard.server.actors.device.DeviceActorToRuleEngineMsg;
 import org.thingsboard.server.actors.ruleChain.RuleChainManagerActor;
 import org.thingsboard.server.actors.service.ContextBasedCreator;
 import org.thingsboard.server.actors.service.DefaultActorService;
@@ -89,9 +88,6 @@ public class TenantActor extends RuleChainManagerActor {
             case QUEUE_TO_RULE_ENGINE_MSG:
                 onQueueToRuleEngineMsg((QueueToRuleEngineMsg) msg);
                 break;
-            case DEVICE_ACTOR_TO_RULE_ENGINE_MSG:
-                onDeviceActorToRuleEngineMsg((DeviceActorToRuleEngineMsg) msg);
-                break;
             case TRANSPORT_TO_DEVICE_ACTOR_MSG:
             case DEVICE_ATTRIBUTES_UPDATE_TO_DEVICE_ACTOR_MSG:
             case DEVICE_CREDENTIALS_UPDATE_TO_DEVICE_ACTOR_MSG:
@@ -131,14 +127,6 @@ public class TenantActor extends RuleChainManagerActor {
         }
     }
 
-    private void onDeviceActorToRuleEngineMsg(DeviceActorToRuleEngineMsg msg) {
-        if (ruleChainManager.getRootChainActor() != null) {
-            ruleChainManager.getRootChainActor().tell(msg, self());
-        } else {
-            log.info("[{}] No Root Chain: {}", tenantId, msg);
-        }
-    }
-
     private void onRuleChainMsg(RuleChainAwareMsg msg) {
         ruleChainManager.getOrCreateActor(context(), msg.getRuleChainId()).tell(msg, self());
     }

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

@@ -102,7 +102,6 @@ public class DefaultTbRuleEngineConsumerService implements TbRuleEngineConsumerS
                         try {
                             TransportProtos.ToRuleEngineMsg toRuleEngineMsg = msg.getValue();
                             TenantId tenantId = new TenantId(new UUID(toRuleEngineMsg.getTenantIdMSB(), toRuleEngineMsg.getTenantIdLSB()));
-                            log.trace("Forwarding message to rule engine {}", toRuleEngineMsg);
                             if (toRuleEngineMsg.getTbMsg() != null && !toRuleEngineMsg.getTbMsg().isEmpty()) {
                                 forwardToRuleEngineActor(tenantId, toRuleEngineMsg.getTbMsg(), callback);
                             } else {
@@ -130,8 +129,8 @@ public class DefaultTbRuleEngineConsumerService implements TbRuleEngineConsumerS
 
     private void forwardToRuleEngineActor(TenantId tenantId, ByteString tbMsgData, TbMsgCallback callback) {
         TbMsg tbMsg = TbMsg.fromBytes(tbMsgData.toByteArray(), callback);
-        log.info("[{}] Received RULE ENGINE msg: {}", tbMsg.getType(), tbMsg);
         actorContext.getAppActor().tell(new QueueToRuleEngineMsg(tenantId, tbMsg), ActorRef.noSender());
+        //TODO: 2.5 before release.
 //        if (statsEnabled) {
 //            stats.log(toDeviceActorMsg);
 //        }

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

@@ -98,7 +98,6 @@ public enum MsgType {
     /**
      * Message that is sent from the Device Actor to Rule Engine. Requires acknowledgement
      */
-    DEVICE_ACTOR_TO_RULE_ENGINE_MSG,
 
     SESSION_TIMEOUT_MSG,
 

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

@@ -19,6 +19,7 @@ import com.google.protobuf.InvalidProtocolBufferException;
 import lombok.AllArgsConstructor;
 import lombok.Builder;
 import lombok.Data;
+import lombok.extern.slf4j.Slf4j;
 import org.thingsboard.server.common.data.id.EntityId;
 import org.thingsboard.server.common.data.id.EntityIdFactory;
 import org.thingsboard.server.common.data.id.RuleChainId;
@@ -35,7 +36,7 @@ import java.util.UUID;
  */
 @Data
 @Builder
-@AllArgsConstructor
+@Slf4j
 public final class TbMsg implements Serializable {
 
     private final UUID id;
@@ -55,6 +56,26 @@ public final class TbMsg implements Serializable {
         this(id, type, originator, metaData, dataType, data, new TbMsgTransactionData(id, originator), ruleChainId, ruleNodeId, callback);
     }
 
+    public TbMsg(UUID id, String type, EntityId originator, TbMsgMetaData metaData, TbMsgDataType dataType, String data,
+                 TbMsgTransactionData transactionData, RuleChainId ruleChainId, RuleNodeId ruleNodeId, TbMsgCallback callback) {
+        this.id = id;
+        this.type = type;
+        this.originator = originator;
+        this.metaData = metaData;
+        this.dataType = dataType;
+        this.data = data;
+        this.transactionData = transactionData;
+        this.ruleChainId = ruleChainId;
+        this.ruleNodeId = ruleNodeId;
+        if (callback != null) {
+            this.callback = callback;
+        } else {
+            log.warn("[{}] Created message with empty callback: {}", originator, type);
+            this.callback = TbMsgCallback.EMPTY;
+        }
+
+    }
+
     public static byte[] toByteArray(TbMsg msg) {
         MsgProtos.TbMsgProto.Builder builder = MsgProtos.TbMsgProto.newBuilder();
         builder.setId(msg.getId().toString());
@@ -116,8 +137,11 @@ public final class TbMsg implements Serializable {
         }
     }
 
-    public TbMsg copy(UUID newId, RuleChainId ruleChainId, RuleNodeId ruleNodeId, TbMsgCallback callback) {
-        return new TbMsg(newId, type, originator, metaData.copy(), dataType, data, transactionData, ruleChainId, ruleNodeId, callback);
+    public TbMsg copyWithRuleChainId(RuleChainId ruleChainId) {
+        return new TbMsg(this.id, this.type, this.originator, this.metaData, this.dataType, this.data, this.transactionData, ruleChainId, null, callback);
     }
 
+    public TbMsg copyWithRuleNodeId(RuleChainId ruleChainId, RuleNodeId ruleNodeId) {
+        return new TbMsg(this.id, this.type, this.originator, this.metaData, this.dataType, this.data, this.transactionData, ruleChainId, ruleNodeId, callback);
+    }
 }

common/queue/src/main/java/org/thingsboard/server/queue/MultipleTbQueueTbMsgCallbackWrapper.java

@@ -0,0 +1,43 @@
+/**
+ * Copyright © 2016-2020 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;
+
+import org.thingsboard.server.common.msg.queue.TbMsgCallback;
+
+import java.util.concurrent.atomic.AtomicInteger;
+
+public class MultipleTbQueueTbMsgCallbackWrapper implements TbQueueCallback {
+
+    private final AtomicInteger tbQueueCallbackCount;
+    private final TbMsgCallback tbMsgCallback;
+
+    public MultipleTbQueueTbMsgCallbackWrapper(int tbQueueCallbackCount, TbMsgCallback tbMsgCallback) {
+        this.tbQueueCallbackCount = new AtomicInteger(tbQueueCallbackCount);
+        this.tbMsgCallback = tbMsgCallback;
+    }
+
+    @Override
+    public void onSuccess(TbQueueMsgMetadata metadata) {
+        if (tbQueueCallbackCount.decrementAndGet() <= 0) {
+            tbMsgCallback.onSuccess();
+        }
+    }
+
+    @Override
+    public void onFailure(Throwable t) {
+        tbMsgCallback.onFailure(t);
+    }
+}

application/src/main/java/org/thingsboard/server/actors/device/DeviceActorToRuleEngineMsg.java → common/queue/src/main/java/org/thingsboard/server/queue/TbQueueTbMsgCallbackWrapper.java

@@ -13,25 +13,27 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package org.thingsboard.server.actors.device;
+package org.thingsboard.server.queue;
 
-import akka.actor.ActorRef;
-import lombok.Data;
-import org.thingsboard.server.common.msg.MsgType;
-import org.thingsboard.server.common.msg.TbActorMsg;
-import org.thingsboard.server.common.msg.TbMsg;
+import org.thingsboard.server.common.msg.queue.TbMsgCallback;
 
-/**
- * Created by ashvayka on 15.03.18.
- */
-@Data
-public final class DeviceActorToRuleEngineMsg implements TbActorMsg {
+import java.util.concurrent.atomic.AtomicInteger;
+
+public class TbQueueTbMsgCallbackWrapper implements TbQueueCallback {
+
+    private final TbMsgCallback tbMsgCallback;
 
-    private final ActorRef callbackRef;
-    private final TbMsg tbMsg;
+    public TbQueueTbMsgCallbackWrapper(TbMsgCallback tbMsgCallback) {
+        this.tbMsgCallback = tbMsgCallback;
+    }
+
+    @Override
+    public void onSuccess(TbQueueMsgMetadata metadata) {
+        tbMsgCallback.onSuccess();
+    }
 
     @Override
-    public MsgType getMsgType() {
-        return MsgType.DEVICE_ACTOR_TO_RULE_ENGINE_MSG;
+    public void onFailure(Throwable t) {
+        tbMsgCallback.onFailure(t);
     }
 }

common/queue/src/main/java/org/thingsboard/server/queue/discovery/ConsistentHashPartitionService.java

@@ -32,6 +32,7 @@ import java.nio.charset.StandardCharsets;
 import java.util.ArrayList;
 import java.util.HashMap;
 import java.util.HashSet;
+import java.util.LinkedHashSet;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
@@ -65,6 +66,7 @@ public class ConsistentHashPartitionService implements PartitionService {
     private ConcurrentMap<ServiceKey, List<Integer>> myPartitions = new ConcurrentHashMap<>();
     //TODO: Fetch this from the database, together with size of partitions for each service for each tenant.
     private ConcurrentMap<TenantId, Set<ServiceType>> isolatedTenants = new ConcurrentHashMap<>();
+    private ConcurrentMap<TopicPartitionInfoKey, TopicPartitionInfo> tpiCache = new ConcurrentHashMap<>();
 
     private Map<String, TopicPartitionInfo> tbCoreNotificationTopics = new HashMap<>();
     private Map<String, TopicPartitionInfo> tbRuleEngineNotificationTopics = new HashMap<>();
@@ -87,12 +89,12 @@ public class ConsistentHashPartitionService implements PartitionService {
     }
 
     @Override
-    public List<TopicPartitionInfo> getCurrentPartitions(ServiceType serviceType) {
+    public Set<TopicPartitionInfo> getCurrentPartitions(ServiceType serviceType) {
         ServiceInfo currentService = serviceInfoProvider.getServiceInfo();
         TenantId tenantId = getSystemOrIsolatedTenantId(currentService);
         ServiceKey serviceKey = new ServiceKey(serviceType, tenantId);
         List<Integer> partitions = myPartitions.get(serviceKey);
-        List<TopicPartitionInfo> topicPartitions = new ArrayList<>();
+        Set<TopicPartitionInfo> topicPartitions = new LinkedHashSet<>();
         for (Integer partition : partitions) {
             TopicPartitionInfo.TopicPartitionInfoBuilder tpi = TopicPartitionInfo.builder();
             tpi.topic(partitionTopics.get(serviceType));
@@ -112,7 +114,9 @@ public class ConsistentHashPartitionService implements PartitionService {
                 .putLong(entityId.getId().getMostSignificantBits())
                 .putLong(entityId.getId().getLeastSignificantBits()).hash().asInt();
         int partition = Math.abs(hash % partitionSizes.get(serviceType));
-        return buildTopicPartitionInfo(serviceType, tenantId, partition);
+        boolean isolatedTenant = isIsolated(serviceType, tenantId);
+        TopicPartitionInfoKey cacheKey = new TopicPartitionInfoKey(serviceType, isolatedTenant ? tenantId : null, partition);
+        return tpiCache.computeIfAbsent(cacheKey, key -> buildTopicPartitionInfo(serviceType, tenantId, partition));
     }
 
     @Override
@@ -156,8 +160,8 @@ public class ConsistentHashPartitionService implements PartitionService {
                 tpiList.add(getNotificationsTopic(serviceKey.getServiceType(), serviceInfoProvider.getServiceId()));
                 applicationEventPublisher.publishEvent(new PartitionChangeEvent(this, serviceKey, tpiList));
             }
-
         });
+        tpiCache.clear();
 
         if (currentOtherServices == null) {
             currentOtherServices = new ArrayList<>(otherServices);
@@ -207,7 +211,7 @@ public class ConsistentHashPartitionService implements PartitionService {
     }
 
     private TopicPartitionInfo buildTopicPartitionInfo(ServiceType serviceType, String serviceId) {
-        return new TopicPartitionInfo(serviceType.name().toLowerCase() + "." + serviceId, null, null);
+        return new TopicPartitionInfo(serviceType.name().toLowerCase() + "." + serviceId, null, null, false);
     }
 
     private TopicPartitionInfo buildTopicPartitionInfo(ServiceKey serviceKey, int partition) {
@@ -215,16 +219,29 @@ public class ConsistentHashPartitionService implements PartitionService {
     }
 
     private TopicPartitionInfo buildTopicPartitionInfo(ServiceType serviceType, TenantId tenantId, int partition) {
-        boolean isolated = isolatedTenants.get(tenantId) != null && isolatedTenants.get(tenantId).contains(serviceType);
         TopicPartitionInfo.TopicPartitionInfoBuilder tpi = TopicPartitionInfo.builder();
         tpi.topic(partitionTopics.get(serviceType));
         tpi.partition(partition);
-        if (isolated) {
+        ServiceKey myPartitionsSearchKey;
+        if (isIsolated(serviceType, tenantId)) {
             tpi.tenantId(tenantId);
+            myPartitionsSearchKey = new ServiceKey(serviceType, tenantId);
+        } else {
+            myPartitionsSearchKey = new ServiceKey(serviceType, new TenantId(TenantId.NULL_UUID));
+        }
+        List<Integer> partitions = myPartitions.get(myPartitionsSearchKey);
+        if (partitions != null) {
+            tpi.myPartition(partitions.contains(partition));
+        } else {
+            tpi.myPartition(false);
         }
         return tpi.build();
     }
 
+    private boolean isIsolated(ServiceType serviceType, TenantId tenantId) {
+        return isolatedTenants.get(tenantId) != null && isolatedTenants.get(tenantId).contains(serviceType);
+    }
+
     private void logServiceInfo(TransportProtos.ServiceInfo server) {
         TenantId tenantId = getSystemOrIsolatedTenantId(server);
         if (tenantId.isNullUid()) {

common/queue/src/main/java/org/thingsboard/server/queue/discovery/PartitionService.java

@@ -20,13 +20,14 @@ import org.thingsboard.server.common.data.id.TenantId;
 import org.thingsboard.server.gen.transport.TransportProtos;
 
 import java.util.List;
+import java.util.Set;
 
 /**
  * Once application is ready or cluster topology changes, this Service will produce {@link PartitionChangeEvent}
  */
 public interface PartitionService {
 
-    List<TopicPartitionInfo> getCurrentPartitions(ServiceType serviceType);
+    Set<TopicPartitionInfo> getCurrentPartitions(ServiceType serviceType);
 
     TopicPartitionInfo resolve(ServiceType serviceType, TenantId tenantId, EntityId entityId);
 

common/queue/src/main/java/org/thingsboard/server/queue/discovery/TopicPartitionInfo.java

@@ -16,6 +16,7 @@
 package org.thingsboard.server.queue.discovery;
 
 import lombok.Builder;
+import lombok.Getter;
 import org.thingsboard.server.common.data.id.TenantId;
 
 import java.util.Objects;
@@ -26,13 +27,17 @@ public class TopicPartitionInfo {
     private final String topic;
     private final TenantId tenantId;
     private final Integer partition;
+    @Getter
     private final String fullTopicName;
+    @Getter
+    private final boolean myPartition;
 
     @Builder
-    public TopicPartitionInfo(String topic, TenantId tenantId, Integer partition) {
+    public TopicPartitionInfo(String topic, TenantId tenantId, Integer partition, boolean myPartition) {
         this.topic = topic;
         this.tenantId = tenantId;
         this.partition = partition;
+        this.myPartition = myPartition;
         String tmp = topic;
         if (tenantId != null) {
             tmp += "." + tenantId.getId().toString();
@@ -40,7 +45,6 @@ public class TopicPartitionInfo {
         if (partition != null) {
             tmp += "." + partition;
         }
-
         this.fullTopicName = tmp;
     }
 
@@ -56,10 +60,6 @@ public class TopicPartitionInfo {
         return Optional.ofNullable(partition);
     }
 
-    public String getFullTopicName() {
-        return fullTopicName;
-    }
-
     @Override
     public boolean equals(Object o) {
         if (this == o) return true;

common/queue/src/main/java/org/thingsboard/server/queue/discovery/TopicPartitionInfoKey.java

@@ -0,0 +1,43 @@
+/**
+ * Copyright © 2016-2020 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.AllArgsConstructor;
+import org.thingsboard.server.common.data.id.TenantId;
+
+import java.util.Objects;
+
+@AllArgsConstructor
+public class TopicPartitionInfoKey {
+    private ServiceType serviceType;
+    private TenantId isolatedTenantId;
+    private int partition;
+
+    @Override
+    public boolean equals(Object o) {
+        if (this == o) return true;
+        if (o == null || getClass() != o.getClass()) return false;
+        TopicPartitionInfoKey that = (TopicPartitionInfoKey) o;
+        return partition == that.partition &&
+                serviceType == that.serviceType &&
+                Objects.equals(isolatedTenantId, that.isolatedTenantId);
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(serviceType, isolatedTenantId, partition);
+    }
+}

common/queue/src/main/java/org/thingsboard/server/queue/kafka/TBKafkaConsumerTemplate.java

@@ -75,7 +75,7 @@ public class TBKafkaConsumerTemplate<T extends TbQueueMsg> implements TbQueueCon
 
     @Override
     public void subscribe() {
-        partitions = Collections.singleton(new TopicPartitionInfo(topic, null, null));
+        partitions = Collections.singleton(new TopicPartitionInfo(topic, null, null, true));
         subscribed = false;
     }