Merge remote-tracking branch 'origin/feature/re-pack-timeout' into feature/mqtt-rate-limits

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

@@ -167,6 +167,11 @@ class DefaultTbContext implements TbContext {
     }
 
     private void enqueue(TopicPartitionInfo tpi, TbMsg tbMsg, Consumer<Throwable> onFailure, Runnable onSuccess) {
+        if (!tbMsg.isValid()) {
+            log.trace("[{}] Skip invalid message: {}", getTenantId(), tbMsg);
+            onFailure.accept(new IllegalArgumentException("Source message is no longer valid!"));
+            return;
+        }
         TransportProtos.ToRuleEngineMsg msg = TransportProtos.ToRuleEngineMsg.newBuilder()
                 .setTenantIdMSB(getTenantId().getId().getMostSignificantBits())
                 .setTenantIdLSB(getTenantId().getId().getLeastSignificantBits())
@@ -235,6 +240,11 @@ class DefaultTbContext implements TbContext {
     }
 
     private void enqueueForTellNext(TopicPartitionInfo tpi, String queueName, TbMsg source, Set<String> relationTypes, String failureMessage, Runnable onSuccess, Consumer<Throwable> onFailure) {
+        if (!source.isValid()) {
+            log.trace("[{}] Skip invalid message: {}", getTenantId(), source);
+            onFailure.accept(new IllegalArgumentException("Source message is no longer valid!"));
+            return;
+        }
         RuleChainId ruleChainId = nodeCtx.getSelf().getRuleChainId();
         RuleNodeId ruleNodeId = nodeCtx.getSelf().getId();
         TbMsg tbMsg = TbMsg.newMsg(source, queueName, ruleChainId, ruleNodeId);

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

@@ -200,17 +200,20 @@ public class RuleChainActorMessageProcessor extends ComponentMsgProcessor<RuleCh
 
     void onQueueToRuleEngineMsg(QueueToRuleEngineMsg envelope) {
         TbMsg msg = envelope.getMsg();
+        if (!checkMsgValid(msg)) {
+            return;
+        }
         log.trace("[{}][{}] Processing message [{}]: {}", entityId, firstId, msg.getId(), msg);
         if (envelope.getRelationTypes() == null || envelope.getRelationTypes().isEmpty()) {
             onTellNext(msg, true);
         } else {
-            onTellNext(envelope.getMsg(), envelope.getMsg().getRuleNodeId(), envelope.getRelationTypes(), envelope.getFailureMessage());
+            onTellNext(msg, envelope.getMsg().getRuleNodeId(), envelope.getRelationTypes(), envelope.getFailureMessage());
         }
     }
 
     private void onTellNext(TbMsg msg, boolean useRuleNodeIdFromMsg) {
         try {
-            checkActive(msg);
+            checkComponentStateActive(msg);
             RuleNodeId targetId = useRuleNodeIdFromMsg ? msg.getRuleNodeId() : null;
             RuleNodeCtx targetCtx;
             if (targetId == null) {
@@ -234,6 +237,10 @@ public class RuleChainActorMessageProcessor extends ComponentMsgProcessor<RuleCh
     }
 
     public void onRuleChainInputMsg(RuleChainInputMsg envelope) {
+        var msg = envelope.getMsg();
+        if (!checkMsgValid(msg)) {
+            return;
+        }
         if (entityId.equals(envelope.getRuleChainId())) {
             onTellNext(envelope.getMsg(), false);
         } else {
@@ -242,6 +249,10 @@ public class RuleChainActorMessageProcessor extends ComponentMsgProcessor<RuleCh
     }
 
     public void onRuleChainOutputMsg(RuleChainOutputMsg envelope) {
+        var msg = envelope.getMsg();
+        if (!checkMsgValid(msg)) {
+            return;
+        }
         if (entityId.equals(envelope.getRuleChainId())) {
             var originatorNodeId = envelope.getTargetRuleNodeId();
             RuleNodeCtx ruleNodeCtx = nodeActors.get(originatorNodeId);
@@ -255,8 +266,12 @@ public class RuleChainActorMessageProcessor extends ComponentMsgProcessor<RuleCh
     }
 
     void onRuleChainToRuleChainMsg(RuleChainToRuleChainMsg envelope) {
+        var msg = envelope.getMsg();
+        if (!checkMsgValid(msg)) {
+            return;
+        }
         try {
-            checkActive(envelope.getMsg());
+            checkComponentStateActive(envelope.getMsg());
             if (firstNode != null) {
                 pushMsgToNode(firstNode, envelope.getMsg(), envelope.getFromRelationType());
             } else {
@@ -268,12 +283,15 @@ public class RuleChainActorMessageProcessor extends ComponentMsgProcessor<RuleCh
     }
 
     void onTellNext(RuleNodeToRuleChainTellNextMsg envelope) {
-        onTellNext(envelope.getMsg(), envelope.getOriginator(), envelope.getRelationTypes(), envelope.getFailureMessage());
+        var msg = envelope.getMsg();
+        if (checkMsgValid(msg)) {
+            onTellNext(msg, envelope.getOriginator(), envelope.getRelationTypes(), envelope.getFailureMessage());
+        }
     }
 
     private void onTellNext(TbMsg msg, RuleNodeId originatorNodeId, Set<String> relationTypes, String failureMessage) {
         try {
-            checkActive(msg);
+            checkComponentStateActive(msg);
             EntityId entityId = msg.getOriginator();
             TopicPartitionInfo tpi = systemContext.resolve(ServiceType.TB_RULE_ENGINE, msg.getQueueName(), tenantId, entityId);
 

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

@@ -27,6 +27,7 @@ import org.thingsboard.server.common.data.id.RuleChainId;
 import org.thingsboard.server.common.data.id.RuleNodeId;
 import org.thingsboard.server.common.data.id.TenantId;
 import org.thingsboard.server.common.msg.TbActorMsg;
+import org.thingsboard.server.common.msg.TbMsg;
 import org.thingsboard.server.common.msg.plugin.ComponentLifecycleMsg;
 import org.thingsboard.server.common.msg.queue.PartitionChangeMsg;
 
@@ -86,12 +87,19 @@ public class RuleNodeActor extends ComponentActor<RuleNodeId, RuleNodeActorMessa
         }
     }
 
-    private void onRuleChainToRuleNodeMsg(RuleChainToRuleNodeMsg msg) {
+    private void onRuleChainToRuleNodeMsg(RuleChainToRuleNodeMsg envelope) {
+        TbMsg msg = envelope.getMsg();
+        if (!msg.isValid()) {
+            if (log.isTraceEnabled()) {
+                log.trace("Skip processing of message: {} because it is no longer valid!", msg);
+            }
+            return;
+        }
         if (log.isDebugEnabled()) {
-            log.debug("[{}][{}][{}] Going to process rule msg: {}", ruleChainId, id, processor.getComponentName(), msg.getMsg());
+            log.debug("[{}][{}][{}] Going to process rule engine msg: {}", ruleChainId, id, processor.getComponentName(), msg);
         }
         try {
-            processor.onRuleChainToRuleNodeMsg(msg);
+            processor.onRuleChainToRuleNodeMsg(envelope);
             increaseMessagesProcessedCount();
         } catch (Exception e) {
             logAndPersist("onRuleMsg", e);

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

@@ -101,7 +101,7 @@ public class RuleNodeActorMessageProcessor extends ComponentMsgProcessor<RuleNod
     }
 
     public void onRuleToSelfMsg(RuleNodeToSelfMsg msg) throws Exception {
-        checkActive(msg.getMsg());
+        checkComponentStateActive(msg.getMsg());
         TbMsg tbMsg = msg.getMsg();
         int ruleNodeCount = tbMsg.getAndIncrementRuleNodeCounter();
         int maxRuleNodeExecutionsPerMessage = getTenantProfileConfiguration().getMaxRuleNodeExecsPerMessage();
@@ -122,7 +122,7 @@ public class RuleNodeActorMessageProcessor extends ComponentMsgProcessor<RuleNod
 
     void onRuleChainToRuleNodeMsg(RuleChainToRuleNodeMsg msg) throws Exception {
         msg.getMsg().getCallback().onProcessingStart(info);
-        checkActive(msg.getMsg());
+        checkComponentStateActive(msg.getMsg());
         TbMsg tbMsg = msg.getMsg();
         int ruleNodeCount = tbMsg.getAndIncrementRuleNodeCounter();
         int maxRuleNodeExecutionsPerMessage = getTenantProfileConfiguration().getMaxRuleNodeExecsPerMessage();

application/src/main/java/org/thingsboard/server/actors/shared/ComponentMsgProcessor.java

@@ -82,7 +82,17 @@ public abstract class ComponentMsgProcessor<T extends EntityId> extends Abstract
         schedulePeriodicMsgWithDelay(context, new StatsPersistTick(), statsPersistFrequency, statsPersistFrequency);
     }
 
-    protected void checkActive(TbMsg tbMsg) throws RuleNodeException {
+    protected boolean checkMsgValid(TbMsg tbMsg) {
+        var valid = tbMsg.isValid();
+        if (!valid) {
+            if (log.isTraceEnabled()) {
+                log.trace("Skip processing of message: {} because it is no longer valid!", tbMsg);
+            }
+        }
+        return valid;
+    }
+
+    protected void checkComponentStateActive(TbMsg tbMsg) throws RuleNodeException {
         if (state != ComponentLifecycleState.ACTIVE) {
             log.debug("Component is not active. Current state [{}] for processor [{}][{}] tenant [{}]", state, entityId.getEntityType(), entityId, tenantId);
             RuleNodeException ruleNodeException = getInactiveException();

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

@@ -66,6 +66,11 @@ public class TbMsgPackCallback implements TbMsgCallback {
         ctx.onFailure(tenantId, id, e);
     }
 
+    @Override
+    public boolean isMsgValid() {
+        return !ctx.isComplete();
+    }
+
     @Override
     public void onProcessingStart(RuleNodeInfo ruleNodeInfo) {
         log.trace("[{}] ON PROCESSING START: {}", id, ruleNodeInfo);

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

@@ -53,6 +53,8 @@ public class TbMsgPackProcessingContext {
     private final ConcurrentMap<TenantId, RuleEngineException> exceptionsMap = new ConcurrentHashMap<>();
 
     private final ConcurrentMap<UUID, RuleNodeInfo> lastRuleNodeMap = new ConcurrentHashMap<>();
+    @Getter
+    private volatile boolean complete = false;
 
     public TbMsgPackProcessingContext(String queueName, TbRuleEngineSubmitStrategy submitStrategy) {
         this.queueName = queueName;
@@ -149,6 +151,7 @@ public class TbMsgPackProcessingContext {
     }
 
     public void cleanup() {
+        complete = true;
         pendingMap.clear();
         successMap.clear();
         failedMap.clear();

application/src/test/java/org/thingsboard/server/rules/flow/AbstractRuleEngineFlowIntegrationTest.java

@@ -144,6 +144,7 @@ public abstract class AbstractRuleEngineFlowIntegrationTest extends AbstractRule
         Thread.sleep(1000);
 
         TbMsgCallback tbMsgCallback = Mockito.mock(TbMsgCallback.class);
+        Mockito.when(tbMsgCallback.isMsgValid()).thenReturn(true);
         TbMsg tbMsg = TbMsg.newMsg("CUSTOM", device.getId(), new TbMsgMetaData(), "{}", tbMsgCallback);
         QueueToRuleEngineMsg qMsg = new QueueToRuleEngineMsg(savedTenant.getId(), tbMsg, null, null);
         // Pushing Message to the system
@@ -256,6 +257,7 @@ public abstract class AbstractRuleEngineFlowIntegrationTest extends AbstractRule
         Thread.sleep(1000);
 
         TbMsgCallback tbMsgCallback = Mockito.mock(TbMsgCallback.class);
+        Mockito.when(tbMsgCallback.isMsgValid()).thenReturn(true);
         TbMsg tbMsg = TbMsg.newMsg("CUSTOM", device.getId(), new TbMsgMetaData(), "{}", tbMsgCallback);
         QueueToRuleEngineMsg qMsg = new QueueToRuleEngineMsg(savedTenant.getId(), tbMsg, null, null);
         // Pushing Message to the system

application/src/test/java/org/thingsboard/server/rules/lifecycle/AbstractRuleEngineLifecycleIntegrationTest.java

@@ -140,6 +140,7 @@ public abstract class AbstractRuleEngineLifecycleIntegrationTest extends Abstrac
         Thread.sleep(1000);
 
         TbMsgCallback tbMsgCallback = Mockito.mock(TbMsgCallback.class);
+        Mockito.when(tbMsgCallback.isMsgValid()).thenReturn(true);
         TbMsg tbMsg = TbMsg.newMsg("CUSTOM", device.getId(), new TbMsgMetaData(), "{}", tbMsgCallback);
         QueueToRuleEngineMsg qMsg = new QueueToRuleEngineMsg(savedTenant.getId(), tbMsg, null, null);
         // Pushing Message to the system

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

@@ -269,7 +269,7 @@ public final class TbMsg implements Serializable {
     }
 
     public TbMsgCallback getCallback() {
-        //May be null in case of deserialization;
+        // May be null in case of deserialization;
         if (callback != null) {
             return callback;
         } else {
@@ -288,4 +288,12 @@ public final class TbMsg implements Serializable {
     public TbMsgProcessingStackItem popFormStack() {
         return ctx.pop();
     }
+
+    /**
+     * Checks if the message is still valid for processing. May be invalid if the message pack is timed-out or canceled.
+     * @return 'true' if message is valid for processing, 'false' otherwise.
+     */
+    public boolean isValid() {
+        return getCallback().isMsgValid();
+    }
 }

common/message/src/main/java/org/thingsboard/server/common/msg/queue/TbMsgCallback.java

@@ -17,6 +17,9 @@ package org.thingsboard.server.common.msg.queue;
 
 import org.thingsboard.server.common.data.id.RuleNodeId;
 
+/**
+ * Should be renamed to TbMsgPackContext, but this can't be changed due to backward-compatibility.
+ */
 public interface TbMsgCallback {
 
     TbMsgCallback EMPTY = new TbMsgCallback() {
@@ -36,11 +39,20 @@ public interface TbMsgCallback {
 
     void onFailure(RuleEngineException e);
 
+    /**
+     * Returns 'true' if rule engine is expecting the message to be processed, 'false' otherwise.
+     * message may no longer be valid, if the message pack is already expired/canceled/failed.
+     *
+     * @return 'true' if rule engine is expecting the message to be processed, 'false' otherwise.
+     */
+    default boolean isMsgValid() {
+        return true;
+    }
+
     default void onProcessingStart(RuleNodeInfo ruleNodeInfo) {
     }
 
     default void onProcessingEnd(RuleNodeId ruleNodeId) {
     }
 
-
 }