Merge branch 'improvements/deduplication-rule-node-cache' of github.com:ShvaykaD/thingsboard into feature/rule-node-cache

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

@@ -36,6 +36,7 @@ import org.thingsboard.script.api.js.JsInvokeService;
 import org.thingsboard.script.api.tbel.TbelInvokeService;
 import org.thingsboard.server.actors.service.ActorService;
 import org.thingsboard.server.actors.tenant.DebugTbRateLimits;
+import org.thingsboard.server.cache.rule.RuleNodeCache;
 import org.thingsboard.server.cluster.TbClusterService;
 import org.thingsboard.server.common.data.event.ErrorEvent;
 import org.thingsboard.server.common.data.event.LifecycleEvent;
@@ -234,6 +235,10 @@ public class ActorSystemContext {
     @Getter
     private RuleNodeStateService ruleNodeStateService;
 
+    @Autowired
+    @Getter
+    private RuleNodeCache ruleNodeCache;
+
     @Autowired
     private PartitionService partitionService;
 

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

@@ -31,6 +31,7 @@ import org.thingsboard.rule.engine.api.RuleEngineAssetProfileCache;
 import org.thingsboard.rule.engine.api.RuleEngineDeviceProfileCache;
 import org.thingsboard.rule.engine.api.RuleEngineRpcService;
 import org.thingsboard.rule.engine.api.RuleEngineTelemetryService;
+import org.thingsboard.rule.engine.api.RuleNodeCacheService;
 import org.thingsboard.rule.engine.api.ScriptEngine;
 import org.thingsboard.rule.engine.api.SmsService;
 import org.thingsboard.rule.engine.api.TbContext;
@@ -98,6 +99,7 @@ import org.thingsboard.server.dao.widget.WidgetsBundleService;
 import org.thingsboard.server.gen.transport.TransportProtos;
 import org.thingsboard.server.queue.TbQueueCallback;
 import org.thingsboard.server.queue.TbQueueMsgMetadata;
+import org.thingsboard.server.service.rule.TbRuleNodeCacheService;
 import org.thingsboard.server.service.script.RuleNodeJsScriptEngine;
 import org.thingsboard.server.service.script.RuleNodeTbelScriptEngine;
 
@@ -119,11 +121,13 @@ class DefaultTbContext implements TbContext {
     private final ActorSystemContext mainCtx;
     private final String ruleChainName;
     private final RuleNodeCtx nodeCtx;
+    private final TbRuleNodeCacheService ruleNodeCacheService;
 
     public DefaultTbContext(ActorSystemContext mainCtx, String ruleChainName, RuleNodeCtx nodeCtx) {
         this.mainCtx = mainCtx;
         this.ruleChainName = ruleChainName;
         this.nodeCtx = nodeCtx;
+        this.ruleNodeCacheService = new DefaultTbRuleNodeCacheService(getSelfId(), mainCtx.getRuleNodeCache());
     }
 
     @Override
@@ -733,6 +737,11 @@ class DefaultTbContext implements TbContext {
         mainCtx.getRuleNodeStateService().removeByRuleNodeId(getTenantId(), getSelfId());
     }
 
+    @Override
+    public RuleNodeCacheService getRuleNodeCacheService() {
+        return ruleNodeCacheService;
+    }
+
     @Override
     public void removeRuleNodeStateForEntity(EntityId entityId) {
         if (log.isDebugEnabled()) {

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

@@ -0,0 +1,90 @@
+/**
+ * Copyright © 2016-2022 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.actors.ruleChain;
+
+import lombok.Data;
+import lombok.RequiredArgsConstructor;
+import org.thingsboard.server.cache.rule.RuleNodeCache;
+import org.thingsboard.server.common.data.id.EntityId;
+import org.thingsboard.server.common.data.id.RuleNodeId;
+import org.thingsboard.server.common.msg.TbMsg;
+import org.thingsboard.server.service.rule.TbRuleNodeCacheService;
+
+import java.util.List;
+import java.util.Set;
+
+@Data
+@RequiredArgsConstructor
+public class DefaultTbRuleNodeCacheService implements TbRuleNodeCacheService {
+
+    private final RuleNodeId ruleNodeId;
+    private final RuleNodeCache cache;
+
+    @Override
+    public void add(String key, String value) {
+        cache.add(toRuleNodeCacheKey(key), value);
+    }
+
+    @Override
+    public void add(String key, EntityId id) {
+        cache.add(toRuleNodeCacheKey(key), id);
+    }
+
+    @Override
+    public void add(String key, TbMsg value) {
+        cache.add(toRuleNodeCacheKey(key), value);
+    }
+
+    @Override
+    public void removeTbMsgList(String key, List<TbMsg> values) {
+        cache.removeTbMsgList(toRuleNodeCacheKey(key), values);
+    }
+
+    @Override
+    public void removeStringList(String key, List<String> values) {
+        cache.removeStringList(toRuleNodeCacheKey(key), values);
+    }
+
+    @Override
+    public void removeEntityIdList(String key, List<EntityId> values) {
+        cache.removeEntityIdList(toRuleNodeCacheKey(key), values);
+    }
+
+    @Override
+    public Set<String> getStrings(String key) {
+        return cache.getStringSetByKey(toRuleNodeCacheKey(key));
+    }
+
+    @Override
+    public Set<EntityId> getEntityIds(String key) {
+        return cache.getEntityIdSetByKey(toRuleNodeCacheKey(key));
+    }
+
+    @Override
+    public Set<TbMsg> getTbMsgs(String key, String queueName) {
+        return cache.getTbMsgSetByKey(toRuleNodeCacheKey(key), queueName);
+    }
+
+    @Override
+    public void evict(String key) {
+        cache.evict(key);
+    }
+
+    private String toRuleNodeCacheKey(String key) {
+        return String.format("%s::%s", ruleNodeId.toString(), key);
+    }
+
+}

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

@@ -88,7 +88,7 @@ public class RuleNodeActorMessageProcessor extends ComponentMsgProcessor<RuleNod
     @Override
     public void stop(TbActorCtx context) {
         if (tbNode != null) {
-            tbNode.destroy();
+            tbNode.destroy(defaultCtx);
             state = ComponentLifecycleState.SUSPENDED;
         }
     }

application/src/main/java/org/thingsboard/server/service/install/DefaultSystemDataLoaderService.java

@@ -203,8 +203,8 @@ public class DefaultSystemDataLoaderService implements SystemDataLoaderService {
         isolatedRuleEngineTenantProfileData.setConfiguration(new DefaultTenantProfileConfiguration());
 
         TenantProfileQueueConfiguration mainQueueConfiguration = new TenantProfileQueueConfiguration();
-        mainQueueConfiguration.setName("Main");
-        mainQueueConfiguration.setTopic("tb_rule_engine.main");
+        mainQueueConfiguration.setName(DataConstants.MAIN_QUEUE_NAME);
+        mainQueueConfiguration.setTopic(DataConstants.MAIN_QUEUE_TOPIC);
         mainQueueConfiguration.setPollInterval(25);
         mainQueueConfiguration.setPartitions(10);
         mainQueueConfiguration.setConsumerPerPartition(true);
@@ -598,12 +598,12 @@ public class DefaultSystemDataLoaderService implements SystemDataLoaderService {
 
     @Override
     public void createQueues() {
-        Queue mainQueue = queueService.findQueueByTenantIdAndName(TenantId.SYS_TENANT_ID, "Main");
+        Queue mainQueue = queueService.findQueueByTenantIdAndName(TenantId.SYS_TENANT_ID, DataConstants.MAIN_QUEUE_NAME);
         if (mainQueue == null) {
             mainQueue = new Queue();
             mainQueue.setTenantId(TenantId.SYS_TENANT_ID);
-            mainQueue.setName("Main");
-            mainQueue.setTopic("tb_rule_engine.main");
+            mainQueue.setName(DataConstants.MAIN_QUEUE_NAME);
+            mainQueue.setTopic(DataConstants.MAIN_QUEUE_TOPIC);
             mainQueue.setPollInterval(25);
             mainQueue.setPartitions(10);
             mainQueue.setConsumerPerPartition(true);
@@ -622,12 +622,12 @@ public class DefaultSystemDataLoaderService implements SystemDataLoaderService {
             queueService.saveQueue(mainQueue);
         }
 
-        Queue highPriorityQueue = queueService.findQueueByTenantIdAndName(TenantId.SYS_TENANT_ID, "HighPriority");
+        Queue highPriorityQueue = queueService.findQueueByTenantIdAndName(TenantId.SYS_TENANT_ID, DataConstants.HP_QUEUE_NAME);
         if (highPriorityQueue == null) {
             highPriorityQueue = new Queue();
             highPriorityQueue.setTenantId(TenantId.SYS_TENANT_ID);
-            highPriorityQueue.setName("HighPriority");
-            highPriorityQueue.setTopic("tb_rule_engine.hp");
+            highPriorityQueue.setName(DataConstants.HP_QUEUE_NAME);
+            highPriorityQueue.setTopic(DataConstants.HP_QUEUE_TOPIC);
             highPriorityQueue.setPollInterval(25);
             highPriorityQueue.setPartitions(10);
             highPriorityQueue.setConsumerPerPartition(true);
@@ -646,12 +646,12 @@ public class DefaultSystemDataLoaderService implements SystemDataLoaderService {
             queueService.saveQueue(highPriorityQueue);
         }
 
-        Queue sequentialByOriginatorQueue = queueService.findQueueByTenantIdAndName(TenantId.SYS_TENANT_ID, "SequentialByOriginator");
+        Queue sequentialByOriginatorQueue = queueService.findQueueByTenantIdAndName(TenantId.SYS_TENANT_ID, DataConstants.SQ_QUEUE_NAME);
         if (sequentialByOriginatorQueue == null) {
             sequentialByOriginatorQueue = new Queue();
             sequentialByOriginatorQueue.setTenantId(TenantId.SYS_TENANT_ID);
-            sequentialByOriginatorQueue.setName("SequentialByOriginator");
-            sequentialByOriginatorQueue.setTopic("tb_rule_engine.sq");
+            sequentialByOriginatorQueue.setName(DataConstants.SQ_QUEUE_NAME);
+            sequentialByOriginatorQueue.setTopic(DataConstants.SQ_QUEUE_TOPIC);
             sequentialByOriginatorQueue.setPollInterval(25);
             sequentialByOriginatorQueue.setPartitions(10);
             sequentialByOriginatorQueue.setPackProcessingTimeout(2000);

application/src/main/java/org/thingsboard/server/service/install/TbRuleEngineQueueConfigService.java

@@ -21,6 +21,7 @@ import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
 import org.springframework.boot.context.properties.ConfigurationProperties;
 import org.springframework.context.annotation.Configuration;
 import org.springframework.context.annotation.Profile;
+import org.thingsboard.server.common.data.DataConstants;
 import org.thingsboard.server.queue.settings.TbRuleEngineQueueConfiguration;
 
 import javax.annotation.PostConstruct;
@@ -39,7 +40,7 @@ public class TbRuleEngineQueueConfigService {
 
     @PostConstruct
     public void validate() {
-        queues.stream().filter(queue -> queue.getName().equals("Main")).findFirst().orElseThrow(() -> {
+        queues.stream().filter(queue -> queue.getName().equals(DataConstants.MAIN_QUEUE_NAME)).findFirst().orElseThrow(() -> {
             log.error("Main queue is not configured in thingsboard.yml");
             return new RuntimeException("No \"Main\" queue configured!");
         });

application/src/main/java/org/thingsboard/server/service/install/update/DefaultDataUpdateService.java

@@ -30,6 +30,7 @@ import org.thingsboard.rule.engine.flow.TbRuleChainInputNode;
 import org.thingsboard.rule.engine.flow.TbRuleChainInputNodeConfiguration;
 import org.thingsboard.rule.engine.profile.TbDeviceProfileNode;
 import org.thingsboard.rule.engine.profile.TbDeviceProfileNodeConfiguration;
+import org.thingsboard.server.common.data.DataConstants;
 import org.thingsboard.server.common.data.EntityView;
 import org.thingsboard.server.common.data.Tenant;
 import org.thingsboard.server.common.data.TenantProfile;
@@ -651,8 +652,8 @@ public class DefaultDataUpdateService implements DataUpdateService {
 
     private TenantProfileQueueConfiguration getMainQueueConfiguration() {
         TenantProfileQueueConfiguration mainQueueConfiguration = new TenantProfileQueueConfiguration();
-        mainQueueConfiguration.setName("Main");
-        mainQueueConfiguration.setTopic("tb_rule_engine.main");
+        mainQueueConfiguration.setName(DataConstants.MAIN_QUEUE_NAME);
+        mainQueueConfiguration.setTopic(DataConstants.MAIN_QUEUE_TOPIC);
         mainQueueConfiguration.setPollInterval(25);
         mainQueueConfiguration.setPartitions(10);
         mainQueueConfiguration.setConsumerPerPartition(true);

application/src/main/java/org/thingsboard/server/service/rule/TbRuleNodeCacheService.java

@@ -0,0 +1,22 @@
+/**
+ * Copyright © 2016-2022 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.service.rule;
+
+import org.thingsboard.rule.engine.api.RuleNodeCacheService;
+
+public interface TbRuleNodeCacheService extends RuleNodeCacheService {
+
+}

application/src/test/java/org/thingsboard/server/controller/BaseTenantControllerTest.java

@@ -30,6 +30,7 @@ import org.mockito.Mockito;
 import org.springframework.test.context.TestPropertySource;
 import org.springframework.test.web.servlet.ResultActions;
 import org.thingsboard.common.util.ThingsBoardExecutors;
+import org.thingsboard.server.common.data.DataConstants;
 import org.thingsboard.server.common.data.StringUtils;
 import org.thingsboard.server.common.data.Tenant;
 import org.thingsboard.server.common.data.TenantInfo;
@@ -427,7 +428,7 @@ public abstract class BaseTenantControllerTest extends AbstractControllerTest {
         tenantProfileData.setConfiguration(new DefaultTenantProfileConfiguration());
         tenantProfile.setProfileData(tenantProfileData);
         tenantProfile.setIsolatedTbRuleEngine(true);
-        addQueueConfig(tenantProfile, "Main");
+        addQueueConfig(tenantProfile, DataConstants.MAIN_QUEUE_NAME);
         addQueueConfig(tenantProfile, "Test");
         tenantProfile = doPost("/api/tenantProfile", tenantProfile, TenantProfile.class);
 
@@ -456,7 +457,7 @@ public abstract class BaseTenantControllerTest extends AbstractControllerTest {
         tenantProfileData2.setConfiguration(new DefaultTenantProfileConfiguration());
         tenantProfile2.setProfileData(tenantProfileData2);
         tenantProfile2.setIsolatedTbRuleEngine(true);
-        addQueueConfig(tenantProfile2, "Main");
+        addQueueConfig(tenantProfile2, DataConstants.MAIN_QUEUE_NAME);
         addQueueConfig(tenantProfile2, "Test");
         addQueueConfig(tenantProfile2, "Test2");
         tenantProfile2 = doPost("/api/tenantProfile", tenantProfile2, TenantProfile.class);

application/src/test/java/org/thingsboard/server/controller/BaseTenantProfileControllerTest.java

@@ -20,6 +20,7 @@ import org.junit.Assert;
 import org.junit.Test;
 import org.mockito.ArgumentMatcher;
 import org.mockito.Mockito;
+import org.thingsboard.server.common.data.DataConstants;
 import org.thingsboard.server.common.data.EntityInfo;
 import org.thingsboard.server.common.data.StringUtils;
 import org.thingsboard.server.common.data.Tenant;
@@ -341,8 +342,8 @@ public abstract class BaseTenantProfileControllerTest extends AbstractController
 
     private void addMainQueueConfig(TenantProfile tenantProfile) {
         TenantProfileQueueConfiguration mainQueueConfiguration = new TenantProfileQueueConfiguration();
-        mainQueueConfiguration.setName("Main");
-        mainQueueConfiguration.setTopic("tb_rule_engine.main");
+        mainQueueConfiguration.setName(DataConstants.MAIN_QUEUE_NAME);
+        mainQueueConfiguration.setTopic(DataConstants.MAIN_QUEUE_TOPIC);
         mainQueueConfiguration.setPollInterval(25);
         mainQueueConfiguration.setPartitions(10);
         mainQueueConfiguration.setConsumerPerPartition(true);

application/src/test/java/org/thingsboard/server/edge/AbstractEdgeTest.java

@@ -28,6 +28,7 @@ import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.test.context.TestPropertySource;
 import org.thingsboard.common.util.JacksonUtil;
 import org.thingsboard.server.cluster.TbClusterService;
+import org.thingsboard.server.common.data.DataConstants;
 import org.thingsboard.server.common.data.Device;
 import org.thingsboard.server.common.data.DeviceProfile;
 import org.thingsboard.server.common.data.OtaPackageInfo;
@@ -427,8 +428,8 @@ abstract public class AbstractEdgeTest extends AbstractControllerTest {
         UUID queueUUID = new UUID(queueUpdateMsg.getIdMSB(), queueUpdateMsg.getIdLSB());
         Queue queue = doGet("/api/queues/" + queueUUID, Queue.class);
         Assert.assertNotNull(queue);
-        Assert.assertEquals("Main", queueUpdateMsg.getName());
-        Assert.assertEquals("tb_rule_engine.main", queueUpdateMsg.getTopic());
+        Assert.assertEquals(DataConstants.MAIN_QUEUE_NAME, queueUpdateMsg.getName());
+        Assert.assertEquals(DataConstants.MAIN_QUEUE_TOPIC, queueUpdateMsg.getTopic());
         Assert.assertEquals(10, queueUpdateMsg.getPartitions());
         Assert.assertEquals(25, queueUpdateMsg.getPollInterval());
         testAutoGeneratedCodeByProtobuf(queueUpdateMsg);

application/src/test/java/org/thingsboard/server/service/queue/DefaultTbClusterServiceTest.java

@@ -24,6 +24,7 @@ import org.springframework.boot.test.mock.mockito.SpyBean;
 import org.springframework.test.context.ContextConfiguration;
 import org.springframework.test.context.junit4.SpringRunner;
 import org.thingsboard.server.cluster.TbClusterService;
+import org.thingsboard.server.common.data.DataConstants;
 import org.thingsboard.server.common.data.id.QueueId;
 import org.thingsboard.server.common.data.id.TenantId;
 import org.thingsboard.server.common.data.queue.Queue;
@@ -239,7 +240,7 @@ public class DefaultTbClusterServiceTest {
         TenantId tenantId = TenantId.SYS_TENANT_ID;
         Queue queue = new Queue(new QueueId(UUID.randomUUID()));
         queue.setTenantId(tenantId);
-        queue.setName("Main");
+        queue.setName(DataConstants.MAIN_QUEUE_NAME);
         queue.setTopic("main");
         queue.setPartitions(10);
         return queue;

application/src/test/java/org/thingsboard/server/service/queue/TbMsgPackProcessingContextTest.java

@@ -22,6 +22,7 @@ import org.junit.Test;
 import org.junit.runner.RunWith;
 import org.mockito.junit.MockitoJUnitRunner;
 import org.thingsboard.common.util.ThingsBoardThreadFactory;
+import org.thingsboard.server.common.data.DataConstants;
 import org.thingsboard.server.gen.transport.TransportProtos;
 import org.thingsboard.server.queue.common.TbProtoQueueMsg;
 import org.thingsboard.server.service.queue.processing.TbRuleEngineSubmitStrategy;
@@ -69,7 +70,7 @@ public class TbMsgPackProcessingContextTest {
         TbRuleEngineSubmitStrategy strategyMock = mock(TbRuleEngineSubmitStrategy.class);
         when(strategyMock.getPendingMap()).thenReturn(messages);
 
-        TbMsgPackProcessingContext context = new TbMsgPackProcessingContext("Main", strategyMock, false);
+        TbMsgPackProcessingContext context = new TbMsgPackProcessingContext(DataConstants.MAIN_QUEUE_NAME, strategyMock, false);
         for (UUID uuid : messages.keySet()) {
             final CountDownLatch readyLatch = new CountDownLatch(parallelCount);
             final CountDownLatch startLatch = new CountDownLatch(1);

common/cache/pom.xml

@@ -95,6 +95,10 @@
             <artifactId>awaitility</artifactId>
             <scope>test</scope>
         </dependency>
+        <dependency>
+            <groupId>org.thingsboard.common</groupId>
+            <artifactId>message</artifactId>
+        </dependency>
     </dependencies>
 
     <build>

common/cache/src/main/java/org/thingsboard/server/cache/rule/DefaultRuleNodeCache.java

@@ -0,0 +1,74 @@
+/**
+ * Copyright © 2016-2022 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.cache.rule;
+
+import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
+import org.springframework.stereotype.Service;
+import org.thingsboard.server.common.data.id.EntityId;
+import org.thingsboard.server.common.msg.TbMsg;
+
+import java.util.Collections;
+import java.util.List;
+import java.util.Set;
+
+@Service
+@ConditionalOnProperty(prefix = "cache", value = "type", havingValue = "caffeine", matchIfMissing = true)
+public class DefaultRuleNodeCache implements RuleNodeCache {
+
+    @Override
+    public void add(String key, String value) {
+    }
+
+    @Override
+    public void add(String key, EntityId value) {
+    }
+
+    @Override
+    public void add(String key, TbMsg value) {
+    }
+
+    @Override
+    public void removeStringList(String key, List<String> values) {
+    }
+
+    @Override
+    public void removeEntityIdList(String key, List<EntityId> values) {
+    }
+
+    @Override
+    public void removeTbMsgList(String key, List<TbMsg> values) {
+    }
+
+    @Override
+    public Set<String> getStringSetByKey(String key) {
+        return Collections.emptySet();
+    }
+
+    @Override
+    public Set<EntityId> getEntityIdSetByKey(String key) {
+        return Collections.emptySet();
+    }
+
+    @Override
+    public Set<TbMsg> getTbMsgSetByKey(String key, String queueName) {
+        return Collections.emptySet();
+    }
+
+    @Override
+    public void evict(String key) {
+    }
+
+}

common/cache/src/main/java/org/thingsboard/server/cache/rule/RedisRuleNodeCache.java

@@ -0,0 +1,153 @@
+/**
+ * Copyright © 2016-2022 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.cache.rule;
+
+import lombok.RequiredArgsConstructor;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
+import org.springframework.data.redis.connection.RedisConnection;
+import org.springframework.data.redis.connection.RedisConnectionFactory;
+import org.springframework.stereotype.Service;
+import org.springframework.util.SerializationUtils;
+import org.thingsboard.server.common.data.id.EntityId;
+import org.thingsboard.server.common.msg.TbMsg;
+import org.thingsboard.server.common.msg.queue.TbMsgCallback;
+
+import java.util.Collections;
+import java.util.List;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+@Service
+@ConditionalOnProperty(prefix = "cache", value = "type", havingValue = "redis")
+@RequiredArgsConstructor
+public class RedisRuleNodeCache implements RuleNodeCache {
+
+    private final RedisConnectionFactory redisConnectionFactory;
+
+    @Override
+    public void add(String key, String value) {
+        processAdd(key, value.getBytes());
+    }
+
+    @Override
+    public void add(String key, EntityId value) {
+        processAdd(key, SerializationUtils.serialize(value));
+    }
+
+    @Override
+    public void add(String key, TbMsg value) {
+        processAdd(key, TbMsg.toByteArray(value));
+    }
+
+    @Override
+    public void removeStringList(String key, List<String> values) {
+        processRemove(key, stringListToBytes(values));
+    }
+
+    @Override
+    public void removeEntityIdList(String key, List<EntityId> values) {
+        processRemove(key, entityIdListToBytes(values));
+    }
+
+    @Override
+    public void removeTbMsgList(String key, List<TbMsg> values) {
+        processRemove(key, tbMsgListToBytes(values));
+    }
+
+    @Override
+    public Set<String> getStringSetByKey(String key) {
+        return toStringSet(processGetMembers(key));
+    }
+
+    @Override
+    public Set<EntityId> getEntityIdSetByKey(String key) {
+        return toEntityIdSet(processGetMembers(key));
+    }
+
+    @Override
+    public Set<TbMsg> getTbMsgSetByKey(String key, String queueName) {
+        return toTbMsgSet(processGetMembers(key), queueName);
+    }
+
+    @Override
+    public void evict(String key) {
+        try (RedisConnection connection = redisConnectionFactory.getConnection()) {
+            connection.del(key.getBytes());
+        }
+    }
+
+    private void processAdd(String key, byte[] value) {
+        try (RedisConnection connection = redisConnectionFactory.getConnection()) {
+            connection.setCommands().sAdd(key.getBytes(), value);
+        }
+    }
+
+    private void processRemove(String key, byte[][] values) {
+        if (values.length == 0) {
+            return;
+        }
+        try (RedisConnection connection = redisConnectionFactory.getConnection()) {
+            connection.setCommands().sRem(key.getBytes(), values);
+        }
+    }
+
+    private Set<byte[]> processGetMembers(String key) {
+        try (RedisConnection connection = redisConnectionFactory.getConnection()) {
+            Set<byte[]> bytes = connection.setCommands().sMembers(key.getBytes());
+            if (bytes == null) {
+                return Collections.emptySet();
+            }
+            return bytes;
+        }
+    }
+
+    private byte[][] stringListToBytes(List<String> values) {
+        return values.stream()
+                .map(String::getBytes)
+                .toArray(byte[][]::new);
+    }
+
+    private byte[][] entityIdListToBytes(List<EntityId> values) {
+        return values.stream()
+                .map(SerializationUtils::serialize)
+                .toArray(byte[][]::new);
+    }
+
+    private byte[][] tbMsgListToBytes(List<TbMsg> values) {
+        return values.stream()
+                .map(TbMsg::toByteArray)
+                .toArray(byte[][]::new);
+    }
+
+    private Set<String> toStringSet(Set<byte[]> values) {
+        return values.stream()
+                .map(String::new)
+                .collect(Collectors.toSet());
+    }
+
+    private Set<EntityId> toEntityIdSet(Set<byte[]> values) {
+        return values.stream()
+                .map(bytes -> (EntityId) SerializationUtils.deserialize(bytes))
+                .collect(Collectors.toSet());
+    }
+
+    private Set<TbMsg> toTbMsgSet(Set<byte[]> values, String queueName) {
+        return values.stream()
+                .map(bytes -> TbMsg.fromBytes(queueName, bytes, TbMsgCallback.EMPTY))
+                .collect(Collectors.toSet());
+    }
+
+}

common/cache/src/main/java/org/thingsboard/server/cache/rule/RuleNodeCache.java

@@ -0,0 +1,45 @@
+/**
+ * Copyright © 2016-2022 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.cache.rule;
+
+import org.thingsboard.server.common.data.id.EntityId;
+import org.thingsboard.server.common.msg.TbMsg;
+
+import java.util.List;
+import java.util.Set;
+
+public interface RuleNodeCache {
+
+    void add(String key, String value);
+
+    void add(String key, EntityId value);
+
+    void add(String key, TbMsg value);
+
+    void removeStringList(String key, List<String> values);
+
+    void removeEntityIdList(String key, List<EntityId> values);
+
+    void removeTbMsgList(String key, List<TbMsg> values);
+
+    Set<String> getStringSetByKey(String key);
+
+    Set<EntityId> getEntityIdSetByKey(String key);
+
+    Set<TbMsg> getTbMsgSetByKey(String key, String queueName);
+
+    void evict(String key);
+}

common/data/src/main/java/org/thingsboard/server/common/data/DataConstants.java

@@ -118,4 +118,12 @@ public class DataConstants {
     public static final String MSG_SOURCE_KEY = "source";
 
     public static final String LAST_CONNECTED_GATEWAY = "lastConnectedGateway";
+
+    public static final String MAIN_QUEUE_NAME = "Main";
+    public static final String MAIN_QUEUE_TOPIC = "tb_rule_engine.main";
+    public static final String HP_QUEUE_NAME = "HighPriority";
+    public static final String HP_QUEUE_TOPIC = "tb_rule_engine.hp";
+    public static final String SQ_QUEUE_NAME = "SequentialByOriginator";
+    public static final String SQ_QUEUE_TOPIC = "tb_rule_engine.sq";
+
 }

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

@@ -17,6 +17,7 @@ package org.thingsboard.server.queue.discovery;
 
 import lombok.AllArgsConstructor;
 import lombok.Data;
+import org.thingsboard.server.common.data.DataConstants;
 import org.thingsboard.server.common.data.id.TenantId;
 import org.thingsboard.server.common.data.queue.Queue;
 import org.thingsboard.server.common.msg.queue.ServiceType;
@@ -24,7 +25,6 @@ import org.thingsboard.server.common.msg.queue.ServiceType;
 @Data
 @AllArgsConstructor
 public class QueueKey {
-    private static final String MAIN = "Main";
 
     private final ServiceType type;
     private final String queueName;
@@ -44,13 +44,13 @@ public class QueueKey {
 
     public QueueKey(ServiceType type, TenantId tenantId) {
         this.type = type;
-        this.queueName = MAIN;
+        this.queueName = DataConstants.MAIN_QUEUE_NAME;
         this.tenantId = tenantId != null ? tenantId : TenantId.SYS_TENANT_ID;
     }
 
     public QueueKey(ServiceType type) {
         this.type = type;
-        this.queueName = MAIN;
+        this.queueName = DataConstants.MAIN_QUEUE_NAME;
         this.tenantId = TenantId.SYS_TENANT_ID;
     }
 

common/queue/src/test/java/org/thingsboard/server/queue/discovery/QueueKeyTest.java

@@ -17,6 +17,7 @@ package org.thingsboard.server.queue.discovery;
 
 import lombok.extern.slf4j.Slf4j;
 import org.junit.jupiter.api.Test;
+import org.thingsboard.server.common.data.DataConstants;
 import org.thingsboard.server.common.data.id.TenantId;
 import org.thingsboard.server.common.msg.queue.ServiceType;
 
@@ -29,7 +30,7 @@ class QueueKeyTest {
 
     @Test
     void testToStringSystemTenant() {
-        QueueKey queueKey = new QueueKey(ServiceType.TB_RULE_ENGINE, "Main", TenantId.SYS_TENANT_ID);
+        QueueKey queueKey = new QueueKey(ServiceType.TB_RULE_ENGINE, DataConstants.MAIN_QUEUE_NAME, TenantId.SYS_TENANT_ID);
         log.info("The queue key is {}",queueKey);
         assertThat(queueKey.toString()).isEqualTo("QK(Main,TB_RULE_ENGINE,system)");
     }
@@ -37,7 +38,7 @@ class QueueKeyTest {
     @Test
     void testToStringCustomTenant() {
         TenantId tenantId = TenantId.fromUUID(UUID.fromString("3ebd39eb-43d4-4911-a818-cdbf8d508f88"));
-        QueueKey queueKey = new QueueKey(ServiceType.TB_RULE_ENGINE, "Main", tenantId);
+        QueueKey queueKey = new QueueKey(ServiceType.TB_RULE_ENGINE, DataConstants.MAIN_QUEUE_NAME, tenantId);
         log.info("The queue key is {}",queueKey);
         assertThat(queueKey.toString()).isEqualTo("QK(Main,TB_RULE_ENGINE,3ebd39eb-43d4-4911-a818-cdbf8d508f88)");
     }

dao/src/main/java/org/thingsboard/server/dao/service/validator/TenantProfileDataValidator.java

@@ -18,6 +18,7 @@ package org.thingsboard.server.dao.service.validator;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.context.annotation.Lazy;
 import org.springframework.stereotype.Component;
+import org.thingsboard.server.common.data.DataConstants;
 import org.thingsboard.server.common.data.StringUtils;
 import org.thingsboard.server.common.data.TenantProfile;
 import org.thingsboard.server.common.data.id.TenantId;
@@ -72,7 +73,7 @@ public class TenantProfileDataValidator extends DataValidator<TenantProfile> {
             Optional<TenantProfileQueueConfiguration> mainQueueConfig =
                     queueConfiguration
                             .stream()
-                            .filter(q -> q.getName().equals("Main"))
+                            .filter(q -> q.getName().equals(DataConstants.MAIN_QUEUE_NAME))
                             .findAny();
             if (mainQueueConfig.isEmpty()) {
                 throw new DataValidationException("Main queue configuration should be specified!");

dao/src/main/java/org/thingsboard/server/dao/sql/queue/JpaQueueDao.java

@@ -20,6 +20,7 @@ import lombok.extern.slf4j.Slf4j;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.data.jpa.repository.JpaRepository;
 import org.springframework.stereotype.Component;
+import org.thingsboard.server.common.data.DataConstants;
 import org.thingsboard.server.common.data.id.TenantId;
 import org.thingsboard.server.common.data.page.PageData;
 import org.thingsboard.server.common.data.page.PageLink;
@@ -70,7 +71,7 @@ public class JpaQueueDao extends JpaAbstractDao<QueueEntity, Queue> implements Q
 
     @Override
     public List<Queue> findAllMainQueues() {
-        List<QueueEntity> entities = Lists.newArrayList(queueRepository.findAllByName("Main"));
+        List<QueueEntity> entities = Lists.newArrayList(queueRepository.findAllByName(DataConstants.MAIN_QUEUE_NAME));
         return DaoUtil.convertDataList(entities);
     }
 

dao/src/test/java/org/thingsboard/server/dao/service/BaseQueueServiceTest.java

@@ -19,6 +19,7 @@ import org.junit.After;
 import org.junit.Assert;
 import org.junit.Before;
 import org.junit.Test;
+import org.thingsboard.server.common.data.DataConstants;
 import org.thingsboard.server.common.data.Tenant;
 import org.thingsboard.server.common.data.TenantProfile;
 import org.thingsboard.server.common.data.id.TenantId;
@@ -54,8 +55,8 @@ public abstract class BaseQueueServiceTest extends AbstractServiceTest {
         tenantProfile.setIsolatedTbRuleEngine(true);
 
         TenantProfileQueueConfiguration mainQueueConfiguration = new TenantProfileQueueConfiguration();
-        mainQueueConfiguration.setName("Main");
-        mainQueueConfiguration.setTopic("tb_rule_engine.main");
+        mainQueueConfiguration.setName(DataConstants.MAIN_QUEUE_NAME);
+        mainQueueConfiguration.setTopic(DataConstants.MAIN_QUEUE_TOPIC);
         mainQueueConfiguration.setPollInterval(25);
         mainQueueConfiguration.setPartitions(10);
         mainQueueConfiguration.setConsumerPerPartition(true);
@@ -495,7 +496,7 @@ public abstract class BaseQueueServiceTest extends AbstractServiceTest {
 
         for (int i = 0; i < loadedQueues.size(); i++) {
             Queue queue = loadedQueues.get(i);
-            if (queue.getName().equals("Main")) {
+            if (queue.getName().equals(DataConstants.MAIN_QUEUE_NAME)) {
                 loadedQueues.remove(queue);
                 break;
             }

dao/src/test/java/org/thingsboard/server/dao/service/BaseTenantProfileServiceTest.java

@@ -19,6 +19,7 @@ import com.fasterxml.jackson.databind.node.NullNode;
 import org.junit.After;
 import org.junit.Assert;
 import org.junit.Test;
+import org.thingsboard.server.common.data.DataConstants;
 import org.thingsboard.server.common.data.EntityInfo;
 import org.thingsboard.server.common.data.Tenant;
 import org.thingsboard.server.common.data.TenantProfile;
@@ -57,8 +58,8 @@ public abstract class BaseTenantProfileServiceTest extends AbstractServiceTest {
         tenantProfile.setIsolatedTbRuleEngine(true);
 
         TenantProfileQueueConfiguration mainQueueConfiguration = new TenantProfileQueueConfiguration();
-        mainQueueConfiguration.setName("Main");
-        mainQueueConfiguration.setTopic("tb_rule_engine.main");
+        mainQueueConfiguration.setName(DataConstants.MAIN_QUEUE_NAME);
+        mainQueueConfiguration.setTopic(DataConstants.MAIN_QUEUE_TOPIC);
         mainQueueConfiguration.setPollInterval(25);
         mainQueueConfiguration.setPartitions(10);
         mainQueueConfiguration.setConsumerPerPartition(true);
@@ -295,8 +296,8 @@ public abstract class BaseTenantProfileServiceTest extends AbstractServiceTest {
 
     private void addMainQueueConfig(TenantProfile tenantProfile) {
         TenantProfileQueueConfiguration mainQueueConfiguration = new TenantProfileQueueConfiguration();
-        mainQueueConfiguration.setName("Main");
-        mainQueueConfiguration.setTopic("tb_rule_engine.main");
+        mainQueueConfiguration.setName(DataConstants.MAIN_QUEUE_NAME);
+        mainQueueConfiguration.setTopic(DataConstants.MAIN_QUEUE_TOPIC);
         mainQueueConfiguration.setPollInterval(25);
         mainQueueConfiguration.setPartitions(10);
         mainQueueConfiguration.setConsumerPerPartition(true);

rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/RuleNodeCacheService.java

@@ -0,0 +1,46 @@
+/**
+ * Copyright © 2016-2022 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.rule.engine.api;
+
+import org.thingsboard.server.common.data.id.EntityId;
+import org.thingsboard.server.common.msg.TbMsg;
+
+import java.util.List;
+import java.util.Set;
+
+public interface RuleNodeCacheService {
+
+    void add(String key, String value);
+
+    void add(String key, EntityId id);
+
+    void add(String key, TbMsg value);
+
+    void removeStringList(String key, List<String> values);
+
+    void removeEntityIdList(String key, List<EntityId> values);
+
+    void removeTbMsgList(String key, List<TbMsg> values);
+
+    Set<String> getStrings(String key);
+
+    Set<EntityId> getEntityIds(String key);
+
+    Set<TbMsg> getTbMsgs(String key, String queueName);
+
+    void evict(String key);
+
+}

rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/TbContext.java

@@ -317,6 +317,8 @@ public interface TbContext {
 
     void clearRuleNodeStates();
 
+    RuleNodeCacheService getRuleNodeCacheService();
+
     void addTenantProfileListener(Consumer<TenantProfile> listener);
 
     void addDeviceProfileListeners(Consumer<DeviceProfile> listener, BiConsumer<DeviceId, DeviceProfile> deviceListener);

rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/TbNode.java

@@ -31,6 +31,11 @@ public interface TbNode {
 
     default void destroy() {}
 
+    // todo: add reason for destroy
+    default void destroy(TbContext ctx) {
+        destroy();
+    }
+
     default void onPartitionChangeMsg(TbContext ctx, PartitionChangeMsg msg) {}
 
 }

rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/TbNodeException.java

@@ -15,8 +15,6 @@
  */
 package org.thingsboard.rule.engine.api;
 
-import com.fasterxml.jackson.core.JsonProcessingException;
-
 /**
  * Created by ashvayka on 19.01.18.
  */

rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/deduplication/DeduplicationData.java

@@ -39,6 +39,10 @@ public class DeduplicationData {
         msgList.add(msg);
     }
 
+    public void addAll(List<TbMsg> tbMsgs) {
+        msgList.addAll(tbMsgs);
+    }
+
     public boolean isEmpty() {
         return msgList.isEmpty();
     }

rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/deduplication/TbMsgDeduplicationNode.java

@@ -39,6 +39,7 @@ import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
+import java.util.Set;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.TimeUnit;
 
@@ -59,6 +60,7 @@ import java.util.concurrent.TimeUnit;
 public class TbMsgDeduplicationNode implements TbNode {
 
     private static final String TB_MSG_DEDUPLICATION_TIMEOUT_MSG = "TbMsgDeduplicationNodeMsg";
+    private static final String DEDUPLICATION_IDS_CACHE_KEY = "deduplication_ids";
     public static final int TB_MSG_DEDUPLICATION_RETRY_DELAY = 10;
     private static final String EMPTY_DATA = "";
     private static final TbMsgMetaData EMPTY_META_DATA = new TbMsgMetaData();
@@ -78,6 +80,9 @@ public class TbMsgDeduplicationNode implements TbNode {
         this.config = TbNodeUtils.convert(configuration, TbMsgDeduplicationNodeConfiguration.class);
         this.deduplicationInterval = TimeUnit.SECONDS.toMillis(config.getInterval());
         this.deduplicationId = config.getId();
+        if (deduplicationMap.isEmpty()) {
+            getDeduplicationDataFromCacheAndSchedule(ctx);
+        }
     }
 
     @Override
@@ -90,8 +95,36 @@ public class TbMsgDeduplicationNode implements TbNode {
     }
 
     @Override
-    public void destroy() {
+    public void destroy(TbContext ctx) {
         deduplicationMap.clear();
+        Set<EntityId> deduplicationIds = getDeduplicationIds(ctx);
+        if (deduplicationIds == null) {
+            return;
+        }
+        deduplicationIds.forEach(id -> ctx.getRuleNodeCacheService().evict(id.toString()));
+        ctx.getRuleNodeCacheService().evict(DEDUPLICATION_IDS_CACHE_KEY);
+    }
+
+    private void getDeduplicationDataFromCacheAndSchedule(TbContext ctx) {
+        Set<EntityId> deduplicationIds = getDeduplicationIds(ctx);
+        if (deduplicationIds == null) {
+            return;
+        }
+        deduplicationIds.forEach(id -> {
+            List<TbMsg> tbMsgs = new ArrayList<>(ctx.getRuleNodeCacheService().getTbMsgs(id.toString(), config.getQueueName()));
+            DeduplicationData deduplicationData = new DeduplicationData();
+            deduplicationData.addAll(tbMsgs);
+            deduplicationMap.put(id, deduplicationData);
+            scheduleTickMsg(ctx, id, deduplicationData);
+        });
+    }
+
+    private Set<EntityId> getDeduplicationIds(TbContext ctx) {
+        Set<EntityId> deduplicationIds = ctx.getRuleNodeCacheService().getEntityIds(TbMsgDeduplicationNode.DEDUPLICATION_IDS_CACHE_KEY);
+        if (deduplicationIds.isEmpty()) {
+            return null;
+        }
+        return deduplicationIds;
     }
 
     private void processOnRegularMsg(TbContext ctx, TbMsg msg) {
@@ -99,7 +132,11 @@ public class TbMsgDeduplicationNode implements TbNode {
         DeduplicationData deduplicationMsgs = deduplicationMap.computeIfAbsent(id, k -> new DeduplicationData());
         if (deduplicationMsgs.size() < config.getMaxPendingMsgs()) {
             log.trace("[{}][{}] Adding msg: [{}][{}] to the pending msgs map ...", ctx.getSelfId(), id, msg.getId(), msg.getMetaDataTs());
+            if (deduplicationMsgs.isEmpty()) {
+                ctx.getRuleNodeCacheService().add(DEDUPLICATION_IDS_CACHE_KEY, id);
+            }
             deduplicationMsgs.add(msg);
+            ctx.getRuleNodeCacheService().add(id.toString(), msg);
             ctx.ack(msg);
             scheduleTickMsg(ctx, id, deduplicationMsgs);
         } else {
@@ -132,13 +169,13 @@ public class TbMsgDeduplicationNode implements TbNode {
         }
         long deduplicationTimeoutMs = System.currentTimeMillis();
         try {
-            List<TbMsg> deduplicationResults = new ArrayList<>();
+            List<TbPair<TbMsg, List<TbMsg>>> deduplicationResults = new ArrayList<>();
             List<TbMsg> msgList = data.getMsgList();
             Optional<TbPair<Long, Long>> packBoundsOpt = findValidPack(msgList, deduplicationTimeoutMs);
             while (packBoundsOpt.isPresent()) {
                 TbPair<Long, Long> packBounds = packBoundsOpt.get();
+                List<TbMsg> pack = new ArrayList<>();
                 if (DeduplicationStrategy.ALL.equals(config.getStrategy())) {
-                    List<TbMsg> pack = new ArrayList<>();
                     for (Iterator<TbMsg> iterator = msgList.iterator(); iterator.hasNext(); ) {
                         TbMsg msg = iterator.next();
                         long msgTs = msg.getMetaDataTs();
@@ -147,12 +184,12 @@ public class TbMsgDeduplicationNode implements TbNode {
                             iterator.remove();
                         }
                     }
-                    deduplicationResults.add(TbMsg.newMsg(
+                    deduplicationResults.add(new TbPair<>(TbMsg.newMsg(
                             config.getQueueName(),
                             config.getOutMsgType(),
                             deduplicationId,
-                            getMetadata(),
-                            getMergedData(pack)));
+                            getMetadata(packBounds.getFirst()),
+                            getMergedData(pack)), pack));
                 } else {
                     TbMsg resultMsg = null;
                     boolean searchMin = DeduplicationStrategy.FIRST.equals(config.getStrategy());
@@ -160,6 +197,7 @@ public class TbMsgDeduplicationNode implements TbNode {
                         TbMsg msg = iterator.next();
                         long msgTs = msg.getMetaDataTs();
                         if (msgTs >= packBounds.getFirst() && msgTs < packBounds.getSecond()) {
+                            pack.add(msg);
                             iterator.remove();
                             if (resultMsg == null
                                     || (searchMin && msg.getMetaDataTs() < resultMsg.getMetaDataTs())
@@ -168,11 +206,14 @@ public class TbMsgDeduplicationNode implements TbNode {
                             }
                         }
                     }
-                    deduplicationResults.add(resultMsg);
+                    if (resultMsg != null) {
+                        resultMsg = TbMsg.transformMsg(resultMsg, config.getQueueName());
+                        deduplicationResults.add(new TbPair<>(resultMsg, pack));
+                    }
                 }
                 packBoundsOpt = findValidPack(msgList, deduplicationTimeoutMs);
             }
-            deduplicationResults.forEach(outMsg -> enqueueForTellNextWithRetry(ctx, outMsg, 0));
+            deduplicationResults.forEach(result -> enqueueForTellNextWithRetry(ctx, result, 0));
         } finally {
             if (!data.isEmpty()) {
                 scheduleTickMsg(ctx, deduplicationId, data);
@@ -199,17 +240,18 @@ public class TbMsgDeduplicationNode implements TbNode {
         });
     }
 
-    private void enqueueForTellNextWithRetry(TbContext ctx, TbMsg msg, int retryAttempt) {
+    private void enqueueForTellNextWithRetry(TbContext ctx, TbPair<TbMsg, List<TbMsg>> result, int retryAttempt) {
         if (config.getMaxRetries() > retryAttempt) {
-            ctx.enqueueForTellNext(msg, TbRelationTypes.SUCCESS,
+            TbMsg outMsg = result.getFirst();
+            List<TbMsg> msgsToRemoveFromCache = result.getSecond();
+            ctx.enqueueForTellNext(outMsg, TbRelationTypes.SUCCESS,
                     () -> {
-                        log.trace("[{}][{}][{}] Successfully enqueue deduplication result message!", ctx.getSelfId(), msg.getOriginator(), retryAttempt);
+                        log.trace("[{}][{}][{}] Successfully enqueue deduplication result message!", ctx.getSelfId(), outMsg.getOriginator(), retryAttempt);
+                        ctx.getRuleNodeCacheService().removeTbMsgList(outMsg.getOriginator().toString(), msgsToRemoveFromCache);
                     },
                     throwable -> {
-                        log.trace("[{}][{}][{}] Failed to enqueue deduplication output message due to: ", ctx.getSelfId(), msg.getOriginator(), retryAttempt, throwable);
-                        ctx.schedule(() -> {
-                            enqueueForTellNextWithRetry(ctx, msg, retryAttempt + 1);
-                        }, TB_MSG_DEDUPLICATION_RETRY_DELAY, TimeUnit.SECONDS);
+                        log.trace("[{}][{}][{}] Failed to enqueue deduplication output message due to: ", ctx.getSelfId(), outMsg.getOriginator(), retryAttempt, throwable);
+                        ctx.schedule(() -> enqueueForTellNextWithRetry(ctx, result, retryAttempt + 1), TB_MSG_DEDUPLICATION_RETRY_DELAY, TimeUnit.SECONDS);
                     });
         }
     }
@@ -229,9 +271,9 @@ public class TbMsgDeduplicationNode implements TbNode {
         return JacksonUtil.toString(mergedData);
     }
 
-    private TbMsgMetaData getMetadata() {
+    private TbMsgMetaData getMetadata(long packStartTs) {
         TbMsgMetaData metaData = new TbMsgMetaData();
-        metaData.putValue("ts", String.valueOf(System.currentTimeMillis()));
+        metaData.putValue("ts", String.valueOf(packStartTs));
         return metaData;
     }
 

rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/deduplication/TbMsgDeduplicationNodeConfiguration.java

@@ -17,6 +17,7 @@ package org.thingsboard.rule.engine.deduplication;
 
 import lombok.Data;
 import org.thingsboard.rule.engine.api.NodeConfiguration;
+import org.thingsboard.server.common.data.DataConstants;
 
 @Data
 public class TbMsgDeduplicationNodeConfiguration implements NodeConfiguration<TbMsgDeduplicationNodeConfiguration> {
@@ -28,10 +29,10 @@ public class TbMsgDeduplicationNodeConfiguration implements NodeConfiguration<Tb
     // Advanced settings:
     private int maxPendingMsgs;
     private int maxRetries;
+    private String queueName;
 
     // only for DeduplicationStrategy.ALL:
     private String outMsgType;
-    private String queueName;
 
     @Override
     public TbMsgDeduplicationNodeConfiguration defaultConfiguration() {
@@ -41,6 +42,7 @@ public class TbMsgDeduplicationNodeConfiguration implements NodeConfiguration<Tb
         configuration.setStrategy(DeduplicationStrategy.FIRST);
         configuration.setMaxPendingMsgs(100);
         configuration.setMaxRetries(3);
+        configuration.setQueueName(DataConstants.MAIN_QUEUE_NAME);
         return configuration;
     }
 }

rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/action/TbMsgDeduplicationNodeTest.java → rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/transform/TbMsgDeduplicationNodeTest.java

@@ -13,10 +13,11 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package org.thingsboard.rule.engine.action;
+package org.thingsboard.rule.engine.transform;
 
 import com.fasterxml.jackson.databind.node.ArrayNode;
 import com.fasterxml.jackson.databind.node.ObjectNode;
+import com.google.protobuf.ByteString;
 import lombok.extern.slf4j.Slf4j;
 import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.Assertions;
@@ -27,6 +28,7 @@ import org.mockito.ArgumentMatchers;
 import org.mockito.stubbing.Answer;
 import org.thingsboard.common.util.JacksonUtil;
 import org.thingsboard.common.util.ThingsBoardThreadFactory;
+import org.thingsboard.rule.engine.api.RuleNodeCacheService;
 import org.thingsboard.rule.engine.api.TbContext;
 import org.thingsboard.rule.engine.api.TbNodeConfiguration;
 import org.thingsboard.rule.engine.api.TbNodeException;
@@ -34,6 +36,7 @@ import org.thingsboard.rule.engine.api.TbRelationTypes;
 import org.thingsboard.rule.engine.deduplication.DeduplicationStrategy;
 import org.thingsboard.rule.engine.deduplication.TbMsgDeduplicationNode;
 import org.thingsboard.rule.engine.deduplication.TbMsgDeduplicationNodeConfiguration;
+import org.thingsboard.server.common.data.DataConstants;
 import org.thingsboard.server.common.data.id.DeviceId;
 import org.thingsboard.server.common.data.id.EntityId;
 import org.thingsboard.server.common.data.id.RuleNodeId;
@@ -43,6 +46,7 @@ import org.thingsboard.server.common.msg.TbMsgMetaData;
 import org.thingsboard.server.common.msg.session.SessionMsgType;
 
 import java.util.ArrayList;
+import java.util.Collections;
 import java.util.List;
 import java.util.Random;
 import java.util.UUID;
@@ -54,13 +58,18 @@ import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicLong;
 import java.util.function.Consumer;
+import java.util.stream.Collectors;
 
 import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyList;
+import static org.mockito.ArgumentMatchers.anyLong;
+import static org.mockito.ArgumentMatchers.anyString;
 import static org.mockito.ArgumentMatchers.eq;
 import static org.mockito.ArgumentMatchers.isNull;
 import static org.mockito.ArgumentMatchers.nullable;
 import static org.mockito.Mockito.doAnswer;
 import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
 import static org.mockito.Mockito.spy;
 import static org.mockito.Mockito.times;
 import static org.mockito.Mockito.verify;
@@ -69,13 +78,12 @@ import static org.mockito.Mockito.when;
 @Slf4j
 public class TbMsgDeduplicationNodeTest {
 
-    private static final String MAIN_QUEUE_NAME = "Main";
-    private static final String HIGH_PRIORITY_QUEUE_NAME = "HighPriority";
     private static final String TB_MSG_DEDUPLICATION_TIMEOUT_MSG = "TbMsgDeduplicationNodeMsg";
+    private static final String DEDUPLICATION_IDS_CACHE_KEY = "deduplication_ids";
 
     private TbContext ctx;
 
-    private final ThingsBoardThreadFactory factory = ThingsBoardThreadFactory.forName("de-duplication-node-test");
+    private final ThingsBoardThreadFactory factory = ThingsBoardThreadFactory.forName("deduplication-node-test");
     private final ScheduledExecutorService executorService = Executors.newSingleThreadScheduledExecutor(factory);
     private final int deduplicationInterval = 1;
 
@@ -84,18 +92,24 @@ public class TbMsgDeduplicationNodeTest {
     private TbMsgDeduplicationNode node;
     private TbMsgDeduplicationNodeConfiguration config;
     private TbNodeConfiguration nodeConfiguration;
+    private RuleNodeCacheService ruleNodeCacheService;
 
     private CountDownLatch awaitTellSelfLatch;
 
     @BeforeEach
     public void init() throws TbNodeException {
         ctx = mock(TbContext.class);
+        ruleNodeCacheService = mock(RuleNodeCacheService.class);
 
         tenantId = TenantId.fromUUID(UUID.randomUUID());
         RuleNodeId ruleNodeId = new RuleNodeId(UUID.randomUUID());
 
         when(ctx.getSelfId()).thenReturn(ruleNodeId);
         when(ctx.getTenantId()).thenReturn(tenantId);
+        when(ctx.getRuleNodeCacheService()).thenReturn(ruleNodeCacheService);
+        when(ruleNodeCacheService.getStrings(anyString())).thenReturn(Collections.emptySet());
+        when(ruleNodeCacheService.getEntityIds(anyString())).thenReturn(Collections.emptySet());
+        when(ruleNodeCacheService.getTbMsgs(anyString(), anyString())).thenReturn(Collections.emptySet());
 
         doAnswer((Answer<TbMsg>) invocationOnMock -> {
             String type = (String) (invocationOnMock.getArguments())[1];
@@ -171,11 +185,23 @@ public class TbMsgDeduplicationNodeTest {
         ArgumentCaptor<Runnable> successCaptor = ArgumentCaptor.forClass(Runnable.class);
         ArgumentCaptor<Consumer<Throwable>> failureCaptor = ArgumentCaptor.forClass(Consumer.class);
 
+        verify(ruleNodeCacheService, times(1)).getEntityIds(anyString());
+        verify(ruleNodeCacheService, times(1)).add(eq(DEDUPLICATION_IDS_CACHE_KEY), eq(deviceId));
+        verify(ruleNodeCacheService, times(msgCount)).add(eq(deviceId.toString()), any(TbMsg.class));
+
+        verify(node, times(msgCount + wantedNumberOfTellSelfInvocation + 1)).onMsg(eq(ctx), any());
+
         verify(ctx, times(msgCount)).ack(any());
         verify(ctx, times(1)).tellFailure(eq(msgToReject), any());
-        verify(node, times(msgCount + wantedNumberOfTellSelfInvocation + 1)).onMsg(eq(ctx), any());
         verify(ctx, times(1)).enqueueForTellNext(newMsgCaptor.capture(), eq(TbRelationTypes.SUCCESS), successCaptor.capture(), failureCaptor.capture());
-        Assertions.assertEquals(inputMsgs.get(0), newMsgCaptor.getValue());
+
+        for (Runnable valueCaptor : successCaptor.getAllValues()) {
+            valueCaptor.run();
+        }
+        verify(ruleNodeCacheService, times(1)).removeTbMsgList(eq(deviceId.toString()), anyList());
+        verify(ctx, never()).schedule(any(), anyLong(), any());
+
+        Assertions.assertEquals(TbMsg.toByteString(inputMsgs.get(0)), TbMsg.toByteString(newMsgCaptor.getValue()));
     }
 
     @Test
@@ -210,11 +236,23 @@ public class TbMsgDeduplicationNodeTest {
         ArgumentCaptor<Runnable> successCaptor = ArgumentCaptor.forClass(Runnable.class);
         ArgumentCaptor<Consumer<Throwable>> failureCaptor = ArgumentCaptor.forClass(Consumer.class);
 
+        verify(ruleNodeCacheService, times(1)).getEntityIds(anyString());
+        verify(ruleNodeCacheService, times(1)).add(eq(DEDUPLICATION_IDS_CACHE_KEY), eq(deviceId));
+        verify(ruleNodeCacheService, times(msgCount)).add(eq(deviceId.toString()), any(TbMsg.class));
+
+        verify(node, times(msgCount + wantedNumberOfTellSelfInvocation + 1)).onMsg(eq(ctx), any());
+
         verify(ctx, times(msgCount)).ack(any());
         verify(ctx, times(1)).tellFailure(eq(msgToReject), any());
-        verify(node, times(msgCount + wantedNumberOfTellSelfInvocation + 1)).onMsg(eq(ctx), any());
         verify(ctx, times(1)).enqueueForTellNext(newMsgCaptor.capture(), eq(TbRelationTypes.SUCCESS), successCaptor.capture(), failureCaptor.capture());
-        Assertions.assertEquals(msgWithLatestTs, newMsgCaptor.getValue());
+
+        for (Runnable valueCaptor : successCaptor.getAllValues()) {
+            valueCaptor.run();
+        }
+        verify(ruleNodeCacheService, times(1)).removeTbMsgList(eq(deviceId.toString()), anyList());
+        verify(ctx, never()).schedule(any(), anyLong(), any());
+
+        Assertions.assertEquals(TbMsg.toByteString(msgWithLatestTs), TbMsg.toByteString(newMsgCaptor.getValue()));
     }
 
     @Test
@@ -227,7 +265,7 @@ public class TbMsgDeduplicationNodeTest {
         config.setInterval(deduplicationInterval);
         config.setStrategy(DeduplicationStrategy.ALL);
         config.setOutMsgType(SessionMsgType.POST_ATTRIBUTES_REQUEST.name());
-        config.setQueueName(HIGH_PRIORITY_QUEUE_NAME);
+        config.setQueueName(DataConstants.HP_QUEUE_NAME);
         nodeConfiguration = new TbNodeConfiguration(JacksonUtil.valueToTree(config));
         node.init(ctx, nodeConfiguration);
 
@@ -245,10 +283,21 @@ public class TbMsgDeduplicationNodeTest {
         ArgumentCaptor<Runnable> successCaptor = ArgumentCaptor.forClass(Runnable.class);
         ArgumentCaptor<Consumer<Throwable>> failureCaptor = ArgumentCaptor.forClass(Consumer.class);
 
-        verify(ctx, times(msgCount)).ack(any());
+        verify(ruleNodeCacheService, times(1)).getEntityIds(anyString());
+        verify(ruleNodeCacheService, times(1)).add(eq(DEDUPLICATION_IDS_CACHE_KEY), eq(deviceId));
+        verify(ruleNodeCacheService, times(msgCount)).add(eq(deviceId.toString()), any(TbMsg.class));
+
         verify(node, times(msgCount + wantedNumberOfTellSelfInvocation)).onMsg(eq(ctx), any());
+
+        verify(ctx, times(msgCount)).ack(any());
         verify(ctx, times(1)).enqueueForTellNext(newMsgCaptor.capture(), eq(TbRelationTypes.SUCCESS), successCaptor.capture(), failureCaptor.capture());
 
+        for (Runnable valueCaptor : successCaptor.getAllValues()) {
+            valueCaptor.run();
+        }
+        verify(ruleNodeCacheService, times(1)).removeTbMsgList(eq(deviceId.toString()), anyList());
+        verify(ctx, never()).schedule(any(), anyLong(), any());
+
         Assertions.assertEquals(1, newMsgCaptor.getAllValues().size());
         TbMsg outMessage = newMsgCaptor.getAllValues().get(0);
         Assertions.assertEquals(getMergedData(inputMsgs), outMessage.getData());
@@ -267,7 +316,7 @@ public class TbMsgDeduplicationNodeTest {
         config.setInterval(deduplicationInterval);
         config.setStrategy(DeduplicationStrategy.ALL);
         config.setOutMsgType(SessionMsgType.POST_ATTRIBUTES_REQUEST.name());
-        config.setQueueName(HIGH_PRIORITY_QUEUE_NAME);
+        config.setQueueName(DataConstants.HP_QUEUE_NAME);
         nodeConfiguration = new TbNodeConfiguration(JacksonUtil.valueToTree(config));
         node.init(ctx, nodeConfiguration);
 
@@ -278,7 +327,7 @@ public class TbMsgDeduplicationNodeTest {
         for (TbMsg msg : firstMsgPack) {
             node.onMsg(ctx, msg);
         }
-        long firstPackDeduplicationPackEndTs =  firstMsgPack.get(0).getMetaDataTs() + TimeUnit.SECONDS.toMillis(deduplicationInterval);
+        long firstPackDeduplicationPackEndTs = firstMsgPack.get(0).getMetaDataTs() + TimeUnit.SECONDS.toMillis(deduplicationInterval);
 
         List<TbMsg> secondMsgPack = getTbMsgs(deviceId, msgCount / 2, firstPackDeduplicationPackEndTs, 500);
         for (TbMsg msg : secondMsgPack) {
@@ -291,10 +340,21 @@ public class TbMsgDeduplicationNodeTest {
         ArgumentCaptor<Runnable> successCaptor = ArgumentCaptor.forClass(Runnable.class);
         ArgumentCaptor<Consumer<Throwable>> failureCaptor = ArgumentCaptor.forClass(Consumer.class);
 
-        verify(ctx, times(msgCount)).ack(any());
+        verify(ruleNodeCacheService, times(1)).getEntityIds(anyString());
+        verify(ruleNodeCacheService, times(1)).add(eq(DEDUPLICATION_IDS_CACHE_KEY), eq(deviceId));
+        verify(ruleNodeCacheService, times(msgCount)).add(eq(deviceId.toString()), any(TbMsg.class));
+
         verify(node, times(msgCount + wantedNumberOfTellSelfInvocation)).onMsg(eq(ctx), any());
+
+        verify(ctx, times(msgCount)).ack(any());
         verify(ctx, times(2)).enqueueForTellNext(newMsgCaptor.capture(), eq(TbRelationTypes.SUCCESS), successCaptor.capture(), failureCaptor.capture());
 
+        for (Runnable valueCaptor : successCaptor.getAllValues()) {
+            valueCaptor.run();
+        }
+        verify(ruleNodeCacheService, times(2)).removeTbMsgList(eq(deviceId.toString()), anyList());
+        verify(ctx, never()).schedule(any(), anyLong(), any());
+
         List<TbMsg> resultMsgs = newMsgCaptor.getAllValues();
         Assertions.assertEquals(2, resultMsgs.size());
 
@@ -345,14 +405,26 @@ public class TbMsgDeduplicationNodeTest {
         ArgumentCaptor<Runnable> successCaptor = ArgumentCaptor.forClass(Runnable.class);
         ArgumentCaptor<Consumer<Throwable>> failureCaptor = ArgumentCaptor.forClass(Consumer.class);
 
-        verify(ctx, times(msgCount)).ack(any());
+        verify(ruleNodeCacheService, times(1)).getEntityIds(anyString());
+        verify(ruleNodeCacheService, times(1)).add(eq(DEDUPLICATION_IDS_CACHE_KEY), eq(deviceId));
+        verify(ruleNodeCacheService, times(msgCount)).add(eq(deviceId.toString()), any(TbMsg.class));
+
         verify(node, times(msgCount + wantedNumberOfTellSelfInvocation)).onMsg(eq(ctx), any());
+
+        verify(ctx, times(msgCount)).ack(any());
         verify(ctx, times(2)).enqueueForTellNext(newMsgCaptor.capture(), eq(TbRelationTypes.SUCCESS), successCaptor.capture(), failureCaptor.capture());
 
+        for (Runnable valueCaptor : successCaptor.getAllValues()) {
+            valueCaptor.run();
+        }
+        verify(ruleNodeCacheService, times(2)).removeTbMsgList(eq(deviceId.toString()), anyList());
+        verify(ctx, never()).schedule(any(), anyLong(), any());
+
         List<TbMsg> resultMsgs = newMsgCaptor.getAllValues();
         Assertions.assertEquals(2, resultMsgs.size());
-        Assertions.assertTrue(resultMsgs.contains(msgWithLatestTsInFirstPack));
-        Assertions.assertTrue(resultMsgs.contains(msgWithLatestTsInSecondPack));
+        List<ByteString> resultMsgsByteStrings = resultMsgs.stream().map(TbMsg::toByteString).collect(Collectors.toList());
+        Assertions.assertTrue(resultMsgsByteStrings.contains(TbMsg.toByteString(msgWithLatestTsInFirstPack)));
+        Assertions.assertTrue(resultMsgsByteStrings.contains(TbMsg.toByteString(msgWithLatestTsInSecondPack)));
     }
 
     private TbMsg getMsgWithLatestTs(List<TbMsg> firstMsgPack) {
@@ -381,7 +453,7 @@ public class TbMsgDeduplicationNodeTest {
         TbMsgMetaData metaData = new TbMsgMetaData();
         metaData.putValue("ts", String.valueOf(ts));
         return TbMsg.newMsg(
-                MAIN_QUEUE_NAME,
+                DataConstants.MAIN_QUEUE_NAME,
                 SessionMsgType.POST_TELEMETRY_REQUEST.name(),
                 deviceId,
                 metaData,